experiment(rotation): 同大类扩充与纳指vs标普替换对比实验
技术修复： - SOCKS5代理IPv6问题：socks5:// → socks5h:// (hybrid_source.py, yfinance_source.py) 目录整理： - scripts/ → 仅保留策略入口(daily_scheduler, run_rotation, run_cci_screener) - 实验脚本移至 tests/experiments/ - 工具脚本移至 tests/utils/ - 实验记录新增 docs/experiments/ - results/ 添加到 gitignore 实验结果： 实验001 - 同大类扩充（添加标普500）： ├─ 累计收益: 1467.35% → 1176.26% (-291%) ├─ CAGR: 48.10% → 43.82% (-4.28%) ├─ 调仓次数: 459 → 501 (+42次) └─ 结论: 添加同大类标的不增加跨类分散，反而侵蚀收益 实验002 - 纳指vs标普替换对比： ├─ 累计收益: 1467.35% → 1118.77% (-348%) ├─ CAGR: 48.10% → 42.87% (-5.22%) ├─ Sharpe: 2.21 → 2.08 (-0.13) ├─ MaxDD: -17.33% → -15.14% (+2.18%) └─ 结论: 纳指100优于标普500，成长风格更适合动量策略 策略建议： - 保持纳指100作为美股大类代表 - 不添加同大类新标的（避免类内切换成本） - 新增标的应优先考虑新大类（增加跨类分散）
This commit is contained in:
183
tests/experiments/ab_test_spx.py
Normal file
183
tests/experiments/ab_test_spx.py
Normal file
@@ -0,0 +1,183 @@
|
||||
"""
|
||||
A/B测试：添加标普500对轮动策略的影响
|
||||
对比：
|
||||
- A组（对照组）：当前11只标的配置
|
||||
- B组（实验组）：添加标普500后的12只标的配置
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from strategies.rotation.engine import RotationStrategy
|
||||
import pandas as pd
|
||||
|
||||
|
||||
def create_config_with_spx(base_config: dict) -> dict:
|
||||
"""在基础配置上添加标普500"""
|
||||
config = base_config.copy()
|
||||
config['code_list'] = base_config['code_list'].copy()
|
||||
|
||||
# 添加标普500（美股大类内）
|
||||
config['code_list']['SPX'] = {
|
||||
'name': '标普500',
|
||||
'etf': '513500.SH',
|
||||
'market': 'US' # 与纳指100同属美股大类
|
||||
}
|
||||
|
||||
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() # result 是 DataFrame
|
||||
|
||||
if result is None or len(result) == 0:
|
||||
return None
|
||||
|
||||
# 从 DataFrame 中直接计算指标
|
||||
strategy_nav = result['轮动策略净值']
|
||||
strategy_ret = result['轮动策略日收益率']
|
||||
benchmark_nav = result['基准净值']
|
||||
benchmark_ret = result['基准日收益率']
|
||||
|
||||
# 累计收益
|
||||
total_return = strategy_nav.iloc[-1] - 1
|
||||
|
||||
# CAGR (交易日口径)
|
||||
days = len(result)
|
||||
years = days / 250
|
||||
cagr = (strategy_nav.iloc[-1] ** (1/years)) - 1
|
||||
|
||||
# Sharpe
|
||||
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
|
||||
calmar = cagr / abs(max_dd) if max_dd < 0 else 0
|
||||
|
||||
# 日胜率
|
||||
win_rate = (strategy_ret > 0).sum() / len(strategy_ret)
|
||||
|
||||
# 提取关键指标
|
||||
metrics = {
|
||||
'label': label,
|
||||
'标的数': len(config['code_list']),
|
||||
'累计收益': total_return,
|
||||
'CAGR': cagr,
|
||||
'Sharpe': sharpe,
|
||||
'MaxDD': max_dd,
|
||||
'Calmar': calmar,
|
||||
'日胜率': win_rate,
|
||||
}
|
||||
|
||||
print(f"\n标的池: {len(config['code_list'])}只")
|
||||
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{'指标':<12} {'A组(无SPX)':<15} {'B组(有SPX)':<15} {'差异':<15}")
|
||||
print("-" * 60)
|
||||
|
||||
metrics_keys = ['标的数', '累计收益', 'CAGR', 'Sharpe', 'MaxDD', 'Calmar', '日胜率']
|
||||
|
||||
for key in metrics_keys:
|
||||
a_val = a_metrics.get(key, 0)
|
||||
b_val = b_metrics.get(key, 0)
|
||||
|
||||
if key == '标的数':
|
||||
diff = b_val - a_val
|
||||
diff_str = f"+{diff}" if diff > 0 else str(diff)
|
||||
else:
|
||||
diff = b_val - a_val
|
||||
if key in ['累计收益', 'CAGR', 'MaxDD', '日胜率']:
|
||||
diff_str = f"{diff*100:+.2f}%"
|
||||
else:
|
||||
diff_str = f"{diff:+.2f}"
|
||||
|
||||
if key in ['累计收益', 'CAGR', 'MaxDD', '日胜率']:
|
||||
a_str = f"{a_val:.2%}"
|
||||
b_str = f"{b_val:.2%}"
|
||||
else:
|
||||
a_str = str(a_val)
|
||||
b_str = str(b_val)
|
||||
|
||||
print(f"{key:<12} {a_str:<15} {b_str:<15} {diff_str:<15}")
|
||||
|
||||
print("-" * 60)
|
||||
|
||||
# 分析美股大类内部切换情况
|
||||
print(f"\n【关键发现】")
|
||||
print(f"添加标普500后：")
|
||||
print(f" - 美股大类从1只→2只（纳指100 + 标普500）")
|
||||
print(f" - 类内竞争：纳指100 vs 标普500，得分高者代表美股大类")
|
||||
print(f" - 跨类分散不变：美股大类还是只输出1只冠军进入Top3")
|
||||
|
||||
if b_metrics['累计收益'] != a_metrics['累计收益']:
|
||||
print(f" - 累计收益变化：{a_metrics['累计收益']:.2%} → {b_metrics['累计收益']:.2%}")
|
||||
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
import yaml
|
||||
|
||||
# 加载基础配置
|
||||
config_path = Path(__file__).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测试：添加标普500对diversified模式的影响")
|
||||
print(f"{'='*60}")
|
||||
print(f"\n测试假设：")
|
||||
print(f" - diversified=true 模式下，每大类只选1只冠军")
|
||||
print(f" - 添加标普500（同属美股大类）不会增加跨类分散")
|
||||
print(f" - 但可能增加类内切换频率和换手率")
|
||||
|
||||
# A组：当前配置（11只，无标普500）
|
||||
a_metrics = run_backtest(base_config, "A组: 当前配置（11只，无标普500）")
|
||||
|
||||
# B组：添加标普500后的配置（12只）
|
||||
config_with_spx = create_config_with_spx(base_config)
|
||||
b_metrics = run_backtest(config_with_spx, "B组: 添加标普500（12只）")
|
||||
|
||||
# 对比结果
|
||||
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 / 'results' / 'ab_test_spx.csv'
|
||||
results_df.to_csv(results_path, index=False)
|
||||
print(f"\n对比结果已保存: {results_path}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user