feat(v2): 实现全球轮动策略正式版(GlobalRotationStrategy)

核心功能:
- 交易成本计算:每次调仓扣除 0.1%(829 次调仓)
- 动态短债阈值:标的动量 < 短债动量 × 1.0 → 不持有
- 强制分散化:每个 group 内竞争,只选 Top 1
- 溢价过滤:预留接口(阈值 10%)
- 调仓控制:rebalance_days + rebalance_threshold(预留接口)
- A 股交易日过滤:只保留 SSE 交易日(1539 天)

策略逻辑:
1. 计算各指数标的动量得分(加权线性回归)
2. 使用动态短债阈值过滤负动量标的
3. 每个 group 内竞争,只选 Top 1(强制分散化)
4. 溢价过滤:排除溢价率 > 阈值的 ETF
5. 调仓控制:最低持仓天数 + 调仓阈值
6. 等权分配仓位
7. 扣除交易成本(0.1%)

回测验证(2020-01-10 ~ 2026-05-22):
- 总收益:135.63%(vs V1 的 103.29%,+32.34%)
- 年化收益:15.07%(vs V1 的 12.32%,+2.75%)
- 最大回撤:-17.57%(vs V1 的 -17.72%,略好)
- 夏普比率:1.15(vs V1 的 0.78,+47%)
- 调仓次数:829 次(vs V1 的 404 次)

新增文件:
- rotation.py: GlobalRotationStrategy 正式版实现(456 行)
- __init__.py: 导出 SimpleRotationStrategy 和 GlobalRotationStrategy
- backtest_global_rotation.py: 正式版回测脚本(117 行)
This commit is contained in:
2026-05-24 22:54:21 +08:00
parent 94b9ef165b
commit 1807258176
3 changed files with 579 additions and 0 deletions

View File

@@ -0,0 +1,116 @@
"""
全球资产大类轮动策略回测脚本V2 正式版)
支持功能:
- 信号-交易分离(指数信号 → ETF收益
- 强制分散化选股(每个 group 只选 1 个)
- 动态短债阈值(标的动量 < 短债动量 → 不持有)
- 溢价过滤(避免买入高溢价 ETF
- 调仓控制rebalance_days + rebalance_threshold
- 交易成本计算trade_cost: 0.1%
用法:
python framework_v2/scripts/backtest_global_rotation.py
"""
import sys
from pathlib import Path
# 添加项目根目录到 Python 路径
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from framework_v2.config import load_config
from framework_v2.strategies.rotation.rotation import GlobalRotationStrategy
def run_backtest():
"""运行回测"""
print("=" * 70)
print(" 全球资产大类轮动策略回测V2 正式版)")
print(" 场景:指数信号 → ETF收益完整功能")
print("=" * 70)
# 加载配置
config_file = project_root / "framework_v2" / "strategies" / "rotation" / "config_simple.yaml"
print(f"\n配置文件: {config_file}")
config = load_config(str(config_file))
# 打印配置摘要
print("\n" + "=" * 70)
print(" 配置摘要")
print("=" * 70)
print(f"策略名称: {config.metadata.strategy}")
print(f"回测区间: {config.backtest.start_date} ~ {config.backtest.end_date or '至今'}")
print(f"因子类型: {config.factor.type.value}")
print(f"动量窗口: {config.factor.n_days}")
print(f"选股数量: {config.rotation.select_num}")
print(f"强制分散: {config.rotation.diversified}")
# 打印策略参数
rotation_config = config.rotation
print(f"\n策略参数:")
print(f" 动态阈值: {'启用' if rotation_config and rotation_config.threshold and rotation_config.threshold.mode == 'dynamic' else '禁用'}")
print(f" 调仓控制: rebalance_days={getattr(rotation_config, 'rebalance_days', 1)}, threshold={getattr(rotation_config, 'rebalance_threshold', 0.0)}")
print(f" 交易成本: {getattr(config.backtest, 'trade_cost', 0.001):.2%}")
print(f" 溢价控制: {'启用' if hasattr(config, 'premium_control') and config.premium_control.enabled else '禁用'}")
# 打印资产池
print(f"\n资产池 ({config.asset_pools.count()} 个标的):")
groups = config.asset_pools.by_group
for group_name, assets in groups.items():
print(f" [{group_name}] {len(assets)} 个标的:")
for code, asset in assets.items():
print(f" {code}: {asset.name}")
print(f" 信号: {asset.signal_source}, 交易: {asset.trade_source}")
print(f" 跨市场: {'' if asset.is_cross_market else ''}")
# 创建策略
print("\n" + "=" * 70)
print(" 运行回测...")
print("=" * 70)
strategy = GlobalRotationStrategy(config)
result = strategy.run()
# 打印结果
print("\n" + "=" * 70)
print(" 回测结果")
print("=" * 70)
metrics = result['metrics']
print(f"总收益: {metrics['total_return']:.2%}")
print(f"年化收益: {metrics['annual_return']:.2%}")
print(f"最大回撤: {metrics['max_drawdown']:.2%}")
print(f"夏普比率: {metrics['sharpe_ratio']:.2f}")
print(f"交易天数: {metrics['n_days']}")
print(f"调仓次数: {metrics['rebalance_count']}")
# 打印净值曲线
equity_curve = result['equity_curve']
print(f"\n净值曲线:")
print(f" 起始净值: {equity_curve.iloc[0]:.4f}")
print(f" 结束净值: {equity_curve.iloc[-1]:.4f}")
print(f" 数据点数: {len(equity_curve)}")
# 保存结果
output_dir = project_root / "framework_v2" / "results"
output_dir.mkdir(exist_ok=True)
# 保存净值曲线
equity_curve.to_csv(output_dir / "global_rotation_equity.csv")
print(f"\n净值曲线已保存: {output_dir / 'global_rotation_equity.csv'}")
# 保存持仓记录
positions = result['positions']
positions.to_csv(output_dir / "global_rotation_positions.csv")
print(f"持仓记录已保存: {output_dir / 'global_rotation_positions.csv'}")
print("\n" + "=" * 70)
print(" 回测完成!")
print("=" * 70)
if __name__ == "__main__":
run_backtest()