diff --git a/Makefile b/Makefile index ce8df3e..dd21057 100644 --- a/Makefile +++ b/Makefile @@ -126,9 +126,13 @@ rebuild-accept: ## 只看判收: 安全垫分布 / 批次账 / 行业集中度 # 实现 B = PMS 自己算 (分日配额 / 分笔 / VWAP / 回踩 / 不追高 / 14:45 兜底) —— 已实现 # 所以跑 A 段不需要动决策系统的任何逻辑, 它这一段本来就不参与。 A ?= http://127.0.0.1:38100 -# --no-ensure-ascii 是必须的: 不加的话中文全被转义成 \uXXXX, 每条 reason 都得自己解码, -# 而这些接口的 reason 恰恰是最该看的东西 (Python 3.9+) -J := python3 -m json.tool --no-ensure-ascii +# 中文必须原样出来: `json.tool` 默认把中文转义成 \uXXXX, 而这些接口的 reason 恰恰是最该 +# 看的东西。**不能用 `json.tool --no-ensure-ascii`** —— 这条管道跑在**宿主机**上不是容器里, +# 宿主机的 python3 可能低于 3.9 (2026-08-03 实机: unrecognized arguments)。下面这行任何 +# python3 都认。解析失败时原样透传, 免得把错误信息也一起吞了。 +J := python3 -c 'import sys,json;\ +d=sys.stdin.read();\ +sys.stdout.write(json.dumps(json.loads(d),ensure_ascii=False,indent=2)+"\n" if d.strip().startswith(("{","[")) else d)' # t-pool 与 t-plans 是**两个完全不同的东西**, 别被名字带偏 (原名 t-plan/t-plans 只差一个 s, # 实测把人绕进去过): diff --git a/README.md b/README.md index 1991657..51dc2c4 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,7 @@ scripts/ test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 32 例 test_batch8_units.py 榜单变化: 名册指纹/三种语义/尾部闸/落库往返 60 例 test_batch9_units.py 成本价体检 / 对账按日推进 / 行业闸 / 取整记账 49 例 - test_batch10_units.py 静默失败专项: 关键路径不许丢返回值 + 八条实例 40 例 + test_batch10_units.py 静默失败专项: 关键路径不许丢返回值 + 八条实例 46 例 test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 58 例 init_db.py 建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检) check_db.py 实机连通性与表结构自检 (需真实 .env) @@ -304,7 +304,7 @@ make t-gate # 随时: 规则闸/研判闸拒了什么、为什么 ## 已实现 / 待开发 -**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**榜单变化提示**（名册快照 + 新进/掉榜/档位升降/覆盖翻转/名次跳变，持仓票单列，榜尾截断噪音闸）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 393 例**。 +**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**榜单变化提示**（名册快照 + 新进/掉榜/档位升降/覆盖翻转/名次跳变，持仓票单列，榜尾截断噪音闸）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 399 例**。 ### 静默失败专项（2026-07-31，八条已修） @@ -346,6 +346,10 @@ make t-gate # 随时: 规则闸/研判闸拒了什么、为什么 `make rebuild` 在「下游一只持仓都没有」时印 `[FAIL]` 并退出码 2，make 报 `Error 2`——可账户空 + 账本空 = 两边一致，**这是终态不是故障**，压根不需要重建这一步。预检现在给出 `verdict`：`READY` / `EMPTY` / `NOT_READY` / `NO_SOURCE`，`EMPTY` 退 0 并明说「没有需要接管的持仓，重建这一步跳过」。只有一种 `EMPTY` 仍然要拦：**账户空而账本非空**，那是顺序反了，得先 `reset_ledger` 清账本，别让对账拿空集去核销已有持仓。 +**第四种变体：测试桩比真依赖宽松，于是单测替真依赖打掩护。** 2026-08-03 切 ws 当天撞到：`executor.run_tick` 给 `pms_repo.update_instruction` 多传了一个 `limit_price=None`，真 repo 没这个形参 → `TypeError` → 被外层 `except Exception` 吞成 `out["errors"]` 里的一条。后果是**单子已经发到 QMT 了，而本端一个字没记**——父指令停在 `PROPOSED`、`children` 空、`dispatch_ref` 空，当日配额恒按 0 算，下一跳会拿同一个子单号 `_D01` 再发一次（出口表唯一索引拦住之后就彻底卡死）。 + +单测全程没看见，因为 `FakeRepo.update_instruction` 写的是 `def update_instruction(self, iid, **kw)`——什么关键字都收。**桩比真依赖宽松，等于单测在替真依赖打掩护。** `test_batch10_units.py` 的 **[M] 组**守这一条：从**源码**（AST）读 `pms_repo` 的真实签名，逐个比对 `FakeRepo` 的同名方法，桩收 `**kw` 而真 repo 是固定签名就判失败。注意必须从源码读——`install_fakes` 已经把 `pms_repo` 的函数替换成桩的绑定方法了，`getattr(pms_repo, name)` 拿到的是桩自己，那样这条守卫会永远绿（第一版就是这么写错的，守卫本身也会静默失效）。`make watch` 里另有一段常驻交叉检查：出口表有委托而父指令还停在 `PROPOSED`、或 `children` 条数跟出口表对不上，直接顶在屏幕上。 + **第三种变体最阴，是「没跑长得像通过」。** 源码是打进镜像的，`docker compose run pms-web python scripts/run_tests.py` 跑的是**镜像里那份**。于是有这么一条路径： ``` diff --git a/app/services/executor.py b/app/services/executor.py index ce4c108..e63f37c 100644 --- a/app/services/executor.py +++ b/app/services/executor.py @@ -236,9 +236,23 @@ def run_tick(*, now=None, dry_run: bool = False) -> dict: prog.setdefault("dispatched_at", now.strftime("%Y-%m-%d %H:%M:%S")) prog["last_decision"] = {"at": now.strftime("%H:%M"), "action": et.ACT_FIRE, "reason": d["reason"]} - pms_repo.update_instruction(ins["instruction_id"], status=ST_DISPATCHED, - limit_price=None, dispatch_ref=res["ref"], - progress=prog) + # 这里曾经多传了一个 limit_price=None (2026-08-03 实机): 真 repo 的 + # update_instruction 没有这个形参 → TypeError → 被本函数外层的 + # `except Exception` 吞成 out["errors"] 里的一条。后果是**单子已经发到 QMT + # 了, 而本端一个字没记**: 父指令停在 PROPOSED、children 空、dispatch_ref 空, + # 于是当日配额恒按 0 算, 下一跳会拿同一个子单号 _D01 再发一次 —— 出口表唯一 + # 索引拦住之后就彻底卡死。单测全程没看见, 因为 FakeRepo.update_instruction + # 是 `**kw` 全收, 比真依赖宽松 (见 test_batch10 的 [M] 组)。 + r = pms_repo.update_instruction(ins["instruction_id"], status=ST_DISPATCHED, + dispatch_ref=res["ref"], progress=prog) + if not r: + # 单子已经出去了, 这里回滚不了; 但绝不能让它悄悄过去 + logger.error("[出手] %s 已下发到通道, 但父指令没更新到任何行 —— " + "本端与出口表将不一致, 请核对 pms_instruction", + ins["instruction_id"]) + out["errors"].append( + f"{ins['instruction_id']} 已下发但父指令未更新 (影响 0 行) —— " + f"出口表与本端不一致") pms_repo.insert_ledger(ts_code=code, action=ins.get("action"), arbiter="rule", verdict="PASS", price_at=day_ctx.get("price") or 0, hard_numbers={**gate["hard_numbers"], "limit": d["limit_price"], diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 5bcc2b5..db1b281 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -14,9 +14,9 @@ test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 (32 例) test_batch8_units.py 榜单变化: 名册指纹/三种语义/尾部闸/落库往返 (60 例) test_batch9_units.py 成本价体检 / 对账按日推进 / 行业闸 / 取整记账 (49 例) - test_batch10_units.py 静默失败专项: 关键路径不许丢返回值 + 八条实例 (40 例) + test_batch10_units.py 静默失败专项: 关键路径不许丢返回值 + 八条实例 (46 例) test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (58 例) - 共 393 例 + 共 399 例 任一子集失败即整体失败 (退出码 1)。 """ import os diff --git a/scripts/test_batch10_units.py b/scripts/test_batch10_units.py index 9e148d3..c5c3f93 100644 --- a/scripts/test_batch10_units.py +++ b/scripts/test_batch10_units.py @@ -830,6 +830,86 @@ def _(): assert "code_fingerprint" in src, "run_tests 要打印代码指纹, 好识别跑的是哪份代码" +# ================================================================ +# [M] 测试桩不许比真依赖宽松 +# ================================================================ +@case("[M1] FakeRepo 的方法签名不许比 pms_repo 宽松 (`**kw` 全收会替真依赖打掩护)") +def _(): + import ast + import inspect + from test_wiring import FakeRepo + + # 2026-08-03 实机: executor 给 update_instruction 多传了一个 limit_price=None, + # 真 repo 没这个形参 → TypeError → 被外层 except 吞掉 → **单子已发到 QMT 而本端 + # 一个字没记**。单测全程没看见, 因为桩是 `def update_instruction(self, iid, **kw)`, + # 什么关键字都收。桩比真依赖宽松, 等于单测在替真依赖打掩护 —— 这一条守住它。 + # + # 真 repo 的签名**必须从源码读**, 不能 getattr(pms_repo, name): install_fakes 已经 + # 把 pms_repo 的函数替换成桩的绑定方法了, 那样等于拿桩跟自己比, 这条守卫会永远绿。 + # (第一版就是这么写错的 —— 守卫本身也会静默失效。) + real = {} + path = os.path.join(ROOT, "app", "repo", "pms_repo.py") + with open(path, encoding="utf-8") as f: + for node in ast.parse(f.read(), path).body: + if isinstance(node, ast.FunctionDef) and not node.name.startswith("_"): + a = node.args + real[node.name] = { + "kw": [x.arg for x in (a.args + a.kwonlyargs)], + "var_kw": a.kwarg is not None} + assert "update_instruction" in real, "没能从源码解析出 pms_repo 的签名" + + bad = [] + for name, r in sorted(real.items()): + fake = getattr(FakeRepo, name, None) + if not callable(fake): + continue + fp = inspect.signature(fake).parameters + fake_var_kw = any(p.kind is inspect.Parameter.VAR_KEYWORD for p in fp.values()) + if fake_var_kw and not r["var_kw"]: + bad.append(f"{name}: 桩收 **kw 而真 repo 是固定签名 " + f"({', '.join(r['kw'])}) —— 调用方多传一个参数, " + f"真机抛 TypeError 而单测照样绿") + assert not bad, ("测试桩比真 repo 宽松, 单测会盖住真机才暴露的错:\n " + + "\n ".join(bad)) + + +@case("[M2] 扫描器本身有效: 造一个宽松桩, 必须被 M1 那套判据抓出来") +def _(): + import inspect + class _Real: + def f(self, a, *, b=None): + return 1 + class _Fake: + def f(self, a, **kw): + return 1 + rp = inspect.signature(_Real.f).parameters + fp = inspect.signature(_Fake.f).parameters + assert any(p.kind is inspect.Parameter.VAR_KEYWORD for p in fp.values()) + assert not any(p.kind is inspect.Parameter.VAR_KEYWORD for p in rp.values()) + + +@case("[M3] 下发成功但父指令没更新到任何行 → 必须冒进 errors, 不许静默") +def _(): + import ast + path = os.path.join(ROOT, "app", "services", "executor.py") + with open(path, encoding="utf-8") as f: + src = f.read() + # 只看真实调用的实参, 不看注释 —— 注释里提这个词是在解释这个坑, 不该被自己抓 + bad = [] + for node in ast.walk(ast.parse(src, path)): + if not isinstance(node, ast.Call): + continue + fn = node.func + if isinstance(fn, ast.Attribute) and fn.attr == "update_instruction": + names = {k.arg for k in node.keywords if k.arg} + extra = names - {"status", "exec_qty", "exec_avg_price", "dispatch_ref", + "progress"} + if extra: + bad.append(f"第 {node.lineno} 行传了 pms_repo 不认的参数 {sorted(extra)}") + assert not bad, "update_instruction 调用点参数不合真 repo 签名:\n " + "\n ".join(bad) + assert "已下发但父指令未更新" in src, "下发成功但更新影响 0 行时没有留出口" + + # ================================================================ def main(): ok = fail = 0 diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index d8be8ad..cc86be9 100644 --- a/scripts/test_wiring.py +++ b/scripts/test_wiring.py @@ -52,7 +52,11 @@ class FakeRepo: def next_command_seq(self, ymd): return len(self.commands) + 1 - def insert_command(self, **kw): + def insert_command(self, *, command_id, cmd_class, cmd_type, ts_code, params, status, + issued_by=None, note=None, progress=None): + kw = {"command_id": command_id, "cmd_class": cmd_class, "cmd_type": cmd_type, + "ts_code": ts_code, "params": params, "status": status, + "issued_by": issued_by, "note": note, "progress": progress} self.commands[kw["command_id"]] = { "id": len(self.commands) + 1, "command_id": kw["command_id"], "cmd_class": kw["cmd_class"], "cmd_type": kw["cmd_type"], "ts_code": kw["ts_code"], @@ -221,7 +225,13 @@ class FakeRepo: "note": note}) return 1 - def insert_instruction(self, **kw): + def insert_instruction(self, *, instruction_id, origin_type, origin_id, ts_code, action, + side, qty, limit_price=None, window_tdays=3, status="PROPOSED", + progress=None): + kw = {"instruction_id": instruction_id, "origin_type": origin_type, + "origin_id": origin_id, "ts_code": ts_code, "action": action, "side": side, + "qty": qty, "limit_price": limit_price, "window_tdays": window_tdays, + "status": status, "progress": progress or {}} kw.setdefault("exec_qty", 0) kw["created_at"] = kw["updated_at"] = datetime.now() self.instructions[kw["instruction_id"]] = kw @@ -237,11 +247,18 @@ class FakeRepo: def get_instruction(self, iid): return self.instructions.get(iid) - def update_instruction(self, iid, **kw): + # **签名必须跟真 repo 一模一样**, 不许写成 `**kw`。2026-08-03 实机: executor 多传了 + # 一个 limit_price=None, 真 repo 抛 TypeError、单子已发出而本端没记账, 而这里 `**kw` + # 全收所以单测一路绿。桩比真依赖宽松 = 单测在替真依赖打掩护。 + # test_batch10 的 [M] 组会扫这一条, 别改回 **kw。 + def update_instruction(self, iid, *, status=None, exec_qty=None, exec_avg_price=None, + dispatch_ref=None, progress=None): i = self.instructions.get(iid) if not i: return 0 - for k, v in kw.items(): + for k, v in (("status", status), ("exec_qty", exec_qty), + ("exec_avg_price", exec_avg_price), ("dispatch_ref", dispatch_ref), + ("progress", progress)): if v is not None: i[k] = v i["updated_at"] = datetime.now() @@ -253,7 +270,11 @@ class FakeRepo: i["exec_qty"] = int(i.get("exec_qty") or 0) + int(qty) return 1 - def insert_proposal(self, **kw): + def insert_proposal(self, *, proposal_id, ts_code, action, qty, hard_numbers, expire_at, + judge_verdict=None, judge_reason=None, status="WAIT_USER"): + kw = {"proposal_id": proposal_id, "ts_code": ts_code, "action": action, "qty": qty, + "hard_numbers": hard_numbers, "expire_at": expire_at, + "judge_verdict": judge_verdict, "judge_reason": judge_reason, "status": status} kw["status"] = kw.get("status", "WAIT_USER") kw["hard_numbers"] = kw.get("hard_numbers") or {} self.proposals[kw["proposal_id"]] = kw @@ -275,7 +296,12 @@ class FakeRepo: def expire_proposals(self, now=None): return 0 - def insert_ledger(self, **kw): + def insert_ledger(self, *, ts_code, action, arbiter, verdict, price_at, hard_numbers=None, + failed_checks=None, reason=None, ref_id=None): + kw = {"ts_code": ts_code, "action": action, "arbiter": arbiter, "verdict": verdict, + "price_at": price_at, "hard_numbers": hard_numbers or {}, + "failed_checks": failed_checks or [], "reason": reason, "ref_id": ref_id, + "decided_at": datetime.now()} self.ledger.append(kw) return 1 @@ -1152,17 +1178,20 @@ def _(): "deadline": "2026-07-27", "status": "EXECUTING", "reason": "收利润"}]) fake.insert_instruction(instruction_id="INS_F", origin_type="plan", origin_id="P_F", ts_code="600000.SH", action="TRIM", side="sell", qty=2000, - status="DISPATCHED", exec_qty=2000, + status="DISPATCHED", progress={"deadline": "2026-07-27", "is_command": True, "children": []}) + # exec_qty 不是 insert_instruction 的形参 (真 repo 没有), 造数据直接改桩里的字段 + fake.instructions["INS_F"]["exec_qty"] = 2000 fake.insert_instruction(instruction_id="INS_P", origin_type="plan", origin_id="P_F", ts_code="600000.SH", action="TRIM", side="sell", qty=2000, - status="DISPATCHED", exec_qty=500, + status="DISPATCHED", progress={"deadline": "2026-07-20", "is_command": True, "children": []}) + fake.instructions["INS_P"]["exec_qty"] = 500 fake.insert_instruction(instruction_id="INS_A2", origin_type="proposal", origin_id="PR1", ts_code="600000.SH", action="ADD", side="buy", qty=1000, - status="DISPATCHED", exec_qty=0, + status="DISPATCHED", progress={"deadline": "2026-07-20", "is_command": False, "children": []}) r = executor.sweep_windows(now=_dt(2026, 7, 27, 15, 10)) @@ -1317,7 +1346,7 @@ def _(): fake = install_fakes(prices={"600000.SH": 10.0}) fake.insert_instruction(instruction_id="INS_R", origin_type="plan", origin_id="P1", ts_code="600000.SH", action="OPEN", side="buy", qty=100, - price_hint=10.0, status="DISPATCHED") + status="DISPATCHED") # 真单: 出口表有行、父指令不带 SMOKE_ fake.qmt.enqueue_order(instruction_id="INS_R-1", parent_id="INS_R", ts_code="600000.SH", side="buy", qty=100, limit_price=10.0, valid_until=0)