akg-factor-bridge/logic_state_coverage.py

"""逻辑状态四态的覆盖率读数：只读，不写库，不改任何判决。

回答一个问题：四态这套东西今天在真实数据上能覆盖多少票、各落到哪个状态。
它是判断"这一路该接多大面"的依据，也是接进计划装配之前必须先看的读数。

三路各自的取数在这里，归一与合成都调 logic_state 里的纯函数——读数脚本绝不能自己
另写一套判据，否则读到的就不是系统真会给出的状态。

  甲路 研报论断   基座 PG   v_factor_logic（经 sources.logic_claims）
  乙路 产业研判   平台 MySQL t_akg_judgement_snapshot（经 judgement.load_previous）
  丙路 券商行动   平台 MySQL gp_report_rc：两个等长窗口的每股收益预测中位数与机构数
  丁路 公司事件   无数据源，恒定缺失

丙路的取数口径（三条都要照做，否则读数是错的）：
  一，两个窗口必须等长。不等长会让八成的票假显示覆盖收缩——实测前 135 天对近 45 天时
      有 907 只票误报。
  二，同一财年同一预测期才可比，按 quarter 精确匹配，跨财年比较没有意义。
  三，同一家机构在窗口里可能发多篇，先按机构取最近一篇再算中位数，否则发得勤的
      机构会被重复计入。

跑法（155 上）：docker exec akg_factor_bridge python3 logic_state_coverage.py [数据日] [计划日]
            计划日不给就取数据日的次日。
"""
from __future__ import annotations

import datetime as dt
import statistics as st
import sys
from collections import Counter, defaultdict

import common
import config
import db
import judgement
import logic_state as ls
import sources

# 丙路两个窗口各自的长度（自然日）。等长是硬要求，见模块说明第一条。
BROKER_WINDOW_DAYS = 45


def broker_paths(codes, ds: str) -> dict:
    """按票算丙路信号。返回前缀码到 logic_state.signal 的字典（算不出的票不进字典）。"""
    end = dt.date.fromisoformat(ds)
    mid = end - dt.timedelta(days=BROKER_WINDOW_DAYS)
    start = end - dt.timedelta(days=BROKER_WINDOW_DAYS * 2)
    # noqa: SLF001 —— _to_dot 是同仓自用的代码格式转换
    dotted = sorted({sources._to_dot(c) for c in codes if c})  # noqa: SLF001
    if not dotted:
        return {}
    out: dict[str, dict] = {}
    # 一次拉两个窗口的全部行，按票在内存里分窗——逐票查库要发几千次请求。
    marks = ",".join(["%s"] * len(dotted))
    try:
        df = db.read_mysql(
            "factor",
            f"SELECT ts_code, report_date, quarter, org_name, eps FROM gp_report_rc "
            f"WHERE ts_code IN ({marks}) AND report_date > %s AND report_date <= %s "
            f"AND eps IS NOT NULL AND quarter IS NOT NULL",
            tuple(dotted) + (start.isoformat(), end.isoformat()))
    except Exception as e:  # noqa: BLE001
        print(f"  （券商研报明细表读取失败，丙路整体缺席: {e!r}）")
        return {}

    # 分票、分窗、分财年地堆起来：{票: {财年: {"now": {机构: (日期, 每股收益)}, "prev": ...}}}
    box: dict = defaultdict(lambda: defaultdict(lambda: {"now": {}, "prev": {}}))
    for r in df.itertuples():
        d = sources._ymd(r.report_date)  # noqa: SLF001 —— 同仓自用
        if not d:
            continue
        win = "now" if d > mid.isoformat() else "prev"
        k = common.to_prefix(str(r.ts_code).strip())
        q = str(r.quarter).strip()
        org = str(r.org_name or "").strip() or "未署名"
        slot = box[k][q][win]
        # 同一家机构在窗口里发了多篇，只留最近一篇（模块说明第三条）。
        if org not in slot or d > slot[org][0]:
            slot[org] = (d, float(r.eps))

    for k, by_q in box.items():
        # 两个窗口都有料的财年里，取行数最多的那个作可比口径（模块说明第二条）。
        usable = [(q, v) for q, v in by_q.items() if v["now"] and v["prev"]]
        if not usable:
            continue
        q, v = max(usable, key=lambda kv: len(kv[1]["now"]) + len(kv[1]["prev"]))
        now = {"eps": st.median([x[1] for x in v["now"].values()]), "firms": len(v["now"])}
        prev = {"eps": st.median([x[1] for x in v["prev"].values()]), "firms": len(v["prev"])}
        s = ls.from_broker(now, prev, as_of=ds)
        s["refs"] = [{**(s["refs"][0] if s["refs"] else {}), "quarter": q}]
        out[k] = s
    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 {}
    if df.empty:
        return {}
    # pandas 把空值读成 NaN，而 NaN 是真值——直接往下传会让"没有上一版倾向"看着像有值。
    # 全部归一成 None，判据那边只认 None。
    import math

    def _n(v):
        if v is None or (isinstance(v, float) and math.isnan(v)):
            return None
        return v

    return {str(r["cluster_key"]): {k: _n(v) for k, v in r.items()}
            for _, r in df.iterrows()}


def main(ds: str | None = None, plan_day: str | None = None) -> None:
    """ds 是数据日（行情与论断按它取），plan_day 是计划日（行业观点快照按它取）。

    这两个日期在生产里本来就差一天：计划日凌晨构建，用的是上一个交易日的数据，
    而当天的行业观点快照按计划日落库。早先这个脚本用同一个日期取三样东西，
    结果是要么读不到当天的快照，要么读不到行情。plan_day 不给就取数据日的次日。
    """
    ds = ds or (dt.date.today() - dt.timedelta(days=1)).isoformat()
    if not plan_day:
        try:
            plan_day = (dt.date.fromisoformat(ds) + dt.timedelta(days=1)).isoformat()
        except ValueError:
            plan_day = ds
    print(f"=== 逻辑状态四态覆盖率读数 · 数据日 {ds} · 计划日 {plan_day} ===\n")

    codes = db.read_pg("SELECT DISTINCT ts_code FROM v_factor_stock_daily "
                       "WHERE trade_date = %s", (ds,))["ts_code"].tolist()
    codes = [common.to_prefix(str(c).strip()) for c in codes]
    print(f"当日有行情的票 {len(codes)} 只\n")

    claims = sources.logic_claims(codes, ds, per_stock=200)
    brokers = broker_paths(codes, ds)
    # 要的是这个计划日当天那一版快照（带迁移与陈旧两列），不是它之前的那一版——
    # load_previous 读的是严格早于计划日的，用它会在快照首日读到空。当天没有就退回上一版。
    snaps = snapshot_of(plan_day) or judgement.load_previous(plan_day)
    print(f"甲路取到 {len(claims)} 只票的论断；丙路算得出 {len(brokers)} 只票；"
          f"乙路快照 {len(snaps)} 个簇\n")

    # 乙路按环节名对上主题：候选卡按环节，产业研判按主题聚簇，两者不在一个命名空间，
    # 这里只做同名匹配，对不上的票乙路就是缺失。这一路的天花板本来就低（实测 1.4%）。
    by_seg = {}
    for r in snaps.values():
        name = str(r.get("segment_name") or r.get("subject_name") or "").strip()
        if name:
            by_seg[name] = r
    seg_of = {}
    try:
        d = db.read_pg("SELECT ts_code, target FROM v_factor_transmission_moved "
                       "WHERE scan_date = %s", (ds,))
        for r in d.itertuples():
            seg_of[common.to_prefix(str(r.ts_code).strip())] = str(r.target)
    except Exception as e:  # noqa: BLE001
        print(f"  （传导视图读取失败，乙路整体缺席: {e!r}）")

    per_path = {ls.PATH_CLAIM: Counter(), ls.PATH_JUDGE: Counter(),
                ls.PATH_BROKER: Counter()}
    states, whys = Counter(), Counter()
    samples: dict = {}
    for k in codes:
        a = ls.from_claims(claims.get(k), ds, stale_days=config.LOGIC_STALE_DAYS)
        b = ls.from_judgement(by_seg.get(seg_of.get(k, "")))
        c = brokers.get(k) or ls.signal(ls.PATH_BROKER, ls.SIG_NONE,
                                        why="两个窗口里算不出可比的预测")
        per_path[ls.PATH_CLAIM][a["signal"]] += 1
        per_path[ls.PATH_JUDGE][b["signal"]] += 1
        per_path[ls.PATH_BROKER][c["signal"]] += 1
        r = ls.compose([a, b, c, ls.from_events()])
        states[r["state"]] += 1
        if r["why"]:
            whys[f"{r['state']}·{r['why']}"] += 1
        # 逻辑存疑与逻辑强化都是稀有且会触发动作的状态，全部列出来逐条看得见；
        # 无法判断那两类只留一条样例，否则几千行没法读。
        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():
        tot = sum(cnt.values())
        line = "、".join(f"{s} {n}（{n / tot:.1%}）" for s, n in cnt.most_common())
        print(f"  {path}：{line}")
    print(f"  {ls.PATH_EVENT}：缺失 {len(codes)}（100.0%，无数据源）\n")

    print("合成后的四态分布")
    for s, n in states.most_common():
        print(f"  {s} {n}（{n / len(codes):.1%}）")
    if whys:
        print("\n  无法判断的子因")
        for w, n in whys.most_common():
            print(f"    {w} {n}")
    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}")


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else None,
         sys.argv[2] if len(sys.argv) > 2 else None)