feat: 新增 standardized_slope (t-statistic) 因子并实验验证

- 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 节详细分析
This commit is contained in:
2026-06-06 16:40:01 +08:00
parent aff04318b1
commit 921f84cb6a
5 changed files with 224 additions and 2 deletions

View File

@@ -0,0 +1,129 @@
"""
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()