读数脚本：乙路改读当日快照，存疑与强化全部列出

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

logic_state_coverage.py

@@ -90,6 +90,18 @@ def broker_paths(codes, ds: str) -> dict:
     return out
 
 
+def snapshot_of(day: str) -> dict:
+    """行业观点快照表里这个计划日的那一版，按簇键索引；表没建或没有当日行返回空字典。"""
+    try:
+        df = db.read_mysql(
+            "factor",
+            f"SELECT * FROM {config.JUDGEMENT_SNAPSHOT_TABLE} WHERE plan_date = %s", (day,))
+    except Exception as e:  # noqa: BLE001
+        print(f"  （行业观点快照表读取失败，乙路整体缺席: {e!r}）")
+        return {}
+    return {str(r["cluster_key"]): dict(r) for _, r in df.iterrows()} if not df.empty else {}
+
+
 def main(ds: str | None = None) -> None:
     ds = ds or (dt.date.today() - dt.timedelta(days=1)).isoformat()
     print(f"=== 逻辑状态四态覆盖率读数 · 数据日 {ds} ===\n")
@@ -101,7 +113,9 @@ def main(ds: str | None = None) -> None:
 
     claims = sources.logic_claims(codes, ds, per_stock=200)
     brokers = broker_paths(codes, ds)
-    snaps = judgement.load_previous(ds)
+    # 要的是这个计划日当天那一版快照（带迁移与陈旧两列），不是它之前的那一版——
+    # load_previous 读的是严格早于计划日的，用它会在快照首日读到空。当天没有就退回上一版。
+    snaps = snapshot_of(ds) or judgement.load_previous(ds)
     print(f"甲路取到 {len(claims)} 只票的论断；丙路算得出 {len(brokers)} 只票；"
           f"乙路快照 {len(snaps)} 个簇\n")
 
@@ -134,11 +148,12 @@ def main(ds: str | None = None) -> None:
         states[r["state"]] += 1
         if r["why"]:
             whys[f"{r['state']}·{r['why']}"] += 1
-        if r["state"] in (ls.STATE_STRONG, ls.STATE_DOUBT) and r["state"] not in samples:
+        # 逻辑存疑与逻辑强化都是稀有且会触发动作的状态，全部列出来逐条看得见；
-            samples[r["state"]] = (k, r["reasons"][:3])
+        # 无法判断那两类只留一条样例，否则几千行没法读。
-        if r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT \
+        if r["state"] in (ls.STATE_STRONG, ls.STATE_DOUBT):
-                and "矛盾" not in samples:
+            samples.setdefault(r["state"], []).append((k, r["reasons"][:3]))
-            samples["矛盾"] = (k, r["reasons"][:3])
+        elif r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT:
+            samples.setdefault("无法判断·证据矛盾", []).append((k, r["reasons"][:3]))
 
     print("每路各自的信号分布")
     for path, cnt in per_path.items():
@@ -154,10 +169,11 @@ def main(ds: str | None = None) -> None:
         print("\n  无法判断的子因")
         for w, n in whys.most_common():
             print(f"    {w} {n}")
-    if samples:
+    for tag, rows in samples.items():
-        print("\n样例")
+        cap = len(rows) if tag in (ls.STATE_STRONG, ls.STATE_DOUBT) else 2
-        for tag, (k, rs) in samples.items():
+        print(f"\n{tag}（{len(rows)} 只{'，全部列出' if cap == len(rows) else '，列前两只'}）")
-            print(f"  {tag} {k}")
+        for k, rs in rows[:cap]:
+            print(f"  {k}")
             for x in rs:
                 print(f"    {x}")