复盘对账后两处口径修正：结论表只取不晚于数据日的最新行（历史重建防前视）；复盘名单加观察档等权；关注环节表补未动环节的成员数与领涨者（不限档位表内）

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

plan.py

@@ -75,8 +75,8 @@ def _names() -> dict:
 def _evidence(ds: str):
     """每股最强一条传导证据：主题、源数、已动比例；另返回涉及的行情快照日。"""
     tr = db.read_pg(
-        "SELECT ts_code, target, n_sources, moved_ratio, mkt_trade_date "
+        "SELECT ts_code, target, n_sources, moved_ratio, mkt_trade_date, "
-        "FROM v_factor_transmission WHERE scan_date = %s", (ds,))
+        "members_total, moved FROM v_factor_transmission WHERE scan_date = %s", (ds,))
     if tr.empty:
         return {}, set()
     tr["k"] = tr["ts_code"].map(common.to_prefix)
@@ -84,7 +84,10 @@ def _evidence(ds: str):
     tr["moved_ratio"] = pd.to_numeric(tr["moved_ratio"], errors="coerce").fillna(0)
     tr["strength"] = tr["n_sources"] * (1.0 - tr["moved_ratio"])
     tr = tr.sort_values("strength", ascending=False).drop_duplicates("k")
-    ev = {r.k: (str(r.target), int(r.n_sources), float(r.moved_ratio))
+    # 元组前三项是既有口径（多处按下标取），第四、五项是环节的成员数与已动数（关注环节表用）
+    ev = {r.k: (str(r.target), int(r.n_sources), float(r.moved_ratio),
+                None if pd.isna(r.members_total) else int(r.members_total),
+                None if pd.isna(r.moved) else int(r.moved))
           for r in tr.itertuples()}
     days = {str(x) for x in tr["mkt_trade_date"].dropna().unique()}
     return ev, days
@@ -161,17 +164,22 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
         if cards.get(k, {}).get("verdict") == card.VERDICT_CANDIDATE:
             s["candidates"] += 1
     for e in ev.values():                       # 只有未动名单的环节也是"被指向"，列入但无已启动
-        segs.setdefault(e[0], {"segment": e[0], "n_sources": e[1], "chain_fit": None,
+        s = segs.setdefault(e[0], {"segment": e[0], "n_sources": e[1], "chain_fit": None,
                                    "members_total": None, "moved": None,
                                    "started": [], "candidates": 0, "leader": None})
+        if s["members_total"] is None and len(e) > 4:
+            s["members_total"], s["moved"] = e[3], e[4]
     for s in segs.values():
         best, best_pct = None, None
         for k in s["started"]:
-            p = cards.get(k, {}).get("pct0")
+            # 领涨者按行情视图找，不限于档位表里的票：被指向且已启动但不在主榜与观察档的
+            # （无覆盖且不在赛道、或被风险闸挡）也要显示，这是"定位对不对"的读数
+            p = (daily.get(k) or {}).get("pct0")
             if p is not None and (best_pct is None or p > best_pct):
                 best, best_pct = k, p
         s["leader"] = {"code": best, "pct0": best_pct} if best else None
         s["started_count"] = len(s["started"])
+        s["started_in_tiers"] = sum(1 for k in s["started"] if k in cards)
     ordered = sorted(segs.values(), key=lambda s: (-s["candidates"], -(s["n_sources"] or 0),
                                                    -(s["chain_fit"] or 0), s["segment"]))
     return cards, ordered

plan_review.py

@@ -183,6 +183,7 @@ def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str
         full = data.get("_full") or {}
         main_rows, obs_rows = full.get("main", []), full.get("observe", [])
         main_codes = [r["code"] for r in main_rows]
+        obs_codes = [r["code"] for r in obs_rows]
         tier_of = {r["code"]: r.get("tier") for r in main_rows}
         verdict_of = {r["code"]: r.get("verdict") for r in main_rows + obs_rows}
         seg_codes = sorted({r["code"] for r in main_rows + obs_rows if r.get("evidence")})
@@ -212,7 +213,8 @@ def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str
                           if reg and reg.get("weak_day") is not None else "无标签")
             lists = {
                 "生产名单": prod, "候选单": cands, "关注单": watch, "环节名单": seg_codes,
-                "主榜等权": main_codes, "全池等权": list(ret.dropna().index),
+                "主榜等权": main_codes, "观察档等权": obs_codes,
+                "全池等权": list(ret.dropna().index),
             }
             for name, codes in lists.items():
                 s = summarize(ret, codes, base_all, base_main)

sources.py

@@ -93,13 +93,18 @@ def night_conclusions(codes, ds: str) -> dict[str, dict]:
     codes = sorted({common.to_prefix(str(c).strip()) for c in codes if c})
     if not codes:
         return {}
+    # 只取不晚于数据日 ds 的结论行：生产上 ds 就是数据日，与"取最新一行"等价；
+    # 复盘重建历史日时这一条防前视（否则历史日会读到今天的吸筹状态）。
+    # trade_date 是整数 YYYYMMDD（数据源盘点 §1a），直接比大小。
+    ds_int = int(ds.replace("-", "")) if ds else 99999999
     try:
         marks = ",".join(["%s"] * len(codes))
         df = db.read_mysql(
             "pms",
             f"SELECT stock_code, trade_date, signal_type, support_level, pressure_level, "
-            f"raw_logic_json FROM strategy_daily_results WHERE stock_code IN ({marks})",
+            f"raw_logic_json FROM strategy_daily_results "
-            tuple(codes))
+            f"WHERE stock_code IN ({marks}) AND trade_date <= %s",
+            tuple(codes) + (ds_int,))
     except Exception as e:  # noqa: BLE001
         print(f"  （决策系统结论表读取失败，吸筹确认线与坏信号风险整体缺席: {e!r}）")
         return {}