# -*- 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, 调用点要接住。 params 暂未用到 (检查点从参数中心直接读), 保留是为对齐规格书签名, 也留给后续包备用。""" 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")) # 先清当天旧的 tech_noread 行 (2026-09-14 评审第三条): 早上 06:30 与 08:40 各重建一次映射, # 06:30 缺读数、08:40 补上的票, 早上那一行本会留着让逐只无读数偏高。先清再写即以最新一次 # 重建为准。map_cover 行走 upsert 覆盖不受影响, 占比一直准。清失败只记警告, 不拦覆盖行落表。 try: consensus_stat_repo.delete_day_kind(day, K_TECH_NOREAD) except Exception as e: # noqa: BLE001 logger.warning("[合议观察] 清当日 tech_noread 失败 (逐只无读数可能偏高, 占比不受影响): %s", e) 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 {}