feat(factor): challenger 三路信号——方向调整等权/ICIR 档案加权/LGBM walk-forward(样外隔离铁律) [nas]

2026-10-10 12:34:10 +08:00
parent 275c66baf5
commit cf451466a6
2 changed files with 135 additions and 1 deletions
@@ -30,3 +30,77 @@ def build_label(close_wide: pd.DataFrame) -> pd.DataFrame:
 def cs_rank_norm(df: pd.DataFrame) -> pd.DataFrame:
    """截面秩归一 (rank_pct-0.5);NaN 透传不占当日截面."""
    return df.rank(axis=1, pct=True) - 0.5
+
+
+LGBM_PARAMS = {  # qlib Alpha158 基准超参起步(spec §4.4;无独立 valid 段,v1 固定轮数)
+    "objective": "mse", "learning_rate": 0.1, "max_depth": 8, "num_leaves": 210,
+    "colsample_bytree": 0.8879, "subsample": 0.8789,
+    "lambda_l1": 205.6999, "lambda_l2": 580.9768,
+    "min_child_samples": 100, "feature_fraction_bynode": 0.8,
+    "seed": 42, "num_threads": 4, "verbose": -1,
+}
+NUM_BOOST_ROUND = 500
+
+
+def _aligned(values: dict[str, pd.DataFrame]) -> pd.DataFrame:
+    """多因子宽表纵向拼接成特征长表(index 对齐,缺失源列 NaN)."""
+    return pd.concat({n: cs_rank_norm(v) for n, v in values.items()}, axis=1)
+
+
+def _dir_sign(pool: dict, name: str) -> float:
+    return -1.0 if pool.get(name, {}).get("direction") == "-" else 1.0
+
+
+def equal_weight_signal(values: dict[str, pd.DataFrame], pool: dict) -> pd.DataFrame:
+    stack = pd.concat([_dir_sign(pool, n) * cs_rank_norm(v) for n, v in values.items()])
+    return stack.groupby(level=0).mean()
+
+
+def icir_signal(values: dict[str, pd.DataFrame], pool: dict) -> pd.DataFrame:
+    total = sum(pool[n].get("weight", 0.0) for n in values)
+    if total <= 0:
+        raise ValueError("ICIR 权重和为零,档案异常")
+    out = None
+    for n, v in values.items():
+        w = pool[n].get("weight", 0.0) * _dir_sign(pool, n)
+        part = w * cs_rank_norm(v)
+        out = part if out is None else out.add(part, fill_value=0.0)
+    return out / total
+
+
+def lgbm_walk_forward(values: dict[str, pd.DataFrame], label: pd.DataFrame,
+                      last_month: str, params: dict | None = None) -> tuple[pd.DataFrame, dict]:
+    """训练窗=last_month 之前全部;样外=last_month 当月(test 隔离铁律).
+
+    返回 (样外日×股票预测宽表, {feature: gain}).特征=CSRankNorm 后各源,
+    label=CSRankNorm 后防缺口收益;日频截面样本(日期,股票)平铺训练.
+    """
+    import lightgbm as lgb
+
+    feat = _aligned(values)
+    lab = cs_rank_norm(label)
+    common = feat.index.intersection(lab.index)
+    feat, lab = feat.loc[common], lab.loc[common]
+
+    oos_mask = feat.index.strftime("%Y-%m") == last_month
+    train_mask = ~oos_mask
+    X_tr = feat[train_mask].stack(future_stack=True).reset_index()
+    X_tr.columns = ["datetime", "vt_symbol", *feat.columns.levels[0]]
+    y_df = lab.stack(future_stack=True).rename("y").reset_index()
+    y_df.columns = ["datetime", "vt_symbol", "y"]  # index 无名时 reset 生成 level_0/1,显式定名
+    tr = X_tr.merge(y_df, on=["datetime", "vt_symbol"]).dropna()
+    X = tr[list(feat.columns.levels[0])]
+    model = lgb.train(params or LGBM_PARAMS, lgb.Dataset(X, label=tr["y"]),
+                      num_boost_round=NUM_BOOST_ROUND)
+    imp = dict(zip(X.columns, model.feature_importance("gain").tolist()))
+
+    X_oos = feat[oos_mask].stack(future_stack=True).reset_index()
+    X_oos.columns = X_tr.columns
+    X_oos = X_oos.dropna(subset=list(feat.columns.levels[0]))
+    X_oos = X_oos.merge(y_df.dropna()[["datetime", "vt_symbol"]],
+                        on=["datetime", "vt_symbol"])  # 只留 label 有效对(可评分样外)
+    if X_oos.empty:
+        return pd.DataFrame(), imp
+    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
@@ -5,7 +5,14 @@ import numpy as np
 import pandas as pd
 import pytest

-from sanguo_factor.challenger_lgbm import build_label, cs_rank_norm, load_pool
+from sanguo_factor.challenger_lgbm import (
+    build_label,
+    cs_rank_norm,
+    equal_weight_signal,
+    icir_signal,
+    load_pool,
+    lgbm_walk_forward,
+)


 def test_load_pool_twelve_sources():
@@ -36,3 +43,56 @@ def test_cs_rank_norm_nan_passthrough_and_centered():
    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)]  # 截面最小/最大
+
+
+def _toy_values(idx, cols):
+    rng = np.random.default_rng(7)
+    return {n: pd.DataFrame(rng.normal(size=(len(idx), len(cols))), index=idx, columns=cols)
+            for n in ("f_pos", "f_neg")}
+
+
+def test_equal_weight_direction_adjusted():
+    idx = pd.date_range("2025-06-01", periods=3, freq="D")
+    cols = ["a", "b"]
+    values = {"f_pos": pd.DataFrame(1.0, index=idx, columns=cols),
+              "f_neg": pd.DataFrame(1.0, index=idx, columns=cols)}
+    pool = {"f_pos": {"direction": "+"}, "f_neg": {"direction": "-"}}
+    sig = equal_weight_signal(values, pool)
+    # +1 与 -1 等权平均=0
+    assert (sig == 0.0).all().all()
+
+
+def test_icir_signal_uses_archive_weights():
+    idx = pd.date_range("2025-06-01", periods=2, freq="D")
+    cols = ["a"]
+    values = {"f_pos": pd.DataFrame(2.0, index=idx, columns=cols),
+              "f_neg": pd.DataFrame(2.0, index=idx, columns=cols)}
+    pool = {"f_pos": {"direction": "+", "weight": 0.75},
+            "f_neg": {"direction": "-", "weight": 0.25}}
+    sig = icir_signal(values, pool)
+    # 常值 2.0 单列截面的 CSRankNorm=0.5,期望=0.5*(+0.75-0.25)
+    # (plan 原期望 2.0*(0.75-0.25) 未算入秩归一,按权威实现修正;
+    #  DataFrame 与 pytest.approx 标量不兼容,用 np.allclose)
+    assert np.allclose(sig.values, 0.5 * (0.75 - 0.25))
+
+
+def test_lgbm_walk_forward_holdout_isolated():
+    """样外月绝不进训练(test 隔离铁律):给训练月与样外月截然不同的
+    因子-收益关系,样外预测应反映训练期学到的关系而非记忆样外."""
+    import pandas as pd
+    from sanguo_factor.challenger_lgbm import build_label, lgbm_walk_forward
+    idx = pd.date_range("2025-01-01", "2025-03-31", freq="B")
+    cols = [f"s{i}" for i in range(5)]
+    rng = np.random.default_rng(42)
+    values = {"s0": pd.DataFrame(rng.normal(size=(len(idx), 5)), index=idx, columns=cols)}
+    label = build_label(pd.DataFrame(100 + rng.normal(scale=0.5, size=(len(idx), 5)),
+                                     index=idx, columns=cols))
+    oos = idx[idx >= "2025-03-01"]
+    # toy 数据(~215 样本)喂生产超参(lambda_l1=205+min_child_samples=100)会零分裂
+    # →gain=0;走函数自带的 params 覆盖入口传轻量超参验证 split/gain 通路
+    light_params = {"objective": "mse", "learning_rate": 0.1, "num_leaves": 4,
+                    "min_child_samples": 5, "lambda_l1": 0.0, "lambda_l2": 1.0,
+                    "seed": 42, "num_threads": 4, "verbose": -1}
+    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