feat(factor): challenger_lgbm 数据层——在役池档案/防缺口 label/双侧 CSRankNorm [nas]

sanguo_factor/challenger_lgbm.py

@@ -0,0 +1,32 @@
+# sanguo_factor/challenger_lgbm.py
+"""LGBM challenger 三路加权对拍(2026-10-10 spec §4.4 定案).
+
+三路=等权(方向调整)/ICIR(quant12_icirfit_v1 档案)/LightGBM(月度重训),
+同池同窗 walk-forward 影子对拍;challenger 永不进生产,对拍件落
+reports/factor_monthly/challenger_lgbm/ 子目录(判定层端点只读根层).
+"""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+import pandas as pd
+
+_PROFILE = Path(__file__).parent / "weight_profiles" / "quant12_icirfit_v1.json"
+
+
+def load_pool() -> dict[str, dict]:
+    """v1 对拍池=在役 12 源(ICIR 档案现成,三路同池才可比;扩池挂后续)."""
+    with open(_PROFILE, encoding="utf-8") as f:
+        return json.load(f)["sources"]
+
+
+def build_label(close_wide: pd.DataFrame) -> pd.DataFrame:
+    """防缺口 label=T+1 收盘→T+2 收盘(信号 T 收盘出,T+1 全天可成交)."""
+    return close_wide.shift(-2) / close_wide.shift(-1) - 1.0
+
+
+def cs_rank_norm(df: pd.DataFrame) -> pd.DataFrame:
+    """截面秩归一 (rank_pct-0.5);NaN 透传不占当日截面."""
+    return df.rank(axis=1, pct=True) - 0.5

tests/factor/test_challenger_lgbm.py

@@ -0,0 +1,38 @@
+"""LGBM challenger 三路对拍(spec §4.4 2026-10-10):数据层."""
+import json
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from sanguo_factor.challenger_lgbm import build_label, cs_rank_norm, load_pool
+
+
+def test_load_pool_twelve_sources():
+    pool = load_pool()
+    assert len(pool) == 12
+    assert pool["vma_60"]["direction"] == "+"
+    assert pool["vol_ma5"]["direction"] == "-"
+    assert 0.0 < pool["alpha16"]["weight"] < 0.2
+
+
+def test_build_label_gap_proof():
+    idx = pd.date_range("2025-01-01", periods=4, freq="D")
+    close = pd.DataFrame({"a": [10.0, 11.0, 12.0, 13.0], "b": [20.0, 20.0, 21.0, 22.0]}, index=idx)
+    lab = build_label(close)
+    # label[t]=close[t+2]/close[t+1]-1:t0 行=12/11-1
+    assert lab.iloc[0]["a"] == pytest.approx(12.0 / 11.0 - 1)
+    # 末两行无可成交区间=NaN
+    assert lab.iloc[-1].isna().all() and lab.iloc[-2].isna().all()
+
+
+def test_cs_rank_norm_nan_passthrough_and_centered():
+    df = pd.DataFrame({"a": [1.0, 2.0, 3.0, np.nan],
+                       "b": [4.0, 3.0, 2.0, 1.0]})
+    out = cs_rank_norm(df)
+    # 行2截面 [a=3.0, b=2.0]: a 为最大, rank pct 1.0 - 0.5 = 0.5
+    # (plan 原断言 0.75 与其自身注释"rank pct 1.0 - 0.5"矛盾, 按权威实现修正)
+    assert out.loc[2, "a"] == pytest.approx(0.5)
+    assert np.isnan(out.loc[3, "a"])                 # NaN 透传不占截面
+    row0 = out.loc[0, ["a", "b"]].tolist()
+    assert row0 == [pytest.approx(0.0), pytest.approx(0.5)]  # 截面最小/最大