From 5a2a51a2d09bb1e22157715627e47507e611032d Mon Sep 17 00:00:00 2001 From: zlt Date: Fri, 4 Sep 2026 09:13:07 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AF=BB=E6=95=B0=E8=84=9A=E6=9C=AC=EF=BC=9A?= =?UTF-8?q?=E4=B9=99=E8=B7=AF=E6=94=B9=E8=AF=BB=E5=BD=93=E6=97=A5=E5=BF=AB?= =?UTF-8?q?=E7=85=A7=EF=BC=8C=E5=AD=98=E7=96=91=E4=B8=8E=E5=BC=BA=E5=8C=96?= =?UTF-8?q?=E5=85=A8=E9=83=A8=E5=88=97=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5 --- logic_state_coverage.py | 36 ++++++++++++++++++++++++++---------- 1 file changed, 26 insertions(+), 10 deletions(-) diff --git a/logic_state_coverage.py b/logic_state_coverage.py index d889e81..7a5a920 100644 --- a/logic_state_coverage.py +++ b/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 \ - and "矛盾" not in samples: - samples["矛盾"] = (k, r["reasons"][:3]) + # 逻辑存疑与逻辑强化都是稀有且会触发动作的状态，全部列出来逐条看得见； + # 无法判断那两类只留一条样例，否则几千行没法读。 + if r["state"] in (ls.STATE_STRONG, ls.STATE_DOUBT): + samples.setdefault(r["state"], []).append((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: - print("\n样例") - for tag, (k, rs) in samples.items(): - print(f" {tag} {k}") + for tag, rows in samples.items(): + cap = len(rows) if tag in (ls.STATE_STRONG, ls.STATE_DOUBT) else 2 + print(f"\n{tag}（{len(rows)} 只{'，全部列出' if cap == len(rows) else '，列前两只'}）") + for k, rs in rows[:cap]: + print(f" {k}") for x in rs: print(f" {x}")