194 lines
8.9 KiB
Python
194 lines
8.9 KiB
Python
# -*- 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)
|
|
# 增持门看只数 (当天被拦的持仓只数)。检查点是两分钟窗口, 累计次数 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 ""))
|
|
|
|
print("\n复核口径提醒: 本报告只陈述读数, 阈值调整要先回台账 006 到 011 写一条, "
|
|
"不按单次复盘读数回调 (原则一)。")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|