""" 运行 CYB50-Pro 回测验证 使用 2018-2025 年历史数据验证目标指标： - 年化收益 25-35% - 最大回撤 <12% - 夏普比率 >1.5 """ import sys import os sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) import pandas as pd import numpy as np from datetime import datetime import json from backtest import CYB50ProBacktester def load_historical_data(data_path: str = None) -> pd.DataFrame: """ 加载历史数据 优先从现有项目加载，如果没有则生成模拟数据 """ # 尝试从现有项目加载 possible_paths = [ data_path, '../../quant/cyb50_historical_data.csv', '../../quant/cyb50_baostock.csv', '../../market-regime-identifier/SZ#399673.txt', '../data/SZ#399673.csv' ] for path in possible_paths: if path and os.path.exists(path): print(f"Loading data from: {path}") if path.endswith('.csv'): df = pd.read_csv(path) elif path.endswith('.txt'): # 处理txt格式 df = pd.read_csv(path, sep='\t', header=None, names=['date', 'open', 'high', 'low', 'close', 'volume']) else: continue # 标准化列名 df.columns = [c.lower() for c in df.columns] # 解析日期 if 'date' in df.columns: df['date'] = pd.to_datetime(df['date']) df.set_index('date', inplace=True) elif 'datetime' in df.columns: df['datetime'] = pd.to_datetime(df['datetime']) df.set_index('datetime', inplace=True) return df # 如果没有找到数据，生成模拟数据 print("No historical data found. Generating synthetic data for testing...") return generate_synthetic_data() def generate_synthetic_data(start='2018-01-01', end='2025-01-01') -> pd.DataFrame: """生成模拟的创业板50指数数据""" np.random.seed(42) dates = pd.date_range(start=start, end=end, freq='D') # 生成带有趋势和波动率聚类的收益率 returns = [] current_regime = 'bull' regime_days = 0 for i in range(len(dates)): # 随机切换市场状态 if regime_days <= 0: current_regime = np.random.choice(['bull', 'bear', 'sideways'], p=[0.5, 0.2, 0.3]) regime_days = np.random.randint(60, 252) # 60天到1年 # 根据状态生成收益 if current_regime == 'bull': daily_return = np.random.normal(0.001, 0.015) elif current_regime == 'bear': daily_return = np.random.normal(-0.001, 0.025) else: # sideways daily_return = np.random.normal(0, 0.012) returns.append(daily_return) regime_days -= 1 returns = np.array(returns) prices = 1000 * np.exp(np.cumsum(returns)) # 生成OHLCV df = pd.DataFrame(index=dates) df['close'] = prices df['open'] = df['close'] * (1 + np.random.normal(0, 0.005, len(dates))) df['high'] = df[['open', 'close']].max(axis=1) * (1 + abs(np.random.normal(0, 0.01, len(dates)))) df['low'] = df[['open', 'close']].min(axis=1) * (1 - abs(np.random.normal(0, 0.01, len(dates)))) df['volume'] = np.random.lognormal(20, 0.5, len(dates)) return df def main(): """主函数""" print("=" * 60) print("CYB50-Pro Backtest Validation") print("Target Metrics:") print(" - Annual Return: 25-35%") print(" - Max Drawdown: <12%") print(" - Sharpe Ratio: >1.5") print("=" * 60) # 加载数据 data = load_historical_data() print(f"\nData loaded: {len(data)} bars from {data.index[0]} to {data.index[-1]}") print(f"Price range: {data['close'].min():.2f} - {data['close'].max():.2f}") # 初始化回测器 backtester = CYB50ProBacktester( initial_capital=1_000_000, commission_rate=0.0005, slippage=0.001 ) # 运行回测 print("\nRunning backtest...") result = backtester.run_backtest( price_data=data, start_date='2018-01-01', end_date='2024-12-31' ) # 打印结果 print("\n" + "=" * 60) print("BACKTEST RESULTS") print("=" * 60) print(f"\n[Period]") print(f" Start: {result.start_date.strftime('%Y-%m-%d')}") print(f" End: {result.end_date.strftime('%Y-%m-%d')}") print(f"\n[Returns]") print(f" Total Return: {result.total_return:+.2%}") print(f" Annual Return: {result.annual_return:+.2%}") print(f"\n[Risk Metrics]") print(f" Max Drawdown: {result.max_drawdown:.2%}") print(f" Max DD Duration: {result.max_drawdown_duration} days") print(f" Volatility (Ann): {result.volatility:.2%}") print(f"\n[Risk-Adjusted Returns]") print(f" Sharpe Ratio: {result.sharpe_ratio:.2f}") print(f" Sortino Ratio: {result.sortino_ratio:.2f}") print(f"\n[Trading Statistics]") print(f" Total Trades: {result.total_trades}") print(f" Win Rate: {result.win_rate:.1%}") print(f" Profit Factor: {result.profit_factor:.2f}") # Target Validation print(f"\n[Target Validation]") targets_met = 0 if 0.25 <= result.annual_return <= 0.35: print(f" [OK] Annual Return 25-35%: {result.annual_return:.2%}") targets_met += 1 else: print(f" [FAIL] Annual Return 25-35%: {result.annual_return:.2%}") if result.max_drawdown > -0.12: print(f" [OK] Max Drawdown <12%: {result.max_drawdown:.2%}") targets_met += 1 else: print(f" [FAIL] Max Drawdown <12%: {result.max_drawdown:.2%}") if result.sharpe_ratio > 1.5: print(f" [OK] Sharpe Ratio >1.5: {result.sharpe_ratio:.2f}") targets_met += 1 else: print(f" [FAIL] Sharpe Ratio >1.5: {result.sharpe_ratio:.2f}") print(f"\n[Summary]") print(f" Targets Met: {targets_met}/3") if targets_met == 3: print(" Status: ALL TARGETS ACHIEVED [OK][OK][OK]") elif targets_met >= 2: print(" Status: MOSTLY SATISFACTORY") else: print(" Status: NEEDS OPTIMIZATION") # Regime Analysis print(f"\n[Regime Performance]") for regime, stats in result.regime_performance.items(): print(f" {regime}: {stats['trades']} trades, " f"Win Rate {stats['win_rate']:.1%}, " f"Avg PnL {stats['avg_pnl']:,.0f}") # Agent Contributions print(f"\n[Agent Contributions]") for agent, pnl in sorted(result.agent_contributions.items(), key=lambda x: x[1], reverse=True): print(f" {agent}: {pnl:+,.0f}") # Generate Report output_dir = os.path.join(os.path.dirname(__file__), '../reports') os.makedirs(output_dir, exist_ok=True) report_path = os.path.join(output_dir, f'backtest_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json') report = backtester.generate_report(result, report_path) print(f"\n[Report Saved]") print(f" {report_path}") print("\n" + "=" * 60) return result if __name__ == "__main__": main()