fix(factor): codex 复审班次二修批——LGBM 训练切分 embargo 2 交易日+11M 显式裁剪/三路统一可评样本 mask/缺源与坏件整月 skip/close loader purge 直读+真实路径单测/cols union/冗余按日循环/LOW 四项(docstring 语义+allow_nan=False+wrapper chal 摘要) [nas]

docs/superpowers/specs/2026-09-22-research-to-trading-pipeline-design.md

@@ -100,7 +100,7 @@ F 环是「管道」与「飞轮」的分界：归因发现→可能=衰减触
   - **池贪心准入（自建；QuantaAlpha 论文有代码无——实验脚本未开源）**：近窗 RankIC 贪心准入+|corr|<0.7 去冗余（高相关对弃低 RankIC）+容量上限（起步=晋级因子全量不设限；饱和点意识挂账——池到几十个再议上限，论文饱和点≈350 因子参考）；
   - 判读：等权≈ICIR=因子相关性高、加权信息少，安心用简单；LGBM 明显跑赢两者=组合方式升级叙事，且 feature importance 天然产出「组合内各因子贡献」（对接 §4.7 归因）；预期两三个完整月度周期才有判读力；
   - 纪律：challenger 永远影子不进生产；若升级仍走决议 K 考场晋级+用户口令，与实盘升级同构。
-  - **班次二落地（2026-10-10）**：`sanguo_factor/challenger_lgbm.py`——v1 池=quant12 在役 12 源（扩池挂后续）；三路=方向调整等权/ICIR 档案加权/LGBM（qlib 基准超参，固定 500 轮——无独立 valid 段不早停，注记在对拍件）；walk-forward=前 11M 训练+末 1M 样外（test 隔离铁律：样外月绝不进训练）；对拍件=`challenger_lgbm/{host}_{as_of}.json`（三路样外 IC/ICIR/Q5-Q1 多空+feature importance+池冗余注记——`|corr|>0.7` 只记录不剔除，首年观察期剔除留人）；挂月度链 stage6（失败不阻链）；判读需累积≥2-3 个样外月。**宽表源现场核实修正**：月度批 `values/` 导出只含注册表因子（fa 族+composite），永不含 quant12 12 源——stage6 `--values-dir` 改指 stage4 截面盘 `/volume1/stock/factor_cross_section`（daily_section 幂等 merge 累积的 12 源主 parquet，12/12 在位实证），零重算同链上游供给。
+  - **班次二落地（2026-10-10）**：`sanguo_factor/challenger_lgbm.py`——v1 池=quant12 在役 12 源（扩池挂后续）；三路=方向调整等权/ICIR 档案加权/LGBM（qlib 基准超参，固定 500 轮——无独立 valid 段不早停，注记在对拍件）；walk-forward=前 11M 训练+末 1M 样外（test 隔离铁律：样外月绝不进训练；codex 复审加固=训练窗末 2 交易日 embargo——build_label 双 shift 前视，不 embargo 则窗口末样本的 label 吃样外月收盘=信息层泄漏；三路统一可评样本 mask、缺源/坏件整月 skip 不产半吊子件）；对拍件=`challenger_lgbm/{host}_{as_of}.json`（三路样外 IC/ICIR/Q5-Q1 多空+feature importance+池冗余注记——`|corr|>0.7` 只记录不剔除，首年观察期剔除留人）；挂月度链 stage6（失败不阻链）；判读需累积≥2-3 个样外月。**宽表源现场核实修正**：月度批 `values/` 导出只含注册表因子（fa 族+composite），永不含 quant12 12 源——stage6 `--values-dir` 改指 stage4 截面盘 `/volume1/stock/factor_cross_section`（daily_section 幂等 merge 累积的 12 源主 parquet，12/12 在位实证），零重算同链上游供给。
 - **test 隔离条款（2026-10-10 成文，QuantaAlpha 附录 E 纪律借鉴）**：**合成层（含 challenger）权重/训练只用截至 as_of 的数据，表现评估只看 as_of 之后——权重窗与成绩窗永不重叠**。现状核实动作=审计 quant12_v2a 权重计算代码确认滚动结构天然合规（每月 T-12~T 算权、T+1 后出成绩），核实结论落本档。防什么=将来实盘与回测打架时，回测成绩有「未被自己污染」的底气；防将来改动（全历史拟合/未来数据调参类）悄悄破坏。
   核实结论（2026-10-10 审计）：quant12_v2a（quant12_icirfit_v1）fit_window=2018-01~2021-06、h2 考场窗=2021-07-01 起（exam_gate DEFAULT_START，同批同 run 对照）——**无重叠**（考场首日恰后于拟合窗末月，首尾相接）；v2c 滚动天然合规：Y 年制度档案 fit [Y-3 年初, Y-1 年末]、考核 Y 年段（walk-forward），且注册路径只挂载档案冻结权重（load_profile→resolve_weights），评估数据不回流拟合。
 - **版本钉住+影子对照闸门（决议 A）**：策略钉住合成层具体版本；合成层变更（晋级/退役/调权）→新旧版影子对照（起步默认 4 周，首年校准）→用户口令→各策略**择期**升级。可回滚、历史可复现，与推 vps 纪律同构。

sanguo_factor/challenger_lgbm.py

@@ -4,6 +4,7 @@
 三路=等权(方向调整)/ICIR(quant12_icirfit_v1 档案)/LightGBM(月度重训),
 同池同窗 walk-forward 影子对拍;challenger 永不进生产,对拍件落
 reports/factor_monthly/challenger_lgbm/ 子目录(判定层端点只读根层).
+对拍件同 host+as_of 幂等覆盖(月度重跑覆写同 key).
 """
 from __future__ import annotations
 
@@ -11,6 +12,7 @@ import json
 import os
 from pathlib import Path
 
+import numpy as np
 import pandas as pd
 
 _PROFILE = Path(__file__).parent / "weight_profiles" / "quant12_icirfit_v1.json"
@@ -68,9 +70,30 @@ def icir_signal(values: dict[str, pd.DataFrame], pool: dict) -> pd.DataFrame:
     return out / total
 
 
+def _split_masks(index: pd.DatetimeIndex, last_month: str,
+                 embargo_days: int = 2, train_months: int = 11) -> tuple[np.ndarray, np.ndarray]:
+    """walk-forward 切分 mask(codex 复审 H-1/H→M-7).
+
+    oos=last_month 当月;train=[last_month 前推 train_months 个自然月之首,
+    oos 首日前 embargo_days 个交易日)。embargo 保证:训练样本 t 的 label
+    (close[t+2]/close[t+1]) 取值严格早于样外月首日——label 信息层不泄漏
+    (build_label 双 shift 前视,窗口末 2 交易的 label 会吃到样外月收盘).
+    训练下界显式裁剪对齐 spec 12M 语义,防截面盘逐年累积导致训练窗漂移.
+    """
+    oos = np.asarray(index.strftime("%Y-%m")) == last_month
+    train = np.zeros(len(index), dtype=bool)
+    if not oos.any():
+        return train, oos
+    first_oos = int(np.argmax(oos))
+    train[:max(first_oos - embargo_days, 0)] = True
+    floor = (pd.Period(last_month, freq="M") - train_months).to_timestamp()
+    train &= np.asarray(index >= floor)
+    return train, oos
+
+
 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 隔离铁律).
+    """训练窗=样外月前推 11 个自然月,末 2 交易日 embargo;样外=last_month 当月.
 
     返回 (样外日×股票预测宽表, {feature: gain}).特征=CSRankNorm 后各源,
     label=CSRankNorm 后防缺口收益;日频截面样本(日期,股票)平铺训练.
@@ -82,13 +105,16 @@ def lgbm_walk_forward(values: dict[str, pd.DataFrame], label: pd.DataFrame,
     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
+    train_mask, oos_mask = _split_masks(feat.index, last_month)
+    if not oos_mask.any():
+        return pd.DataFrame(), {}
     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()
+    if tr.empty:
+        return pd.DataFrame(), {}
     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)
@@ -107,16 +133,21 @@ def lgbm_walk_forward(values: dict[str, pd.DataFrame], label: pd.DataFrame,
 
 
 def _load_close_wide(vnpy_db: str, columns_ref: pd.Index, start: str, end: str) -> pd.DataFrame:
-    """按因子宽表列(股票)拉收盘价.生产实现走 universe.load_universe_bars 轻量列.
+    """按因子宽表列(股票)拉收盘价(universe.load_universe_bars 轻量列).
 
-    真实 loader 必传 start/end(universe.py:45);+45d 前向缓冲覆盖 label 的 t+2.
+    真实 loader 必传 start/end(universe.py);+45d 前向缓冲覆盖 label 的 t+2。
+    直读前 purge 同窗缓存(daily_section 同款纪律:防旧缓存把新到 bar 截在
+    缓存生成日)——challenger 月度重跑不复用旧缓存件,选型=每次全量重读,
+    月频成本可接受。
     """
-    from sanguo_data.config import find_config_path, load_config
-    from .universe import load_universe_bars
+    from .universe import load_universe_bars, purge_cache
 
-    cfg = load_config(find_config_path())
-    bars = load_universe_bars(vnpy_db or cfg.data_paths["vnpy_db"], start, end,
-                              symbols=list(columns_ref))
+    db = vnpy_db
+    if not db:
+        from sanguo_data.config import find_config_path, load_config
+        db = load_config(find_config_path()).data_paths["vnpy_db"]
+    purge_cache(db, start, end)
+    bars = load_universe_bars(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)
@@ -126,7 +157,9 @@ def _load_close_wide(vnpy_db: str, columns_ref: pd.Index, start: str, end: str)
 def score_signal(signal: pd.DataFrame, label: pd.DataFrame) -> dict:
     """样外评分:日 IC 均值/ICIR/五分位多空累计/有效天数.
 
-    days=进评分的天数(截面≥5 对);IC 全 NaN(常数信号)时 ic_mean/icir=None.
+    days=进入评分的天数(截面≥5 对)——非有效 IC 天数(常数信号 IC=NaN 也计入).
+    q5q1=20%/80% 分位阈值的近似多空累计(tie 语义:quantile 阈值含等值边界,
+    小截面下是近似而非严格五分位分组);IC 全 NaN 时 ic_mean/icir=None.
     """
     common = signal.index.intersection(label.index)
     ics = []
@@ -149,51 +182,109 @@ def score_signal(signal: pd.DataFrame, label: pd.DataFrame) -> dict:
             "q5q1": round(float(sum(ls_rets)), 6), "days": len(ics)}
 
 
+def _read_value_parquet(path: str) -> pd.DataFrame | None:
+    """读因子宽表 parquet;坏件/空件/索引不可转 datetime → None(codex 复审 M-5).
+
+    结构化剔除走缺源 skip 语义,不裸抛 AttributeError/ArrowInvalid.
+    """
+    try:
+        df = pd.read_parquet(path)
+        if df.empty or df.columns.empty:
+            return None
+        idx = pd.to_datetime(df.index)
+    except Exception as exc:
+        print(f"[challenger] ⚠️ parquet 不可读: {path} ({exc!r})")
+        return None
+    return df.set_axis(idx)
+
+
+def _redundancy_flags(values: dict[str, pd.DataFrame], threshold: float = 0.7) -> list[dict]:
+    """池冗余注记:按日源间 corr 均值 |corr|>threshold 只记录不剔除(首年观察期).
+
+    按日循环构造 股票×源 帧后 corr=源×源(codex 复审 M-8:替代全历史 stack 的
+    MultiIndex 大表——12 源×12M×全A 在 NAS 2G 内存机不可行);逐对 NaN 感知
+    累积(缺数据日只计入有值对).
+    """
+    names = list(values)
+    normed = {n: cs_rank_norm(v) for n, v in values.items()}
+    all_dates = sorted(set().union(*(set(v.index) for v in normed.values())))
+    sum_c = None
+    cnt_c = None
+    for d in all_dates:
+        day = pd.DataFrame({n: normed[n].loc[d] for n in names if d in normed[n].index})
+        if day.shape[0] < 2 or day.shape[1] < 2:
+            continue
+        c = day.corr(method="spearman")
+        sum_c = c if sum_c is None else sum_c.add(c, fill_value=0.0)
+        cnt = c.notna().astype(float)
+        cnt_c = cnt if cnt_c is None else cnt_c.add(cnt, fill_value=0.0)
+    flagged = []
+    if sum_c is None:
+        return flagged
+    means = sum_c / cnt_c.replace(0.0, np.nan)
+    for n1 in names:
+        for n2 in names:
+            if n1 < n2 and n1 in means.index and n2 in means.columns:
+                v = means.loc[n1, n2]
+                if pd.notna(v) and abs(v) > threshold:
+                    flagged.append({"a": n1, "b": n2, "corr": round(float(v), 4)})
+    return flagged
+
+
 def run_challenge(values_dir: str, vnpy_db: str, as_of: str, out_dir: str,
                   host: str = "nas") -> str:
-    """月度链 stage6 入口:读导出宽表→三路→对拍件(append-only 子目录)."""
+    """月度链 stage6 入口:读导出宽表→三路→对拍件(同 host+as_of 幂等覆盖).
+
+    池完整性铁律(codex 复审 H-3):任何源缺件/坏件 → FileNotFoundError 走
+    main 的 skip 路径,不产半吊子对拍件.
+    """
     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")
+    loaded: dict[str, pd.DataFrame] = {}
+    for n in names:
+        p = os.path.join(values_dir, f"{n}.parquet")
+        if not os.path.exists(p):
+            continue
+        df = _read_value_parquet(p)
+        if df is not None:
+            loaded[n] = df
+    missing = [n for n in names if n not in loaded]
+    if missing:
+        raise FileNotFoundError(
+            f"missing sources: expected={len(pool)} loaded={len(loaded)} "
+            f"missing={missing} (values_dir={values_dir})")
+
+    cols_ref = pd.Index(sorted(set().union(*(set(v.columns) for v in loaded.values()))))
+    start = min(v.index.min() for v in loaded.values()).strftime("%Y-%m-%d")
+    end = max(v.index.max() for v in loaded.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)
+    signals = {"equal": equal_weight_signal(loaded, pool),
+               "icir": icir_signal(loaded, pool)}
+    pred, imp = lgbm_walk_forward(loaded, label, last_month)
     if not pred.empty:
         signals["lgbm"] = pred
 
+    # 统一可评样本 mask(codex 复审 H-2):OOS 月内全部源非空且 label 非空的
+    # date×stock 集合=唯一评分集,三路全部 reindex 到该集合再 score
     oos_label = label[label.index.strftime("%Y-%m") == last_month]
-    scored = {k: score_signal(v, oos_label) for k, v in signals.items()}
+    mask = pd.DataFrame(True, index=oos_label.index, columns=cols_ref)
+    for v in loaded.values():
+        mask &= v.reindex(index=oos_label.index, columns=cols_ref).notna()
+    mask &= oos_label.reindex(index=oos_label.index, columns=cols_ref).notna()
+    scored = {k: score_signal(sig.reindex(index=oos_label.index, columns=cols_ref).where(mask),
+                              oos_label)
+              for k, sig 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)})
+    flagged = _redundancy_flags(loaded)
 
     doc = {"as_of": as_of, "generated_at": pd.Timestamp.now().isoformat(),
-           "host": host, "pool": {"names": sorted(values), "size": len(values)},
+           "host": host, "pool": {"names": sorted(loaded), "size": len(loaded)},
            "split": {"oos_month": last_month,
-                     "note": "训练=样外月前全部(test 隔离铁律);无独立 valid,固定轮数"},
+                     "note": "训练=样外月前推 11 个自然月(末 2 交易日 embargo,"
+                             "label 信息层隔离);无独立 valid,固定轮数"},
            "signals": scored, "feature_importance": imp,
            "redundancy_flagged": flagged,
            "lgbm_params": {**LGBM_PARAMS, "num_boost_round": NUM_BOOST_ROUND}}
@@ -201,7 +292,7 @@ def run_challenge(values_dir: str, vnpy_db: str, as_of: str, out_dir: str,
     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)
+        json.dump(doc, f, ensure_ascii=False, indent=2, allow_nan=False)
     return path
 
 

scripts/nas_sync/run_factor_monthly_standalone.sh

@@ -180,6 +180,6 @@ PY
   [ "$rc_review" -eq 1 ] && final=1
   [ "$rc_gap" -eq 1 ] && final=1
   [ "$rc_attr" -ne 0 ] && final=1
-  echo "=== $(date '+%F %T') factor-monthly done final_rc=$final (batch=$rc_batch review=$rc_review gap=$rc_gap attr=$rc_attr replay=$rc_replay) ==="
+  echo "=== $(date '+%F %T') factor-monthly done final_rc=$final (batch=$rc_batch review=$rc_review gap=$rc_gap attr=$rc_attr replay=$rc_replay chal=$rc_chal) ==="
   exit "$final"
 } >> "$LOG" 2>&1

tests/factor/test_challenger_lgbm.py

@@ -164,3 +164,175 @@ def test_cli_smoke(tmp_path, monkeypatch, capsys):
                   "--out-dir", str(tmp_path), "--vnpy-db", "fake.db"])
     assert rc == 0
     assert (tmp_path / "challenger_lgbm" / "nas_2025-03-31.json").exists()
+
+
+# ———— codex 复审班次二修批(2026-10-10):H1/H7 embargo+训练窗裁拆 ————
+
+def test_split_masks_embargo_and_floor():
+    """H-1/H→M-7:训练窗=[样外月前推 11 个自然月首,样外首日前 2 交易日).
+
+    embargo 2 交易的数学保证:训练样本 t 的 label 取 close[t+1],close[t+2]
+    (build_label shift(-2)/(--1)),故 max(train_pos)+2 < first_oos_pos
+    ⇔ 样外月收盘价不参与任何训练样本的 label 计算.
+    """
+    from sanguo_factor.challenger_lgbm import _split_masks
+    idx = pd.date_range("2024-01-01", "2025-03-31", freq="B")
+    train_mask, oos_mask = _split_masks(idx, "2025-03")
+    oos_pos = np.flatnonzero(oos_mask)
+    tr_pos = np.flatnonzero(train_mask)
+    assert oos_pos[0] == int(np.argmax(oos_mask))
+    # 样外月收盘绝不进训练 label:max 训练位 +2 严格早于样外首日
+    assert tr_pos.max() + 2 < oos_pos[0]
+    assert tr_pos.max() == oos_pos[0] - 3  # embargo=2 恰好剔够
+    assert not train_mask[oos_pos].any()   # 训练/样外不重叠
+    # 训练下界=2025-03 前推 11 个自然月之首(2024-04-01)
+    assert idx[tr_pos].min() >= pd.Timestamp("2024-04-01")
+    # floor 之上只剔样外月+embargo 两天
+    above = np.flatnonzero(np.asarray(idx >= pd.Timestamp("2024-04-01")))
+    assert len(above) == len(tr_pos) + len(oos_pos) + 2
+    # 无样外月:训练也空(防全量误训)
+    t0, o0 = _split_masks(idx, "2026-01")
+    assert not o0.any() and not t0.any()
+
+
+# ———— H-2:三路统一可评样本 mask ————
+
+def test_unified_scoring_mask_same_sample_set(tmp_path, monkeypatch):
+    """某源缺 2 股:未统一时 equal/icir 按 6 股评分、lgbm 只剩 4 股(<5
+    跳过)→days 分叉;统一 mask 后三路同一 date×stock 集合→days 一致."""
+    from sanguo_factor import challenger_lgbm as cl
+    idx = pd.date_range("2025-01-01", "2025-03-31", freq="B")
+    cols = ["a", "b", "c", "d", "e", "f"]
+    rng = np.random.default_rng(11)
+    s0 = pd.DataFrame(rng.normal(size=(len(idx), 6)), index=idx, columns=cols)
+    s0[["b", "c"]] = np.nan                      # 源 s0 缺 b/c 两股
+    s1 = pd.DataFrame(rng.normal(size=(len(idx), 6)), index=idx, columns=cols)
+    vdir = tmp_path / "vals"; vdir.mkdir()
+    s0.to_parquet(vdir / "s0.parquet")
+    s1.to_parquet(vdir / "s1.parquet")
+    close = pd.DataFrame(100 + np.cumsum(rng.normal(scale=0.4, size=(len(idx), 6)), axis=0),
+                         index=idx, columns=cols)
+    monkeypatch.setattr(cl, "load_pool", lambda: {
+        "s0": {"direction": "+", "weight": 0.6},
+        "s1": {"direction": "-", "weight": 0.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())
+    days = {k: v["days"] for k, v in doc["signals"].items()}
+    assert len(set(days.values())) == 1          # 三路同一评分集
+    assert set(days.values()) == {0}             # 统一后每日常截面=4 股<5,全跳
+
+
+# ———— H-3/M-5:缺源与坏件走 skip 语义 ————
+
+def test_missing_sources_skip_no_artifact(tmp_path, monkeypatch, capsys):
+    from sanguo_factor import challenger_lgbm as cl
+    idx = pd.date_range("2025-01-01", "2025-03-31", freq="B")
+    vdir = tmp_path / "vals"; vdir.mkdir()
+    pd.DataFrame(np.zeros((len(idx), 2)), index=idx,
+                 columns=["a", "b"]).to_parquet(vdir / "s0.parquet")
+    monkeypatch.setattr(cl, "load_pool", lambda: {
+        "s0": {"direction": "+", "weight": 0.5},
+        "s1": {"direction": "+", "weight": 0.5}})
+    rc = cl.main(["--values-dir", str(vdir), "--as-of", "2025-03-31",
+                  "--out-dir", str(tmp_path), "--vnpy-db", "fake.db"])
+    assert rc == 0
+    assert not (tmp_path / "challenger_lgbm").exists()   # 不产半吊子件
+    out = capsys.readouterr().out
+    assert "missing sources" in out and "expected=2" in out and "loaded=1" in out
+    assert "s1" in out
+
+
+def test_bad_parquet_structured_skip(tmp_path, monkeypatch, capsys):
+    """坏件(非 parquet 字节)/空件(0 行)结构化剔除不裸抛,汇入缺源 skip."""
+    from sanguo_factor import challenger_lgbm as cl
+    idx = pd.date_range("2025-01-01", "2025-03-31", freq="B")
+    vdir = tmp_path / "vals"; vdir.mkdir()
+    pd.DataFrame(np.zeros((len(idx), 2)), index=idx,
+                 columns=["a", "b"]).to_parquet(vdir / "s0.parquet")
+    (vdir / "s1.parquet").write_bytes(b"definitely not a parquet")
+    pd.DataFrame().to_parquet(vdir / "s2.parquet")       # 0 行空件
+    monkeypatch.setattr(cl, "load_pool", lambda: {
+        n: {"direction": "+", "weight": 1.0 / 3} for n in ("s0", "s1", "s2")})
+    rc = cl.main(["--values-dir", str(vdir), "--as-of", "2025-03-31",
+                  "--out-dir", str(tmp_path), "--vnpy-db", "fake.db"])
+    assert rc == 0
+    assert not (tmp_path / "challenger_lgbm").exists()
+    out = capsys.readouterr().out
+    assert "expected=3" in out and "loaded=1" in out
+    assert "s1" in out and "s2" in out
+
+
+# ———— H-4:_load_close_wide 真实路径(monkeypatch universe.load_universe_bars) ————
+
+def test_load_close_wide_real_path(monkeypatch):
+    import polars as pl
+    import sanguo_factor.universe as uni
+    from sanguo_factor import challenger_lgbm as cl
+    calls = {}
+
+    def fake_loader(db, start, end, symbols=None, limit=None):
+        calls.update(db=db, start=start, end=end, symbols=list(symbols or []))
+        return pl.DataFrame({
+            "vt_symbol": ["a.SSE", "a.SSE", "b.SSE", "b.SSE"],
+            "datetime": [pd.Timestamp("2025-01-02"), pd.Timestamp("2025-01-03")] * 2,
+            "close": [10.0, 11.0, 20.0, 21.0],
+        })
+
+    monkeypatch.setattr(uni, "load_universe_bars", fake_loader)
+    wide = cl._load_close_wide("fake.db", pd.Index(["a.SSE", "b.SSE"]),
+                               "2025-01-01", "2025-01-31")
+    assert calls["db"] == "fake.db"
+    assert calls["start"] == "2025-01-01" and calls["end"] == "2025-01-31"
+    assert calls["symbols"] == ["a.SSE", "b.SSE"]
+    assert isinstance(wide.index, pd.DatetimeIndex)
+    assert list(wide.columns) == ["a.SSE", "b.SSE"]
+    assert wide.loc[pd.Timestamp("2025-01-03"), "a.SSE"] == pytest.approx(11.0)
+    # 反向:loader 空帧 → 空宽表不炸
+    monkeypatch.setattr(uni, "load_universe_bars", lambda *a, **k: pl.DataFrame(
+        schema={"vt_symbol": pl.Utf8, "datetime": pl.Datetime, "close": pl.Float64}))
+    wide0 = cl._load_close_wide("fake.db", pd.Index([]), "2025-01-01", "2025-01-31")
+    assert wide0.empty
+
+
+# ———— M-6:cols_ref=池内 union ————
+
+def test_cols_ref_is_union_of_pool_columns(tmp_path, monkeypatch):
+    from sanguo_factor import challenger_lgbm as cl
+    idx = pd.date_range("2025-01-01", "2025-03-31", freq="B")
+    rng = np.random.default_rng(5)
+    s0 = pd.DataFrame(rng.normal(size=(len(idx), 2)), index=idx, columns=["a", "b"])
+    s1 = pd.DataFrame(rng.normal(size=(len(idx), 2)), index=idx, columns=["b", "c"])
+    vdir = tmp_path / "vals"; vdir.mkdir()
+    s0.to_parquet(vdir / "s0.parquet")
+    s1.to_parquet(vdir / "s1.parquet")
+    captured = {}
+
+    def fake_close(db, cols_ref, start, end):
+        captured["cols"] = list(cols_ref)
+        return pd.DataFrame(100 + rng.normal(scale=0.4, size=(len(idx), len(cols_ref))),
+                            index=idx, columns=list(cols_ref))
+
+    monkeypatch.setattr(cl, "load_pool", lambda: {
+        "s0": {"direction": "+", "weight": 0.5},
+        "s1": {"direction": "-", "weight": 0.5}})
+    monkeypatch.setattr(cl, "_load_close_wide", fake_close)
+    cl.run_challenge(str(vdir), "fake.db", "2025-03-31", str(tmp_path))
+    assert captured["cols"] == ["a", "b", "c"]
+
+
+# ———— M-8:池冗余按日循环 ————
+
+def test_redundancy_flags_per_day_unit():
+    from sanguo_factor.challenger_lgbm import _redundancy_flags
+    idx = pd.date_range("2025-01-01", periods=60, freq="B")
+    cols = [f"s{i}" for i in range(20)]
+    rng = np.random.default_rng(9)
+    x = pd.DataFrame(rng.normal(size=(60, 20)), index=idx, columns=cols)
+    z = pd.DataFrame(rng.normal(size=(60, 20)), index=idx, columns=cols)
+    flagged = _redundancy_flags({"x": x, "y": x.copy(), "z": z})
+    pairs = {(f["a"], f["b"]) for f in flagged}
+    assert ("x", "y") in pairs                     # 完全同源必标
+    assert ("x", "z") not in pairs and ("y", "z") not in pairs  # 独立源不标
+    xy = next(f for f in flagged if (f["a"], f["b"]) == ("x", "y"))
+    assert xy["corr"] == pytest.approx(1.0, abs=1e-6)