aszerW
b564a47a1b
- simple_rotation.py: 新增3种score函数（vol_adjusted_momentum, slope_r2, momentum） - config_loader.py: FactorType枚举新增VOL_ADJUSTED_MOMENTUM - config_simple.yaml: factor.type 切换为 slope_r2 - experiments/factor_comparison.py: 4种因子对比实验脚本 - experiments/output: 实验结果（slope_r2全面胜出）
145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
"""
|
||
因子类型对比实验
|
||
|
||
测试 4 种动量因子的回测表现：
|
||
1. weighted_momentum: 加权线性回归动量 (当前默认)
|
||
2. vol_adjusted_momentum: 波动率调整动量 (Moskowitz TSMOM)
|
||
3. slope_r2: 斜率 × R² (未加权)
|
||
4. momentum: 简单收益率
|
||
|
||
运行方式:
|
||
cd /Users/aszer/code/etf
|
||
python3 rotation/experiments/factor_comparison.py
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import yaml
|
||
import copy
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
|
||
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
from rotation.simple_rotation import SimpleRotationStrategy
|
||
from rotation.config_loader import FactorType
|
||
|
||
FACTOR_TYPES = [
|
||
("weighted_momentum", "加权线性回归动量 (年化收益×R²)"),
|
||
("vol_adjusted_momentum", "波动率调整动量 (Sharpe-like×R²)"),
|
||
("slope_r2", "斜率×R² (未加权归一化)"),
|
||
("momentum", "简单收益率 (last/first-1)"),
|
||
]
|
||
|
||
|
||
def load_config():
|
||
"""Load base config"""
|
||
config_path = PROJECT_ROOT / "rotation" / "config_simple.yaml"
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
return yaml.safe_load(f)
|
||
|
||
|
||
def run_factor_experiment(factor_type: str):
|
||
"""Run backtest with a specific factor type"""
|
||
print(f"\n{'='*60}")
|
||
print(f" Testing: {factor_type}")
|
||
print(f"{'='*60}")
|
||
|
||
# Modify config to use this factor type
|
||
config_path = PROJECT_ROOT / "rotation" / "config_simple.yaml"
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
config = yaml.safe_load(f)
|
||
|
||
original_type = config['factor']['type']
|
||
config['factor']['type'] = factor_type
|
||
|
||
# Write temporary config
|
||
with open(config_path, 'w', encoding='utf-8') as f:
|
||
yaml.dump(config, f, allow_unicode=True, default_flow_style=False)
|
||
|
||
try:
|
||
# Run strategy
|
||
strategy = SimpleRotationStrategy()
|
||
result = strategy.run()
|
||
|
||
if result:
|
||
metrics = result.get('metrics', {})
|
||
return {
|
||
'factor_type': factor_type,
|
||
'annual_return': metrics.get('annual_return', 0),
|
||
'total_return': metrics.get('total_return', 0),
|
||
'sharpe_ratio': metrics.get('sharpe_ratio', 0),
|
||
'max_drawdown': metrics.get('max_drawdown', 0),
|
||
'win_rate': metrics.get('win_rate', 0),
|
||
'rebalance_count': metrics.get('rebalance_count', 0),
|
||
'calmar_ratio': metrics.get('calmar_ratio', 0),
|
||
}
|
||
finally:
|
||
# Restore original config
|
||
config['factor']['type'] = original_type
|
||
with open(config_path, 'w', encoding='utf-8') as f:
|
||
yaml.dump(config, f, allow_unicode=True, default_flow_style=False)
|
||
|
||
return None
|
||
|
||
|
||
def main():
|
||
if 'FLASK_API_URL' not in os.environ:
|
||
os.environ['FLASK_API_URL'] = 'https://k3s.tokenpluse.xyz'
|
||
|
||
print("="*60)
|
||
print(" ETF轮动策略 - 因子类型对比实验")
|
||
print("="*60)
|
||
|
||
results = []
|
||
|
||
for factor_type, description in FACTOR_TYPES:
|
||
print(f"\n>>> {description}")
|
||
result = run_factor_experiment(factor_type)
|
||
if result:
|
||
results.append(result)
|
||
print(f" ✓ {factor_type}: 年化={result['annual_return']:.2%}, "
|
||
f"夏普={result['sharpe_ratio']:.2f}, 回撤={result['max_drawdown']:.2%}")
|
||
else:
|
||
print(f" ✗ {factor_type}: 运行失败")
|
||
|
||
# Summary table
|
||
print(f"\n{'='*60}")
|
||
print(" 对比结果汇总")
|
||
print(f"{'='*60}")
|
||
print(f"{'因子类型':<25} {'年化收益':>10} {'夏普比率':>8} {'最大回撤':>10} {'调仓次数':>8} {'胜率':>6}")
|
||
print("-"*60)
|
||
|
||
for r in results:
|
||
print(f"{r['factor_type']:<25} "
|
||
f"{r['annual_return']:>9.2%} "
|
||
f"{r['sharpe_ratio']:>8.2f} "
|
||
f"{r['max_drawdown']:>9.2%} "
|
||
f"{r['rebalance_count']:>8d} "
|
||
f"{r['win_rate']:>5.1%}")
|
||
|
||
# Find best
|
||
if results:
|
||
best = max(results, key=lambda x: x['sharpe_ratio'])
|
||
print(f"\n★ 最优因子 (按夏普): {best['factor_type']}")
|
||
print(f" 年化收益: {best['annual_return']:.2%}")
|
||
print(f" 夏普比率: {best['sharpe_ratio']:.2f}")
|
||
print(f" 最大回撤: {best['max_drawdown']:.2%}")
|
||
|
||
# Save results
|
||
output_dir = PROJECT_ROOT / "rotation" / "experiments" / "output"
|
||
output_dir.mkdir(exist_ok=True)
|
||
output_path = output_dir / "factor_comparison_results.json"
|
||
with open(output_path, 'w', encoding='utf-8') as f:
|
||
json.dump({
|
||
'timestamp': datetime.now().isoformat(),
|
||
'results': results
|
||
}, f, ensure_ascii=False, indent=2)
|
||
print(f"\n结果已保存: {output_path}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|