feat(api): monthly-review/ic-trend 接真——10-01 用户拍板「接真不删页」P1 件 [vps]

审计 10-01 P1-12 三 mock 端点之二接真(读现成月度批评产物,hypotheses 留 P4):
- GET /pipeline/monthly-review: 本机 monthly dir 最新报告 JSON 原文→
  period/runDate/factorsEvaluated/consensusMedianT(collective.median)/
  consensusThreshold(collective.floor);首班 10-04 前 null(前端 PanelCard
  隐藏不造数);crossCheck 仅同 as_of 双 host 报告同 dir 才出双行(单机=空
  列表,不伪造"一致"),maxAbsDiff=同因子 t 差最大值,>0.05(起步默认)判分歧
- GET /pipeline/factors/{name}/ic-trend: 最新报告 monthly_points 月度 t
  序列(t=None 月如实跳过);baseline=晋级 promotion_t×0.5(相对降幅过半
  红线,衰减条件①),未晋级=2.0 绝对地板(条件②);无报告/无点 404
- mock/index.ts: 两端点条目退役(逐端点删条目纪律,composite 先例);
  PIPELINE_MOCK 仅剩 hypotheses(P4);全接真后删机制=在案退出条件
- spec §7 数据行接真进度同步(档随码走)
- 测试 tests/api/test_routes_pipeline_monthly.py 8 例(单机/双机一致/分歧/
  baseline 两态/404 两态)

Co-Authored-By: Claude Code <noreply@anthropic.com>

sanguo_api/routes_pipeline.py

@@ -394,6 +394,101 @@ def factors() -> dict:
     return {"items": items}
 
 
+_CROSS_DIFF_MAX = 0.05   # 双机对拍分歧阈(起步默认,首年校准;spec 对拍规则未定数)
+
+
+def _monthly_reports() -> list[tuple[str, str, dict]]:
+    """monthly dir 全部报告 [(host, as_of, doc)];按 as_of 降序,坏文件跳过.
+
+    文件名={host}_{as_of}.json(monthly_review.main 落盘原文)。
+    """
+    d = _monthly_dir()
+    if not os.path.isdir(d):
+        return []
+    out: list[tuple[str, str, dict]] = []
+    for fn in os.listdir(d):
+        if not fn.endswith(".json"):
+            continue
+        host, _, as_of = fn[:-5].rpartition("_")
+        if not host or len(as_of) != 10 or as_of[4] != "-":
+            continue
+        try:
+            with open(os.path.join(d, fn), encoding="utf-8") as f:
+                out.append((host, as_of, json.load(f)))
+        except (OSError, json.JSONDecodeError):
+            continue
+    out.sort(key=lambda t: t[1], reverse=True)
+    return out
+
+
+def _machine_label(host: str) -> str:
+    return "NAS" if "nas" in host.lower() else "VPS"
+
+
+@router.get("/pipeline/monthly-review")
+def monthly_review_view() -> dict | None:
+    """月度批评卡(10-01 用户拍板接真):读本机 monthly dir 最新报告 JSON 原文.
+
+    首班(10-04)未跑→null(前端 PanelCard 隐藏,不造数);
+    crossCheck 只在同 as_of 双 host 报告都落在本机 dir 时才出双行
+    (对拍件汇合后自动生效),单机=空列表(不伪造"一致");
+    maxAbsDiff=两机同因子 t 差绝对值最大值,> _CROSS_DIFF_MAX 判"分歧"。
+    """
+    reports = _monthly_reports()
+    if not reports:
+        return None
+    host, as_of, doc = reports[0]
+    factors = doc.get("factors") or {}
+    col = doc.get("collective") or {}
+    peers = {h: d for h, a, d in reports if a == as_of}
+    cross: list[dict[str, Any]] = []
+    if len(peers) >= 2:
+        hosts = sorted(peers)
+        fa = peers[hosts[0]].get("factors") or {}
+        fb = peers[hosts[1]].get("factors") or {}
+        common = [f for f in fa.keys() & fb.keys()
+                  if (fa.get(f) or {}).get("t") is not None
+                  and (fb.get(f) or {}).get("t") is not None]
+        max_diff = max((abs(fa[f]["t"] - fb[f]["t"]) for f in common),
+                       default=0.0)
+        verdict = "一致" if max_diff <= _CROSS_DIFF_MAX else "分歧"
+        cross = [{"machine": _machine_label(h),
+                  "factors": len(peers[h].get("factors") or {}),
+                  "maxAbsDiff": round(max_diff, 6), "verdict": verdict}
+                 for h in hosts[:2]]
+    return {"period": as_of[:7], "runDate": as_of,
+            "factorsEvaluated": len(factors),
+            "consensusMedianT": col.get("median"),
+            "consensusThreshold": col.get("floor"),
+            "crossCheck": cross}
+
+
+@router.get("/pipeline/factors/{name}/ic-trend")
+def factor_ic_trend(name: str) -> dict:
+    """IC 趋势图(10-01 用户拍板接真):最新报告 monthly_points 的月度 t 序列.
+
+    baseline 红线=已晋级因子 promotion_t×0.5(相对降幅过半位置,触点②
+    衰减条件①);未晋级/无基准=2.0 绝对地板(衰减条件②)——衰减告警需
+    两条同时满足+连续 3 月,t 为 None 的月如实跳过。
+    """
+    reports = _monthly_reports()
+    if not reports:
+        raise HTTPException(404, "月度批评首班未跑,无 monthly_points")
+    _, _, doc = reports[0]
+    pts = (doc.get("monthly_points") or {}).get(name)
+    if not pts:
+        raise HTTPException(404, f"因子无月度点: {name}")
+    pairs = [(p["month"], p["t"]) for p in pts if p.get("t") is not None]
+    from sanguo_factor import version_registry as vr
+    from sanguo_portfolio.strategy_registry import ensure_runtime_registry
+    reg = vr.load_registry(ensure_runtime_registry(
+        _factor_registry_path(), os.path.join("config", "factor_registry.yaml")))
+    promotion_t = (reg["factors"].get(name) or {}).get("promotion_t")
+    baseline = round(promotion_t * 0.5, 4) if promotion_t else 2.0
+    return {"dates": [m for m, _ in pairs],
+            "rollingT": [t for _, t in pairs], "baseline": baseline}
+
+
 @router.post("/pipeline/factors/{name}/verdict")
 def factor_verdict(name: str, body: dict) -> dict:
     """晋级评审三按钮(触点②):promote/graveyard 走 version_registry 状态机;