# -*- coding: utf-8 -*- """三源合议观察读数复核 (2026-09-14 观察读数包, 台账 013)。只读, 不写任何表, 随时可跑。 回答台账 006 到 011 复核要的四个读数, 打四张表: 一, 逐日: 相关票数、技术面无读数占比、基本面无读数只数、择时无读数只数 (占比用 map_cover 算)。 二, 逐日: 等开口只数、等盘中确认只数、跳过原因分布、交人只数、放行只数、增持门拦截次数。 三, 转空离场逐条: 日期、代码、确认与否、数量。 四, 合议方向与人工裁决比对: 带合议六键且已被人裁决的提议里, 合议看多而人驳回、 合议交人而人采纳的, 逐只列出。 台账里的对照阈值 (只作提示, 不下结论 —— 调阈值要先回台账): 技术面无读数占比应低于 3%; 每天等技术面开口应在 1 到 15 只; 转空退出每周 0 到 3 条。 运行 (模拟仓 155): make consensus-review # 默认近 3 个交易日 make consensus-review DAYS=7 # 近 7 个交易日 或直接: docker compose run --rm --no-deps pms-web python scripts/consensus_review.py --days 3 """ import argparse import json import os import sys from datetime import datetime, timedelta sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.repo import consensus_stat_repo, pms_repo # noqa: E402 from app.services import consensus_stats as cst # noqa: E402 # 人裁决的终态 (提议服务现有枚举, 见 ddl_pms_v1.sql 的 pms_proposal.status; 不另造) ACCEPTED, DECLINED = "ACCEPTED", "DECLINED" def _date_range(days: int): """[date_from, date_to]: date_to 是今天 (北京), date_from 往前第 days-1 个交易日。 交易日历取不到就退回自然日。""" to_ymd = int(datetime.now().strftime("%Y%m%d")) try: from app.core import tradedays as td from_ymd = int(td.ymd(td.prev_trade_day(to_ymd, max(0, int(days) - 1)))) except Exception: d = datetime.now() - timedelta(days=max(0, int(days) - 1)) from_ymd = int(d.strftime("%Y%m%d")) return from_ymd, to_ymd def _loads(v): if isinstance(v, dict): return v try: return json.loads(v) if v else {} except (TypeError, ValueError): return {} def _fmt_pct(x): return f"{x:.1%}" if x is not None else "—" def _dates_in(counts, extra=None): """出现过读数的日期, 升序 (两个来源的日期并集)。""" ds = set(counts or {}) for d in (extra or []): ds.add(d) return sorted(ds) def main(): ap = argparse.ArgumentParser() ap.add_argument("--days", type=int, default=3, help="复核窗口 (交易日, 默认 3)") args = ap.parse_args() a, b = _date_range(args.days) print("=" * 68) print("三源合议观察读数复核 %s (只读; 窗口 %s ~ %s, %d 个交易日)" % (datetime.now().strftime("%Y-%m-%d %H:%M:%S"), a, b, args.days)) print("=" * 68) try: counts = consensus_stat_repo.count_by_kind(a, b) map_rows = consensus_stat_repo.list_range(a, b, kinds=[cst.K_MAP_COVER]) exit_rows = consensus_stat_repo.list_range(a, b, kinds=[cst.K_TECH_EXIT]) except Exception as e: # noqa: BLE001 —— 连不上库也说人话, 不甩栈 print(f"\n✗ 观察读数表读取失败: {type(e).__name__}: {e}\n" f" (本脚本要在服务器容器里跑: make consensus-review)") sys.exit(1) # map_cover 逐日 detail: {日期: {codes, states}} cover = {} for r in map_rows: d = _loads(r.get("detail")) cover[int(r["stat_date"])] = {"codes": int(d.get("codes") or 0), "states": int(d.get("states") or 0)} # ---- 一、逐日无读数 ---- print("\n【一】逐日无读数 (占比用 map_cover 算; 台账线: 技术面无读数占比 < 3%)") print(" 日期 相关票 技术面覆盖 技术面无读数占比 基本面无读数 择时无读数") for d in _dates_in(counts, cover): cv = cover.get(d) or {} n_rel, n_have = cv.get("codes"), cv.get("states") ratio = None if n_rel: ratio = (n_rel - n_have) / n_rel c = counts.get(d) or {} fn = (c.get(cst.K_FUND_NOREAD) or {}).get("rows", 0) tn = (c.get(cst.K_TIMING_NOREAD) or {}).get("rows", 0) flag = " ⚠ 高于 3%" if (ratio is not None and ratio > 0.03) else "" print(" %-8s %5s %8s %14s %10d %8d%s" % (d, n_rel if n_rel is not None else "—", n_have if n_have is not None else "—", _fmt_pct(ratio), fn, tn, flag)) if not _dates_in(counts, cover): print(" 窗口内没有观察读数 —— 观察读数包还没部署, 或映射还没重建过。") # ---- 二、逐日路由分布 ---- print("\n【二】逐日合议路由 (等开口台账线: 每天 1 到 15 只)") print(" 日期 等开口 等盘中确认 跳过·无评析 跳过·看空 交人 放行 增持门拦") for d in _dates_in(counts): c = counts.get(d) or {} def _n(k): return (c.get(k) or {}).get("rows", 0) wt = _n(cst.K_OPEN_WAIT_TECH) wc = _n("open_wait_confirm") # 盘中确认包后续才有, 现在恒 0 sf = _n(cst.K_OPEN_SKIP_FUND) sb = _n(cst.K_OPEN_SKIP_BEAR) cf = _n(cst.K_OPEN_CONFIRM) ps = _n(cst.K_OPEN_PASS) gb = (c.get(cst.K_GATE_BLOCK) or {}).get("rounds", 0) # 拦截次数用累计次数 flag = " ⚠ 等开口超 15" if wt > 15 else "" print(" %-8s %5d %8d %9d %8d %4d %4d %6d%s" % (d, wt, wc, sf, sb, cf, ps, gb, flag)) if not _dates_in(counts): print(" 窗口内没有路由读数。") # ---- 三、转空离场逐条 ---- print("\n【三】转空离场逐条 (台账线: 每周 0 到 3 条; 合议链第一天应为零)") if not exit_rows: print(" 窗口内没有转空离场实弹 —— 符合第一天为零的预期。") for r in exit_rows: d = _loads(r.get("detail")) conf = "确认清仓" if d.get("confirm") else "未确认减仓" print(" %-8s %-10s %s %s 股 (记到 %s 次) %s" % (r["stat_date"], r["ts_code"], conf, d.get("qty"), r.get("rounds"), (r.get("reason") or "")[:40])) # ---- 四、合议方向与人工裁决比对 ---- print("\n【四】合议方向与人工裁决对不上的 (合议看多而你驳回 / 合议交人而你采纳)") try: props = pms_repo.list_proposals(statuses=(ACCEPTED, DECLINED), limit=500, include_archived=True) except Exception as e: # noqa: BLE001 print(f" ✗ 提议表读取失败: {type(e).__name__}: {e}") props = [] mismatches = [] for p in props: decided = str(p.get("decided_at") or "")[:10].replace("-", "") if not decided or not (str(a) <= decided <= str(b)): continue con = (p.get("hard_numbers") or {}).get("consensus") or {} direction, route = con.get("direction"), con.get("route") if not direction and not route: continue # 没有合议六键的提议 (旧提议), 不比对 status = p.get("status") if direction == "看多" and status == DECLINED: mismatches.append((decided, p.get("ts_code"), "合议看多, 你驳回了", con)) elif route == "交人" and status == ACCEPTED: mismatches.append((decided, p.get("ts_code"), "合议交人, 你采纳了", con)) if not mismatches: print(" 窗口内没有对不上的 (或还没有带合议六键、已被裁决的提议)。") for d, code, why, con in sorted(mismatches): print(" %-8s %-10s %s (合议: %s)" % (d, code, why, con.get("reason") or "")) print("\n复核口径提醒: 本报告只陈述读数, 阈值调整要先回台账 006 到 011 写一条, " "不按单次复盘读数回调 (原则一)。") if __name__ == "__main__": main()