docs(plans): 班次一/二实施计划归档——判定弹药+AST防换皮+test隔离 / LGBM challenger三路对拍 [nas]

docs/superpowers/plans/2026-10-10-ammo-astp-match-isolation.md

@@ -0,0 +1,623 @@
+# 判定弹药三件+AST 防换皮+test 隔离核实 Implementation Plan（班次一）
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 给月度批评判定人补三件弹药（全期合成 t/正 IC 占比/逐年 IC）+注册时防换皮提示+quant12_v2a 权重窗隔离审计核实。
+
+**Architecture:** 聚合源头在 `monthly_review.build_report`（monthly_points 已有逐点 ic_mean/t，只差聚合口径）→ 报告 JSON 新增 `factor_stats` 块 → detail 端点透传 → 前端展示。防换皮=新模块 `expression_match`（Python ast 方言内实现交换律归一化+最大公共子树占比，QuantaAlpha 算法思想不移植解析器）→ decompose 注册路径调用存 `entry.similarity`（提示非硬拒）。审计=读 weight_profiles 档案与考场窗核实无重叠，结论落 spec。
+
+**Tech Stack:** Python 3.11 stdlib（ast/statistics 零新依赖）、Vue3+Element Plus（既有栈）、pytest/vitest。
+
+## Global Constraints
+
+- spec=2026-09-22 流水线总设计 §4.2「判定弹药三件」「AST 防换皮」+§4.4「test 隔离条款」（2026-10-10 补丁块，本次已更新）
+- **push 纪律**：今晚 21:30 双机日度首班——改动触及 monthly_review（daily 两段链 stage2）。**全部 task 本地 commit 完成，push 等明早首班验证通过后**（用户已知情）。
+- commit 标签：触及 factors 端点+前端 → `[vps]`（VPS 常驻 api+前端也触发）；monthly_review 批链 → `[nas]`。每条 commit 末尾二选一或并存，CI enforce-label 必过。
+- 测试铁律：凡 decompose/registry 测试注册因子必须 `monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(tmp))`；前端 `vue-tsc` 管道退出码显式门控。
+- 口径注记（plan 内定为规范）：monthly point 的 `ic_mean`=该批 12M 窗内日均 IC 均值；跨月点等权均值≈全史等权近似（相邻窗重叠 11/12），月度链根层每月一点（daily/ 子目录不进 monthly_points，判定层月末快照语义不变）。
+- 阈值常量：相似度提示阈值 `SIM_FLOOR = 0.6`（起步默认，首年校准点）。
+
+---
+
+### Task 1: monthly_review 弹药聚合（factor_stats）
+
+**Files:**
+- Modify: `sanguo_factor/monthly_review.py`（build_report 内新函数+JSON 字段+render_markdown 增节）
+- Test: `tests/factor/test_monthly_review.py`（新建，目录已有 conftest.py）
+
+**Interfaces:**
+- Consumes: `build_report` 既有 `monthly_points: dict[str, list[dict]]`（每点 `{"month": "YYYY-MM", "t": float|None, "ic_mean": float|None, "count": int}`）
+- Produces: `factor_stats(points: list[dict]) -> {"icAll": float|None, "tAll": float|None, "positiveRatio": float|None, "byYear": dict[str, float]}`；报告 JSON 顶层新键 `"factor_stats": {name: <上述>}`
+
+- [ ] **Step 1: 写失败测试**
+
+```python
+# tests/factor/test_monthly_review.py
+"""判定弹药三件聚合口径(spec §4.2 2026-10-10):全期合成t/正IC占比/逐年."""
+import math
+
+from sanguo_factor.monthly_review import factor_stats
+
+
+def _pts(*pairs):
+    return [{"month": m, "ic_mean": v, "t": None, "count": 1} for m, v in pairs]
+
+
+def test_factor_stats_basic():
+    s = factor_stats(_pts(("2025-09", 0.04), ("2025-10", 0.06), ("2026-09", 0.02)))
+    assert s["icAll"] == round((0.04 + 0.06 + 0.02) / 3, 6)
+    assert s["positiveRatio"] == 1.0
+    assert s["byYear"] == {"2025": round((0.04 + 0.06) / 2, 6), "2026": 0.02}
+    mean = 0.04
+    var = ((0.0) + (0.02) + (-0.02)) / 2  # ddof=1
+    t_all = mean / (math.sqrt(var) / math.sqrt(3))
+    assert s["tAll"] == round(t_all, 4)
+
+
+def test_factor_stats_negative_and_short():
+    s = factor_stats(_pts(("2025-09", 0.05), ("2025-10", -0.01)))
+    assert s["positiveRatio"] == 0.5
+    assert s["tAll"] is not None
+    single = factor_stats(_pts(("2025-09", 0.05)))
+    assert single["tAll"] is None      # n<2 无合成 t
+    assert single["positiveRatio"] == 1.0
+
+
+def test_factor_stats_empty_and_dirty():
+    assert factor_stats([]) == {"icAll": None, "tAll": None,
+                                "positiveRatio": None, "byYear": {}}
+    dirty = [{"month": "2025-09", "ic_mean": None, "t": None},
+             {"month": "", "ic_mean": 0.1, "t": None}]
+    s = factor_stats(dirty)
+    assert s["icAll"] == 0.1
+    assert s["byYear"] == {}           # 空 month 不进逐年
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `python3 -m pytest tests/factor/test_monthly_review.py -v`
+Expected: FAIL `ImportError: cannot import name 'factor_stats'`
+
+- [ ] **Step 3: 实现 factor_stats + build_report 接线**
+
+在 `sanguo_factor/monthly_review.py` 的 `build_report` 函数**之前**加入：
+
+```python
+def factor_stats(points: list[dict]) -> dict:
+    """判定弹药聚合(2026-10-10 spec §4.2):全期合成 t/正 IC 占比/逐年.
+
+    口径:各月点 ic_mean(该批 12M 窗日均 IC 均值)的等权均值≈全史近似;
+    合成 t=mean/(std(ddof=1)/√n),n<2 或 std=0 时如实 None(小样本判读弱
+    的诚实注记在 spec,判定动作不因此自动化).
+    """
+    ics = [p["ic_mean"] for p in points if p.get("ic_mean") is not None]
+    if not ics:
+        return {"icAll": None, "tAll": None, "positiveRatio": None, "byYear": {}}
+    n = len(ics)
+    mean_ic = sum(ics) / n
+    t_all = None
+    if n >= 2:
+        var = sum((v - mean_ic) ** 2 for v in ics) / (n - 1)
+        if var > 0:
+            t_all = round(mean_ic / (var ** 0.5 / n ** 0.5), 4)
+    by_year: dict[str, list[float]] = {}
+    for p in points:
+        m, v = str(p.get("month") or ""), p.get("ic_mean")
+        if m and v is not None:
+            by_year.setdefault(m[:4], []).append(v)
+    return {"icAll": round(mean_ic, 6), "tAll": t_all,
+            "positiveRatio": round(sum(1 for v in ics if v > 0) / n, 4),
+            "byYear": {y: round(sum(v) / len(v), 6)
+                       for y, v in sorted(by_year.items())}}
+```
+
+`build_report` 返回 dict 处（现含 `"monthly_points": monthly_points` 的字面量）加一行：
+
+```python
+        "factor_stats": {name: factor_stats(pts)
+                         for name, pts in monthly_points.items()},
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `python3 -m pytest tests/factor/test_monthly_review.py -v`
+Expected: 3 PASS
+
+- [ ] **Step 5: render_markdown 加弹药节**
+
+在 `render_markdown` 内（verdicts 表渲染之后、collective 节之前）插入：
+
+```python
+    stats = report.get("factor_stats") or {}
+    if stats:
+        lines.append("")
+        lines.append("## 判定弹药（全期口径）")
+        lines.append("")
+        lines.append("| 因子 | 全期IC | 合成t | 正IC占比 | 逐年IC |")
+        lines.append("|------|--------|-------|----------|--------|")
+        for name in sorted(stats):
+            s = stats[name]
+            years = ", ".join(f"{y}:{v:.4f}" for y, v in s["byYear"].items()) or "—"
+
+            def _f(v, nd=4):
+                return "—" if v is None else f"{v:.{nd}f}"
+
+            lines.append(f"| {name} | {_f(s['icAll'], 6)} | {_f(s['tAll'])} "
+                         f"| {_f(s['positiveRatio'])} | {years} |")
+```
+
+（`lines` 为该函数既有的输出行列表变量名，按现场适配。）
+
+- [ ] **Step 6: 回归+commit**
+
+Run: `python3 -m pytest tests/factor/ -q`
+Expected: 全绿
+
+```bash
+git add sanguo_factor/monthly_review.py tests/factor/test_monthly_review.py
+git commit -m "feat(factor): 月度批评报告判定弹药三件——factor_stats 聚合(全期合成t/正IC占比/逐年IC)+md 渲染 [nas]"
+```
+
+---
+
+### Task 2: detail 端点透传 icStats
+
+**Files:**
+- Modify: `sanguo_api/routes_pipeline.py:582`（factor_detail 函数）
+- Test: `tests/api/test_routes_pipeline_factors.py`（已有文件追加）
+
+**Interfaces:**
+- Consumes: Task 1 报告 JSON 的 `factor_stats` 块；既有 `_monthly_reports() -> list[tuple[host, as_of, doc]]`
+- Produces: detail 响应 `factor.icStats: {"icAll","tAll","positiveRatio","byYear"} | None`
+
+- [ ] **Step 1: 写失败测试**
+
+在 `tests/api/test_routes_pipeline_factors.py` 追加（沿用该文件既有的 client/tmp registry fixture 风格；报告件用 tmp 目录+`SANGUO_FACTOR_MONTHLY_DIR` 环境变量 monkeypatch——若该 env 尚不存在则经 `_monthly_dir` 的真实 env 名适配，见 Step 3 附注）：
+
+```python
+def test_factor_detail_ic_stats(monkeypatch, tmp_path):
+    """detail 端点带出 factor_stats(全期弹药),无报告时如实 None."""
+    import json as _json
+    # 注册一个活因子(沿用文件内既有 helper/fixture 注册路径)
+    ...
+    rep_dir = tmp_path / "reports" / "factor_monthly"
+    rep_dir.mkdir(parents=True)
+    doc = {"as_of": "2026-09-30",
+           "monthly_points": {"fa_x": [
+               {"month": "2025-09", "ic_mean": 0.04, "t": 2.1, "count": 1},
+               {"month": "2025-10", "ic_mean": 0.06, "t": 2.4, "count": 1},
+               {"month": "2026-09", "ic_mean": 0.02, "t": 1.1, "count": 1}]},
+           "factor_stats": {"fa_x": {"icAll": 0.04, "tAll": 1.7321,
+                                     "positiveRatio": 1.0,
+                                     "byYear": {"2025": 0.05, "2026": 0.02}}}}
+    (rep_dir / "nas_2026-09-30.json").write_text(_json.dumps(doc), "utf-8")
+    monkeypatch.setenv("SANGUO_FACTOR_MONTHLY_DIR", str(rep_dir))
+    r = client.get("/pipeline/factors/fa_x/detail")
+    assert r.status_code == 200
+    got = r.json()["factor"]["icStats"]
+    assert got["icAll"] == 0.04 and got["tAll"] == 1.7321
+    assert got["byYear"]["2025"] == 0.05
+```
+
+（注册段 `...` 处复用本文件既有测试的注册代码块原样抄——每个测试自含注册是本文件既有惯例。）
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `python3 -m pytest tests/api/test_routes_pipeline_factors.py::test_factor_detail_ic_stats -v`
+Expected: FAIL `KeyError: 'icStats'`（或 assert None 异常）
+
+- [ ] **Step 3: 实现**
+
+`factor_detail` 内 `return` 前加：
+
+```python
+    stats = None
+    reports = _monthly_reports()
+    if reports:
+        stats = (reports[0][2].get("factor_stats") or {}).get(name)
+```
+
+返回字面量 `factor` dict 中 `"icRecentT"` 行后加：
+
+```python
+        "icStats": stats,
+```
+
+**附注**：`_monthly_dir()` 若未读 env，则加 `os.environ.get("SANGUO_FACTOR_MONTHLY_DIR", ...)` 前缀（与 `SANGUO_FACTOR_EVAL_DB` 同款覆盖纪律，生产缺省行为不变）：
+
+```python
+def _monthly_dir() -> str:
+    return os.environ.get("SANGUO_FACTOR_MONTHLY_DIR",
+                          os.path.join("reports", "factor_monthly"))
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `python3 -m pytest tests/api/test_routes_pipeline_factors.py -v`
+Expected: 全 PASS（含既有用例无回归）
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sanguo_api/routes_pipeline.py tests/api/test_routes_pipeline_factors.py
+git commit -m "feat(api): 因子详情端点透传 icStats 全期弹药(无报告如实 None)+monthly_dir env 覆盖 [vps]"
+```
+
+---
+
+### Task 3: expression_match 模块（防换皮算法）
+
+**Files:**
+- Create: `sanguo_factor/expression_match.py`
+- Test: `tests/factor/test_expression_match.py`（新建）
+
+**Interfaces:**
+- Consumes: Python `ast`；表达式方言=factor_guard 白名单（`ts_mean(x, 5)` 函数调用式 + `+ - * /` binop）
+- Produces:
+  - `similarity(expr_a: str, expr_b: str) -> float | None`（最大公共子树节点数/较小表达式节点数；解析失败 None）
+  - `top_similar(expression: str, candidates: dict[str, str], floor: float = 0.6) -> list[{"name": str, "ratio": float}]`（按 ratio 降序）
+
+- [ ] **Step 1: 写失败测试**
+
+```python
+# tests/factor/test_expression_match.py
+"""AST 防换皮(spec §4.2 2026-10-10):交换律归一化+最大公共子树占比."""
+from sanguo_factor.expression_match import similarity, top_similar
+
+
+def test_identical_and_commutative():
+    assert similarity("close/open", "close/open") == 1.0
+    # 交换律:a+b ≡ b+a(归一化后同构)
+    assert similarity("close + open", "open + close") == 1.0
+    assert similarity("ts_mean(close, 5) * volume",
+                      "volume * ts_mean(close, 5)") == 1.0
+
+
+def test_partial_common_subtree():
+    # 公共子树=ts_mean(close,5)(5节点);分母=较小表达式(7节点)
+    r = similarity("ts_mean(close, 5) / volume",
+                   "ts_mean(close, 5) * turnover")
+    assert r is not None and 0.6 < r < 1.0
+
+
+def test_no_common_and_invalid():
+    assert similarity("close / open", "volume * turnover") == 0.0
+    assert similarity("close +++", "close/open") is None  # 解析失败如实 None
+    assert similarity("", "close") is None
+
+
+def test_top_similar_floor():
+    cands = {"fa_a": "close + open", "fa_b": "open + close",
+             "fa_c": "volume * turnover"}
+    hits = top_similar("close + open", cands, floor=0.6)
+    assert [h["name"] for h in hits] == ["fa_a", "fa_b"]
+    assert all(h["ratio"] >= 0.6 for h in hits)
+    # 空表达式/坏 candidates 不炸
+    assert top_similar("close", {"fa_x": ""}) == []
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `python3 -m pytest tests/factor/test_expression_match.py -v`
+Expected: FAIL `ModuleNotFoundError: No module named 'sanguo_factor.expression_match'`
+
+- [ ] **Step 3: 实现模块**
+
+```python
+# sanguo_factor/expression_match.py
+"""表达式结构相似度(AST 防换皮,2026-10-10 spec §4.2).
+
+QuantaAlpha factor_ast 算法思想(最大公共子树+交换律)的方言内实现:
+不移植其 qlib 式解析器,直接用 Python ast——与 factor_guard 白名单
+同方言,注册的新因子表达式必然可解析.判定永远人做:本模块只产提示.
+"""
+from __future__ import annotations
+
+import ast
+
+_COMMUTATIVE = (ast.Add, ast.Mult)
+
+
+def _node_size(node: ast.AST) -> int:
+    return 1 + sum(_node_size(c) for c in ast.iter_child_nodes(node))
+
+
+def _key(node: ast.AST) -> str:
+    """规范化结构签名:交换律 binop 左右子树排序后拼接."""
+    if isinstance(node, ast.BinOp) and isinstance(node.op, _COMMUTATIVE):
+        lk, rk = _key(node.left), _key(node.right)
+        a, b = sorted((lk, rk))
+        return f"({a}|{b}|{type(node.op).__name__})"
+    if isinstance(node, ast.BinOp):
+        return f"({_key(node.left)}>{_key(node.right)}|{type(node.op).__name__})"
+    if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
+        args = ",".join(_key(a) for a in node.args)
+        return f"{node.func.id}({args})"
+    if isinstance(node, ast.Name):
+        return f"#{node.id}"
+    if isinstance(node, ast.Constant):
+        return f"#{node.value!r}"
+    return f"?{type(node).__name__}"
+
+
+def _all_subtree_keys(node: ast.AST) -> dict[str, int]:
+    """子树签名→节点数(同签名取最大)."""
+    out: dict[str, int] = {}
+    for sub in ast.walk(node):
+        k = _key(sub)
+        out[k] = max(out.get(k, 0), _node_size(sub))
+    return out
+
+
+def similarity(expr_a: str, expr_b: str) -> float | None:
+    """最大公共子树节点数 / 较小表达式节点数;任一解析失败=None."""
+    try:
+        ta = ast.parse(expr_a, mode="eval")
+        tb = ast.parse(expr_b, mode="eval")
+    except (SyntaxError, ValueError):
+        return None
+    sa, sb = _node_size(ta), _node_size(tb)
+    if sa == 0 or sb == 0:
+        return None
+    keys_a = _all_subtree_keys(ta)
+    best = 0
+    for sub in ast.walk(tb):
+        k = _key(sub)
+        if k in keys_a:
+            best = max(best, min(keys_a[k], _node_size(sub)))
+    return round(best / min(sa, sb), 4)
+
+
+def top_similar(expression: str, candidates: dict[str, str],
+                floor: float = 0.6) -> list[dict]:
+    """对候选池按相似度排序,过滤低于 floor 的(提示非硬拒)."""
+    out: list[dict] = []
+    for name, expr in candidates.items():
+        if not expression or not expr:
+            continue
+        r = similarity(expression, expr)
+        if r is not None and r >= floor:
+            out.append({"name": name, "ratio": r})
+    out.sort(key=lambda h: h["ratio"], reverse=True)
+    return out
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `python3 -m pytest tests/factor/test_expression_match.py -v`
+Expected: 4 PASS
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sanguo_factor/expression_match.py tests/factor/test_expression_match.py
+git commit -m "feat(factor): expression_match 防换皮模块——Python ast 交换律归一化+最大公共子树占比(QuantaAlpha 算法思想方言内实现) [nas]"
+```
+
+---
+
+### Task 4: decompose 注册链接入 similarity + 端点带出
+
+**Files:**
+- Modify: `sanguo_api/routes_pipeline.py:1165` 附近（decompose 注册路径，`vr.upsert_factor(...)` 调用处）+ `factor_detail`/`factors` 端点
+- Test: `tests/api/test_routes_pipeline_factors.py`（追加）
+
+**Interfaces:**
+- Consumes: Task 3 `top_similar`；registry 条目 `versions[0].params.expression`
+- Produces: registry 新条目可选键 `similarity: [{"name": str, "ratio": float}]`（仅建条落盘幂等不覆盖——与 description 补丁同纪律）；detail 响应 `factor.similarity`
+
+- [ ] **Step 1: 写失败测试**
+
+```python
+def test_decompose_register_similarity_hint(monkeypatch, tmp_path):
+    """新因子注册时对既有池比对,疑似换皮(≥0.6)存 entry.similarity(提示非硬拒)."""
+    # 复用本文件既有注册 helper:先注册 fa_old(expression="close + open")
+    ...
+    # 走 decompose 注册路径注册 fa_new(expression="open + close")
+    # (复用既有 decompose job 测试的注册调用;注册后重读 registry)
+    reg = _load_runtime_registry(tmp_path)
+    entry = reg["factors"]["fa_new"]
+    assert entry.get("similarity") == [{"name": "fa_old", "ratio": 1.0}]
+
+    r = client.get("/pipeline/factors/fa_new/detail")
+    assert r.json()["factor"]["similarity"] == [{"name": "fa_old", "ratio": 1.0}]
+```
+
+（两处 `...` 复用本文件既有 decompose/注册测试代码块——file 内已有 decompose job 注册用例，抄其 setup 原样。）
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `python3 -m pytest tests/api/test_routes_pipeline_factors.py::test_decompose_register_similarity_hint -v`
+Expected: FAIL（similarity 键不存在）
+
+- [ ] **Step 3: 实现（注册点 + 端点）**
+
+`routes_pipeline.py:1165` 的 `vr.upsert_factor(reg, cand["name"], ...)` 之后、落盘 save 之前加：
+
+```python
+                from sanguo_factor import expression_match as em
+                _expr = str((cand.get("params") or {}).get("expression") or "")
+                if _expr:
+                    _cands = {n: str(((v.get("versions") or [{}])[0]
+                                      .get("params") or {}).get("expression") or "")
+                              for n, v in reg["factors"].items()
+                              if n != cand["name"]}
+                    _sim = em.top_similar(_expr, _cands)
+                    if _sim:
+                        reg["factors"][cand["name"]]["similarity"] = _sim
+```
+
+（`cand`/`reg` 为该处既有变量名，按现场适配；`upsert_factor` 建新条目后 `reg["factors"][name]` 直接可写。）
+
+`factor_detail` 返回字面量加：
+
+```python
+        "similarity": e.get("similarity"),
+```
+
+- [ ] **Step 4: 跑测试确认通过+回归**
+
+Run: `python3 -m pytest tests/api/test_routes_pipeline_factors.py tests/api/ -q`
+Expected: 全 PASS
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sanguo_api/routes_pipeline.py tests/api/test_routes_pipeline_factors.py
+git commit -m "feat(api): decompose 注册时防换皮比对存 similarity 提示(非硬拒,judgement 人做)+detail 带出 [vps]"
+```
+
+---
+
+### Task 5: 前端展示（弹药卡+换皮徽标）
+
+**Files:**
+- Modify: `frontend/src/api/pipeline.ts`（FactorDetail 接口+factor 列表行接口）
+- Modify: `frontend/src/views/pipeline/FactorDetail.vue`（弹药行+相似提示）
+- Modify: `frontend/src/views/pipeline/FactorFactory.vue`（列表现有来源列旁加 ⚠ 徽标）
+- Test: `frontend/src/views/pipeline/FactorDetail.spec.ts`、`FactorFactory.spec.ts`（追加用例）
+
+**Interfaces:**
+- Consumes: detail `factor.icStats`/`factor.similarity`（Task 2/4）
+- Produces: 展示（无新数据流）
+
+- [ ] **Step 1: 类型+展示实现**
+
+`pipeline.ts` FactorDetail 的 factor 接口加：
+
+```typescript
+  icStats?: { icAll: number | null; tAll: number | null;
+              positiveRatio: number | null; byYear: Record<string, number> } | null
+  similarity?: Array<{ name: string; ratio: number }> | null
+```
+
+`FactorDetail.vue` 档案卡 `icRecentT` 行后加（沿用既有 `kv` 行结构）：
+
+```vue
+        <div class="kv"><span class="k">全期弹药</span>
+          <span class="v">
+            <template v-if="detail.factor.icStats && detail.factor.icStats.tAll !== null">
+              全期IC {{ detail.factor.icStats.icAll?.toFixed(4) }} ·
+              合成t {{ detail.factor.icStats.tAll?.toFixed(2) }} ·
+              正IC占比 {{ ((detail.factor.icStats.positiveRatio ?? 0) * 100).toFixed(0) }}% ·
+              逐年 {{ Object.entries(detail.factor.icStats.byYear).map(([y, v]) => `${y}:${v.toFixed(4)}`).join(' ') || '—' }}
+            </template>
+            <template v-else>—（月度点不足或首班未跑）</template>
+          </span></div>
+        <div class="kv" v-if="detail.factor.similarity?.length"><span class="k">疑似换皮</span>
+          <span class="v warn">
+            <span v-for="s in detail.factor.similarity" :key="s.name"
+                  class="sim-chip" @click="goFactor(s.name)">
+              {{ s.name }} · {{ (s.ratio * 100).toFixed(0) }}%</span>
+          </span></div>
+```
+
+（`goFactor` 复用本组件既有跳转 factor 详情的方法名；`.sim-chip`/`.warn` 样式按本文件既有 style 块补最小两条：cursor:pointer + 底色。）
+
+`FactorFactory.vue` 表格「来源」列的链接标识后加换皮徽标（行数据经 factors 端点带出 `similarity`——列表端点在 Task 4 一并带出：`factors()` 列表项加 `"similarity": fe.get("similarity")`）：
+
+```vue
+        <el-tooltip v-if="row.similarity?.length"
+                    :content="`疑似换皮: ${row.similarity.map(s => s.name).join(', ')}`">
+          <span class="dup-badge" @click="goDetail(row.name)">⚠</span>
+        </el-tooltip>
+```
+
+- [ ] **Step 2: 写前端测试（两文件各追加一用例）**
+
+`FactorDetail.spec.ts`：
+
+```typescript
+it('renders ammo stats and similarity chips', () => {
+  const detail = makeDetail()
+  detail.factor.icStats = { icAll: 0.04, tAll: 1.73, positiveRatio: 1.0,
+                            byYear: { 2025: 0.05, 2026: 0.02 } }
+  detail.factor.similarity = [{ name: 'fa_old', ratio: 1.0 }]
+  mount(Detail, { props: { name: 'fa_x' } })
+  expect(document.body.textContent).toContain('合成t 1.73')
+  expect(document.body.textContent).toContain('正IC占比 100%')
+  expect(document.body.textContent).toContain('fa_old')
+})
+```
+
+（`makeDetail`/mount 结构抄本文件既有用例；无 icStats 时显示 `—` 的反向用例一并加。）
+
+`FactorFactory.spec.ts`：
+
+```typescript
+it('shows dup badge when similarity present', async () => {
+  const rows = makeRows()
+  rows[0].similarity = [{ name: 'fa_old', ratio: 0.9 }]
+  mount(Factory)
+  await flushPromises()
+  expect(document.querySelector('.dup-badge')).toBeTruthy()
+})
+```
+
+- [ ] **Step 3: 跑测试+类型检查**
+
+Run: `cd frontend && npx vitest run src/views/pipeline/FactorDetail.spec.ts src/views/pipeline/FactorFactory.spec.ts && npx vue-tsc --noEmit; echo "rc=$?"`
+Expected: vitest 全 PASS；vue-tsc rc=0（管道退出码显式门控）
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add frontend/src/api/pipeline.ts frontend/src/views/pipeline/FactorDetail.vue frontend/src/views/pipeline/FactorFactory.vue frontend/src/views/pipeline/FactorDetail.spec.ts frontend/src/views/pipeline/FactorFactory.spec.ts
+git commit -m "feat(frontend): 因子详情全期弹药行+疑似换皮徽标(工厂列表/详情两触点) [vps]"
+```
+
+---
+
+### Task 6: test 隔离审计 + spec 落账 + 收尾
+
+**Files:**
+- Read-only: `sanguo_factor/weight_profiles/quant12_icirfit_v1.json`（fit_window）、`sanguo_factor/composite_weighting.py`、`sanguo_factor/exam_gate.py`（考场窗逻辑）、`sanguo_factor/composite_rolling.py`
+- Modify: `docs/superpowers/specs/2026-09-22-research-to-trading-pipeline-design.md`（§4.4 test 隔离条款尾部落审计结论）
+
+**Interfaces:** 无代码——审计+文档。
+
+- [ ] **Step 1: 审计**
+
+1. 读 `weight_profiles/quant12_icirfit_v1.json` 的 `fit_window`（ICIR 拟合窗）与 `baseline`；
+2. 读 `exam_gate.py` 确认 h2 样外考场窗与 fit_window 的关系（考场必须完全后于拟合窗）；
+3. 读 `composite_rolling.py` 确认滚动档案（quant12_roll_20XX.json）的年滚动是否每次只用当年之前数据拟合、之后数据考核；
+4. 产出三行结论：fit 窗=【】、考场窗=【】、是否重叠=【否/是+风险描述】。
+
+- [ ] **Step 2: 结论落 spec**
+
+§4.4 test 隔离条款段尾追加一行：
+
+```markdown
+  核实结论（2026-10-10 审计）：quant12_v2a（quant12_icirfit_v1）fit_window=【实际值】、h2 考场窗=【实际值】——【无重叠，滚动结构天然合规/发现 X，已修或将修】。
+```
+
+- [ ] **Step 3: 全量回归**
+
+Run: `python3 -m pytest tests/factor tests/api -q && cd frontend && npx vitest run src/views/pipeline/ 2>&1 | tail -5`
+Expected: 全绿
+
+- [ ] **Step 4: Commit（spec 落账随本 task）**
+
+```bash
+git add docs/superpowers/specs/2026-09-22-research-to-trading-pipeline-design.md
+git commit -m "docs: 流水线 spec 判定弹药/AST防换皮/三路对拍/test隔离四块增补+quant12_v2a 权重窗审计结论 [no-doc]"
+```
+
+（spec 增补主体若已在开工前随本地 commit 落盘则本条合并入上一条；`[no-doc]` 用于纯落账——若同 push 含 sanguo_factor 设计变更则改 `[nas]` 并确认 spec 同 push 已更新，档随码走铁律。）
+
+- [ ] **Step 5: push 纪律（人工闸门）**
+
+**今晚 21:30 双机日度首班跑完并验证通过后（明早）**，统一 push：
+
+```bash
+git log --oneline origin/master..HEAD   # 报清单给用户
+git push origin master                   # 用户确认后
+```
+
+---
+
+## Self-Review 结论
+
+1. **Spec 覆盖**：§4.2 判定弹药三件→Task 1/2/5；§4.2 AST 防换皮→Task 3/4/5；§4.4 test 隔离→Task 6。§4.4 三路对拍（LGBM challenger）**不在本 plan**——班次二单独开计划（spec 已定案）。
+2. **占位符**：Task 1 Step 5 的 `lines` 变量名、Task 4 的 `cand`/`reg` 变量名、前端 `goFactor`/`makeDetail` 复用点均为「按现场既有名适配」的显式指令（非 TBD），代码块本身完整。
+3. **类型一致**：`factor_stats` 返回键 `icAll/tAll/positiveRatio/byYear` 在 Task 1（产生）、Task 2（透传）、Task 5（前端 `icStats` 类型）三处一致；`similarity` 的 `{name, ratio}` 在 Task 3/4/5 一致。

docs/superpowers/plans/2026-10-10-lgbm-challenger-shadow.md

@@ -0,0 +1,579 @@
+# LGBM Challenger 三路加权对拍 Implementation Plan（班次二）
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 同池同窗三路组合信号（等权/ICIR/LightGBM）月度 walk-forward 影子对拍，产出对拍件+特征贡献，回答「ICIR 加权是不是最优」——challenger 永不进生产。
+
+**Architecture:** 新模块 `sanguo_factor/challenger_lgbm.py`：读月度批已导出的因子值宽表（`monthly_batch --factor-values-out` 产物，每因子一份 datetime×vt_symbol parquet）+ vnpy_db close 构造防缺口 label → 池内三路信号（等权带方向翻转/ICIR 档案权重/LGBM 月度重训）→ 样外月 walk-forward 滚动 → 对拍件 JSON（三路样外 IC/ICIR/分层多空+feature importance+池准入表）。挂 NAS 月度链 stage6（失败不阻链）。
+
+**Tech Stack:** lightgbm==4.6.0（docker 依赖已在 requirements-docker.txt:84；本机 venv 需装）、pandas/polars 既有栈、pytest。
+
+## Global Constraints
+
+- spec=2026-09-22 流水线总设计 §4.4「三路加权对拍（2026-10-10 定案）」+「test 隔离条款」。
+- **v1 对拍池=quant12_icirfit_v1.sources 的 12 源**（vma_60/alpha16/alpha83/alpha12/alpha2/alpha42/vol_ma5/wvma_20/klow/cord_5/kup/alpha81）——在役晋级池，ICIR 权重档案现成，三路同池才可比；fa_* 新因子扩池挂后续。
+- **test 隔离铁律（本班直接适用）**：LGBM 训练窗=样外月之前的全部数据，样外月数据绝不进训练；对拍件如实记录切分。
+- **纪律**：challenger 只落 `reports/factor_monthly/challenger_lgbm/` 子目录（判定层端点只读根层零影响）；永不进生产，升级走决议 K 考场+用户口令。
+- **池贪心准入 v1 形态**：12 源全量进（在役池已过晋级闸），corr 去冗余只**记录**不剔除（`|corr|>0.7` 对标记注在对拍件，剔除动作留人判断——首年观察期）。
+- label 口径：`close.shift(-2)/close.shift(-1)-1`（T 信号→T+1 收盘可成交→T+2 收盘卖，QuantaAlpha 防缺口同构保守口径）；**双侧 CSRankNorm**（特征与 label 截面秩归一 `(rank(pct)-0.5)`，NaN 保持 NaN 不参与当日截面）。
+- 等权信号=**方向调整后等权**（`direction=='-'` 的源先取负再平均，否则对负向因子不公平）；ICIR 同理带 direction；LGBM 不带（模型自学）。
+- 环境：NAS docker 镜像已含 lightgbm；**本机 venv310 需 `pip install lightgbm==4.6.0`**（dev/test 用）；VPS 不跑本链（纯 NAS 影子）→ commit 标签 `[nas]`。
+- 月度批导出件 12M 窗；walk-forward=前 11M 训练+最后 1M 样外，每月滚动，样外月序列逐月累积。
+- LGBM 超参（qlib Alpha158 基准起步）：`loss=mse, lr=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`；early stopping=**无独立 valid 段时用固定 num_boost_round=500**（12M 窗内再切 valid 会挤占训练数据，v1 从简，注记在对拍件）。
+
+---
+
+### Task 1: challenger_lgbm 数据层（装载+label+CSRankNorm）
+
+**Files:**
+- Create: `sanguo_factor/challenger_lgbm.py`
+- Test: `tests/factor/test_challenger_lgbm.py`
+
+**Interfaces:**
+- Consumes: `sanguo_factor/weight_profiles/quant12_icirfit_v1.json`（sources 结构 `{name: {direction, fit_icir, weight}}`）
+- Produces:
+  - `load_pool() -> dict[str, dict]`（12 源档案）
+  - `build_label(close_wide: pd.DataFrame) -> pd.DataFrame`（防缺口 label 宽表）
+  - `cs_rank_norm(df: pd.DataFrame) -> pd.DataFrame`（截面秩归一，NaN 透传）
+
+- [ ] **Step 1: venv 装 lightgbm**
+
+```bash
+venv310/bin/pip install lightgbm==4.6.0
+venv310/bin/python -c "import lightgbm; print(lightgbm.__version__)"
+```
+Expected: `4.6.0`（装不上则报告并停——后续 task 全依赖它）。
+
+- [ ] **Step 2: 写失败测试**
+
+```python
+# tests/factor/test_challenger_lgbm.py
+"""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)
+    assert out.loc[2, "a"] == pytest.approx(0.75)   # rank pct 1.0 - 0.5
+    assert np.isnan(out.loc[3, "a"])                 # NaN 透传不占截面
+    row0 = out.loc[0, ["a", "b"]].tolist()
+    assert row0 == [pytest.approx(-0.25), pytest.approx(0.25)]  # 双侧中心化
+```
+
+- [ ] **Step 3: 跑测试确认失败**
+
+Run: `venv310/bin/python -m pytest tests/factor/test_challenger_lgbm.py -v`
+Expected: FAIL `ModuleNotFoundError`/`ImportError`
+
+- [ ] **Step 4: 实现数据层**
+
+```python
+# 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
+```
+
+- [ ] **Step 5: 跑测试确认通过并 commit**
+
+Run: `venv310/bin/python -m pytest tests/factor/test_challenger_lgbm.py -v`
+Expected: 3 PASS
+
+```bash
+git add sanguo_factor/challenger_lgbm.py tests/factor/test_challenger_lgbm.py
+git commit -m "feat(factor): challenger_lgbm 数据层——在役池档案/防缺口 label/双侧 CSRankNorm [nas]"
+```
+
+---
+
+### Task 2: 三路信号构造（等权/ICIR/LGBM walk-forward）
+
+**Files:**
+- Modify: `sanguo_factor/challenger_lgbm.py`
+- Test: `tests/factor/test_challenger_lgbm.py`（追加）
+
+**Interfaces:**
+- Consumes: Task 1 三函数；因子值字典 `{name: pd.DataFrame}`（宽表，统一对齐列/索引）
+- Produces:
+  - `equal_weight_signal(values: dict[str, pd.DataFrame], pool: dict) -> pd.DataFrame`（方向调整等权）
+  - `icir_signal(values: dict, pool: dict) -> pd.DataFrame`（档案权重加权）
+  - `lgbm_walk_forward(values: dict, label: pd.DataFrame, last_month: str, params: dict | None = None) -> tuple[pd.DataFrame, dict]`（返回样外月预测+feature importance；训练窗=样外月之前全部）
+
+- [ ] **Step 1: 写失败测试**
+
+```python
+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)
+    assert (sig == pytest.approx(2.0 * (0.75 - 0.25))).all().all()
+
+
+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"]
+    pred, imp = lgbm_walk_forward(values, label, last_month="2025-03")
+    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
+```
+
+- [ ] **Step 2: 跑测试确认失败** — `ImportError`（三函数未定义）
+
+- [ ] **Step 3: 实现**
+
+```python
+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()
+    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]))
+    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
+```
+
+（`stack(future_stack=True)` 为 pandas≥2.1 语义——本仓 pandas 版本若 <2.1 改 `stack(dropna=True)`，跑 Step 4 时确认。）
+
+- [ ] **Step 4: 跑测试确认通过并 commit**
+
+Run: `venv310/bin/python -m pytest tests/factor/test_challenger_lgbm.py -v`
+Expected: 6 PASS
+
+```bash
+git add sanguo_factor/challenger_lgbm.py tests/factor/test_challenger_lgbm.py
+git commit -m "feat(factor): challenger 三路信号——方向调整等权/ICIR 档案加权/LGBM walk-forward(样外隔离铁律) [nas]"
+```
+
+---
+
+### Task 3: 对拍评估与产物件
+
+**Files:**
+- Modify: `sanguo_factor/challenger_lgbm.py`
+- Test: `tests/factor/test_challenger_lgbm.py`（追加）
+
+**Interfaces:**
+- Consumes: Task 2 三信号+`build_label`
+- Produces:
+  - `score_signal(signal: pd.DataFrame, label: pd.DataFrame) -> dict`（`{"ic_mean","icir","q5q1","days"}`——日 IC 均值/ICIR/分层多空累计/有效天数）
+  - `run_challenge(values_dir: str, vnpy_db: str, as_of: str, out_dir: str) -> str`（主入口：读月度批导出宽表+拉 close→三路→对拍件 JSON 落 `out_dir/challenger_lgbm/{host}_{as_of}.json`，返回件路径）
+
+- [ ] **Step 1: 写失败测试**
+
+```python
+def test_score_signal_perfect_and_flat():
+    idx = pd.date_range("2025-03-03", periods=10, freq="B")
+    cols = ["a", "b", "c"]
+    sig = pd.DataFrame(np.linspace(-1, 1, 30).reshape(10, 3), 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
+
+
+def test_run_challenge_end_to_end(tmp_path, monkeypatch):
+    """端到端:合成 3 因子×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)
+    monkeypatch.setattr(cl, "_load_close_wide", lambda db, cols_ref: 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"
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+- [ ] **Step 3: 实现**
+
+```python
+def _load_close_wide(vnpy_db: str, columns_ref: pd.Index) -> pd.DataFrame:
+    """按因子宽表列(股票)拉收盘价.生产实现走 load_universe_bars 轻量列."""
+    from sanguo_data.config import load_config, find_config_path
+    from .batch_eval import load_universe_bars
+    cfg = load_config(find_config_path())
+    bars = load_universe_bars(vnpy_db or cfg.data_paths["vnpy_db"],
+                              symbols=list(columns_ref), limit=None)
+    _ = 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/五分位多空累计/有效天数."""
+    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),
+            "icir": round(float(icir), 6) if icir is not None else None,
+            "q5q1": round(float(sum(ls_rets)), 6), "days": len(ser)}
+
+
+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
+    close = _load_close_wide(vnpy_db, cols_ref)
+    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 只记录不剔除(首年观察期,剔除留人)
+    flat = pd.concat({n: cs_rank_norm(v) for n, v in values.items()})
+    daily_corr = flat.groupby(level=0).apply(
+        lambda g: g.T.corr(method="spearman") if g.shape[0] > 1 else None)
+    flagged = []
+    means = daily_corr.groupby(level=1).mean() if daily_corr is not None else None
+    if means is not None:
+        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)})
+
+    import platform
+    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
+```
+
+- [ ] **Step 4: 跑测试确认通过并 commit**
+
+Run: `venv310/bin/python -m pytest tests/factor/test_challenger_lgbm.py -v`
+Expected: 8 PASS
+
+```bash
+git add sanguo_factor/challenger_lgbm.py tests/factor/test_challenger_lgbm.py
+git commit -m "feat(factor): challenger 对拍评估——三路样外 IC/ICIR/分层多空+特征贡献+池冗余注记件 [nas]"
+```
+
+---
+
+### Task 4: CLI 入口+月度链 stage6 接线
+
+**Files:**
+- Modify: `sanguo_factor/challenger_lgbm.py`（main）
+- Modify: `scripts/nas_sync/run_factor_monthly_standalone.sh`（stage5 replay 后加 stage6）
+- Test: `tests/factor/test_challenger_lgbm.py`（CLI 冒烟一用例）
+
+**Interfaces:**
+- Consumes: Task 3 `run_challenge`；月度批 `--factor-values-out` 导出目录（NAS wrapper 需确认 monthly_batch CLI 透传参数名——读 monthly_batch main 后按实际名接）
+- Produces: `python -m sanguo_factor.challenger_lgbm --values-dir ... --as-of ... --out-dir ... [--vnpy-db ...] [--host nas]`；stage6 产物 `reports/factor_monthly/challenger_lgbm/nas_{as_of}.json`
+
+- [ ] **Step 1: 写 CLI 冒烟失败测试**
+
+```python
+def test_cli_smoke(tmp_path, monkeypatch, capsys):
+    from sanguo_factor import challenger_lgbm as cl
+    idx = pd.date_range("2025-01-01", "2025-03-31", freq="B")
+    close = pd.DataFrame(100 + np.cumsum(np.random.default_rng(1).normal(0.4, size=(len(idx), 2)), axis=0),
+                         index=idx, columns=["a", "b"])
+    monkeypatch.setattr(cl, "_load_close_wide", lambda db, c: close)
+    vdir = tmp_path / "vals"; vdir.mkdir()
+    for n in ("s0", "s1"):
+        pd.DataFrame(np.random.default_rng(2).normal(size=(len(idx), 2)),
+                     index=idx, columns=["a", "b"]).to_parquet(vdir / f"{n}.parquet")
+    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 (tmp_path / "challenger_lgbm" / "nas_2025-03-31.json").exists()
+```
+
+- [ ] **Step 2: 确认失败 → 实现 main + 接线**
+
+main（模块尾）：
+
+```python
+def main(argv=None):
+    import argparse
+    ap = argparse.ArgumentParser(description="LGBM challenger 三路对拍(月度链 stage6)")
+    ap.add_argument("--values-dir", required=True,
+                    help="月度批 --factor-values-out 导出目录(每因子一 parquet)")
+    ap.add_argument("--as-of", required=True)
+    ap.add_argument("--out-dir", required=True, help="报告根(factor_monthly)")
+    ap.add_argument("--vnpy-db", default=None)
+    ap.add_argument("--host", default="nas")
+    a = ap.parse_args(argv)
+    try:
+        path = run_challenge(a.values_dir, a.vnpy_db, a.as_of, a.out_dir, host=a.host)
+    except FileNotFoundError as e:
+        print(f"[challenger] skip: {e}")
+        return 0          # 无宽表=非错误(月度批未开导出),不阻链
+    print(f"[challenger] 对拍件: {path}")
+    return 0
+```
+
+**wrapper 接线**（`run_factor_monthly_standalone.sh` stage5 段后，最终 rc 聚合前）：
+
+```bash
+  # stage6 challenger 三路对拍(影子实验,失败不阻月度链;产物落子目录
+  # 判定层端点只读根层零影响)
+  "$DOCKER" run --rm --name sanguo-factor-chal --user "${UID_ADMIN}:${GID_ADMIN}" \
+    --group-add "${GID_ADMINS}" --no-healthcheck --entrypoint python \
+    -e HOME=/tmp -e MPLCONFIGDIR=/tmp/mpl \
+    -v /volume1/stock:/volume1/stock \
+    -v "$APP":/app:ro \
+    -w /app \
+    sanguo_vnpy_v2:lock-aligned \
+    -m sanguo_factor.challenger_lgbm \
+    --values-dir "$BASE/reports/factor_values" \
+    --as-of "$AS_OF" --out-dir "$OUT_DIR" --host nas
+  rc_chal=$?
+  echo "=== $(date '+%F %T') stage6 challenger exit=$rc_chal (影子,不进 final) ==="
+```
+
+（`$BASE/reports/factor_values` 为月度批 `--factor-values-out` 实际目录——**执行时先读 `monthly_batch.py` CLI 与 wrapper 现状确认导出是否已开+目录名**；若月度批尚未传导出参数，本 task 同时给 wrapper 的 stage1 命令加透传并验证 monthly_batch CLI 支持该 flag，不支持则本 task 给 monthly_batch 补 `--factor-values-out` CLI 参数透传到 run_batch_eval——已存在函数参数，只补 argparse。）
+
+- [ ] **Step 3: 跑测试+本地冒烟并 commit**
+
+Run: `venv310/bin/python -m pytest tests/factor/test_challenger_lgbm.py -v && bash -n scripts/nas_sync/run_factor_monthly_standalone.sh`
+Expected: 9 PASS + bash -n 语法过
+
+```bash
+git add sanguo_factor/challenger_lgbm.py scripts/nas_sync/run_factor_monthly_standalone.sh tests/factor/test_challenger_lgbm.py
+git commit -m "feat(factor): challenger CLI+月度链 stage6 接线(影子失败不阻链;月度批因子宽表导出透传) [nas]"
+```
+
+---
+
+### Task 5: spec 落账+全量回归+真数据首跑预约
+
+**Files:**
+- Modify: `docs/superpowers/specs/2026-09-22-research-to-trading-pipeline-design.md`（§4.4 三路对拍块补「班次二落地」一行）
+- Test: 全量回归
+
+- [ ] **Step 1: spec 落账**
+
+§4.4 三路对拍块尾追加：
+
+```markdown
+  - **班次二落地（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 个样外月。
+```
+
+- [ ] **Step 2: 全量回归**
+
+Run: `venv310/bin/python -m pytest tests/factor tests/api -q && cd frontend && npx vitest run src/views/pipeline/ 2>&1 | tail -3`
+Expected: 全绿（challenger 新 9 用例含其中）
+
+- [ ] **Step 3: commit**
+
+```bash
+git add docs/superpowers/specs/2026-09-22-research-to-trading-pipeline-design.md
+git commit -m "docs: 三路对拍班次二落地落账 spec §4.4 [no-doc]"
+```
+
+（若 Task 4 同 push 含 sanguo_factor 设计变更，spec 同 push 已更新则本条合并标签纪律照旧。）
+
+- [ ] **Step 4: 真数据首跑预约（不跑）**
+
+真数据首跑=**11-01 月度链首班自动带 stage6**（NAS wrapper 生效位需 `scp -O scripts/nas_sync/run_factor_monthly_standalone.sh sanguo-nas:/volume1/stock/sanguo_vnpy_v2/run_factor_monthly.sh` 同步——**push 过 nas-verify 后执行**，与既往生效位纪律一致）。本班不本地跑真数据（vnpy_db 全量在 NAS）。
+
+---
+
+## Self-Review 结论
+
+1. **Spec 覆盖**：§4.4 三路（等权 Task 2/ICIR Task 2/LGBM Task 2）、月度重训（walk-forward Task 2）、CSRankNorm 双侧（Task 1）、防缺口 label（Task 1）、池贪心准入 v1=全量+冗余注记（Task 3，首年观察期口径已在 Global Constraints 声明）、feature importance（Task 2/3）、challenger 子目录隔离（Task 3）、影子纪律（Task 4 stage6 不阻链）——全覆盖。**spec 的「贪心准入」v1 简化为记录不剔除**（在役池已过晋级闸，重复剔除是减法实验，留观察期后决定）——与 spec「起步=晋级因子全量不设限」一致。
+2. **占位符**：无 TBD；Task 4 的「执行时确认 factor-values-out 目录名」是显式适配指令（月度批 CLI 需现场核对）。
+3. **类型一致**：`cs_rank_norm/build_label/load_pool`（Task 1）→三信号（Task 2）→`score_signal/run_challenge`（Task 3）→`main`（Task 4）签名逐级消费一致；对拍件 JSON 键 `signals/pool/split/feature_importance/redundancy_flagged/lgbm_params` 在 Task 3 产生、测试断言同键。