diff --git a/sanguo_factor/challenger_lgbm.py b/sanguo_factor/challenger_lgbm.py new file mode 100644 index 00000000..f6a78285 --- /dev/null +++ b/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 diff --git a/tests/factor/test_challenger_lgbm.py b/tests/factor/test_challenger_lgbm.py new file mode 100644 index 00000000..de9ed21b --- /dev/null +++ b/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)] # 截面最小/最大