etf/archive/tests/verify_mode_b.py

#!/usr/bin/env python3
"""
验证 Mode B: 指数信号 → ETF收益

文档预期结果:
  CAGR: 28.07%, 最大回撤: -13.34%, 夏普: 1.685, Calmar: 2.104
"""

import sys
from pathlib import Path
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))

from dotenv import load_dotenv
load_dotenv()

import pandas as pd
import numpy as np
import yaml
from datetime import datetime
from strategies.rotation.strategy import RotationStrategy


def calculate_metrics(nav: pd.Series) -> dict:
    """计算绩效指标"""
    start_date = nav.index[0]
    end_date = nav.index[-1]
    days = (end_date - start_date).days
    years = days / 365
    
    total_return = nav.iloc[-1] - 1
    cagr = (nav.iloc[-1] / nav.iloc[0]) ** (1/years) - 1
    
    daily_ret = nav.pct_change().dropna()
    sharpe = daily_ret.mean() / daily_ret.std() * np.sqrt(252) if daily_ret.std() > 0 else 0
    
    peak = nav.cummax()
    drawdown = (nav - peak) / peak
    max_dd = drawdown.min()
    
    calmar = cagr / abs(max_dd) if max_dd != 0 else 0
    win_rate = (daily_ret > 0).sum() / len(daily_ret)
    
    return {
        'start_date': start_date.strftime('%Y-%m-%d'),
        'end_date': end_date.strftime('%Y-%m-%d'),
        'years': years,
        'days': len(nav),
        'total_return': total_return,
        'cagr': cagr,
        'max_dd': max_dd,
        'sharpe': sharpe,
        'calmar': calmar,
        'win_rate': win_rate
    }


def run_mode_b_backtest(data: dict, signals: pd.DataFrame, valid_codes: list, 
                         etf_code_map: dict, a_share_dates: pd.DatetimeIndex,
                         trade_cost: float, select_num: int) -> dict:
    """
    Mode B: 使用ETF价格计算收益
    
    Args:
        data: 包含 etf_data 的数据字典
        signals: 指数生成的信号
        valid_codes: 指数代码列表
        etf_code_map: {指数代码: ETF代码} 映射
        a_share_dates: A股交易日历
        trade_cost: 交易成本
        select_num: 选股数量
    """
    from framework.execution import BacktestExecutor
    
    etf_data = data.get('etf_data')
    if etf_data is None:
        print("❌ ETF数据不可用")
        return {'result': None}
    
    # 将信号对齐到 A 股日历
    if a_share_dates is not signals.index:
        signals = signals.reindex(a_share_dates, method='ffill').dropna(subset=[signals.columns[0]])
    
    # 使用ETF收盘价计算收益率
    returns_data = {}
    for code in valid_codes:
        etf_code = etf_code_map.get(code)
        if etf_code and etf_code in etf_data.columns:
            etf_close = etf_data[etf_code].dropna()
            # 对齐到A股日历
            etf_aligned = etf_close.reindex(a_share_dates, method='ffill')
            returns_aligned = etf_aligned.pct_change(fill_method=None)
            # 使用指数代码作为列名（与信号匹配）
            returns_data[f'日收益率_{code}'] = returns_aligned
        else:
            # 没有ETF映射的标的，回退使用指数数据
            index_data = data.get('index_data', {})
            if code in index_data and 'close' in index_data[code].columns:
                close_series = index_data[code]['close'].dropna()
                close_aligned = close_series.reindex(a_share_dates, method='ffill')
                returns_data[f'日收益率_{code}'] = close_aligned.pct_change(fill_method=None)
    
    returns_df = pd.DataFrame(returns_data)
    
    # 对齐日期
    common_dates = signals.index.intersection(returns_df.index)
    signals = signals.loc[common_dates]
    returns_df = returns_df.loc[common_dates]
    
    print(f"  Mode B 对齐后日期: {len(common_dates)} 天")
    print(f"  使用ETF计算收益: {len([c for c in valid_codes if etf_code_map.get(c)])} 只")
    
    executor = BacktestExecutor(
        initial_capital=100000,
        trade_cost=trade_cost,
        select_num=select_num
    )
    
    portfolio = executor.execute(signals, returns_df)
    
    if hasattr(portfolio, 'backtest_result'):
        return {'result': portfolio.backtest_result, 'portfolio': portfolio}
    
    return {'result': None}


def main():
    # 加载配置
    config_path = project_root / 'strategies/rotation/config.yaml'
    with open(config_path, 'r') as f:
        config = yaml.safe_load(f)
    
    # 设置回测区间
    config['start_date'] = '2020-01-02'
    config['end_date'] = '2026-05-19'
    
    print('='*70)
    print('Mode B 验证: 指数信号 → ETF收益')
    print('='*70)
    
    # 初始化策略
    strategy = RotationStrategy(config)
    
    # 获取数据
    print('\n获取数据...')
    data = strategy.get_data(use_flask_api=False)
    
    # 计算因子（使用指数数据）
    print('\n计算因子（指数信号）...')
    factor_df = strategy.compute_factors(data)
    
    # 生成信号
    print('\n生成信号...')
    signals = strategy.generate_signals(factor_df)
    
    # 执行 Mode B 回测
    print('\n执行 Mode B 回测（ETF收益）...')
    result_b = run_mode_b_backtest(
        data=data,
        signals=signals,
        valid_codes=data['valid_codes'],
        etf_code_map=data['etf_code_map'],
        a_share_dates=data.get('a_share_dates'),
        trade_cost=config.get('trade_cost', 0.001),
        select_num=config.get('select_num', 3)
    )
    
    if result_b.get('result') is None:
        print('❌ Mode B 回测未生成结果')
        return
    
    # 计算指标
    nav_b = result_b['result']['策略净值']
    metrics_b = calculate_metrics(nav_b)
    
    # 输出结果
    print('\n' + '='*70)
    print('Mode B 回测结果')
    print('='*70)
    print(f"回测区间: {metrics_b['start_date']} ~ {metrics_b['end_date']}")
    print(f"回测年数: {metrics_b['years']:.2f} 年")
    print(f"交易天数: {metrics_b['days']} 天")
    print('-'*70)
    print(f"CAGR:        {metrics_b['cagr']:.2%}")
    print(f"最大回撤:    {metrics_b['max_dd']:.2%}")
    print(f"夏普比率:    {metrics_b['sharpe']:.3f}")
    print(f"Calmar比率:  {metrics_b['calmar']:.3f}")
    print(f"日胜率:      {metrics_b['win_rate']:.2%}")
    print(f"累计收益:    {metrics_b['total_return']:.2%}")
    print('='*70)
    
    # 文档预期对比
    print('\n文档预期 (Mode B):')
    print('  CAGR: 28.07%, MaxDD -13.34%, Sharpe 1.685, Calmar 2.104')
    
    cagr_diff = abs(metrics_b['cagr'] - 0.2807)
    print(f'\nCAGR差异: {cagr_diff:.2%}')
    
    if cagr_diff < 0.05:
        print('✓ 结果与文档预期基本一致')
    else:
        print('⚠ 结果与文档预期有差异')
    
    return metrics_b


if __name__ == '__main__':
    main()