""" A/B测试：法国CAC40与德国DAX类内竞争对比 对比： - A组（对照组）：仅德国DAX代表欧洲(EU) - B组（实验组）：德国DAX + 法国CAC40 同属EU大类（类内竞争） 核心问题： - 法国放在欧洲大类下（market='EU'），与德国竞争 - diversified模式下，EU大类只输出1个冠军 - 验证：类内竞争选择最优，而非盲目增加大类 对比003/004实验： - 003/004：法国/印度作为独立大类 → 收益下降 - 本次：法国放EU大类内 → 类内竞争，不增加大类数量 """ import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from strategies.rotation.engine import RotationStrategy import pandas as pd import yaml def create_config_with_france_in_eu(base_config: dict) -> dict: """在基础配置上添加法国，但放入EU大类""" config = base_config.copy() config['code_list'] = base_config['code_list'].copy() # 添加法国CAC40，但market='EU'（与德国DAX同大类） # diversified模式下：德国 vs 法国 → 选动量最强的1个进入Top3 config['code_list']['^FCHI'] = { 'name': '法国CAC40', 'etf': '513080.SH', 'market': 'EU' # 与德国DAX同属欧洲大类 } return config def run_backtest(config: dict, label: str) -> dict: """运行回测并返回关键指标""" print(f"\n{'='*60}") print(f" {label}") print(f"{'='*60}") strategy = RotationStrategy(config) result = strategy.run() if result is None or len(result) == 0: return None # 计算指标 strategy_nav = result['轮动策略净值'] strategy_ret = result['轮动策略日收益率'] total_return = strategy_nav.iloc[-1] - 1 days = len(result) years = days / 250 cagr = (strategy_nav.iloc[-1] ** (1/years)) - 1 if years > 0 else 0 excess_ret = strategy_ret.mean() * 250 vol = strategy_ret.std() * (250 ** 0.5) sharpe = excess_ret / vol if vol > 0 else 0 rolling_max = strategy_nav.cummax() drawdown = (strategy_nav - rolling_max) / rolling_max max_dd = drawdown.min() calmar = cagr / abs(max_dd) if max_dd < 0 else 0 win_rate = (strategy_ret > 0).sum() / len(strategy_ret) # 统计大类数量 markets = set() for code_info in config['code_list'].values(): markets.add(code_info.get('market', 'A')) # 统计EU大类标的数量 eu_count = sum(1 for c in config['code_list'].values() if c.get('market') == 'EU') metrics = { 'label': label, '大类数量': len(markets), 'EU大类标的数': eu_count, '累计收益': total_return, 'CAGR': cagr, 'Sharpe': sharpe, 'MaxDD': max_dd, 'Calmar': calmar, '日胜率': win_rate, } print(f"\n大类数量: {metrics['大类数量']}") print(f"EU大类标的数: {metrics['EU大类标的数']}") print(f"累计收益: {metrics['累计收益']:.2%}") print(f"CAGR: {metrics['CAGR']:.2%}") print(f"Sharpe: {metrics['Sharpe']:.2f}") print(f"MaxDD: {metrics['MaxDD']:.2%}") print(f"Calmar: {metrics['Calmar']:.2f}") print(f"日胜率: {metrics['日胜率']:.2%}") return metrics def compare_results(a_metrics: dict, b_metrics: dict): """对比两组结果""" print(f"\n{'='*60}") print(f" 对比结果") print(f"{'='*60}") print(f"\n{'指标':<15} {'A组(仅德国)':<15} {'B组(德国+法国)':<15} {'差异':<15}") print("-" * 60) metrics_keys = ['大类数量', 'EU大类标的数', '累计收益', 'CAGR', 'Sharpe', 'MaxDD', 'Calmar', '日胜率'] for key in metrics_keys: a_val = a_metrics.get(key, 0) b_val = b_metrics.get(key, 0) diff = b_val - a_val if key in ['累计收益', 'CAGR', 'MaxDD', '日胜率']: a_str = f"{a_val:.2%}" b_str = f"{b_val:.2%}" diff_str = f"{diff*100:+.2f}%" elif key in ['大类数量', 'EU大类标的数']: a_str = str(a_val) b_str = str(b_val) diff_str = f"+{diff}" if diff > 0 else str(diff) else: a_str = f"{a_val:.2f}" b_str = f"{b_val:.2f}" diff_str = f"{diff:+.2f}" print(f"{key:<15} {a_str:<15} {b_str:<15} {diff_str:<15}") print("-" * 60) print(f"\n【关键发现】") print(f"法国放入EU大类（与德国类内竞争）：") if b_metrics['大类数量'] == a_metrics['大类数量']: print(f" ✓ 大类数量不变（{b_metrics['大类数量']}大类）") print(f" → 不增加跨类分散，避免占用Top3权重") if b_metrics['EU大类标的数'] > a_metrics['EU大类标的数']: print(f" ✓ EU大类标的增加 {b_metrics['EU大类标的数'] - a_metrics['EU大类标的数']}") print(f" → 德国 vs 法国 类内竞争，选动量最强") if b_metrics['累计收益'] > a_metrics['累计收益']: print(f" ✓ 累计收益提升 {b_metrics['累计收益'] - a_metrics['累计收益']:.2%}") print(f" → 类内竞争选择最优，比新大类添加更有效") elif b_metrics['累计收益'] < a_metrics['累计收益']: loss = a_metrics['累计收益'] - b_metrics['累计收益'] print(f" ✗ 累计收益下降 {loss:.2%}") if loss < 0.05: # 下降5%以内 print(f" → 收益损失较小（vs 004实验下降167%）") else: print(f" → 法国动量信号不如德国，类内竞争仍选德国") print(f"\n【与004实验对比】") print(f" 004实验（法国作为独立FR大类）：") print(f" ├─ 大类数量: 7 → 8 (+1)") print(f" └─ 累计收益: -166.80%") print(f" ") print(f" 本次实验（法国放入EU大类）：") print(f" ├─ 大类数量: 保持{b_metrics['大类数量']}不变") print(f" ├─ EU类内竞争：德国DAX vs 法国CAC40") print(f" └─ 累计收益变化: {b_metrics['累计收益'] - a_metrics['累计收益']:.2%}") print(f"\n【策略建议】") if b_metrics['累计收益'] >= a_metrics['累计收益'] * 0.98: print(f" 建议：法国放入EU大类可行") print(f" 原因：类内竞争自动选最优，不增加大类数量") elif abs(b_metrics['累计收益'] - a_metrics['累计收益']) < 0.05: print(f" 建议：法国放入EU大类可考虑") print(f" 原因：收益损失小于5%，可接受") else: print(f" 建议：保持仅德国DAX") print(f" 原因：法国动量信号不如德国") def main(): """主函数""" config_path = Path(__file__).parent.parent.parent / 'config' / 'strategies' / 'rotation.yaml' with open(config_path, 'r') as f: base_config = yaml.safe_load(f) # 添加 end_date from datetime import datetime base_config['end_date'] = datetime.now().strftime('%Y-%m-%d') print(f"\n{'='*60}") print(f" A/B测试：法国CAC40放入EU大类（类内竞争）") print(f"{'='*60}") print(f"\n研究问题：") print(f" - 法国CAC40放入EU大类（market='EU'）") print(f" - 与德国DAX类内竞争，选动量最强的1个") print(f" - 不增加大类数量，避免占用Top3权重") print(f" - 对比004实验（法国独立大类）的差异") # A组：仅德国DAX a_metrics = run_backtest(base_config, "A组: 仅德国DAX代表欧洲") # B组：德国+法国（同属EU大类） config_with_france = create_config_with_france_in_eu(base_config) b_metrics = run_backtest(config_with_france, "B组: 德国+法国（EU类内竞争）") # 对比 if a_metrics and b_metrics: compare_results(a_metrics, b_metrics) # 保存结果 results_df = pd.DataFrame([a_metrics, b_metrics]) results_path = Path(__file__).parent.parent.parent / 'results' / 'ab_test_france_in_eu.csv' results_df.to_csv(results_path, index=False) print(f"\n对比结果已保存: {results_path}") if __name__ == '__main__': main()