""" 策略迭代 A/B 对比实验脚本 量化三个维度的改进贡献度： 1. 标的池: 原始全市场池 vs. 精选11只核心池 2. 评分公式: 简单斜率(slope_r2) vs. 年化收益率*R2 (weighted_momentum) 3. 观察窗口: 固定25日窗口 vs. 动态ATR窗口 (20-60天) """ import sys import pandas as pd import numpy as np from pathlib import Path from datetime import datetime # 添加项目根目录 sys.path.insert(0, str(Path(__file__).parent.parent)) from strategies.rotation.engine import RotationStrategy import matplotlib.pyplot as plt # ==================== 标的池定义 ==================== ORIGINAL_POOL = { "000300.SH": {"name": "沪深300", "market": "A", "etf": "510300.SH"}, "000905.SH": {"name": "中证500", "market": "A", "etf": "510500.SH"}, "000852.SH": {"name": "中证1000", "market": "A", "etf": "512100.SH"}, "399006.SZ": {"name": "创业板指", "market": "A", "etf": "159915.SZ"}, "000015.SH": {"name": "上证红利", "market": "A", "etf": "510880.SH"}, "399986.SZ": {"name": "中证银行", "market": "A", "etf": "516310.SH"}, "399997.SZ": {"name": "中证白酒", "market": "A", "etf": "512690.SH"}, "399989.SZ": {"name": "中证医疗", "market": "A", "etf": "512170.SH"}, "399395.SZ": {"name": "国证有色", "market": "A", "etf": "159880.SZ"}, "399998.SZ": {"name": "中证煤炭", "market": "A", "etf": "515220.SH"}, "399967.SZ": {"name": "中证军工", "market": "A", "etf": "512660.SH"}, "HSTECH.HK": {"name": "恒生科技", "market": "HK", "etf": "513180.SH"}, "NDX": {"name": "纳指100", "market": "US", "etf": "513100.SH"}, "AU.SHF": {"name": "黄金", "market": "COMMODITY", "etf": "518880.SH"} } FINAL_POOL = { "399006.SZ": {"name": "创业板指", "market": "A", "etf": "159915.SZ"}, "H30269.CSI": {"name": "中证红利低波", "market": "A", "etf": "512890.SH"}, "000015.SH": {"name": "上证红利", "market": "A", "etf": "510880.SH"}, "NDX": {"name": "纳指100", "market": "US", "etf": "513100.SH"}, "N225": {"name": "日经225", "market": "JP", "etf": "513520.SH"}, "GDAXI": {"name": "德国DAX", "market": "EU", "etf": "513030.SH"}, "HSI": {"name": "恒生指数", "market": "HK", "etf": "159920.SZ"}, "HSTECH.HK": {"name": "恒生科技", "market": "HK", "etf": "513130.SH"}, "AU.SHF": {"name": "黄金", "market": "COMMODITY", "etf": "518880.SH"}, "CL.NYM": {"name": "原油", "market": "COMMODITY", "etf": "160723.SZ"}, "931862.CSI": {"name": "30年国债", "market": "BOND", "etf": "511090.SH"} } # ==================== 实验配置 ==================== ITERATIONS = [ { "label": "1. 原始基准 (原始池+简单评分+固定窗口)", "config": { "code_list": ORIGINAL_POOL, "factor_type": "slope_r2", "auto_day": False, "n_days": 25, "diversified": False } }, { "label": "2. 标的池优化 (精选池+简单评分+固定窗口)", "config": { "code_list": FINAL_POOL, "factor_type": "slope_r2", "auto_day": False, "n_days": 25, "diversified": True # 开启跨大类分散 } }, { "label": "3. 评分公式优化 (精选池+加权评分+固定窗口)", "config": { "code_list": FINAL_POOL, "factor_type": "weighted_momentum", "auto_day": False, "n_days": 25, "diversified": True } }, { "label": "4. 终极版本 (精选池+加权评分+动态窗口)", "config": { "code_list": FINAL_POOL, "factor_type": "weighted_momentum", "auto_day": True, "n_days": 25, # 提供默认窗口作为 fallback "min_days": 20, "max_days": 60, "diversified": True } } ] COMMON_CONFIG = { "start_date": "2019-01-01", "end_date": datetime.now().strftime('%Y-%m-%d'), "select_num": 3, "rebalance_days": 1, "rebalance_threshold": 0.0, "trade_cost": 0.001, "premium_control": {"enabled": True, "default_threshold": 0.10}, "use_cache": True, "ssh_tunnel": {"enabled": True, "host": "8.218.167.69", "port": 22, "username": "root", "key_path": "hk_ecs.pem", "local_port": 1080} } def run_experiment(): results = [] for i, item in enumerate(ITERATIONS): print(f"\n{'='*80}") print(f"运行实验 {item['label']}") print(f"{'='*80}") cfg = COMMON_CONFIG.copy() cfg.update(item['config']) strategy = RotationStrategy(cfg) try: res_df = strategy.run() # 计算指标 nav = res_df['轮动策略净值'] total_ret = nav.iloc[-1] - 1 days = (nav.index[-1] - nav.index[0]).days cagr = (1 + total_ret)**(365.25/days) - 1 daily_ret = res_df['轮动策略日收益率'] sharpe = daily_ret.mean() / daily_ret.std() * np.sqrt(252) if daily_ret.std() > 0 else 0 peak = nav.cummax() dd = (nav - peak) / peak max_dd = dd.min() results.append({ "label": item['label'], "total_ret": total_ret, "cagr": cagr, "max_dd": max_dd, "sharpe": sharpe, "nav": nav }) print(f"完成: CAGR={cagr:.2%}, MaxDD={max_dd:.2%}, Sharpe={sharpe:.2f}") except Exception as e: print(f"实验失败: {e}") import traceback traceback.print_exc() # ==================== 汇总报告 ==================== print(f"\n\n{'='*100}") print(f"{'策略迭代对比报告':^100}") print(f"{'='*100}") print(f"{'版本':<40} | {'累计收益':>10} | {'年化(CAGR)':>10} | {'最大回撤':>10} | {'夏普比率':>8} | {'贡献增量':>10}") print(f"{'-'*100}") prev_cagr = 0 for i, r in enumerate(results): delta = f"+{(r['cagr'] - prev_cagr)*100:>.2f}%" if i > 0 else "-" print(f"{r['label']:<40} | {r['total_ret']:>10.2%} | {r['cagr']:>10.2%} | {r['max_dd']:>10.2%} | {r['sharpe']:>8.2f} | {delta:>10}") prev_cagr = r['cagr'] print(f"{'='*100}") # ==================== 绘图 ==================== plt.figure(figsize=(15, 8)) for r in results: plt.plot(r['nav'].index, r['nav'], label=r['label'], linewidth=1.5) plt.yscale('log') plt.title("策略迭代 A/B 对比 - 净值曲线 (对数坐标)", fontsize=14) plt.legend() plt.grid(True, alpha=0.3) output_path = Path(__file__).parent.parent / "results" / "ab_test_iterations.png" plt.savefig(output_path) print(f"\n对比图表已保存至: {output_path}") if __name__ == "__main__": run_experiment()