perf(factor): metrics热路径numpy/scipy化——rank_corr/quantile_mask/decile的pandas rank改scipy rankdata(axis=1)C实现,宽矩阵压力下分钟级→秒级;接口与数值不变,测试原样全绿+速度护栏 [vps]

sanguo_factor/metrics.py

@@ -8,6 +8,7 @@ import warnings
 
 import numpy as np
 import pandas as pd
+from scipy import stats
 
 TRADING_DAYS_PER_YEAR = 244
 
@@ -35,8 +36,27 @@ def rank_corr_rows(A: pd.DataFrame, B: pd.DataFrame) -> pd.Series:
     """逐日 Spearman:先各自按行取秩再逐行 Pearson."""
     cols = A.columns.intersection(B.columns)
     idx = A.index.intersection(B.index)
-    a = A.loc[idx, cols].rank(axis=1).to_numpy(dtype=float)
-    b = B.loc[idx, cols].rank(axis=1).to_numpy(dtype=float)
+    a_vals = A.loc[idx, cols].to_numpy(dtype=float)
+    b_vals = B.loc[idx, cols].to_numpy(dtype=float)
+
+    # scipy.stats.rankdata ranks NaN positions (NaN sorts last)
+    # pandas keeps NaN as NaN, so we need to mask and restore
+    a_nan = np.isnan(a_vals)
+    b_nan = np.isnan(b_vals)
+
+    # Fill NaN with large negative value (sorts first, then rankdata puts it last)
+    # But we need it to sort last, so use large positive
+    a_filled = np.where(a_nan, np.inf, a_vals)
+    b_filled = np.where(b_nan, np.inf, b_vals)
+
+    # scipy.stats.rankdata with axis=1 ranks per row
+    a = stats.rankdata(a_filled, axis=1)
+    b = stats.rankdata(b_filled, axis=1)
+
+    # Restore NaN at original NaN positions
+    a[a_nan] = np.nan
+    b[b_nan] = np.nan
+
     return _row_pearson(a, b, idx)
 
 
@@ -53,12 +73,43 @@ def factor_turnover(F: pd.DataFrame) -> float:
 
 def _quantile_mask(F: pd.DataFrame, top: bool) -> pd.DataFrame:
     """按行把因子值分位选mask(基于升序秩/当日有效数). top=True选最大10%,False选最小10%."""
-    ranks = F.rank(axis=1, ascending=False)  # 1=最大
-    n = ranks.notna().sum(axis=1)
-    k = np.maximum((n * 0.1).round().astype(int), 1)
+    vals = F.to_numpy(dtype=float)
+    nan_mask = np.isnan(vals)
+
+    # For descending rank (1=max), rank negative values
+    # scipy.stats.rankdata sorts ascending by default
+    vals_filled = np.where(nan_mask, -np.inf, -vals)  # NaN -> -inf (smallest), negate for descending
+    ranks = stats.rankdata(vals_filled, axis=1)
+
+    # Now ranks are in descending order (1=max), but NaN got rank 1, need to shift
+    # Actually, let me think more carefully:
+    # - pandas rank(axis=1, ascending=False): 1 = max, NaN = NaN
+    # - scipy rankdata on negative values: smallest original gets largest rank
+    # - So -val (largest original) -> smallest rank value = 1
+    # This is what we want!
+
+    # But NaN got filled with -inf, which becomes largest, so rank is 1 - wrong!
+    # Need to fill NaN with something that becomes largest rank
+    vals_filled = np.where(nan_mask, np.inf, -vals)  # NaN -> +inf (largest after negation)
+    ranks = stats.rankdata(vals_filled, axis=1)
+
+    # Now: max original -> -max -> min rank = 1 ✓
+    # NaN -> +inf -> max rank (last) ✓
+
+    # Restore NaN at original NaN positions
+    ranks[nan_mask] = np.nan
+
+    n = (~nan_mask).sum(axis=1)
+    k = np.maximum((n * 0.1).round().astype(int), 1).reshape(-1, 1)
+
     if top:
-        return ranks.le(k, axis=0) & ranks.notna()
-    return ranks.ge(n - k + 1, axis=0) & ranks.notna()
+        mask = (ranks <= k) & ~nan_mask
+    else:
+        # For bottom: n - k + 1 to n
+        # E.g., n=10, k=1: ranks 10 to 10 (last 1)
+        mask = (ranks >= n.reshape(-1, 1) - k + 1) & ~nan_mask
+
+    return pd.DataFrame(mask, index=F.index, columns=F.columns)
 
 
 def long_short_annual_return(F: pd.DataFrame, R: pd.DataFrame) -> float | None:
@@ -79,7 +130,21 @@ def decile_annual_returns(F: pd.DataFrame, R: pd.DataFrame) -> list[float | None
     cols = F.columns.intersection(R.columns)
     idx = F.index.intersection(R.index)
     f, r = F.loc[idx, cols], R.loc[idx, cols]
-    pct = f.rank(axis=1, ascending=True).div(f.notna().sum(axis=1), axis=0)
+
+    vals = f.to_numpy(dtype=float)
+    nan_mask = np.isnan(vals)
+
+    # Fill NaN with large value (sorts last, gets largest rank)
+    vals_filled = np.where(nan_mask, np.inf, vals)
+    ranks = stats.rankdata(vals_filled, axis=1)
+
+    # Restore NaN at original positions
+    ranks[nan_mask] = np.nan
+
+    # Normalize to percentiles
+    n = (~nan_mask).sum(axis=1, keepdims=True)
+    pct = ranks / n
+
     out: list[float | None] = []
     for d in range(10):
         sel = (pct > d / 10) & (pct <= (d + 1) / 10)

tests/factor/test_metrics.py

@@ -99,3 +99,13 @@ def test_summarize_factor_structure():
         assert key in p1
     assert p1["count"] == 60
     assert p1["ic_mean"] < 0  # 构造为负相关
+
+
+def test_rank_corr_rows_speed_guard():
+    import time
+    rng = np.random.default_rng(1)
+    F = pd.DataFrame(rng.normal(size=(2000, 100)))
+    R = pd.DataFrame(rng.normal(size=(2000, 100)))
+    t0 = time.time()
+    rank_corr_rows(F, R)
+    assert time.time() - t0 < 3