diff --git a/logic_state_coverage.py b/logic_state_coverage.py new file mode 100644 index 0000000..d889e81 --- /dev/null +++ b/logic_state_coverage.py @@ -0,0 +1,166 @@ +"""逻辑状态四态的覆盖率读数：只读，不写库，不改任何判决。 + +回答一个问题：四态这套东西今天在真实数据上能覆盖多少票、各落到哪个状态。 +它是判断"这一路该接多大面"的依据，也是接进计划装配之前必须先看的读数。 + +三路各自的取数在这里，归一与合成都调 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 main(ds: str | None = None) -> None: + ds = ds or (dt.date.today() - dt.timedelta(days=1)).isoformat() + print(f"=== 逻辑状态四态覆盖率读数 · 数据日 {ds} ===\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) + snaps = judgement.load_previous(ds) + print(f"甲路取到 {len(claims)} 只票的论断；丙路算得出 {len(brokers)} 只票；" + f"乙路快照 {len(snaps)} 个簇\n") + + # 乙路按环节名对上主题：候选卡按环节，产业研判按主题聚簇，两者不在一个命名空间， + # 这里只做同名匹配，对不上的票乙路就是缺失。这一路的天花板本来就低（实测 1.4%）。 + by_seg = {(r.get("segment_name") or r.get("subject_name") or "").strip(): r + for r in snaps.values()} + 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) 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]) + + 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}") + if samples: + print("\n样例") + for tag, (k, rs) in samples.items(): + print(f" {tag} {k}") + for x in rs: + print(f" {x}") + + +if __name__ == "__main__": + main(sys.argv[1] if len(sys.argv) > 1 else None)