From f2d6b37fd2372e97f43262c1cd659300971fe44c Mon Sep 17 00:00:00 2001 From: zlt Date: Mon, 14 Sep 2026 10:16:46 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=82=E5=AF=9F=E8=AF=BB=E6=95=B0=E5=8C=85?= =?UTF-8?q?=EF=BC=9A=E5=90=88=E8=AE=AE=E5=88=A4=E5=AE=9A=E8=90=BD=E8=A1=A8?= =?UTF-8?q?=20pms=5Fconsensus=5Fstat=EF=BC=8C=E5=8A=A0=E5=A4=8D=E6=A0=B8?= =?UTF-8?q?=E8=84=9A=E6=9C=AC=E4=B8=8E=E6=97=A5=E6=8A=A5=E5=B0=8F=E8=8A=82?= =?UTF-8?q?=EF=BC=88=E5=8F=B0=E8=B4=A6013=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 每轮扫描把合议相关的判定归类落表，供台账006到011复核，之前这些数只活在 扫描返回值与页面即时快照里，复核日期到了拿不出数。 做了什么： - 新表 pms_consensus_stat（第21张），一天一票一类一行，重复记到只加次数。 - 动作引擎给合议跳过项（跳过分fund/bear、观察wait_tech）与增持门（gate_block 带fill/add/dca）打类别标签，只在对应开关开着的分支里打，关掉逐字回旧。 - consensus_stats.record_round 按检查点归类落表：转空实弹随时写，路由与装配两类 只在检查点写，试算不写；映射重建时 record_map_cover 记覆盖只数与逐只无读数。 - 复核脚本 scripts/consensus_review.py 与 make consensus-review 打出台账要的四个 读数；日报加合议观察小节。 - 新参数 PMS_CONSENSUS_STAT_TIMES 控制检查点（默认0935,1030,1330,1445）。 哨兵：DDL表数20改21；record_round进SOFT_FAIL；第31批登记进SUITES与例数表。 开发机全量单测840例 ALL SUITES PASS。未在155部署（收盘后另行审批）。 Co-Authored-By: Claude Opus 4.8 --- Makefile | 6 +- app/core/action_engine.py | 23 +- app/repo/consensus_stat_repo.py | 68 ++++++ app/services/consensus_stats.py | 194 ++++++++++++++++ app/services/ledger_service.py | 7 + app/services/param_store.py | 4 + app/services/proposal_service.py | 28 ++- app/services/tech_service.py | 10 + ddl_pms_v1.sql | 21 ++ docs/复盘决定台账.md | 18 ++ scripts/consensus_review.py | 180 +++++++++++++++ scripts/run_tests.py | 15 +- scripts/test_batch10_units.py | 1 + scripts/test_batch31_units.py | 366 +++++++++++++++++++++++++++++++ scripts/test_batch6_units.py | 3 +- 15 files changed, 933 insertions(+), 11 deletions(-) create mode 100644 app/repo/consensus_stat_repo.py create mode 100644 app/services/consensus_stats.py create mode 100644 scripts/consensus_review.py create mode 100644 scripts/test_batch31_units.py diff --git a/Makefile b/Makefile index 9e716d5..2b19be8 100644 --- a/Makefile +++ b/Makefile @@ -27,7 +27,8 @@ RUN := docker compose run --rm --no-deps pms-web .PHONY: help deploy deploy-local build up down ps logs test stale initdb check health \ probe changes industry ws-status rebuild rebuild-accept \ watch loop purge-dead-orders t-pool t-plan t-pre t-issue t-cmd t-cmds \ - t-catalog t-plans t-mat t-dry t-tick t-ins t-book t-gate reset-ledger shell + t-catalog t-plans t-mat t-dry t-tick t-ins t-book t-gate reset-ledger shell \ + consensus-review help: ## 列出所有目标 @grep -hE '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \ @@ -61,6 +62,9 @@ test: ## 全部单测 (先查镜像新旧再跑; 应输出 ALL SUITES PASS) @$(MAKE) --no-print-directory stale $(RUN) python scripts/run_tests.py +consensus-review: ## 合议观察读数复核 (只读; 加 DAYS=7 看更多天, 默认 3)。台账 006-011 复核用 + $(RUN) python scripts/consensus_review.py --days $(or $(DAYS),3) + stale: ## 容器里的代码跟工作树对不对得上 (git pull 之后没 build 的话这里会喊) @host=$$(python3 scripts/code_fingerprint.py 2>/dev/null || echo UNKNOWN); \ img=$$($(RUN) python scripts/code_fingerprint.py 2>/dev/null | tr -d '\r' | tail -1); \ diff --git a/app/core/action_engine.py b/app/core/action_engine.py index e123ab7..4e668e3 100644 --- a/app/core/action_engine.py +++ b/app/core/action_engine.py @@ -52,6 +52,15 @@ BUY, SELL = "buy", "sell" # 页面把它渲染成「等技术面开口」。与既有的 would / deny 并列, 走 disposition_snapshot 那条路。 DISP_WAIT_TECH = "wait_tech" +# 观察读数包 (2026-09-14, 台账 013): 合议相关跳过项打的类别标签, 值即 pms_consensus_stat.kind。 +# consensus_stats.record_round 认这些标签落表。**只在合议开着 (con 非 None) 或增持门开着 +# (cblocks 非 None) 的分支里打** —— 开关关掉时那些分支一行都不执行, tag 键不出现, 逐字回旧成立。 +# open_skip 分两类: 没有买方评析 → fund; 方向看空等其余原因 → bear (按路由原因文字分)。 +TAG_OPEN_SKIP_FUND = "open_skip_fund" # 合议判跳过, 因没有买方评析 +TAG_OPEN_SKIP_BEAR = "open_skip_bear" # 合议判跳过, 其余原因 (方向看空) +TAG_OPEN_WAIT_TECH = "open_wait_tech" # 合议判观察 (等技术面开口/转向) +TAG_GATE_BLOCK = "gate_block" # 持仓增持门拦下 (detail 里带 gate_kind: fill/add/dca) + # 同一轮里出现多条减持时留哪一条 (2026-09-03): 数字小的优先。用户自己设的目标价到价排在 # 系统按规则算出来的保垫减仓前面 —— 人已经说了到价就清, 这一轮就不该再自作主张先卖一部分。 # 只在同一只票的同一轮里比较, 不影响不同票, 也不影响下一轮。 @@ -676,7 +685,11 @@ def scan(*, positions: list, params: dict, market: dict, skip=None, if cblocks and action not in SELL_SIDE_ACTIONS: gate_why = consensus_gate_why(cblocks, action) if gate_why: - skipped.append({"ts_code": code, "action": action, "why": gate_why}) + # 观察读数包: 打 gate_block 标签, detail 记被拦的是哪一类增持腿 (fill/add/dca)。 + # 只在 cblocks 为真 (增持门开着) 才走到这里, 开关关掉时不打 tag, 逐字回旧。 + skipped.append({"ts_code": code, "action": action, "why": gate_why, + "tag": TAG_GATE_BLOCK, + "detail": {"gate_kind": str(action).lower()}}) continue try: c = fn(p, params, mkt) @@ -910,12 +923,16 @@ def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float, # 处置词打 wait_tech, 页面据此显示「等技术面开口」。交人与放行落到后面正常产出候选那条路。 con = c.get("consensus") if consensus_on else None if con and con.get("route") == "跳过": + # tag 按路由原因分两类 (观察读数包): 没有买方评析归 fund, 方向看空等其余归 bear。 + _rr = con.get("route_reason") or "" skipped.append({"ts_code": code, "action": A_OPEN, - "why": con.get("route_reason") or "三源合议判为跳过"}) + "why": _rr or "三源合议判为跳过", + "tag": TAG_OPEN_SKIP_FUND if "买方评析" in _rr else TAG_OPEN_SKIP_BEAR}) continue if con and con.get("route") == "观察": skipped.append({"ts_code": code, "action": A_OPEN, "disp": DISP_WAIT_TECH, - "why": con.get("route_reason") or "三源合议判为观察，等技术面开口"}) + "why": con.get("route_reason") or "三源合议判为观察，等技术面开口", + "tag": TAG_OPEN_WAIT_TECH}) continue if slots <= 0: skipped.append({"ts_code": code, "action": A_OPEN, diff --git a/app/repo/consensus_stat_repo.py b/app/repo/consensus_stat_repo.py new file mode 100644 index 0000000..45a90d3 --- /dev/null +++ b/app/repo/consensus_stat_repo.py @@ -0,0 +1,68 @@ +# -*- coding: utf-8 -*- +"""pms_consensus_stat 单表访问 (2026-09-14 观察读数包, 台账 013)。 + +三源合议观察读数的落表与回看。**严格单表访问**: 每个函数只碰 pms_consensus_stat 一张表。 +落表走 execute_many + ON DUPLICATE KEY UPDATE —— 同一 (北京日期, 代码, 类别) 重复记到时 +只把 rounds 加一、刷新 reason/detail/last_at, 不新增行 (一天一票一类一行)。 +回看按 (stat_date, kind) 索引取某段日期的行, 供复核脚本与日报聚合。 +""" +from __future__ import annotations + +from app.db.session import execute_many, fetch_all + +# 落表的业务列 (与 ddl_pms_v1.sql 的 pms_consensus_stat 对齐)。 +# rounds 不在插入列里 —— 插入时用 DDL 的 DEFAULT 1, 重复时在更新子句里 +1。 +_COLS = ("stat_date", "ts_code", "kind", "reason", "detail", "first_at", "last_at") + + +def upsert_many(rows: list) -> int: + """批量落观察读数。rows 每项是 {列名: 值}, 至少含 stat_date / ts_code / kind。 + 同 (日, 码, 类) 重复即累加次数 (rounds+1), 不新增行。返回受影响行数。""" + rows = [r for r in (rows or []) + if r.get("stat_date") and r.get("ts_code") and r.get("kind")] + if not rows: + return 0 + payload = [{c: r.get(c) for c in _COLS} for r in rows] + cols = ", ".join(_COLS) + vals = ", ".join(f":{c}" for c in _COLS) + # ON DUPLICATE 子句只用 VALUES(列), **绝不带绑定参数 (:col)** —— 2026-09-11 真机踩过: + # pymysql 的 executemany 对 INSERT ... ON DUPLICATE 做多行合并, 只展开 VALUES 子句的 + # 占位符, UPDATE 子句里的 :col 不展开却仍算参数, 批量时参数错位报 1064。 + # first_at 有意不进更新子句 —— 首次记到的时刻要保住; rounds 在库里自增, 不从外面传。 + updates = ("reason = VALUES(reason), detail = VALUES(detail), " + "last_at = VALUES(last_at), rounds = rounds + 1") + return execute_many( + f"INSERT INTO pms_consensus_stat ({cols}) VALUES ({vals}) " + f"ON DUPLICATE KEY UPDATE {updates}", payload) + + +def list_range(date_from: int, date_to: int, kinds=None) -> list: + """取 [date_from, date_to] (含两端) 的观察读数行, 升序。kinds 非空时只取这些类别。 + kinds 的 IN 占位符手动展开 (照 tech_repo.history_multi)。""" + p = {"a": int(date_from), "b": int(date_to)} + where = "stat_date >= :a AND stat_date <= :b" + ks = [k for k in (kinds or []) if k] + if ks: + keys = [] + for i, k in enumerate(ks): + keys.append(f":k{i}") + p[f"k{i}"] = k + where += f" AND kind IN ({', '.join(keys)})" + return fetch_all( + f"SELECT * FROM pms_consensus_stat WHERE {where} " + f"ORDER BY stat_date ASC, kind ASC, ts_code ASC", p) + + +def count_by_kind(date_from: int, date_to: int) -> dict: + """[date_from, date_to] 内按 (日期, 类别) 的计数, 返回 {stat_date: {kind: {rows, rounds}}}。 + rows 是行数 (几只票), rounds 是累计记到的次数 (同一票记了几轮)。""" + rows = fetch_all( + "SELECT stat_date, kind, COUNT(*) AS n_rows, SUM(rounds) AS n_rounds " + "FROM pms_consensus_stat WHERE stat_date >= :a AND stat_date <= :b " + "GROUP BY stat_date, kind", {"a": int(date_from), "b": int(date_to)}) + out: dict = {} + for r in rows: + d = int(r["stat_date"]) + out.setdefault(d, {})[r["kind"]] = { + "rows": int(r["n_rows"] or 0), "rounds": int(r["n_rounds"] or 0)} + return out diff --git a/app/services/consensus_stats.py b/app/services/consensus_stats.py new file mode 100644 index 0000000..6e895cc --- /dev/null +++ b/app/services/consensus_stats.py @@ -0,0 +1,194 @@ +# -*- coding: utf-8 -*- +"""三源合议观察读数落表 (2026-09-14 观察读数包, 台账 013)。 + +把每轮扫描里合议相关的判定归类落进 pms_consensus_stat, 让台账 006 到 011 的复核有据。 +一天一票一类一行, 重复记到只加 rounds。落表**只加不改行为**: 失败只记警告、返回 ok 为假, +绝不拦扫描 (登记进 test_batch10 的 SOFT_FAIL)。 + +类别 kind (与 pms_consensus_stat.kind、规格书附录甲一致): + open_skip_fund 新建仓因没有买方评析被合议判跳过 —— 动作引擎 scan_open 打 tag + open_skip_bear 新建仓因方向看空 (等) 被合议判跳过 —— 动作引擎 scan_open 打 tag + open_wait_tech 合议判观察 (等技术面开口/转向) —— 动作引擎 scan_open 打 tag + open_wait_confirm 等盘中确认 (盘中确认包, 后续) —— 动作引擎打 tag + open_breakout 盘中收口突破视为开口 (盘中确认包, 后续) —— 盘中确认包 + open_confirm 合议判交人 —— 从 out["consensus_seen"] 生成 + open_pass 合议判放行 —— 从 out["consensus_seen"] 生成 + fund_noread 基本面无读数 —— 从 out["consensus_seen"] 生成 + timing_noread 择时无读数 —— 从 out["consensus_seen"] 生成 + gate_block 持仓增持门拦下 (detail 带 gate_kind) —— 动作引擎 scan 打 tag + tech_exit 转空离场实弹 (执行或入队, 未被闸拒) —— 从 scanned["candidates"] 生成 + tech_noread 相关票无当日技术面读数 —— 映射重建时 (record_map_cover) + map_cover 映射覆盖 (ts_code=*, detail 记 codes/states) —— 映射重建时 (record_map_cover) + +检查点 (PMS_CONSENSUS_STAT_TIMES, 默认 0935,1030,1330,1445): 合议路由与装配那两类 +(上面前十项里非 tech_exit 的) 只在检查点及其下一分钟才写, 一天最多四次; tech_exit 随时写; +map_cover 与 tech_noread 由映射重建时写 (调度位本就一天一两次, 不另设检查点)。dry_run 一律不写。 +""" +from __future__ import annotations + +import json +import logging +from datetime import datetime + +from app.core import action_engine as ae +from app.repo import consensus_stat_repo +from app.services import param_store + +logger = logging.getLogger("pms.consensus_stat") + +STAT_TIMES_KEY = "PMS_CONSENSUS_STAT_TIMES" + +# 动作引擎打在跳过项上的 tag 即 kind, 一处定义 (core), 这里直接引用, 不另写一份免得分叉。 +K_OPEN_SKIP_FUND = ae.TAG_OPEN_SKIP_FUND +K_OPEN_SKIP_BEAR = ae.TAG_OPEN_SKIP_BEAR +K_OPEN_WAIT_TECH = ae.TAG_OPEN_WAIT_TECH +K_GATE_BLOCK = ae.TAG_GATE_BLOCK +# 从 consensus_seen / scanned / 映射重建派生的类别 (动作引擎不打这些 tag)。 +K_OPEN_CONFIRM = "open_confirm" +K_OPEN_PASS = "open_pass" +K_FUND_NOREAD = "fund_noread" +K_TIMING_NOREAD = "timing_noread" +K_TECH_EXIT = "tech_exit" +K_TECH_NOREAD = "tech_noread" +K_MAP_COVER = "map_cover" + + +# ================================================================ 时刻与行 +def _parse_times(times_str) -> list: + """"0935,1030" → [575, 630] (自零点起的分钟)。认不出的段跳过, 不抛。""" + out = [] + for t in str(times_str or "").split(","): + t = t.strip() + if len(t) == 4 and t.isdigit(): + hh, mm = int(t[:2]), int(t[2:]) + if 0 <= hh < 24 and 0 <= mm < 60: + out.append(hh * 60 + mm) + return out + + +def _at_checkpoint(now, times_str) -> bool: + """now 落在任一检查点或它的下一分钟 (扫描每分钟一跳, 两分钟窗口保证记到一次)。 + 空串 → 无检查点, 路由与装配两类一律不写 (只留转空实弹)。""" + cur = now.hour * 60 + now.minute + return any(cur == cp or cur == cp + 1 for cp in _parse_times(times_str)) + + +def _row(day, code, kind, *, reason=None, detail=None, now=None) -> dict: + now = now or datetime.now() # 北京时间 (容器时钟), 与项目写记录口径一致 + return {"stat_date": int(day), "ts_code": code or "*", "kind": kind, + "reason": (str(reason)[:255] if reason else None), + "detail": (json.dumps(detail, ensure_ascii=False) if detail else None), + "first_at": now, "last_at": now} + + +def _flush(rows) -> dict: + """同一轮里同 (日, 码, 类) 只留一条 (一轮记一次, rounds 跨轮由库自增), 再批量 upsert。 + 写失败只记警告、返回 ok 为假 —— 观察读数是加不改, 绝不拦扫描 (SOFT_FAIL)。""" + dedup = {} + for r in rows: + dedup[(r["stat_date"], r["ts_code"], r["kind"])] = r + rows = list(dedup.values()) + if not rows: + return {"ok": True, "rows": 0} + try: + n = consensus_stat_repo.upsert_many(rows) + return {"ok": True, "rows": len(rows), "written": n} + except Exception as e: # noqa: BLE001 + logger.warning("[合议观察] 落表失败 (不影响扫描): %s", e) + return {"ok": False, "rows": len(rows), "error": f"{type(e).__name__}: {e}"} + + +# ================================================================ 逐类别取行 +def _seen_rows(out, day, now) -> list: + """从 out["consensus_seen"] 生成 open_pass / open_confirm 与 fund_noread / timing_noread。 + 紧凑记录由 proposal_service._attach_consensus 逐只候选追加 (ts_code/fund/tech/timing/direction/route/phase)。""" + rows = [] + for s in out.get("consensus_seen") or []: + code = s.get("ts_code") + if not code: + continue + route = s.get("route") + detail = {"fund": s.get("fund"), "tech": s.get("tech"), "timing": s.get("timing"), + "direction": s.get("direction"), "route": route, "phase": s.get("phase")} + if route == "放行": + rows.append(_row(day, code, K_OPEN_PASS, reason=s.get("direction"), + detail=detail, now=now)) + elif route == "交人": + rows.append(_row(day, code, K_OPEN_CONFIRM, reason=s.get("direction"), + detail=detail, now=now)) + if s.get("fund") == "无读数": + rows.append(_row(day, code, K_FUND_NOREAD, reason="基本面无读数", now=now)) + if s.get("timing") == "无读数": + rows.append(_row(day, code, K_TIMING_NOREAD, reason="择时无读数", now=now)) + return rows + + +def _skip_tag_rows(out, day, now) -> list: + """从 out["skipped"] 里带 tag 的项生成对应类别行 (tag 即 kind)。 + 覆盖 open_skip_* / open_wait_* / gate_block, 以及盘中确认包后续的 open_wait_confirm 等。""" + rows = [] + for s in out.get("skipped") or []: + tag = s.get("tag") + code = s.get("ts_code") + if not tag or not code: + continue + rows.append(_row(day, code, tag, reason=s.get("why"), detail=s.get("detail"), now=now)) + return rows + + +def _tech_exit_rows(out, scanned, day, now) -> list: + """从 scanned["candidates"] 里来源是 tech_exit 且没被闸拒的项生成 tech_exit 行。 + 确认转空清仓走 EXIT, 未确认减三分之一走 TRIM —— 据动作名记 confirm。""" + rows = [] + rej = {(r.get("ts_code"), r.get("action")) for r in (out.get("rejected") or [])} + for c in (scanned or {}).get("candidates") or []: + if c.get("source") != ae.SRC_TECH_EXIT: + continue + code, action = c.get("ts_code"), c.get("action") + if not code or (code, action) in rej: + continue + detail = {"action": action, "qty": c.get("qty"), "confirm": action == ae.A_EXIT} + rows.append(_row(day, code, K_TECH_EXIT, reason=c.get("reason"), detail=detail, now=now)) + return rows + + +# ================================================================ 对外入口 +def record_round(out, scanned, params, now=None) -> dict: + """一轮扫描的合议判定归类落表。调用点在 proposal_service.scan_and_route 末尾 (去重集写回之后)。 + + tech_exit 随时写 (稀有实弹); 路由与装配两类只在检查点写 (每分钟太密); dry_run 一律不写。 + 返回 {ok, rows, ...}; 失败只记警告不抛 —— 登记进 test_batch10 的 SOFT_FAIL, 调用点要接住。""" + now = now or datetime.now() + if not isinstance(out, dict) or out.get("dry_run"): + return {"ok": True, "rows": 0, "note": "dry_run 或无 out, 不落表"} + day = int(now.strftime("%Y%m%d")) # 北京日期 (容器时钟) + rows = list(_tech_exit_rows(out, scanned, day, now)) + times = param_store.get(STAT_TIMES_KEY, "0935,1030,1330,1445") + if _at_checkpoint(now, times): + rows += _seen_rows(out, day, now) + rows += _skip_tag_rows(out, day, now) + return _flush(rows) + + +def record_map_cover(*, relevant_codes, state_codes, now=None) -> dict: + """映射重建时写 map_cover 一行 (ts_code=*, detail 记相关票数与覆盖只数) 与逐只 tech_noread。 + 调用点在 tech_service.build_map 写完映射之后。失败只记警告、返回 ok 为假, 不拦建映射。""" + now = now or datetime.now() + day = int(now.strftime("%Y%m%d")) + rel = [c for c in (relevant_codes or []) if c] + have = set(state_codes or []) + rows = [_row(day, "*", K_MAP_COVER, + detail={"codes": len(rel), "states": len(have)}, now=now)] + for c in rel: + if c not in have: + rows.append(_row(day, c, K_TECH_NOREAD, reason="相关票无当日技术面读数", now=now)) + return _flush(rows) + + +def daily_summary(ymd) -> dict: + """某北京日期的按类别计数 (给运营日报的 consensus 小节)。取不到返回空 dict, 不抛。""" + try: + return consensus_stat_repo.count_by_kind(int(ymd), int(ymd)).get(int(ymd), {}) + except Exception as e: # noqa: BLE001 + logger.warning("[合议观察] 日报小节取数失败 (写空): %s", e) + return {} diff --git a/app/services/ledger_service.py b/app/services/ledger_service.py index 33f5fc0..3cb2b0f 100644 --- a/app/services/ledger_service.py +++ b/app/services/ledger_service.py @@ -1481,6 +1481,13 @@ def build_daily_report(ymd: int = None) -> dict: for x in v["held"]}, "recon": recon_state, } + # 合议观察小节 (2026-09-14 观察读数包, 台账 013): 当天按类别的计数。取不到写空, 不拖垮日报。 + try: + from app.services import consensus_stats + report["consensus"] = consensus_stats.daily_summary(ymd) + except Exception as e: # noqa: BLE001 + logger.warning("[日报] 合议观察小节取数失败 (写空): %s", e) + report["consensus"] = {} try: pms_repo.upsert_report(ymd, report) except Exception as e: diff --git a/app/services/param_store.py b/app/services/param_store.py index 92c86af..702f055 100644 --- a/app/services/param_store.py +++ b/app/services/param_store.py @@ -126,6 +126,10 @@ RUNTIME_EXTRA = { # 弱基本面试探仓的紧止盈自动挂载 (2026-09-11 工作包三 part 4, 台账 011): 基本面看空加技术面看多的试探仓次日自动挂。 "PMS_TECH_TIGHT_TRAIL_GIVEBACK": (0.03, float, "弱基本面试探仓紧止盈的回撤比例 (从高点回落这么多即全清; 默认百分之三)"), "PMS_TECH_TIGHT_TRAIL_TARGET": (0.08, float, "弱基本面试探仓紧止盈的硬目标 (浮盈到这么多直接全清; 默认百分之八)"), + # ── 2026-09-14 观察读数包 (台账 013): 合议判定落表 pms_consensus_stat 的检查点。 + # 四位时刻逗号分隔; consensus_stats.record_round 只在这些时刻及其下一分钟写路由与装配两类 + # (一天最多四次), 转空离场实弹随时写。设为空串即不写检查点行, 只留转空实弹。 + "PMS_CONSENSUS_STAT_TIMES": ("0935,1030,1330,1445", str, "合议观察读数落表的检查点 (四位时刻逗号分隔; 空串=只记转空实弹)"), } # **读不到时必须按"已暂停"处理的键 (fail-closed)。** diff --git a/app/services/proposal_service.py b/app/services/proposal_service.py index 5c47185..1f5d262 100644 --- a/app/services/proposal_service.py +++ b/app/services/proposal_service.py @@ -224,6 +224,17 @@ def scan_and_route(*, now=None, dry_run: bool = False) -> dict: _save_tech_exit_done(set(params.get("tech_exit_done") or set()) | _acted) except Exception as e: # noqa: BLE001 logger.warning("[技术面离场] 去重集写回失败 (下轮可能重评一次): %s", e) + # 三源合议观察读数落表 (2026-09-14 观察读数包, 台账 013): 把本轮合议相关的判定归类落表, + # 供台账 006 到 011 的复核。只加不改行为 —— 失败只记警告、不抛 (SOFT_FAIL); dry_run 内部会跳过。 + # 返回值接住 (test_batch10 [A1] 查丢弃返回值)。 + if not dry_run: + try: + from app.services import consensus_stats + rec = consensus_stats.record_round(out, scanned, params, now) + if not rec.get("ok"): + logger.warning("[合议观察] 本轮落表未全成: %s", rec.get("error")) + except Exception as e: # noqa: BLE001 + logger.warning("[合议观察] 落表出错 (不影响扫描): %s", e) out["ok"] = not out["errors"] return out @@ -356,7 +367,7 @@ def _scan_open(view, params, stock_params, skip, mkt, out) -> list: # 开关关掉整段不做, cands 原样进 scan_open —— 与接入前逐字相同。装配失败只记日志、不拦扫描 # (合议是加不改: 装不上就当没有合议, scan_open 里 con 为 None 自然退回旧路)。 if params.get("consensus_route"): - _attach_consensus(cands) + _attach_consensus(cands, out) res = ae.scan_open(candidates=cands, params=params, caps=portfolio.caps_ctx(view), room_amt=room, slots=slots, skip=skip) out["skipped"].extend(res["skipped"]) @@ -375,15 +386,19 @@ def _scan_open(view, params, stock_params, skip, mkt, out) -> list: return res["candidates"] -def _attach_consensus(cands) -> None: +def _attach_consensus(cands, out=None) -> None: """给每只候选装配三源合议, 挂三样到候选上 (原地改): consensus —— 合议块 (direction / votes / strength / reason / route / route_reason), scan_open 分流读它; consensus_blocks —— 四块意见 (fund / tech / timing / consensus), scan_open 定档时取 fund/tech 喂 advise_v2; consensus_hard —— 提议硬数字六键 (方案附录丁), 进评审账本与提议卡, 不进送研判名单。 批量取一次昨夜定性与技术面映射; 盘中转多留痕从候选自带的 sig_buy 取时刻。任何一步失败都不抛 —— - 装不上就当这只票没有合议 (scan_open 里 con 为 None 自然退回旧路), 合议是加不改。""" + 装不上就当这只票没有合议 (scan_open 里 con 为 None 自然退回旧路), 合议是加不改。 + + 每装配一只候选往 out["consensus_seen"] 追加一条紧凑记录 (观察读数包用, 也给候选处置快照用): + ts_code / fund / tech / timing / direction / route / phase。out 为 None 或缺该键时只装配不留痕。""" if not cands: return + seen = out.setdefault("consensus_seen", []) if isinstance(out, dict) else None try: codes = [c.get("ts_code") for c in cands if c.get("ts_code")] nightly = consensus_service.nightly_map(codes) @@ -402,6 +417,13 @@ def _attach_consensus(cands) -> None: c["consensus"] = blocks["consensus"] c["consensus_blocks"] = blocks c["consensus_hard"] = consensus_service.hard_keys(blocks) + if seen is not None: + con, fund, tech, tm = (blocks["consensus"], blocks["fund"], + blocks["tech"], blocks["timing"]) + seen.append({"ts_code": code, "fund": fund.get("stance"), + "tech": tech.get("stance"), "timing": tm.get("stance"), + "direction": con.get("direction"), "route": con.get("route"), + "phase": tech.get("phase")}) except Exception as e: # noqa: BLE001 logger.warning("[合议] 装配失败 %s (本票按无合议): %s", code, e) diff --git a/app/services/tech_service.py b/app/services/tech_service.py index c9164b4..942e8d8 100644 --- a/app/services/tech_service.py +++ b/app/services/tech_service.py @@ -358,6 +358,16 @@ def build_map(*, now=None) -> dict: out["error"] = f"映射写入失败: {r.get('error')}" out["states"] = len(states) out["by_stance"] = dict(Counter(s.get("stance") or "空" for s in states.values())) + # 观察读数包 (2026-09-14, 台账 013): 映射写完后记 map_cover 一行 (相关票数与覆盖只数) 与 + # 逐只 tech_noread —— 复核脚本的「技术面无读数占比」用 map_cover 算。失败只记警告, 不拦建映射。 + try: + from app.services import consensus_stats + cov = consensus_stats.record_map_cover(relevant_codes=codes, + state_codes=set(states.keys()), now=now) + if not cov.get("ok"): + logger.warning("[技术面] 映射覆盖落表未全成: %s", cov.get("error")) + except Exception as e: # noqa: BLE001 + logger.warning("[技术面] 映射覆盖落表出错 (不影响建映射): %s", e) return out diff --git a/ddl_pms_v1.sql b/ddl_pms_v1.sql index 028effe..9734e5a 100644 --- a/ddl_pms_v1.sql +++ b/ddl_pms_v1.sql @@ -456,3 +456,24 @@ CREATE TABLE IF NOT EXISTS pms_tech_daily ( UNIQUE KEY uk_date_code (data_date, ts_code), KEY idx_code_date (ts_code, data_date) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='全市场每日技术面读数 (保留 40 个交易日, 供回看判震荡与开口)'; + +-- 21. 三源合议观察读数 (2026-09-14 观察读数包, 台账 013) +-- 台账 006 到 011 的复核要量化读数, 但合议的判定原来只活在每分钟扫描的返回值与页面 +-- 即时快照里, 不落任何表, 复核日期到了拿不出数。这张表把每轮扫描里合议相关的判定落下来: +-- 一天一票一类一行 (uk_day_code_kind), 重复记到只加 rounds 不新增行, 一天最多写四次 +-- (检查点由 PMS_CONSENSUS_STAT_TIMES 控制; 转空离场实弹随时写)。复核脚本 +-- scripts/consensus_review.py 从这张表算台账要的四个读数。类别 kind 的含义见 +-- app/services/consensus_stats.py 顶部与规格书附录甲。 +CREATE TABLE IF NOT EXISTS pms_consensus_stat ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + stat_date INT NOT NULL COMMENT '北京日期 yyyymmdd', + ts_code VARCHAR(16) NOT NULL COMMENT '代码; 全局行 (如映射覆盖) 用 *', + kind VARCHAR(24) NOT NULL COMMENT '类别, 见 consensus_stats 顶部', + reason VARCHAR(255) NULL COMMENT '一句原因, 取跳过原因原文截断', + detail TEXT NULL COMMENT 'JSON: 三票/方向/强弱/相位/路由/盘中确认 等', + first_at DATETIME NOT NULL COMMENT '北京时间, 首次记到', + last_at DATETIME NOT NULL COMMENT '北京时间, 最近记到', + rounds INT NOT NULL DEFAULT 1 COMMENT '记到的次数', + UNIQUE KEY uk_day_code_kind (stat_date, ts_code, kind), + KEY idx_day_kind (stat_date, kind) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='三源合议观察读数 (台账复核底本, 一天一票一类一行)'; diff --git a/docs/复盘决定台账.md b/docs/复盘决定台账.md index 5cfd1df..b491e3c 100644 --- a/docs/复盘决定台账.md +++ b/docs/复盘决定台账.md @@ -187,3 +187,21 @@ **预期。** 策略票在 SAR 翻空的当天被止损线保护。弱基本面试探仓入场次日就有紧止盈兜底。 **复核日期。** 上线后每周复核一次，看紧止盈有没有过早止损、SAR 线有没有误杀。 + +## 013 · 2026-09-14 · 观察读数与复核口径 + +**改动。** 把每轮扫描里合议相关的判定落进新表 pms_consensus_stat，让台账 006 到 011 的复核有据。之前这些数只活在每分钟扫描的返回值与页面即时快照里，不落任何表，复核日期到了拿不出数。 + +落表一天一票一类一行，重复记到只加次数。类别有十来种：新建仓因没有买方评析或方向看空被跳过、等技术面开口、合议交人、合议放行、基本面无读数、择时无读数、增持门拦下、转空离场实弹、技术面无读数、映射覆盖。检查点由参数控制，一天最多写四次；转空离场实弹随时写；映射覆盖在早上重建映射时写。试算一律不写。 + +复核脚本 scripts/consensus_review.py 从这张表打四张表：逐日的无读数占比与只数、逐日的路由分布、转空离场逐条、合议方向与人工裁决对不上的逐只。命令是 make consensus-review，加 DAYS 看更多天。日报也加了一节合议观察。 + +**依据。** 没有读数就没有复核，没有复核就不能把这套东西部署到真实仓。这是评审认定的下一阶段第一优先级。落表只加不改行为：失败只记警告、不抛、不拦扫描；开关关掉时动作引擎那几处标签一行都不打，逐字回旧。 + +**口径。** 技术面无读数占比用映射覆盖那一行算，等于相关票数减覆盖只数再除以相关票数。增持门拦截次数用累计次数算，不是行数。转空离场的确认与否按动作名判，确认转空清仓走 EXIT，未确认减三分之一走 TRIM。合议与人工裁决的比对只看带合议六键且已被人裁决的提议，合议看多而人驳回、合议交人而人采纳的逐只列出。 + +**参数。** 落表检查点 PMS_CONSENSUS_STAT_TIMES 默认 0935,1030,1330,1445，四位时刻逗号分隔，页面可改。设为空串即不写检查点行，只留转空离场实弹。 + +**台账线（只作提示，调阈值先回台账）。** 技术面无读数占比应低于百分之三。每天等技术面开口应在一到十五只。转空退出每周零到三条。合议链上线第一天转空离场应为零。 + +**复核日期。** 观察读数包部署后的下一个交易日先跑一次 consensus-review 看有没有读数落下来。台账 006 到 009 的第一次正式复核在合议分流上线满三个交易日，也就是 09-17。 diff --git a/scripts/consensus_review.py b/scripts/consensus_review.py new file mode 100644 index 0000000..9acdc55 --- /dev/null +++ b/scripts/consensus_review.py @@ -0,0 +1,180 @@ +# -*- 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() diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 29de35c..ad98bb1 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -86,20 +86,28 @@ 策略票只看目标价按函数判(技术面转空对策略票不评)/跟踪止盈盘中 SAR 止损线 (09:45 后跌破缓冲全清、当日一次、没刷进不判、开关关掉不触发)/SAR 线刷新腿只动 自动跟踪止盈且无读数撤旧线/弱基本面试探仓紧止盈自动挂载(入场标记识别、回撤3%硬目标8%带SAR线、不占名额、冷却让路)/参数登记 (33 例) + test_batch31_units.py 观察读数包 (2026-09-14 台账 013): 动作引擎给合议跳过项打标签 + (跳过分 fund/bear、观察 wait_tech、增持门 gate_block 带 gate_kind, + 开关关掉不打标签)/record_round 归类落表 (检查点写路由与装配两类、 + 转空实弹随时写、试算不写、被闸拒不记、confirm 据动作名、空串检查点)/ + record_map_cover 记覆盖与逐只无读数/日报小节失败不拖垮/upsert 更新子句 + 只用 VALUES 列不带绑定参数且单表合规/参数登记与检查点窗口 (20 例) test_page_enum_guard.py 页面文案守卫 (静态扫描, 不连库不起浏览器): 枚举字段不许 直接印到页面上 / 判据码显示前必须剥前缀 / 不许把整个对象 打给交易员看 / 翻译兜底不许让英文码单独当句子 (1 例) test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) + 目标价到价必定入队 (档位 full 也不自动卖) + 用户设的止损价与目标价单独成列显示 (70 例) - 共 820 例 + 共 840 例 (总数按实跑逐批相加校正过两次: 曾写 649 是笔误, 实为 650; 09-03 先后加了同轮只发一条 - 减持与研究理由两键各一例, 到 652; 09-04 加了仅展示跳过原因与空候选说明各一例, 到 654; 又加了页面文案守卫一例, 到 655; 09-07 审查修复加了跨轮减持等五例, 到 660; 第二件低把握驳回交人一例, 到 661; 第三件逻辑状态接入第二十二批十五例, 到 676; 第四件安全边际整句透传一例, 到 677; 参考目标价一例, 到 678; 催化事件与定价状态透传一例, 到 679; 两个期限的头一例, 现为 680; 09-10 建议档位对齐第二十六批十七例到 697; 09-11 技术面接入工作包一第二十七批二十七例到 724; 三源合议工作包二纯逻辑第二十八批三十七例到 763; 接入下单链路第二十九批二十四例到 787; 工作包三离场纪律第三十批十八例到 805; part3 盘中 SAR 止损线补九例到 814; part4 弱基本面紧止盈补六例, 现为 820) + 减持与研究理由两键各一例, 到 652; 09-04 加了仅展示跳过原因与空候选说明各一例, 到 654; 又加了页面文案守卫一例, 到 655; 09-07 审查修复加了跨轮减持等五例, 到 660; 第二件低把握驳回交人一例, 到 661; 第三件逻辑状态接入第二十二批十五例, 到 676; 第四件安全边际整句透传一例, 到 677; 参考目标价一例, 到 678; 催化事件与定价状态透传一例, 到 679; 两个期限的头一例, 现为 680; 09-10 建议档位对齐第二十六批十七例到 697; 09-11 技术面接入工作包一第二十七批二十七例到 724; 三源合议工作包二纯逻辑第二十八批三十七例到 763; 接入下单链路第二十九批二十四例到 787; 工作包三离场纪律第三十批十八例到 805; part3 盘中 SAR 止损线补九例到 814; part4 弱基本面紧止盈补六例到 820; + 09-14 观察读数包第三十一批二十例, 现为 840) 任一子集失败即整体失败 (退出码 1)。 哨兵位置清单 (2026-09-03 抄录; 改了对应的东西就得来这些地方改断言, 断言不动就是漏了): DDL 表数 scripts/test_batch6_units.py 用例「DDL 文件本身体检通过」(约第 580 行), - 断言 `eq(len([... k == "table"]), 19)` (约第 593 行) —— 加表就 +1 + 断言 `eq(len([... k == "table"]), 21)` (约第 596 行) —— 加表就 +1 + (2026-09-11 tech_daily → 20; 2026-09-14 consensus_stat → 21) 调度位 scripts/test_wiring.py 用例「装配·调度表覆盖设计 §10 全部调度位」(约第 654 行), beat_schedule 名字集合与 pms.* 任务名两处 路由清单 scripts/test_wiring.py 用例「装配·API 路由齐全 (四块页面 + 运维)」(约第 634 行) @@ -147,6 +155,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py", "test_batch28_units.py", "test_batch29_units.py", "test_batch30_units.py", + "test_batch31_units.py", "test_page_enum_guard.py", "test_page_wiring_guard.py", "test_wiring.py"] diff --git a/scripts/test_batch10_units.py b/scripts/test_batch10_units.py index 4bdfc01..d142744 100644 --- a/scripts/test_batch10_units.py +++ b/scripts/test_batch10_units.py @@ -56,6 +56,7 @@ SOFT_FAIL = { "dispatch": {"dispatcher"}, # → {"ok": False, "error": ...} "cancel": {"dispatcher"}, # → {"ok": False, "error": ...} "update_order": {"qmt_repo"}, # → 影响行数, 0 = 出口表里没这行 + "record_round": {"consensus_stats"}, # → {"ok": False, "error": ...} 观察读数落表 } # 确实可以丢的调用点写在这里, **必须带理由**。空着比乱加强。 diff --git a/scripts/test_batch31_units.py b/scripts/test_batch31_units.py new file mode 100644 index 0000000..a5a4686 --- /dev/null +++ b/scripts/test_batch31_units.py @@ -0,0 +1,366 @@ +# -*- coding: utf-8 -*- +"""观察读数包 (2026-09-14 方案第三节, 台账 013)。全部离线, 不连库。 + +把每轮扫描里合议相关的判定落进 pms_consensus_stat, 让台账 006 到 011 的复核有据。 + A 动作引擎打标签: 合议判跳过分 fund/bear、判观察打 wait_tech、增持门打 gate_block; + 开关关掉时标签不出现 (逐字回旧); tag 值与 consensus_stats 认的 kind 对齐。 + B 归类落表 record_round: 检查点写路由与装配两类、转空实弹随时写、试算不写、被闸拒不记、 + 确认与否据动作名; daily_summary 取数失败返回空不抛。 + C 映射覆盖 record_map_cover: 记 map_cover 与逐只 tech_noread。 + D 仓库 upsert: 更新子句只用 VALUES 列不带绑定参数 (pymysql 批量陷阱), SQL 单表合规。 + E 参数登记与检查点窗口。 +""" +import json +import os +import sys +import traceback +from datetime import datetime + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.core import action_engine as ae # noqa: E402 +from app.services import consensus_stats as cst # noqa: E402 +from app.repo import consensus_stat_repo as csr # noqa: E402 + +RESULTS = [] + + +def case(name): + def deco(fn): + RESULTS.append((name, fn)) + return fn + return deco + + +# ================================================================ 公共桩 +def _copen(code, route, rr, direction="中性"): + """一条挂了合议块的候选 (scan_open 用)。""" + return {"ts_code": code, "score": 100, "rank": 1, + "consensus": {"route": route, "route_reason": rr, "direction": direction}} + + +def _caps(): + return {"max_names": 20, "sector_source_ready": True} + + +def _pos_con(code, direction): + """一条挂了合议块的持仓行 (scan 增持门用)。""" + return {"ts_code": code, "total_qty": 6000, "avail_qty": 6000, "price": 10.0, + "avg_cost": 10.0, "base_qty": 6000, "add_qty": 0, "dca_qty": 0, + "cushion_pct": 0.0, "cushion_peak": 0.0, "frozen_reason": "NONE", + "price_ok": True, "target_pct": 0.06, + "consensus_blocks": {"consensus": {"direction": direction}, + "tech": {"stance": direction}, "fund": {"stance": direction}}} + + +def _sp(**kw): + p = {"cushion_solid": 0.03, "trim_peak": 0.06, "trim_giveback": 0.5, + "dca_triggers": (-0.08, -0.15), "dca_deep_confirm": -0.15, "dca_max_ratio": 0.5, + "no_chase_ma5": 0.06, "build_window_tdays": 10, "fill_max_loss": -0.03, + "batch_split": (0.5, 0.25, 0.25), "stock_target_default": 0.06, "scale": 2_000_000, + "logic_state_route": False, "tech_gate_increase": True} + p.update(kw) + return p + + +class _patched: + """把 record_round 会碰的两处外部依赖换成桩: param_store.get 返回固定检查点串, + consensus_stat_repo.upsert_many 捕获落表行 (不连库)。with 结束自动还原。""" + + def __init__(self, times="0935,1030,1330,1445"): + self.times = times + self.captured = [] + + def __enter__(self): + self._o_get = cst.param_store.get + self._o_up = cst.consensus_stat_repo.upsert_many + cst.param_store.get = lambda k, d=None: (self.times if k == cst.STAT_TIMES_KEY + else self._o_get(k, d)) + cst.consensus_stat_repo.upsert_many = lambda rows: (self.captured.extend(rows) + or len(rows)) + return self.captured + + def __exit__(self, *a): + cst.param_store.get = self._o_get + cst.consensus_stat_repo.upsert_many = self._o_up + return False + + +def _by_kind(rows): + out = {} + for r in rows: + out.setdefault(r["kind"], []).append(r) + return out + + +def _sample_out(): + return {"dry_run": False, "rejected": [], + "consensus_seen": [ + {"ts_code": "600000.SH", "fund": "看多", "tech": "看多", "timing": "看多", + "direction": "看多", "route": "放行", "phase": "趋势多"}, + {"ts_code": "600001.SH", "fund": "无读数", "tech": "中性", "timing": "看多", + "direction": "看多", "route": "交人", "phase": None}, + {"ts_code": "600010.SH", "fund": "看多", "tech": "看多", "timing": "无读数", + "direction": "看多", "route": "放行", "phase": "趋势多"}], + "skipped": [ + {"ts_code": "600002.SH", "action": "OPEN", "why": "没有买方评析", + "tag": "open_skip_fund"}, + {"ts_code": "600003.SH", "action": "OPEN", "why": "方向看空（…）", + "tag": "open_skip_bear"}, + {"ts_code": "600004.SH", "action": "OPEN", "why": "等技术面开口", + "tag": "open_wait_tech", "disp": "wait_tech"}, + {"ts_code": "600005.SH", "action": "FILL", "why": "技术面看空", + "tag": "gate_block", "detail": {"gate_kind": "fill"}}, + {"ts_code": "600009.SH", "why": "宏观偏热闸生效"}]} # 无 tag, 应被忽略 + + +def _sample_scanned(): + return {"candidates": [ + {"ts_code": "600006.SH", "action": "EXIT", "source": "tech_exit", "qty": 1000, + "reason": "确认转空清仓"}, + {"ts_code": "600007.SH", "action": "TRIM", "source": "tech_exit", "qty": 300, + "reason": "未确认减三分之一"}, + {"ts_code": "600008.SH", "action": "FILL", "source": "engine", "qty": 500}]} + + +# ================================================================ A 动作引擎打标签 +@case("A scan_open·合议判跳过·没有买方评析 → open_skip_fund") +def _(): + res = ae.scan_open(candidates=[_copen("600000.SH", "跳过", "没有买方评析")], + params={"consensus_route": True}, caps=_caps(), + room_amt=1e6, slots=10, skip={}) + assert len(res["skipped"]) == 1 + assert res["skipped"][0]["tag"] == "open_skip_fund" + assert res["skipped"][0]["ts_code"] == "600000.SH" + + +@case("A scan_open·合议判跳过·方向看空 → open_skip_bear") +def _(): + res = ae.scan_open(candidates=[_copen("600000.SH", "跳过", "方向看空（基本面看空、技术面看空、择时中性）")], + params={"consensus_route": True}, caps=_caps(), + room_amt=1e6, slots=10, skip={}) + assert res["skipped"][0]["tag"] == "open_skip_bear" + + +@case("A scan_open·合议判观察 → open_wait_tech 且 disp wait_tech") +def _(): + res = ae.scan_open(candidates=[_copen("600000.SH", "观察", "等技术面开口")], + params={"consensus_route": True}, caps=_caps(), + room_amt=1e6, slots=10, skip={}) + assert res["skipped"][0]["tag"] == "open_wait_tech" + assert res["skipped"][0]["disp"] == ae.DISP_WAIT_TECH + + +@case("A scan_open·合议开关关掉·跳过项不带 tag (开着才有, 逐字回旧)") +def _(): + cand = _copen("600000.SH", "跳过", "没有买方评析") + on = ae.scan_open(candidates=[dict(cand)], params={"consensus_route": True}, + caps=_caps(), room_amt=1e6, slots=10, skip={}) + # 关掉合议: con 为 None, 那只候选不再被合议分流; 用 skip 在 eval_open 之前接住它, 免起重设备 + off = ae.scan_open(candidates=[dict(cand)], params={"consensus_route": False}, + caps=_caps(), room_amt=1e6, slots=10, + skip={("600000.SH", "OPEN"): "已在途"}) + assert on["skipped"][0].get("tag") == "open_skip_fund" + assert all("tag" not in s for s in off["skipped"]) + + +@case("A scan·增持门拦下 → gate_block 且 detail.gate_kind ∈ {fill,add,dca}") +def _(): + res = ae.scan(positions=[_pos_con("600000.SH", "看空")], params=_sp(tech_gate_increase=True), + market={"600000.SH": {}}, skip={}) + gb = [s for s in res["skipped"] if s.get("tag") == "gate_block"] + assert gb, "增持门看空应打 gate_block 标签" + assert all((s.get("detail") or {}).get("gate_kind") in ("fill", "add", "dca") for s in gb) + + +@case("A scan·增持门开关关掉 → 无 gate_block tag (逐字回旧)") +def _(): + res = ae.scan(positions=[_pos_con("600000.SH", "看空")], params=_sp(tech_gate_increase=False), + market={"600000.SH": {}}, skip={}) + assert all(s.get("tag") != "gate_block" for s in res["skipped"]) + + +@case("A 哨兵·动作引擎 tag 与 consensus_stats 认的 kind 对齐") +def _(): + assert cst.K_OPEN_SKIP_FUND == ae.TAG_OPEN_SKIP_FUND == "open_skip_fund" + assert cst.K_OPEN_SKIP_BEAR == ae.TAG_OPEN_SKIP_BEAR == "open_skip_bear" + assert cst.K_OPEN_WAIT_TECH == ae.TAG_OPEN_WAIT_TECH == "open_wait_tech" + assert cst.K_GATE_BLOCK == ae.TAG_GATE_BLOCK == "gate_block" + + +# ================================================================ B 归类落表 record_round +@case("B record_round·检查点写全部类别 (放行/交人/无读数/跳过/观察/增持门/转空实弹)") +def _(): + with _patched() as cap: + r = cst.record_round(_sample_out(), _sample_scanned(), {}, datetime(2026, 9, 15, 9, 35)) + assert r["ok"] + bk = _by_kind(cap) + assert {x["ts_code"] for x in bk["open_pass"]} == {"600000.SH", "600010.SH"} + assert {x["ts_code"] for x in bk["open_confirm"]} == {"600001.SH"} + assert {x["ts_code"] for x in bk["fund_noread"]} == {"600001.SH"} + assert {x["ts_code"] for x in bk["timing_noread"]} == {"600010.SH"} + assert {x["ts_code"] for x in bk["open_skip_fund"]} == {"600002.SH"} + assert {x["ts_code"] for x in bk["open_skip_bear"]} == {"600003.SH"} + assert {x["ts_code"] for x in bk["open_wait_tech"]} == {"600004.SH"} + assert {x["ts_code"] for x in bk["gate_block"]} == {"600005.SH"} + assert {x["ts_code"] for x in bk["tech_exit"]} == {"600006.SH", "600007.SH"} + assert all(x["stat_date"] == 20260915 for x in cap) # 北京日期 + + +@case("B record_round·gate_block 带上 detail.gate_kind; 无 tag 的跳过项被忽略") +def _(): + with _patched() as cap: + cst.record_round(_sample_out(), _sample_scanned(), {}, datetime(2026, 9, 15, 10, 30)) + bk = _by_kind(cap) + assert json.loads(bk["gate_block"][0]["detail"])["gate_kind"] == "fill" + assert "600009.SH" not in {x["ts_code"] for x in cap} # 无 tag 的宏观闸跳过不落 + + +@case("B record_round·转空实弹据动作名记 confirm (EXIT 真 / TRIM 假)") +def _(): + with _patched() as cap: + cst.record_round(_sample_out(), _sample_scanned(), {}, datetime(2026, 9, 15, 9, 35)) + ex = {x["ts_code"]: json.loads(x["detail"]) for x in _by_kind(cap)["tech_exit"]} + assert ex["600006.SH"]["confirm"] is True and ex["600006.SH"]["qty"] == 1000 + assert ex["600007.SH"]["confirm"] is False and ex["600007.SH"]["qty"] == 300 + + +@case("B record_round·非检查点只写转空实弹 (路由与装配两类不写)") +def _(): + with _patched() as cap: + cst.record_round(_sample_out(), _sample_scanned(), {}, datetime(2026, 9, 15, 10, 0)) + bk = _by_kind(cap) + assert set(bk) == {"tech_exit"} + assert {x["ts_code"] for x in bk["tech_exit"]} == {"600006.SH", "600007.SH"} + + +@case("B record_round·试算 (dry_run) 一律不写") +def _(): + out = {**_sample_out(), "dry_run": True} + with _patched() as cap: + r = cst.record_round(out, _sample_scanned(), {}, datetime(2026, 9, 15, 9, 35)) + assert cap == [] and r["rows"] == 0 + + +@case("B record_round·转空离场被闸拒的不记 tech_exit") +def _(): + out = {"dry_run": False, "consensus_seen": [], "skipped": [], + "rejected": [{"ts_code": "600006.SH", "action": "EXIT"}]} + with _patched() as cap: + cst.record_round(out, _sample_scanned(), {}, datetime(2026, 9, 15, 10, 0)) + assert {x["ts_code"] for x in _by_kind(cap).get("tech_exit", [])} == {"600007.SH"} + + +@case("B record_round·空串检查点 → 只写转空实弹") +def _(): + with _patched(times="") as cap: + cst.record_round(_sample_out(), _sample_scanned(), {}, datetime(2026, 9, 15, 9, 35)) + assert set(_by_kind(cap)) == {"tech_exit"} + + +@case("B daily_summary·取数失败返回空 dict 不抛") +def _(): + o = cst.consensus_stat_repo.count_by_kind + + def _boom(a, b): + raise RuntimeError("db down") + cst.consensus_stat_repo.count_by_kind = _boom + try: + assert cst.daily_summary(20260915) == {} + finally: + cst.consensus_stat_repo.count_by_kind = o + + +# ================================================================ C 映射覆盖 +@case("C record_map_cover·记 map_cover (ts_code=*) 与逐只 tech_noread") +def _(): + with _patched() as cap: + cst.record_map_cover(relevant_codes=["600000.SH", "600001.SH", "600002.SH"], + state_codes={"600000.SH", "600001.SH"}, + now=datetime(2026, 9, 15, 8, 40)) + bk = _by_kind(cap) + mc = bk["map_cover"][0] + assert mc["ts_code"] == "*" + assert json.loads(mc["detail"]) == {"codes": 3, "states": 2} + assert {x["ts_code"] for x in bk["tech_noread"]} == {"600002.SH"} + + +# ================================================================ D 仓库 upsert +@case("D upsert_many·更新子句只用 VALUES 列不带绑定参数; SQL 单表合规") +def _(): + from app.db.session import assert_single_table + grabbed = {} + o = csr.execute_many + csr.execute_many = lambda sql, payload, **kw: (grabbed.update(sql=sql, payload=payload) + or len(payload)) + try: + n = csr.upsert_many([{"stat_date": 20260915, "ts_code": "600000.SH", "kind": "open_pass", + "reason": "看多", "detail": None, + "first_at": datetime.now(), "last_at": datetime.now()}]) + finally: + csr.execute_many = o + sql = grabbed["sql"] + upd = sql.split("ON DUPLICATE KEY UPDATE", 1)[1] + assert ":" not in upd, "更新子句不许有绑定参数 (:col) —— 2026-09-11 pymysql 批量陷阱" + assert "VALUES(reason)" in upd and "rounds = rounds + 1" in upd + assert_single_table(sql) # 不抛即单表合规 (守卫会摘掉 ON DUPLICATE 尾巴) + assert n == 1 + + +@case("D upsert_many·缺主键三件套的行被丢掉") +def _(): + grabbed = {} + o = csr.execute_many + csr.execute_many = lambda sql, payload, **kw: (grabbed.update(payload=payload) + or len(payload)) + try: + # 只有一行齐全, 另两行分别缺 kind 与 ts_code + n = csr.upsert_many([ + {"stat_date": 20260915, "ts_code": "600000.SH", "kind": "open_pass", + "first_at": datetime.now(), "last_at": datetime.now()}, + {"stat_date": 20260915, "ts_code": "600001.SH"}, + {"stat_date": 20260915, "kind": "open_pass"}]) + finally: + csr.execute_many = o + assert n == 1 and len(grabbed["payload"]) == 1 + + +# ================================================================ E 参数与检查点窗口 +@case("E 参数·PMS_CONSENSUS_STAT_TIMES 登记, 默认四检查点, 类型 str") +def _(): + from app.services import param_store as ps + assert cst.STAT_TIMES_KEY == "PMS_CONSENSUS_STAT_TIMES" + dv, typ, _desc = ps.RUNTIME_EXTRA[cst.STAT_TIMES_KEY] + assert dv == "0935,1030,1330,1445" and typ is str + + +@case("E _at_checkpoint·命中检查点及其下一分钟, 其余与空串不命中") +def _(): + t = "0935,1030,1330,1445" + assert cst._at_checkpoint(datetime(2026, 9, 15, 9, 35), t) is True + assert cst._at_checkpoint(datetime(2026, 9, 15, 9, 36), t) is True # 下一分钟 + assert cst._at_checkpoint(datetime(2026, 9, 15, 9, 37), t) is False + assert cst._at_checkpoint(datetime(2026, 9, 15, 14, 45), t) is True + assert cst._at_checkpoint(datetime(2026, 9, 15, 9, 35), "") is False # 空串无检查点 + + +def main(): + ok = 0 + for name, fn in RESULTS: + try: + fn() + ok += 1 + print(" ok " + name) + except Exception: + print(" FAIL " + name) + traceback.print_exc() + print("-" * 60) + if ok == len(RESULTS): + print("ALL PASS (%d cases)" % ok) + return 0 + print("FAILED %d/%d" % (len(RESULTS) - ok, len(RESULTS))) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/test_batch6_units.py b/scripts/test_batch6_units.py index 1ba8a23..093351e 100644 --- a/scripts/test_batch6_units.py +++ b/scripts/test_batch6_units.py @@ -592,7 +592,8 @@ def run(): # (2026-08-18: +pms_macro_signal 宏观信号日快照 → 18) # (2026-08-28: +pms_nav_daily 每日净值快照·公示导出 → 19) # (2026-09-11: +pms_tech_daily 全市场每日技术面读数 → 20) - eq(len([1 for k, _, _ in stmts if k == "table"]), 20) + # (2026-09-14: +pms_consensus_stat 三源合议观察读数·观察读数包 → 21) + eq(len([1 for k, _, _ in stmts if k == "table"]), 21) eq(len([1 for k, _, _ in stmts if k == "seed"]), 1) @case("通道三表的 SQL 全部单表合规")