feat(factor): challenger 对拍评估——三路样外 IC/ICIR/分层多空+特征贡献+池冗余注记件 [nas]

sanguo_factor/challenger_lgbm.py

@@ -104,3 +104,102 @@ def lgbm_walk_forward(values: dict[str, pd.DataFrame], label: pd.DataFrame,
     preds = model.predict(X_oos[list(feat.columns.levels[0])])
     out = X_oos[["datetime", "vt_symbol"]].assign(p=preds)
     return out.pivot(index="datetime", columns="vt_symbol", values="p"), imp
+
+
+def _load_close_wide(vnpy_db: str, columns_ref: pd.Index, start: str, end: str) -> pd.DataFrame:
+    """按因子宽表列(股票)拉收盘价.生产实现走 universe.load_universe_bars 轻量列.
+
+    真实 loader 必传 start/end(universe.py:45);+45d 前向缓冲覆盖 label 的 t+2.
+    """
+    from sanguo_data.config import find_config_path, load_config
+    from .universe import load_universe_bars
+
+    cfg = load_config(find_config_path())
+    bars = load_universe_bars(vnpy_db or cfg.data_paths["vnpy_db"], start, end,
+                              symbols=list(columns_ref))
+    _ = bars.select(["datetime", "vt_symbol", "close"]).to_pandas()
+    wide = _.pivot(index="datetime", columns="vt_symbol", values="close").sort_index()
+    wide.index = pd.to_datetime(wide.index)
+    return wide
+
+
+def score_signal(signal: pd.DataFrame, label: pd.DataFrame) -> dict:
+    """样外评分:日 IC 均值/ICIR/五分位多空累计/有效天数.
+
+    days=进评分的天数(截面≥5 对);IC 全 NaN(常数信号)时 ic_mean/icir=None.
+    """
+    common = signal.index.intersection(label.index)
+    ics = []
+    ls_rets = []
+    for d in common:
+        s, y = signal.loc[d], label.loc[d]
+        pair = pd.concat([s, y], axis=1, keys=["s", "y"]).dropna()
+        if len(pair) < 5:
+            continue
+        ics.append(pair["s"].corr(pair["y"], method="spearman"))
+        q = pair["s"].quantile([0.2, 0.8])
+        lo, hi = pair[pair["s"] <= q[0.2]]["y"].mean(), pair[pair["s"] >= q[0.8]]["y"].mean()
+        ls_rets.append((hi - lo) if (lo is not None and hi is not None) else 0.0)
+    if not ics:
+        return {"ic_mean": None, "icir": None, "q5q1": None, "days": 0}
+    ser = pd.Series(ics).dropna()
+    icir = (ser.mean() / ser.std()) if len(ser) > 1 and ser.std() > 0 else None
+    return {"ic_mean": round(float(ser.mean()), 6) if len(ser) else None,
+            "icir": round(float(icir), 6) if icir is not None else None,
+            "q5q1": round(float(sum(ls_rets)), 6), "days": len(ics)}
+
+
+def run_challenge(values_dir: str, vnpy_db: str, as_of: str, out_dir: str,
+                  host: str = "nas") -> str:
+    """月度链 stage6 入口:读导出宽表→三路→对拍件(append-only 子目录)."""
+    pool = load_pool()
+    names = sorted(pool)
+    values = {n: pd.read_parquet(os.path.join(values_dir, f"{n}.parquet"))
+              for n in names if os.path.exists(os.path.join(values_dir, f"{n}.parquet"))}
+    if not values:
+        raise FileNotFoundError(f"因子宽表目录无池内因子: {values_dir}")
+    cols_ref = next(iter(values.values())).columns
+    start = min(v.index.min() for v in values.values()).strftime("%Y-%m-%d")
+    end = max(v.index.max() for v in values.values()).strftime("%Y-%m-%d")
+    close = _load_close_wide(vnpy_db, cols_ref, start, end)
+    label = build_label(close)
+    last_month = as_of[:7]
+
+    signals = {"equal": equal_weight_signal(values, pool),
+               "icir": icir_signal(values, pool)}
+    pred, imp = lgbm_walk_forward(values, label, last_month)
+    if not pred.empty:
+        signals["lgbm"] = pred
+
+    oos_label = label[label.index.strftime("%Y-%m") == last_month]
+    scored = {k: score_signal(v, oos_label) for k, v in signals.items()}
+
+    # 池冗余注记:|corr|>0.7 只记录不剔除(首年观察期,剔除留人).
+    # 按日构造 股票×因子 矩阵→corr=因子×因子(源间相关,plan 原稿 groupby
+    # level=0 按因子分组会算成股票对,与下游 n1<n2 因子对语义矛盾,已重构)
+    by_date = pd.concat({n: cs_rank_norm(v).stack(future_stack=True)
+                         for n, v in values.items()}, axis=1)
+    daily_corr = by_date.groupby(level=0).apply(
+        lambda g: g.corr(method="spearman") if g.shape[0] > 1 else None)
+    flagged = []
+    if daily_corr is not None and not daily_corr.empty:
+        means = daily_corr.groupby(level=1).mean()
+        for n1 in means.index:
+            for n2 in means.columns:
+                if n1 < n2 and abs(means.loc[n1, n2]) > 0.7:
+                    flagged.append({"a": n1, "b": n2,
+                                    "corr": round(float(means.loc[n1, n2]), 4)})
+
+    doc = {"as_of": as_of, "generated_at": pd.Timestamp.now().isoformat(),
+           "host": host, "pool": {"names": sorted(values), "size": len(values)},
+           "split": {"oos_month": last_month,
+                     "note": "训练=样外月前全部(test 隔离铁律);无独立 valid,固定轮数"},
+           "signals": scored, "feature_importance": imp,
+           "redundancy_flagged": flagged,
+           "lgbm_params": {**LGBM_PARAMS, "num_boost_round": NUM_BOOST_ROUND}}
+    sub = os.path.join(out_dir, "challenger_lgbm")
+    os.makedirs(sub, exist_ok=True)
+    path = os.path.join(sub, f"{host}_{as_of}.json")
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(doc, f, ensure_ascii=False, indent=2)
+    return path

tests/factor/test_challenger_lgbm.py

@@ -1,5 +1,6 @@
 """LGBM challenger 三路对拍(spec §4.4 2026-10-10):数据层."""
 import json
+from pathlib import Path
 
 import numpy as np
 import pandas as pd
@@ -12,6 +13,8 @@ from sanguo_factor.challenger_lgbm import (
     icir_signal,
     load_pool,
     lgbm_walk_forward,
+    run_challenge,
+    score_signal,
 )
 
 
@@ -96,3 +99,49 @@ def test_lgbm_walk_forward_holdout_isolated():
     pred, imp = lgbm_walk_forward(values, label, last_month="2025-03", params=light_params)
     assert list(pred.index) == [d for d in oos if d in label.index and label.loc[d].notna().any()]
     assert set(imp.keys()) == {"s0"} and imp["s0"] > 0
+
+
+def test_score_signal_perfect_and_flat():
+    # 5 列:实现有最小截面阈值(len(pair)<5 skip),plan 原稿 3 列进不了评分
+    idx = pd.date_range("2025-03-03", periods=10, freq="B")
+    cols = ["a", "b", "c", "d", "e"]
+    sig = pd.DataFrame(np.linspace(-1, 1, 50).reshape(10, 5), index=idx, columns=cols)
+    lab = sig * 1.0          # 完美信号
+    s = score_signal(sig, lab)
+    assert s["ic_mean"] == pytest.approx(1.0, abs=1e-6)
+    assert s["q5q1"] > 0
+    flat = pd.DataFrame(0.0, index=idx, columns=cols)
+    s0 = score_signal(flat, lab)
+    assert s0["days"] == 10  # 进评分的天数(IC=NaN 但天在)
+
+
+def test_run_challenge_end_to_end(tmp_path, monkeypatch):
+    """端到端:合成 2 因子×40 日数据落宽表 parquet+合成 close db 太重——
+    本用例走 values_dir 真文件+vnpy_db 用 monkeypatch 替换拉取函数."""
+    from sanguo_factor import challenger_lgbm as cl
+    idx = pd.date_range("2025-01-01", "2025-03-31", freq="B")
+    cols = ["a", "b"]
+    rng = np.random.default_rng(3)
+    values = {n: pd.DataFrame(rng.normal(size=(len(idx), 2)), index=idx, columns=cols)
+              for n in ("s0", "s1")}
+    vdir = tmp_path / "factor_values"
+    vdir.mkdir()
+    for n, v in values.items():
+        v.to_parquet(vdir / f"{n}.parquet")
+    close = pd.DataFrame(100 + np.cumsum(rng.normal(scale=0.4, size=(len(idx), 2)), axis=0),
+                         index=idx, columns=cols)
+    # s0/s1 不在 quant12 档案内,plan 原稿缺此 patch 必 FileNotFoundError
+    monkeypatch.setattr(cl, "load_pool", lambda: {
+        "s0": {"direction": "+", "weight": 0.6},
+        "s1": {"direction": "-", "weight": 0.4}})
+    # 真实 loader(universe.load_universe_bars)必传 start/end,签名 4 参
+    monkeypatch.setattr(cl, "_load_close_wide", lambda db, c, s, e: close)
+    out = cl.run_challenge(str(vdir), "fake.db", "2025-03-31", str(tmp_path))
+    doc = json.loads(Path(out).read_text())
+    assert doc["as_of"] == "2025-03-31"
+    assert set(doc["signals"]) == {"equal", "icir", "lgbm"}
+    for k, v in doc["signals"].items():
+        assert {"ic_mean", "icir", "q5q1", "days"} <= set(v)
+    assert doc["pool"]["names"] == ["s0", "s1"]
+    assert len(doc["feature_importance"]) == 2
+    assert doc["split"]["oos_month"] == "2025-03"