# -*- coding: utf-8 -*- """三源合议观察读数复核 (2026-09-14 观察读数包, 台账 013)。只读, 不写任何表, 随时可跑。 回答台账 006 到 011、015 复核要的读数, 打六张表: 一, 逐日: 相关票数、技术面无读数占比、基本面无读数只数、择时无读数只数 (占比用 map_cover 算)。 二, 逐日: 等开口只数、等盘中确认只数、跳过原因分布、交人只数、放行只数、增持门拦截次数。 三, 转空离场逐条: 日期、代码、确认与否、数量。 四, 合议方向与人工裁决比对: 带合议六键且已被人裁决的提议里, 合议看多而人驳回、 合议交人而人采纳的, 逐只列出。 五, 系统自决逐条 (台账 015): 日期、代码、决定 (采纳/放弃/交人)、档位、规则、合议、研判把握度。 六, 研判不可用占比: 自决触及的候选里研判不可用的比例 (台账线低于两成)。 台账里的对照阈值 (只作提示, 不下结论 —— 调阈值要先回台账): 技术面无读数占比应低于 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) # 增持门看只数 (当天被拦的持仓只数)。检查点是两分钟窗口, 累计次数 rounds 会把同一只翻倍 # (一个检查点记两次、一天四个点最多八次), 只当诊断用, 不当拦截次数 (2026-09-14 评审第一条)。 gb = _n(cst.K_GATE_BLOCK) 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【四】合议与人工裁决的差异") # 上限按天数放大 (2026-09-14 评审第四条): 三天窗口五百条够用, 三十天可能不够。 _lim = min(5000, max(500, int(args.days) * 200)) try: props = pms_repo.list_proposals(statuses=(ACCEPTED, DECLINED), limit=_lim, include_archived=True) except Exception as e: # noqa: BLE001 print(f" ✗ 提议表读取失败: {type(e).__name__}: {e}") props = [] divergences, handoff_ok = [], [] 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: divergences.append((decided, p.get("ts_code"), con)) elif route == "交人" and status == ACCEPTED: handoff_ok.append((decided, p.get("ts_code"), con)) # 真分歧: 合议看多而人驳回。这才是台账要盯的「方向对不上」。 print(" 真分歧 (合议看多而你驳回):") if not divergences: print(" 无 (或还没有带合议六键、已被裁决的提议)。") for d, code, con in sorted(divergences): print(" %-8s %-10s (合议: %s)" % (d, code, con.get("reason") or "")) # 信息行: 合议交人而人采纳。交人本就是让人定, 人采纳不是分歧 (2026-09-14 评审第二条), # 单列出来是看合议是不是偏保守 (交人后多半被放行则可考虑放宽)。 print(" 信息行 (合议交人而你采纳; 交人本就让你定, 不算分歧):") if not handoff_ok: print(" 无。") for d, code, con in sorted(handoff_ok): print(" %-8s %-10s (合议: %s)" % (d, code, con.get("reason") or "")) # ---- 五、系统自决逐条 (2026-09-14 台账 015) ---- print("\n【五】系统自决逐条 (日期 代码 决定 档位 规则 合议 研判把握度)") try: sd_rows = consensus_stat_repo.list_range( a, b, kinds=[cst.K_AUTO_ACCEPT, cst.K_AUTO_DECLINE, cst.K_AUTO_CONFIRM, cst.K_AUTO_WOULD_ACCEPT, cst.K_AUTO_WOULD_DECLINE]) except Exception as e: # noqa: BLE001 print(f" ✗ 自决读数读取失败: {type(e).__name__}: {e}") sd_rows = [] _DEC_CN = {cst.K_AUTO_ACCEPT: "采纳", cst.K_AUTO_DECLINE: "放弃", cst.K_AUTO_CONFIRM: "交人", cst.K_AUTO_WOULD_ACCEPT: "本会采纳", cst.K_AUTO_WOULD_DECLINE: "本会放弃"} if not sd_rows: print(" 窗口内没有系统自决记录 (自决还没开, 或没有归自决管的候选)。") for r in sd_rows: d = _loads(r.get("detail")) con = "%s/%s" % (d.get("direction") or "—", d.get("route") or "—") conf = d.get("conf") print(" %-8s %-10s %-6s %-5s %-16s %-7s 把握度 %s (记到 %s 次)" % (r["stat_date"], r["ts_code"], _DEC_CN.get(r["kind"], r["kind"]), d.get("tier") or "—", d.get("rule") or "—", con, conf if conf is not None else "—", r.get("rounds"))) # ---- 六、研判不可用占比 (2026-09-14; 台账线: 超两成报警) ---- print("\n【六】研判不可用占比 (judge_unavail / 自决触及的候选; 台账线: < 20%)") print(" 日期 研判不可用 自决触及 占比") for d in _dates_in(counts): c = counts.get(d) or {} def _n2(k): return (c.get(k) or {}).get("rows", 0) un = _n2(cst.K_JUDGE_UNAVAIL) touched = sum(_n2(k) for k in (cst.K_AUTO_ACCEPT, cst.K_AUTO_DECLINE, cst.K_AUTO_CONFIRM, cst.K_AUTO_WOULD_ACCEPT, cst.K_AUTO_WOULD_DECLINE)) ratio = (un / touched) if touched else None flag = " ⚠ 超 20%" if (ratio is not None and ratio > 0.2) else "" if un or touched: print(" %-8s %8d %8d %6s%s" % (d, un, touched, _fmt_pct(ratio), flag)) if not any((counts.get(d) or {}).get(cst.K_JUDGE_UNAVAIL) or (counts.get(d) or {}).get(cst.K_AUTO_CONFIRM) for d in _dates_in(counts)): print(" 窗口内没有自决读数 (自决还没开)。") print("\n复核口径提醒: 本报告只陈述读数, 阈值调整要先回台账 006 到 011、015 写一条, " "不按单次复盘读数回调 (原则一)。") if __name__ == "__main__": main()