PMS 内部那一截——候选池 → 建仓/升仓命令 → 方案生成 → 指令 → 规则闸终检 → 择时分日配额

Makefile

@@ -22,7 +22,9 @@ DC       := docker compose $(PROFILES)
 RUN      := docker compose run --rm --no-deps pms-web
 
 .PHONY: help deploy deploy-local build up down ps logs test initdb check health \
-        probe changes industry ws-status rebuild rebuild-accept reset-ledger shell
+        probe changes industry ws-status rebuild rebuild-accept \
+        t-plan t-pre t-cmd t-plans t-mat t-dry t-tick t-ins t-book t-gate \
+        reset-ledger shell
 
 help:   ## 列出所有目标
 	@grep -hE '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
@@ -85,6 +87,43 @@ rebuild: ## 账本重建预检 (只读)。加 GO=1 才真的建账, FORCE=1 连
 rebuild-accept: ## 只看判收: 安全垫分布 / 批次账 / 行业集中度
 	$(RUN) python scripts/rebuild_ledger.py --accept
 
+# ---------------------------------------------------------------- 联测 A 段 (shadow 下逐跳)
+# 一步一停, 每步都能看清楚 —— 别开 beat, 那会让几个调度位同时动, 出了岔子分不清是谁干的。
+# 全程 PMS_DISPATCH_MODE=shadow: 指令照常生成、照常过规则闸、照常记账, **只是不写下游**。
+# 顺序就是生产里那几个调度位的顺序, 只是改成手动触发。
+A ?= http://127.0.0.1:38100
+J := python3 -m json.tool
+
+t-plan:    ## A1 拉候选池 (= 08:40 调度位)
+	@curl -s -X POST '$(A)/api/ops/plan-refresh' | $(J)
+
+t-pre:     ## A2 盘前准备: T+1 可卖重置 / 参考位 / 刹车结算 (= 08:50 调度位)
+	@curl -s -X POST '$(A)/api/ops/premarket' | $(J)
+
+t-cmd:     ## A3 命令轮询: 新命令 → 方案生成 → 状态机推进 (= 每分钟调度位)
+	@curl -s -X POST '$(A)/api/ops/plan-pending' | $(J)
+
+t-plans:   ## A4 看方案 (方案是"买什么、多少股、分几批", 还没成指令)
+	@curl -s '$(A)/api/plans?limit=50' | $(J)
+
+t-mat:     ## A5 方案 → 指令 (先记账后动作: 指令先落表)
+	@curl -s -X POST '$(A)/api/ops/materialize' | $(J)
+
+t-dry:     ## A6 择时试算 (dry_run: 规则闸判什么、今天该出多少, 一股都不发)
+	@curl -s -X POST '$(A)/api/ops/exec-tick?dry_run=true' | $(J)
+
+t-tick:    ## A7 真出手 (shadow 下 = 置 DISPATCHED 但不写下游, 你照着在 QMT 手工执行)
+	@curl -s -X POST '$(A)/api/ops/exec-tick' | $(J)
+
+t-ins:     ## A8 看指令与子单状态
+	@curl -s '$(A)/api/instructions?limit=50' | $(J)
+
+t-book:    ## A9 看账本 (持仓 / 批次 / 安全垫 / 行业)
+	@curl -s '$(A)/api/positions' | $(J)
+
+t-gate:    ## 看评审账本: 规则闸/研判闸拒了什么、为什么 (联测时最该盯的一张表)
+	@curl -s '$(A)/api/ledger?limit=50' | $(J)
+
 # ---------------------------------------------------------------- 危险操作 (要确认)
 reset-ledger: ## 清空账本重来 (影子期专用; 必须 CONFIRM=1)
 	@if [ "$(CONFIRM)" != "1" ]; then \

README.md

@@ -66,7 +66,7 @@ scripts/
   test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 65 例
   test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫   32 例
   test_batch8_units.py 榜单变化: 名册指纹/三种语义/尾部闸/落库往返   60 例
-  test_batch9_units.py 接管持仓前的成本价体检: 阻断判据/情形覆盖     23 例
+  test_batch9_units.py 成本价体检 + 连续不一致按日推进               29 例
   test_wiring.py       装配自检: 服务层→核心→落表 全链路 (内存桩)   58 例
   init_db.py           建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检)
   check_db.py          实机连通性与表结构自检 (需真实 .env)
@@ -215,6 +215,34 @@ make rebuild-accept     # 只看判收：安全垫分布 / 批次账 / 行业集
 
 判收里最硬的一条是**安全垫分布**：如果重建后每一只票的安全垫都是 0，那就是踩了这个坑——页面上每个数都合理，只有这一处露馅。行业集中度那一项顺带回答 `PMS_SECTOR_MAX_RATIO=40%` 在三级粒度下偏不偏松（分母与 `sizer.check_caps` 一致，是**组合持仓市值**不是总规模，两边用不同分母会得出两个都自称「行业集中度」的数）。
 
+**这道闸判不了来历。** 它体检的是数据可不可信（成本价合不合理、可用量对不对），判不了「这笔持仓是不是联调残留的测试数据」——那个信息不在数据里。2026-07-31 就撞到一次：`600000.SH 1100 股 @ 9.273` 成本价完全正常、闸放行了，但它正是 `QMT_SIDE_S3_CLOSEOUT.md` §5.1 问过对端、至今没答复的那笔联调残留。**建账的第一笔要人工确认来历**，闸只保证「数据本身没毛病」。
+
+另外：清 PMS 账本只是一半。对端 `trading_position` / ws 快照里那笔还在的话，下次日终结算或 `make rebuild GO=1` 会把它再认领回来——要清得两边一起清。
+
+## 联测 A 段：shadow 下走完整链路
+
+候选池 → 建仓/升仓命令 → 方案生成 → 指令 → 规则闸终检 → 择时分日配额，这一截**从来没有用真实候选池加真实账本跑过**，只有单测——而单测覆盖的是 planner / rule_gate / exec_timing 各自的算数，没覆盖过「真实候选池的形状撞上真实持仓的上下文」。
+
+`shadow` 正好是这一段的天然测试环境，不是妥协：指令照常生成、照常过规则闸、照常记账，**只是不写下游**。所以这一段不用碰任何开关，也不需要 ws 通道就绪。
+
+**别开 beat**——那会让几个调度位同时动，出了岔子分不清是谁干的。`Makefile` 里有一组 `t-*` 目标，就是把那几个调度位改成手动逐跳触发，顺序与生产一致：
+
+```bash
+make t-plan     # A1 拉候选池      (= 08:40 调度位)
+make t-pre      # A2 盘前准备      (= 08:50 调度位)
+# —— 这里在页面「命令台」下一条建仓或升仓命令 ——
+make t-cmd      # A3 命令轮询: 新命令 → 方案生成
+make t-plans    # A4 看方案: 买什么、多少股、分几批 (还没成指令)
+make t-mat      # A5 方案 → 指令 (先记账后动作)
+make t-dry      # A6 择时试算 (dry_run: 规则闸判什么、今天该出多少, 一股都不发)
+make t-tick     # A7 真出手 (shadow 下 = 置 DISPATCHED 但不写下游)
+make t-ins      # A8 看指令与子单
+make t-book     # A9 看账本
+make t-gate     # 随时: 规则闸/研判闸拒了什么、为什么
+```
+
+**`make t-gate` 是联测时最该盯的一张表。** 方案里的票被拦掉时，它是唯一能回答「为什么这只没进去」的地方——单看方案和指令只能看到「少了几只」，看不到是撞了单股上限、一手不可行、行业集中度，还是不追高。
+
 ## 自主提议的分流（设计 §6 / §7）
 
 动作引擎每分钟扫一遍持仓，产出 FILL / ADD / DCA / TRIM 四类候选，然后依次过三道：
@@ -235,7 +263,7 @@ make rebuild-accept     # 只看判收：安全垫分布 / 批次账 / 行业集
 
 ## 已实现 / 待开发
 
-**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**榜单变化提示**（名册快照 + 新进/掉榜/档位升降/覆盖翻转/名次跳变，持仓票单列，榜尾截断噪音闸）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 327 例**。
+**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**榜单变化提示**（名册快照 + 新进/掉榜/档位升降/覆盖翻转/名次跳变，持仓票单列，榜尾截断噪音闸）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 333 例**。
 
 ### 下一步（按可动工顺序）
 

app/core/recon.py

@@ -334,6 +334,34 @@ def recon_severity(consecutive_days: int, alarm_days: int = 3) -> str:
     return SEV_ERROR if n >= int(alarm_days) else SEV_WARN
 
 
+def advance_streak(prev_streak, prev_ymd, today_ymd, has_diff: bool) -> dict:
+    """连续不一致的**按日**推进。返回 {streak, ymd, changed}。
+
+    2026-07-31 修: 原来是每调一次 `reconcile()` 就 +1。而盘中轻对账**每分钟**调一次, 于是
+    设计里「连续 3 日不一致 → ERROR 待人工」实际变成「连续 3 分钟」—— 一个持续存在的差异
+    三分钟就升到 ERROR, 日报关注区还会写出「连续 175 日」这种数 (项目 07-14 才开工)。
+    假警报天天响, 真告警就被埋掉了; 这跟 `_capped` 拿 counts 比返回条数、榜尾闸不区分截断
+    噪音, 是同一类毛病的第三个面。
+
+    两条规矩:
+      1. **同一天内反复对账不重复计数** —— 判据是 `today_ymd != prev_ymd`。手工点五次
+         「对账」也只算一天。
+      2. **只有权威那一趟才推进** (调用方保证: `apply_fix=True` 的那次)。盘中轻对账是
+         `apply_fix=False` 的只读观察, 不该写状态 —— 它照常读当前 streak 报 severity,
+         那本来就是"截至上一次日结"的事实。
+
+    差异消失时立即归零, 不必等到下一天。
+    """
+    prev_streak = int(prev_streak or 0)
+    prev_ymd = int(prev_ymd or 0)
+    today_ymd = int(today_ymd or 0)
+    if not has_diff:
+        return {"streak": 0, "ymd": today_ymd, "changed": prev_streak != 0}
+    if today_ymd and today_ymd == prev_ymd:
+        return {"streak": prev_streak, "ymd": prev_ymd, "changed": False}
+    return {"streak": prev_streak + 1, "ymd": today_ymd, "changed": True}
+
+
 def summarize_recon(diffs: list, consecutive_days: int = 0, alarm_days: int = 3) -> dict:
     kinds = {}
     for d in diffs or []:

app/services/ledger_service.py

@@ -33,6 +33,8 @@ CF_FEE, CF_CALIBRATE = "FEE", "CALIBRATE"   # pms_cash_flow.kind
 CURSOR_KEY = "PMS_REPLAY_CURSOR"
 CURSOR_ALL = "ALL"          # 游标设成这个值 = 显式要求从头全量回放 (见 _seed_cursor)
 STREAK_KEY = "PMS_RECON_STREAK"
+# 该 streak 最后一次推进是哪个交易日 —— 同一天内反复对账不重复计数的判据
+STREAK_YMD_KEY = "PMS_RECON_STREAK_YMD"
 LIVE_INSTR = ("DISPATCHED", "JUDGE_PASSED", "RULE_PASSED")
 
 
@@ -613,10 +615,19 @@ def reconcile(*, apply_fix: bool = True, force: bool = False) -> dict:
                                      for r in ds["rows"]])
     out["diffs"] = diffs
 
+    # 连续不一致计数 —— **按交易日推进, 且只有权威那一趟才写**。
+    # 盘中轻对账 (apply_fix=False) 每分钟跑一次, 让它推进的话「连续 3 日」就成了「连续
+    # 3 分钟」, 日报还会写出「连续 175 日」这种数。详见 rc.advance_streak 的注释。
     streak = param_store.get_int(STREAK_KEY, 0)
-    streak = streak + 1 if diffs else 0
+    if apply_fix:
-    # 必须走 ParamStore 写入: 直接写库不会失效缓存, 会导致连续天数一直读到旧值
+        st = rc.advance_streak(streak, param_store.get_int(STREAK_YMD_KEY, 0),
-    param_store.set_param(STREAK_KEY, streak, "system")
+                               int(datetime.now().strftime("%Y%m%d")), bool(diffs))
+        if st["changed"] or st["ymd"] != param_store.get_int(STREAK_YMD_KEY, 0):
+            # 必须走 ParamStore 写入: 直接写库不会失效缓存, 连续天数会一直读到旧值
+            param_store.set_param(STREAK_KEY, st["streak"], "system")
+            param_store.set_param(STREAK_YMD_KEY, st["ymd"], "system")
+        streak = st["streak"]
+    out["streak"] = streak
     out["severity"] = rc.recon_severity(streak if diffs else 0,
                                         param_store.get_int("PMS_RECON_ALARM_DAYS", 3))
 

scripts/run_tests.py

@@ -13,9 +13,9 @@
   test_batch6_units.py  ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账    (65 例)
   test_batch7_units.py  上游选股计划: 解析/新鲜度/候选筛选/取数守卫      (32 例)
   test_batch8_units.py  榜单变化: 名册指纹/三种语义/尾部闸/落库往返      (60 例)
-  test_batch9_units.py  接管持仓前的成本价体检: 阻断判据/情形覆盖        (23 例)
+  test_batch9_units.py  成本价体检 + 连续不一致按日推进                  (29 例)
   test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩)      (58 例)
-                                                                  共 327 例
+                                                                  共 333 例
 任一子集失败即整体失败 (退出码 1)。
 """
 import os

scripts/test_batch9_units.py

@@ -226,6 +226,60 @@ def t_d3():
     assert len(set(vs)) == len(vs)
 
 
+# ================================================================ [E] 连续不一致按日推进
+# 2026-07-31 实机暴露: 日报关注区写出「连续 175 日不一致」, 而项目 07-14 才开工。
+# 原因是每调一次 reconcile() 就 +1, 而盘中轻对账每分钟调一次 —— 设计里「连续 3 日 → ERROR
+# 待人工」实际成了「连续 3 分钟」。假警报天天响, 真告警就被埋掉。
+from app.core import recon as rc                                               # noqa: E402
+
+
+@case("[E1] 同一天内反复对账不重复计数")
+def t_e1():
+    s = rc.advance_streak(0, 0, 20260731, True)
+    assert s["streak"] == 1 and s["changed"] is True
+    for _ in range(5):                       # 手工点五次「对账」
+        s = rc.advance_streak(s["streak"], s["ymd"], 20260731, True)
+    assert s["streak"] == 1, "同一天点几次都只算一天 —— 否则三分钟就升 ERROR"
+    assert s["changed"] is False
+
+
+@case("[E2] 跨交易日才 +1")
+def t_e2():
+    s = rc.advance_streak(1, 20260731, 20260801, True)
+    assert s["streak"] == 2 and s["ymd"] == 20260801
+    s = rc.advance_streak(s["streak"], s["ymd"], 20260803, True)   # 跨周末
+    assert s["streak"] == 3
+
+
+@case("[E3] 差异消失立刻归零, 不必等下一天")
+def t_e3():
+    s = rc.advance_streak(7, 20260731, 20260731, False)
+    assert s["streak"] == 0 and s["changed"] is True and s["ymd"] == 20260731
+
+
+@case("[E4] 本来就是 0 且没差异 → 什么都没变, 不必写库")
+def t_e4():
+    assert rc.advance_streak(0, 20260731, 20260731, False)["changed"] is False
+
+
+@case("[E5] 三日门槛与 severity 对得上")
+def t_e5():
+    assert rc.recon_severity(0) == rc.SEV_OK
+    assert rc.recon_severity(1) == rc.SEV_WARN and rc.recon_severity(2) == rc.SEV_WARN
+    assert rc.recon_severity(3) == rc.SEV_ERROR
+    # 走满三个交易日才该到 ERROR —— 这条串起来验, 免得两边各改一半
+    s = {"streak": 0, "ymd": 0}
+    for i, d in enumerate((20260731, 20260801, 20260803), start=1):
+        s = rc.advance_streak(s["streak"], s["ymd"], d, True)
+        assert rc.recon_severity(s["streak"]) == (rc.SEV_ERROR if i >= 3 else rc.SEV_WARN)
+
+
+@case("[E6] 缺日期时退化成每次都推进, 但不会把已有计数弄丢")
+def t_e6():
+    s = rc.advance_streak(2, 0, 0, True)     # 两个 ymd 都拿不到
+    assert s["streak"] == 3, "判不了是不是同一天就按保守走(照常推进), 别把计数清零"
+
+
 def main():
     import logging
     logging.disable(logging.CRITICAL)