aszerW
921f84cb6a
- simple_rotation.py: 新增 standardized_slope_score 函数 (slope/SE) - config_loader.py: FactorType 枚举新增 STANDARDIZED_SLOPE - 对比实验结果: standardized_slope 年化 13.73% vs slope_r2 19.84% - 结论: t-statistic 过度惩罚高波动资产的有效趋势信号，不适合本场景 - 文档更新: 动量因子对比调研报告新增 3.3 节详细分析
130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
"""
|
||
slope_r2 vs standardized_slope 对比实验
|
||
|
||
测试两种信号质量优化的回测表现：
|
||
1. slope_r2: slope × R² (当前默认)
|
||
2. standardized_slope: slope / SE(slope) (t-statistic)
|
||
|
||
运行方式:
|
||
cd /Users/aszer/code/etf
|
||
set -a && source .env && set +a
|
||
python3 rotation/experiments/std_slope_test.py
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import yaml
|
||
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
|
||
|
||
FACTOR_TYPES = [
|
||
("slope_r2", "slope_r2 (slope×R², 当前默认)"),
|
||
("standardized_slope", "standardized_slope (t-statistic)"),
|
||
]
|
||
|
||
|
||
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}")
|
||
|
||
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
|
||
|
||
with open(config_path, 'w', encoding='utf-8') as f:
|
||
yaml.dump(config, f, allow_unicode=True, default_flow_style=False)
|
||
|
||
try:
|
||
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:
|
||
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(" slope_r2 vs standardized_slope 对比实验")
|
||
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}: 运行失败")
|
||
|
||
print(f"\n{'='*60}")
|
||
print(" 对比结果汇总")
|
||
print(f"{'='*60}")
|
||
print(f"{'因子类型':<25} {'年化收益':>10} {'夏普比率':>8} {'最大回撤':>10} {'Calmar':>8} {'调仓次数':>8} {'胜率':>6}")
|
||
print("-"*80)
|
||
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['calmar_ratio']:>8.2f} "
|
||
f"{r['rebalance_count']:>8d} "
|
||
f"{r['win_rate']:>5.1%}")
|
||
|
||
if len(results) >= 2:
|
||
base = results[0]
|
||
new = results[1]
|
||
print(f"\n{'='*60}")
|
||
print(" 变化对比 (standardized_slope vs slope_r2)")
|
||
print(f"{'='*60}")
|
||
print(f" 年化收益: {base['annual_return']:.2%} → {new['annual_return']:.2%} "
|
||
f"(Δ={new['annual_return']-base['annual_return']:+.2%})")
|
||
print(f" 夏普比率: {base['sharpe_ratio']:.2f} → {new['sharpe_ratio']:.2f} "
|
||
f"(Δ={new['sharpe_ratio']-base['sharpe_ratio']:+.2f})")
|
||
print(f" 最大回撤: {base['max_drawdown']:.2%} → {new['max_drawdown']:.2%} "
|
||
f"(Δ={new['max_drawdown']-base['max_drawdown']:+.2%})")
|
||
print(f" 调仓次数: {base['rebalance_count']} → {new['rebalance_count']} "
|
||
f"(Δ={new['rebalance_count']-base['rebalance_count']:+d})")
|
||
|
||
output_dir = PROJECT_ROOT / "rotation" / "experiments" / "output"
|
||
output_dir.mkdir(exist_ok=True)
|
||
output_path = output_dir / "std_slope_test_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()
|