test(experiments): add France CAC40 and SEA ETF experiments
- Add France CAC40 market test (004) - Add SEA ETF limited test (005) - Add France in EU category test (006) - Update experiment README with new results - Modify emerging market test description
This commit is contained in:
@@ -2,7 +2,7 @@
|
||||
A/B测试：添加新兴市场大类的影响
|
||||
对比：
|
||||
- A组（对照组）：当前配置（无新兴市场）
|
||||
- B组（实验组）：添加印度作为新兴市场大类
|
||||
- B组（实验组）：添加印度/法国作为新兴/欧洲市场大类
|
||||
|
||||
核心问题：添加新大类是否增加跨类分散、提升绩效
|
||||
"""
|
||||
|
||||
200
tests/experiments/ab_test_france.py
Normal file
200
tests/experiments/ab_test_france.py
Normal file
@@ -0,0 +1,200 @@
|
||||
"""
|
||||
A/B测试：添加法国CAC40市场大类的影响
|
||||
对比：
|
||||
- A组（对照组）：当前配置（无法国）
|
||||
- B组（实验组）：添加法国CAC40作为新大类
|
||||
|
||||
核心问题：法国市场是否能有效补充欧洲分散
|
||||
"""
|
||||
|
||||
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(base_config: dict) -> dict:
|
||||
"""在基础配置上添加法国市场"""
|
||||
config = base_config.copy()
|
||||
config['code_list'] = base_config['code_list'].copy()
|
||||
|
||||
# 添加法国CAC40（新大类）
|
||||
# 当前已有德国DAX(EU)，法国可以增加欧洲内部的多样性
|
||||
# 注意：德国和法国同属EU大类，但这是不同的指数
|
||||
config['code_list']['^FCHI'] = {
|
||||
'name': '法国CAC40',
|
||||
'etf': '513080.SH', # 法国ETF华安
|
||||
'market': 'FR' # 法国大类（独立于德国）
|
||||
}
|
||||
|
||||
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'))
|
||||
|
||||
metrics = {
|
||||
'label': label,
|
||||
'大类数量': len(markets),
|
||||
'累计收益': total_return,
|
||||
'CAGR': cagr,
|
||||
'Sharpe': sharpe,
|
||||
'MaxDD': max_dd,
|
||||
'Calmar': calmar,
|
||||
'日胜率': win_rate,
|
||||
}
|
||||
|
||||
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%}")
|
||||
|
||||
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 == '大类数量':
|
||||
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"添加法国CAC40大类效果：")
|
||||
|
||||
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}")
|
||||
|
||||
print(f"\n【与德国DAX对比分析】")
|
||||
print(f" 当前配置：德国DAX (513030.SH) → EU大类")
|
||||
print(f" 新增配置：法国CAC40 (513080.SH) → FR大类")
|
||||
print(f" ")
|
||||
print(f" 德国 vs 法国特点：")
|
||||
print(f" ├─ 德国DAX：工业权重高（汽车、机械）")
|
||||
print(f" ├─ 法国CAC：奢侈品权重高（LV、欧莱雅）")
|
||||
print(f" └─ 两者风格不同，可能互补")
|
||||
|
||||
print(f"\n【策略建议】")
|
||||
if b_metrics['累计收益'] > a_metrics['累计收益'] and b_metrics['Sharpe'] >= a_metrics['Sharpe'] * 0.95:
|
||||
print(f" 建议：添加法国CAC40（欧洲分散有效）")
|
||||
elif b_metrics['累计收益'] < a_metrics['累计收益'] * 0.95:
|
||||
print(f" 建议：暂不添加法国（收益损失较大）")
|
||||
print(f" 原因：法国动量信号可能不如德国DAX强")
|
||||
else:
|
||||
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市场大类")
|
||||
print(f"{'='*60}")
|
||||
print(f"\n研究问题：")
|
||||
print(f" - 添加法国CAC40作为新大类（FR）")
|
||||
print(f" - 与德国DAX（EU）形成欧洲内部分散")
|
||||
print(f" - 德国工业风格 vs 法国奢侈品风格")
|
||||
print(f" - 验证欧洲分散是否有效")
|
||||
|
||||
# A组：当前配置
|
||||
a_metrics = run_backtest(base_config, "A组: 当前配置（仅德国DAX）")
|
||||
|
||||
# B组：添加法国CAC40
|
||||
config_with_france = create_config_with_france(base_config)
|
||||
b_metrics = run_backtest(config_with_france, "B组: 添加法国CAC40")
|
||||
|
||||
# 对比
|
||||
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.csv'
|
||||
results_df.to_csv(results_path, index=False)
|
||||
print(f"\n对比结果已保存: {results_path}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
218
tests/experiments/ab_test_france_in_eu.py
Normal file
218
tests/experiments/ab_test_france_in_eu.py
Normal file
@@ -0,0 +1,218 @@
|
||||
"""
|
||||
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()
|
||||
208
tests/experiments/ab_test_sea_etf.py
Normal file
208
tests/experiments/ab_test_sea_etf.py
Normal file
@@ -0,0 +1,208 @@
|
||||
"""
|
||||
A/B测试：添加东南亚科技ETF的影响（受限测试）
|
||||
对比：
|
||||
- A组（对照组）：当前配置（无东南亚）
|
||||
- B组（实验组）：添加东南亚科技ETF
|
||||
|
||||
限制说明：
|
||||
- 东南亚科技ETF（513730.SH）2023年12月上市，数据仅约2年
|
||||
- 新交所泛东南亚科技指数在YFinance中暂无数据
|
||||
- 本次测试使用ETF价格作为信号源（非最佳实践，仅作参考）
|
||||
- 回测时间范围将被缩短
|
||||
|
||||
核心问题：新兴市场ETF流动性是否优于印度LOF
|
||||
"""
|
||||
|
||||
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_sea(base_config: dict) -> dict:
|
||||
"""在基础配置上添加东南亚科技"""
|
||||
config = base_config.copy()
|
||||
config['code_list'] = base_config['code_list'].copy()
|
||||
|
||||
# 添加东南亚科技（新大类）
|
||||
# 注意：由于指数数据不可用，使用ETF价格作为信号源
|
||||
# 513730.SH 同时作为指数代码和ETF代码
|
||||
config['code_list']['513730.SH'] = {
|
||||
'name': '东南亚科技',
|
||||
'etf': '513730.SH', # 华泰柏瑞东南亚科技ETF
|
||||
'market': 'SEA' # 东南亚大类
|
||||
}
|
||||
|
||||
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'))
|
||||
|
||||
metrics = {
|
||||
'label': label,
|
||||
'大类数量': len(markets),
|
||||
'回测天数': days,
|
||||
'回测年数': years,
|
||||
'累计收益': total_return,
|
||||
'CAGR': cagr,
|
||||
'Sharpe': sharpe,
|
||||
'MaxDD': max_dd,
|
||||
'Calmar': calmar,
|
||||
'日胜率': win_rate,
|
||||
}
|
||||
|
||||
print(f"\n大类数量: {metrics['大类数量']}")
|
||||
print(f"回测天数: {metrics['回测天数']}")
|
||||
print(f"回测年数: {metrics['回测年数']:.2f}")
|
||||
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 = ['大类数量', '回测天数', '回测年数', '累计收益', '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(int(a_val))
|
||||
b_str = str(int(b_val))
|
||||
diff_str = f"+{int(diff)}" if diff > 0 else str(int(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" ⚠ 本次测试数据量受限（东南亚ETF仅2年数据）")
|
||||
print(f" ⚠ 使用ETF价格作为信号源（指数数据暂不可用）")
|
||||
print(f" ⚠ 结果仅供参考，不建议直接用于决策")
|
||||
|
||||
print(f"\n【关键发现】")
|
||||
|
||||
if b_metrics['大类数量'] > a_metrics['大类数量']:
|
||||
print(f" ✓ 大类数量增加 {b_metrics['大类数量'] - a_metrics['大类数量']}")
|
||||
|
||||
if b_metrics['累计收益'] > a_metrics['累计收益']:
|
||||
print(f" ✓ 累计收益提升 {b_metrics['累计收益'] - a_metrics['累计收益']:.2%}")
|
||||
else:
|
||||
print(f" ✗ 累计收益下降 {a_metrics['累计收益'] - b_metrics['累计收益']:.2%}")
|
||||
|
||||
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}")
|
||||
|
||||
print(f"\n【策略建议】")
|
||||
print(f" 建议：等待东南亚科技ETF积累更多数据后再测试")
|
||||
print(f" 原因：")
|
||||
print(f" 1. 数据量不足（仅{b_metrics['回测年数']:.1f}年）")
|
||||
print(f" 2. 指数信号源暂不可用")
|
||||
print(f" 3. 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)
|
||||
|
||||
# 设置回测结束日期
|
||||
from datetime import datetime
|
||||
base_config['end_date'] = datetime.now().strftime('%Y-%m-%d')
|
||||
|
||||
# ⚠ 重要：由于东南亚ETF数据从2023年12月开始
|
||||
# 需要调整start_date以匹配数据可用范围
|
||||
# 本次测试将使用较短的时间窗口
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f" A/B测试：添加东南亚科技ETF（受限测试）")
|
||||
print(f"{'='*60}")
|
||||
print(f"\n⚠ 限制说明：")
|
||||
print(f" - 东南亚科技ETF（513730.SH）2023年12月上市")
|
||||
print(f" - 数据仅约2年，回测时间范围受限")
|
||||
print(f" - 指数数据暂不可用，使用ETF价格作为信号源")
|
||||
print(f" - 结果仅供参考，不建议直接用于决策")
|
||||
|
||||
# A组：当前配置（使用较短时间窗口）
|
||||
config_a = base_config.copy()
|
||||
config_a['start_date'] = '2024-01-01' # 调整为东南亚ETF有数据的起始时间
|
||||
a_metrics = run_backtest(config_a, "A组: 当前配置（2024年起）")
|
||||
|
||||
# B组：添加东南亚科技
|
||||
config_b = create_config_with_sea(base_config)
|
||||
config_b['start_date'] = '2024-01-01'
|
||||
b_metrics = run_backtest(config_b, "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_sea_etf.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