From 6749f8cf619c0ea33e9343604448715dabccf205 Mon Sep 17 00:00:00 2001 From: aszerW Date: Mon, 25 May 2026 00:29:49 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20GlobalRotationStrategy=20=E4=BD=BF?= =?UTF-8?q?=E7=94=A8=20CrossMarketAligner=20=E8=BF=9B=E8=A1=8C=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=AF=B9=E9=BD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心改进： - 替换手动对齐逻辑为框架标准的 CrossMarketAligner - 修复收益率计算顺序：先对齐价格 → 再计算收益率 - 修复首日收益率 NaN 问题（填充为 0%） - 添加 Pydantic Schema 验证（数据质量保证） 对齐逻辑变更： 修复前（错误）： 1. returns = close_df.pct_change() # 在原始日历计算 2. returns = returns[returns.index.isin(trading_calendar)] # 过滤 修复后（正确）： 1. aligner = CrossMarketAligner(target_calendar) 2. returns_df = aligner.align_multi_asset(close_dict) - 内部：先 ffill 价格到 A 股日历 - 内部：再计算收益率（休市日 = 0%） - 内部：填充首日 NaN 为 0% - 内部：Pydantic Schema 验证 回测验证（2020-01-10 ~ 2026-05-22）： - 修复前：总收益 135.63%，年化 15.07%，夏普 1.15 - 修复后：总收益 137.88%，年化 15.25%，夏普 1.16 - 收益提升：+2.25%（修复首日 NaN 和跨日收益率问题） 关键修复： 1. 首日收益率从 NaN 改为 0%（避免收益丢失） 2. 休市日收益率正确 = 0%（价格 ffill 后不变） 3. 消除跨多日收益率被当作单日收益率的 bug 4. 统一使用框架标准组件（CrossMarketAligner） --- framework_v2/strategies/rotation/rotation.py | 44 +++++++++++--------- 1 file changed, 25 insertions(+), 19 deletions(-) diff --git a/framework_v2/strategies/rotation/rotation.py b/framework_v2/strategies/rotation/rotation.py index 5bae48a..80e580b 100644 --- a/framework_v2/strategies/rotation/rotation.py +++ b/framework_v2/strategies/rotation/rotation.py @@ -18,6 +18,7 @@ from datetime import datetime, timedelta from framework_v2.core.strategy import StrategyBase from framework_v2.config.schemas import StrategyConfig from framework_v2.shared.factors import MomentumFactor +from framework_v2.shared.data.alignment import CrossMarketAligner class GlobalRotationStrategy(StrategyBase): @@ -275,7 +276,7 @@ class GlobalRotationStrategy(StrategyBase): def _execute_backtest(self, positions: pd.DataFrame, data: Dict[str, pd.DataFrame]) -> Dict[str, any]: """ - 执行回测（包含交易成本和调仓控制） + 执行回测（使用 CrossMarketAligner 进行正确的数据对齐） Args: positions: 仓位 DataFrame @@ -287,33 +288,38 @@ class GlobalRotationStrategy(StrategyBase): # 获取信号→交易映射 signal_to_trade = self.config.asset_pools.get_signal_to_trade_mapping() - # 提取交易标的的收盘价 - close_prices = {} + # 获取 A 股交易日历 + print("\n [对齐] 获取 A 股交易日历...") + trading_calendar = self._get_trading_calendar() + print(f" [日历] A 股交易日: {len(trading_calendar)} 天 ({trading_calendar[0]} ~ {trading_calendar[-1]})") + + # 创建对齐器 + aligner = CrossMarketAligner(target_calendar=trading_calendar) + + # 提取交易标的的收盘价，并对齐到 A 股日历 + print(" [对齐] 对齐 ETF 价格到 A 股日历...") + close_dict = {} for signal_code, trade_code in signal_to_trade.items(): if trade_code in data: - close_prices[signal_code] = data[trade_code]['close'] + # 提取收盘价 + close_series = data[trade_code]['close'] + # 使用 signal_code 作为键（与 positions 列名一致） + close_dict[signal_code] = close_series else: print(f" 警告: {trade_code} 数据不存在，跳过") - close_df = pd.DataFrame(close_prices) + # 使用 CrossMarketAligner 对齐多标的收益率 + # 内部逻辑：先 ffill 价格到 A 股日历，再计算收益率 + print(" [对齐] 计算收益率（先对齐价格，再计算）...") + returns_df = aligner.align_multi_asset(close_dict) + print(f" [对齐] 收益率数据: {len(returns_df)} 天, {len(returns_df.columns)} 个标的") - # 计算收益率 - returns = close_df.pct_change() - - # 获取 A 股交易日历并过滤 - print("\n [过滤] 获取 A 股交易日历...") - trading_calendar = self._get_trading_calendar() - - # 过滤到 A 股交易日 - original_days = len(returns) - returns = returns[returns.index.isin(trading_calendar)] - positions = positions[positions.index.isin(trading_calendar)] - filtered_days = len(returns) - print(f" [过滤] 原始数据: {original_days} 天 -> A 股交易日: {filtered_days} 天 (过滤 {original_days - filtered_days} 天)") + # 对齐 positions 到 A 股日历 + positions = positions.reindex(trading_calendar, method='ffill') # 计算策略收益（仓位加权，T+1 执行） positions_delayed = positions.shift(1).fillna(0) - strategy_returns = (positions_delayed * returns).sum(axis=1) + strategy_returns = (positions_delayed * returns_df).sum(axis=1) # 扣除交易成本 strategy_returns, rebalance_count = self._apply_trade_cost(