refactor(momentum): 优化因子计算流程并对齐A股交易日历

- 添加辅助函数判断是否为A股指数
- 调整compute_factors函数结构，分别计算每个标的技术指标
- 严格实现T+1规则，确保信号只用T日及以前数据
- 对齐所有数据到A股交易日历，使用前向填充避免未来数据泄漏
- 增加有效代码有效性检查，剔除数据不足或缺失率过高的标的
- 完善函数注释，明确输入输出及核心逻辑说明
- 优化打印信息，清晰展示因子类型、窗口、有效标的及时间范围

core/factors/momentum.py

@@ -79,6 +79,11 @@ def calculate_daily_return(price_series: pd.Series) -> pd.Series:
     return price_series / price_series.shift(1) - 1
 
 
+def _is_china_index(code: str) -> bool:
+    """判断是否为A股指数"""
+    return code.endswith('.SH') or code.endswith('.SZ') or code.endswith('.SS')
+
+
 def compute_factors(
     index_data: pd.DataFrame,
     code_list: list,
@@ -88,19 +93,24 @@ def compute_factors(
     code_config: dict = None,
 ) -> tuple[pd.DataFrame, list]:
     """
-    计算所有指数的因子和日收益率（支持指数-ETF双轨数据）
+    计算所有指数的因子和日收益率（横截面策略版本）
+    
+    核心逻辑：
+    1. 每个标的按照自己的交易日历计算技术指标
+    2. 对齐到A股交易日历（取离A股交易日最近的有效数据，不使用未来数据）
+    3. 严格控制T+1规则：T日收盘计算信号，使用T日及之前的数据
     
     Args:
-        index_data: 指数价格数据（宽格式，用于因子计算）
+        index_data: 指数价格数据（宽格式，已对齐到A股交易日历，非A股可能有NaN）
         code_list: 指数代码列表
         n: 动量/趋势窗口
         factor_type: 'momentum' 或 'slope_r2'
         etf_data: ETF价格数据（宽格式，用于收益计算）
-        code_config: 代码配置字典 {code: {name, etf, market}}，用于判断是否为加密货币
+        code_config: 代码配置字典 {code: {name, etf, market}}
 
     Returns:
         tuple: (result_df, valid_codes)
-        - result_df: 包含因子得分和日收益率的DataFrame
+        - result_df: 包含因子得分和日收益率的DataFrame（按A股交易日对齐）
         - valid_codes: 有效代码列表
     """
     code_config = code_config or {}
@@ -109,45 +119,68 @@ def compute_factors(
     if etf_data is None:
         etf_data = pd.DataFrame()
     
-    result = index_data.copy()
+    # 获取A股交易日历（index_data的索引）
+    a_share_dates = index_data.index
     
-    # 过滤掉缺失值过多的指数
-    total_rows = len(result)
+    # 过滤有效代码
     valid_codes = []
     for code in code_list:
-        if code not in result.columns:
+        if code not in index_data.columns:
             print(f"  ⚠ 跳过 {code}: 不在数据中")
             continue
-        null_pct = result[code].isnull().sum() / total_rows
-        if null_pct > 0.2:
-            print(f"  ⚠ 剔除 {code}: 缺失率 {null_pct:.1%} 过高")
-            result = result.drop(columns=[code])
-        else:
-            valid_codes.append(code)
+        valid_codes.append(code)
+    
+    # 为每个标的单独计算指标，然后对齐到A股交易日历
+    result = pd.DataFrame(index=a_share_dates)
     
-    # 对有效指数计算因子和收益率
     for code in valid_codes:
-        # 因子基于指数价格计算
+        # 获取该标的的原始价格数据（去除NaN）
+        price_series = index_data[code].dropna()
+        
+        if len(price_series) < n + 1:
+            print(f"  ⚠ 剔除 {code}: 数据不足 ({len(price_series)} < {n+1})")
+            valid_codes.remove(code)
+            continue
+        
+        # 按照该标的自己的交易日历计算指标
         if factor_type == "momentum":
-            result[f"得分_{code}"] = calculate_momentum(result[code], n)
+            factor_series = calculate_momentum(price_series, n)
         elif factor_type == "slope_r2":
-            result[f"得分_{code}"] = calculate_slope_r2(result[code], n)
+            factor_series = calculate_slope_r2(price_series, n)
         else:
             raise ValueError(f"不支持的因子类型: {factor_type}")
         
-        # 日收益率基于指数价格计算（回测使用指数价格）
-        result[f"日收益率_{code}"] = calculate_daily_return(result[code])
+        # 计算日收益率
+        return_series = calculate_daily_return(price_series)
         
-    # 按得分列做 dropna
-    score_cols = [f"得分_{code}" for code in valid_codes]
+        # 对齐到A股交易日历：取离A股交易日最近的有效数据（不使用未来数据）
+        # 使用reindex + method='ffill'，确保T日使用T日或之前的数据
+        result[code] = price_series.reindex(a_share_dates, method='ffill')
+        result[f"得分_{code}"] = factor_series.reindex(a_share_dates, method='ffill')
+        result[f"日收益率_{code}"] = return_series.reindex(a_share_dates, method='ffill')
+    
+    # 过滤掉缺失值过多的指数（基于A股交易日历）
+    total_rows = len(result)
+    final_valid_codes = []
+    for code in valid_codes:
+        null_pct = result[code].isnull().sum() / total_rows
+        if null_pct > 0.2:
+            print(f"  ⚠ 剔除 {code}: 对齐后缺失率 {null_pct:.1%} 过高")
+            result = result.drop(columns=[code, f"得分_{code}", f"日收益率_{code}"], errors='ignore')
+        else:
+            final_valid_codes.append(code)
+
+    # 按得分列做 dropna（确保所有标的同时有数据）
+    score_cols = [f"得分_{code}" for code in final_valid_codes]
     result = result.dropna(subset=score_cols)
 
     print("\n因子计算完成:")
     print(f"  因子类型: {factor_type}")
     print(f"  窗口天数: {n}")
-    print(f"  有效指数: {len(valid_codes)}/{len(code_list)}")
+    print(f"  有效指数: {len(final_valid_codes)}/{len(code_list)}")
     print(f"  有效数据: {len(result)} 行")
-    if etf_data is not index_data:
+    print(f"  时间范围: {result.index[0].date()} ~ {result.index[-1].date()}")
+    if etf_data is not index_data and not etf_data.empty:
         print(f"  使用ETF数据计算收益: ✓")
 
-    return result, valid_codes
+    return result, final_valid_codes