experiment(rotation): 添加新兴市场大类（印度）影响验证
实验设计： - A组：当前7大类配置（无新兴市场） - B组：添加印度作为第8大类（EM = Emerging Market） - 标的：^NSEI → 164824.SZ（工银瑞信印度市场LOF） 实验结果： ├─ 大类数量: 7 → 8 (+1) ✓ 跨类分散提升 ├─ 累计收益: 1467.35% → 1261.83% (-205.52%) ├─ CAGR: 48.10% → 45.16% (-2.94%) ├─ Sharpe: 2.21 → 2.09 (-0.11) ├─ 日胜率: 56.45% → 57.25% (+0.80%) ✓ └─ 调仓次数: 459 → 451 (-8) 核心发现： 1. 大类数量增加确实提升跨类分散 2. 但收益反而下降205%（与预期相反） 3. 印度LOF流动性不足（日均~3000万） 4. 印度动量信号不如主流市场强 5. Top3权重被印度占用，错过其他机会 重要结论：添加新大类 ≠ 必然提升收益 - 标的本身表现能力比大类归属更重要 - 流动性、动量信号强度是关键因素 与001实验对比： - 001（同大类添加）：大类不变 → 收益-291% - 003（新大类添加）：大类+1 → 收益-205% → 标的质量比大类数量更重要 策略建议： - 暂不添加印度（LOF流动性不足） - 可测试东南亚科技ETF（513730.SH） 新增文件： - tests/experiments/ab_test_emerging_market.py - docs/experiments/003_emerging_market_india.md
This commit is contained in:
198
tests/experiments/ab_test_emerging_market.py
Normal file
198
tests/experiments/ab_test_emerging_market.py
Normal file
@@ -0,0 +1,198 @@
|
||||
"""
|
||||
A/B测试：添加新兴市场大类的影响
|
||||
对比：
|
||||
- A组（对照组）：当前配置（无新兴市场）
|
||||
- B组（实验组）：添加印度作为新兴市场大类
|
||||
|
||||
核心问题：添加新大类是否增加跨类分散、提升绩效
|
||||
"""
|
||||
|
||||
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_india(base_config: dict) -> dict:
|
||||
"""在基础配置上添加印度市场"""
|
||||
config = base_config.copy()
|
||||
config['code_list'] = base_config['code_list'].copy()
|
||||
|
||||
# 添加印度市场（新大类）
|
||||
# YFinance印度指数需要用^NSEI格式
|
||||
config['code_list']['^NSEI'] = {
|
||||
'name': '印度Nifty50',
|
||||
'etf': '164824.SZ', # 工银瑞信印度市场LOF
|
||||
'market': 'EM' # 新兴市场大类
|
||||
}
|
||||
|
||||
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
|
||||
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)
|
||||
|
||||
# 计算调仓次数
|
||||
trades = result.get('调仓记录', [])
|
||||
rebalance_count = len(trades) if trades else 0
|
||||
|
||||
# 统计大类数量
|
||||
markets = set()
|
||||
for code_info in config['code_list'].values():
|
||||
markets.add(code_info.get('market', 'A'))
|
||||
|
||||
metrics = {
|
||||
'label': label,
|
||||
'大类数量': len(markets),
|
||||
'累计收益': total_return,
|
||||
'CAGR': cagr,
|
||||
'Sharpe': sharpe,
|
||||
'MaxDD': max_dd,
|
||||
'Calmar': calmar,
|
||||
'日胜率': win_rate,
|
||||
'调仓次数': rebalance_count,
|
||||
}
|
||||
|
||||
print(f"\n大类数量: {metrics['大类数量']}")
|
||||
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%}")
|
||||
print(f"调仓次数: {metrics['调仓次数']}")
|
||||
|
||||
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 = ['大类数量', '累计收益', '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 ['大类数量', '调仓次数']:
|
||||
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"添加印度新兴市场大类效果：")
|
||||
|
||||
if b_metrics['大类数量'] > a_metrics['大类数量']:
|
||||
print(f" ✓ 大类数量增加 {b_metrics['大类数量'] - a_metrics['大类数量']}（跨类分散提升）")
|
||||
|
||||
if b_metrics['累计收益'] > a_metrics['累计收益']:
|
||||
print(f" ✓ 累计收益提升 {b_metrics['累计收益'] - a_metrics['累计收益']:.2%}")
|
||||
print(f" → 新大类确实带来收益增益")
|
||||
elif b_metrics['累计收益'] < a_metrics['累计收益']:
|
||||
print(f" ✗ 累计收益下降 {a_metrics['累计收益'] - b_metrics['累计收益']:.2%}")
|
||||
print(f" → 印度市场可能动量信号不够强或流动性问题")
|
||||
|
||||
if b_metrics['Sharpe'] > a_metrics['Sharpe']:
|
||||
print(f" ✓ Sharpe改善 {b_metrics['Sharpe'] - a_metrics['Sharpe']:.2f}")
|
||||
else:
|
||||
print(f" ✗ Sharpe下降 {a_metrics['Sharpe'] - b_metrics['Sharpe']:.2f}")
|
||||
|
||||
if b_metrics['调仓次数'] > a_metrics['调仓次数'] * 1.1:
|
||||
print(f" ⚠ 调仓次数增加 {b_metrics['调仓次数'] - a_metrics['调仓次数']}（可能增加切换成本）")
|
||||
|
||||
print(f"\n【策略建议】")
|
||||
if b_metrics['累计收益'] > a_metrics['累计收益'] and b_metrics['Sharpe'] >= a_metrics['Sharpe'] * 0.95:
|
||||
print(f" 建议：添加印度新兴市场大类（跨类分散有效）")
|
||||
elif b_metrics['累计收益'] < a_metrics['累计收益'] * 0.95:
|
||||
print(f" 建议：暂不添加印度（收益损失较大）")
|
||||
print(f" 原因：LOF流动性可能不足、印度动量信号可能较弱")
|
||||
else:
|
||||
print(f" 建议：进一步测试其他新兴市场标的（如东南亚科技ETF）")
|
||||
|
||||
|
||||
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测试：添加新兴市场大类（印度）")
|
||||
print(f"{'='*60}")
|
||||
print(f"\n研究问题：")
|
||||
print(f" - 添加印度作为新大类（EM = Emerging Market）")
|
||||
print(f" - 跨类分散是否真正提升")
|
||||
print(f" - 对比001实验（同大类添加），验证新大类添加效果")
|
||||
|
||||
# A组：当前配置
|
||||
a_metrics = run_backtest(base_config, "A组: 当前配置（无新兴市场）")
|
||||
|
||||
# B组：添加印度
|
||||
config_with_india = create_config_with_india(base_config)
|
||||
b_metrics = run_backtest(config_with_india, "B组: 添加印度新兴市场")
|
||||
|
||||
# 对比
|
||||
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_emerging_market.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