cyb50-quant/cyb50-pro/scripts/run_backtest.py

"""
运行 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()