From e9afc7a3a33706b851eb594e77c450c22227dddb Mon Sep 17 00:00:00 2001 From: baobao Date: Thu, 15 Oct 2026 11:58:16 +0800 Subject: [PATCH] =?UTF-8?q?=E6=88=90=E4=BA=A4=E5=9B=9E=E6=94=BE=EF=BC=9A?= =?UTF-8?q?=E6=8C=89=E5=A7=94=E6=89=98=E5=BD=92=E5=B1=9E=E9=9A=94=E7=A6=BB?= =?UTF-8?q?=E9=80=90=E7=AC=94=E4=B8=8E=E6=B1=87=E6=80=BB=E6=9D=A5=E6=BA=90?= =?UTF-8?q?=EF=BC=8C=E9=81=BF=E5=85=8D=E9=87=8D=E5=A4=8D=E5=85=A5=E8=B4=A6?= =?UTF-8?q?=EF=BC=88=E5=8F=B0=E8=B4=A6=20019=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一、修复汇总回放的缺口 155 模拟仓八个委托同时由 ws 逐笔与 trading_order 汇总入账，形成三项持仓差异。出口身份相同的汇总镜像只推进游标，账本由原逐笔通道负责。 二、失败守成与兼容 查询失败、代码方向冲突、券商号冲突及多记录命中保留待重试。部分成交、逐笔迟到和模式切换不再重复记账；外部人工成交保持原回放。不改历史账本、交易参数和策略。 三、验证与部署边界 新增十二例先复现再通过，配置隔离及禁止联网的四十三批回归全部通过。155 部署及自然成交判收单独记录，不以离线测试冒充历史差异已修复。 --- DEVLOG.md | 10 ++ app/repo/qmt_repo.py | 17 ++++ app/services/ledger_service.py | 43 ++++++++- docs/PMS三项对账差异排查_2026-10-15.md | 38 ++++++++ docs/复盘决定台账.md | 11 +++ scripts/run_tests.py | 3 +- scripts/test_batch39_units.py | 123 +++++++++++++++++++++++++ scripts/test_wiring.py | 7 +- 8 files changed, 249 insertions(+), 3 deletions(-) create mode 100644 docs/PMS三项对账差异排查_2026-10-15.md create mode 100644 scripts/test_batch39_units.py diff --git a/DEVLOG.md b/DEVLOG.md index b73e014..1cb20a0 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -1839,3 +1839,13 @@ docs/交接_2026-09-20.md（新）、docs/dev_preview/mock_preview.py 与 sample **部署方式。** 本节点不重启 PMS。桥从 Git 快进，已提交源码构建候选镜像，33 组隔离断网契约通过后切换。用户批准的配置只用于保持桥原行为。 **还欠着什么。** 首个完整晨版在下个交易日核验后，建立 XXL 五分钟任务并开启模拟仓 PMS_PLAN_VERSION_RECHECK；当前仍为 False。五日观察与传播时延未判收，188 未部署。此前历史出口缺父指令和轻量对账三项差异仍待查，本轮未改账。 + + +## 2026-10-15 · 三项对账差异定位与防双记修复 + +用户要求先排查 155 模拟 PMS 的三项对账差异，取消额外五日模拟观察，下一交易日晨版通过后启用修订和重核。 +八个委托的出口号、券商号、ws 逐笔合计与 trading_order 完整对应，确认回放缺少归属过滤。 +新增 qmt_repo.orders_for_fill 与 ledger_service 的身份分流，新增第 39 批 12 例并补内存替身与 SUITES。 +先跑回归复现 11 失败、1 新接口未实现错误；修复后 12 例和完整 43 批隔离、禁止联网测试通过。 +历史证据：155 的 ~/pms_logs/reconciliation-evidence-20261015-before-fix-v2.json，340982 字节，SHA256 8e0badf4d771fbf8a3227250c6cc2c2bf3c8234dda2fc33a31f0ab00920f5501。初次导出因方法名拼错失败，零字节初稿保留，v2 才是有效证据。 +不修改历史批次、成本、已实现盈亏或指令状态。部署与自然回放判收待追加；历史修复的前提见 docs/PMS三项对账差异排查_2026-10-15.md。 diff --git a/app/repo/qmt_repo.py b/app/repo/qmt_repo.py index 7a9ab8a..3ee71a6 100644 --- a/app/repo/qmt_repo.py +++ b/app/repo/qmt_repo.py @@ -355,6 +355,23 @@ def get_order(instruction_id: str): {"iid": instruction_id}) +def orders_for_fill(*, order_id, broker_order_id=None) -> list: + """汇总成交的精确归属；包含终态和联调委托，不能按当前模式或状态裁剪。""" + clauses, params = [], {} + if order_id not in (None, ""): + clauses.append("instruction_id = :iid") + params["iid"] = str(order_id) + if broker_order_id not in (None, ""): + clauses.append("broker_order_id = :bid") + params["bid"] = str(broker_order_id) + if not clauses: + raise ValueError("成交缺少可核验的委托身份") + # 最多两行即可识别冲突；一条 SQL 只读本实例出口表。 + return fetch_all("SELECT instruction_id, parent_id, ts_code, side, broker_order_id " + "FROM pms_qmt_order WHERE (" + " OR ".join(clauses) + + ") ORDER BY instruction_id LIMIT 2", params) + + def list_orders(*, statuses=None, parent_id=None, ts_code=None, limit: int = 200) -> list: where, p = [], {"n": int(limit)} if statuses: diff --git a/app/services/ledger_service.py b/app/services/ledger_service.py index 725f563..e45cf0d 100644 --- a/app/services/ledger_service.py +++ b/app/services/ledger_service.py @@ -339,6 +339,23 @@ def calibrate_fees(*, ymd=None, actual_fee=None) -> dict: return out +def _ws_owner_for_fill(fill: dict): + """汇总表不能再次入账本 PMS 的 ws 委托。归属不可核验时抛错，禁止猜成外部单。""" + broker_id = (fill.get("raw") or {}).get("qmt_order_id") + owners = qmt_repo.orders_for_fill(order_id=fill.get("order_id"), broker_order_id=broker_id) + if not owners: + return None + if len(owners) != 1: + raise ValueError("委托身份命中多个出口记录") + owner = owners[0] + for field in ("ts_code", "side"): + if str(owner.get(field) or "") != str(fill.get(field) or ""): + raise ValueError(f"汇总成交与出口委托 {field} 不一致") + if broker_id and owner.get("broker_order_id") and str(broker_id) != str(owner["broker_order_id"]): + raise ValueError("汇总成交与出口委托券商编号不一致") + return owner + + def replay_fills(*, limit: int = 500) -> dict: """成交回放一跳 = ws 逐笔入账 + trading_order 增量回放 (幂等)。 @@ -375,6 +392,31 @@ def replay_fills(*, limit: int = 500) -> dict: out["ok"] = not out["errors"] # 早退也要按 errors 定 ok (ws 那半的错不能吞) return out + # 2026-10-15：155 模拟仓三项差异均来自同一 ws 委托被汇总表再次回放。 + # 以出口身份分流，不等逐笔齐全才排除；否则汇总先到或逐笔迟到仍会双记。 + external, ws_owned = [], [] + t_max = str(state["t"]) + for fill in fresh: + key = _fill_key(fill) + try: + owner = _ws_owner_for_fill(fill) + except Exception as e: + out["errors"].append(f"order_id={key} 成交归属核验失败，保留待重试: {type(e).__name__}: {e}") + continue + if owner is None: + external.append(fill) + continue + ft = str(fill.get("done_time") or "")[:19] or t_max + seen[key] = ft + t_max = max(t_max, ft) + ws_owned.append({"order_id": key, "instruction_id": owner["instruction_id"], + "ts_code": fill.get("ts_code"), "qty": fill.get("qty"), + "reason": "本实例 ws 委托，汇总镜像不重复入账"}) + out["ws_owned"] = ws_owned + if ws_owned: + logger.info("[回放] %s 笔本实例 ws 委托的汇总镜像跳过，逐笔通道负责入账", len(ws_owned)) + fresh = external + try: instrs = _open_instructions() except Exception as e: @@ -391,7 +433,6 @@ def replay_fills(*, limit: int = 500) -> dict: by_key.setdefault(str(act.get("order_id")), []).append(act) time_by_key = {_fill_key(f): str(f.get("done_time") or "")[:19] for f in fresh} booked_codes = set() - t_max = str(state["t"]) for key, acts in by_key.items(): ok_acts = 0 err = None diff --git a/docs/PMS三项对账差异排查_2026-10-15.md b/docs/PMS三项对账差异排查_2026-10-15.md new file mode 100644 index 0000000..ad73813 --- /dev/null +++ b/docs/PMS三项对账差异排查_2026-10-15.md @@ -0,0 +1,38 @@ +# PMS 三项对账差异排查与修复规格 + +2026-10-15，北京时间。对象是 155 模拟 PMS，ws 下发、propose_only 新建仓。188 真实仓不在本次范围。 + +## 事实与预判 + +只读 reconcile(apply_fix=False) 发现 002050.SZ 账本 800 / QMT 700，002979.SZ 账本 0 / QMT 1900，688100.SH 账本 2400 / QMT 1400。 +八个委托的 trading_order.order_id 与 pms_qmt_order.instruction_id 相同，qmt_order_id 与 broker_order_id 也全部一致。 +WebSocket 逐笔回报均 processed=1，汇总委托全部已进入 PMS_REPLAY_CURSOR.seen。 + +| 股票 | 当日重复回放的汇总数量 | 结果 | +| --- | --- | --- | +| 002050.SZ | 买入 100 | 净多 100；批次 1361 以委托限价 31.560 重复入账，逐笔实际为 73 股 30.690 和 27 股 30.650 | +| 002979.SZ | 卖出 1300、500、100、100，合计 2000 | 重复核销后数量扣到零，与 QMT 1900 不符；数量下限截断使差额不是 2000 | +| 688100.SH | 买入 700、卖出 400、买入 700 | 重复净买入 1000；部分重复买入又被错误认作外部 BASE | + +根因在 ledger_service.replay_fills：注释承诺 trading_order 只兜外部/人工成交，实际没有排除本 PMS 的 WebSocket 委托。 +这不是新计划版本造成的；三项当日账本更新时间均早于 10:42 PMS 部署。成交价格也因汇总表回退委托价格而被污染。 + +## 修复原理及边界 + +在 trading_order 回放认领前，以委托编号及券商编号精确查询本实例 pms_qmt_order，不按股票、时间和数量猜测重复。 +属于本实例出口表的委托只由 WebSocket 逐笔入账，汇总镜像只推进既有 seen 游标并留跳过原因。 +即使逐笔迟到、部分到达、联调单或当前模式已经切回 shadow，也不得从汇总表补记同一委托。 +身份查不到的外部/人工成交保留原回放。身份读取失败、两个标识对应不同委托、代码/方向/券商身份冲突时，不入账、不记 seen，留错供下一跳重试。 +单表 SQL，不新增表、调度或交易参数。历史账本与历史指令进度不由此补丁自动回写。 + +## 验收 + +先用合成数据复现双来源买入、卖出、部分成交与乱序。新增第 39 批测试；执行配置隔离与禁止联网的全量测试。 +部署前再核验实例、在途委托与容器指纹；匹配候选镜像断网全量通过后，按本聊天授权重建 PMS 四进程。 +真机检查健康、通道、回放无重复动作与只读身份分类。已有错账单独标记，不能把修复代码通过说成历史账已恢复。 + +## 历史修复约束 + +日终 RECON 只调数量，并会把差额按行情价格当成新增或核销，不能修复被污染的原始成本、已实现盈亏和指令 exec_qty。 +因此不为消除告警直接 force 对账。先保存当前批次、原逐笔成交、汇总回放游标及指令快照，评估能否完整重建。 +历史已有 RECON 与过往双记相互叠加；冻结证据不足时不能伪造无误差历史。任何后续修复须另存可逆备份与调整依据。 diff --git a/docs/复盘决定台账.md b/docs/复盘决定台账.md index 9f2d2fa..25bca4a 100644 --- a/docs/复盘决定台账.md +++ b/docs/复盘决定台账.md @@ -263,3 +263,14 @@ **复核日期。** 获准部署后连续五个交易日记录顺序、版本、传播时延、重核结果和重复指令；具体首日待部署确定。本地测试不能替代真机判收。 **部署补记（2026-10-15）。** 用户授权后 10:42 在 155 模拟仓部署 6a5ae12，42 批镜像断网回归通过；收到和消费记录分别落库，既有参数保持。新版本核验仍关闭，桥完整版本和五日观察待后续启用。188 未部署。 + + +## 019 · 2026-10-15 · 成交回放按委托归属隔离双来源 + +**改动。** trading_order 回放前用本实例出口表的委托号或券商号核对归属。已归属 WebSocket 的汇总镜像只记跳过及 seen，逐笔迟到仍等待原通道；读源失败或身份冲突不猜作外部成交。外部人工成交保留原回放。 + +**依据。** 155 模拟仓三项差异由八个委托的两套来源重复入账产生，两个身份字段逐笔一致。002050 重复买 100，688100 重复净买 1000，002979 重复卖 2000 后数量下限截断成零。发生于本次计划功能部署之前。 + +**预期。** 新成交不再双记数量、成本与指令进度，不改变交易参数、候选及执行规则。历史成本和盈亏不由数量对账冒充修复；原始证据已保存在 155 宿主日志目录。 + +**复核日期。** 2026-10-15 部署后核验身份分类与自然回放；2026-10-16 晨版核验同步检查。12 个新增用例及全量 43 批离线回归通过；部署结果追加到 DEVLOG。 diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 2af1a68..7811d89 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -5,6 +5,7 @@ 运行: docker compose run --rm pms-web python scripts/run_tests.py 包含 (例数按 2026-10-12 隔离回归实跑校正): + test_batch39_units.py 成交双来源归属、迟到、部分成交与失败守成 (12 例) test_batch38_units.py 每日计划顺序、版本核验、重新确认、并发采纳、历史保留 (15 例) test_core_units.py 仓位规划器 / 安全垫与成本账 (14 例) test_batch2_units.py 命令状态机 / 方案生成器 / 回放对账纯逻辑 (35 例) @@ -188,7 +189,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py", "test_batch35_units.py", "test_batch36_units.py", "test_batch37_units.py", - "test_batch38_units.py", + "test_batch38_units.py", "test_batch39_units.py", "test_page_copy_guard.py", "test_page_enum_guard.py", "test_page_wiring_guard.py", diff --git a/scripts/test_batch39_units.py b/scripts/test_batch39_units.py new file mode 100644 index 0000000..e386ad6 --- /dev/null +++ b/scripts/test_batch39_units.py @@ -0,0 +1,123 @@ +"""本地：python scripts/unit_test_guard.py scripts/test_batch39_units.py。禁止联网。""" +import unit_test_guard # noqa: F401 +import json +import unittest +from unittest.mock import patch +from test_wiring import install_fakes +from app.services import ledger_service as ls +from app.repo import downstream_repo as ds, qmt_repo as q + +CODE = '600000.SH' +IID = 'INS_R_D01' + + +class ReplayOwnershipTest(unittest.TestCase): + def setUp(self): + self.fake = install_fakes(prices={CODE: 10.0}) + self.fake.insert_instruction(instruction_id='INS_R', origin_type='plan', origin_id='P1', + ts_code=CODE, action='OPEN', side='buy', qty=100, status='DISPATCHED') + self.fake.instructions['INS_R']['created_at'] = '2026-10-15 09:00:00' + self.fake.qmt.enqueue_order(instruction_id=IID, parent_id='INS_R', ts_code=CODE, + side='buy', qty=100, limit_price=10, valid_until=0) + self.fake.qmt.orders[IID]['broker_order_id'] = 'BROKER-1' + self.fills = [dict(order_id=IID, ts_code=CODE, side='buy', qty=100, price=11, + done_time='2026-10-15 10:00:00', raw={'qmt_order_id': 'BROKER-1'})] + self.patcher = patch.object(ds, 'fetch_filled_orders', side_effect=lambda **kw: self.fills) + self.patcher.start() + self.addCleanup(self.patcher.stop) + + def trade(self, seq=1, qty=100, side='buy'): + self.fake.qmt.put_trade(seq=seq, instruction_id=IID, ts_code=CODE, + side=side, qty=qty, price=10, fee=0) + + def qty(self): + return self.fake.positions.get(CODE, {}).get('total_qty', 0) + + def test_buy_in_both_sources_is_booked_once(self): + self.trade() + result = ls.replay_fills() + self.assertTrue(result['ok'], result) + self.assertEqual(self.qty(), 100) + self.assertEqual(self.fake.instructions['INS_R']['exec_qty'], 100) + self.assertEqual(ls.replay_fills()['actions'], 0) + self.assertEqual(self.qty(), 100) + + def test_sell_in_both_sources_is_booked_once(self): + ls._apply_action(dict(kind='BUY', ts_code=CODE, qty=300, price=9)) + self.fake.qmt.orders[IID]['side'] = 'sell' + self.fake.instructions['INS_R'].update(side='sell', action='EXIT') + self.fills[0]['side'] = 'sell' + self.trade(side='sell') + self.assertTrue(ls.replay_fills()['ok']) + self.assertEqual(self.qty(), 200) + + def test_table_before_ws_waits_for_exact_trades(self): + self.assertTrue(ls.replay_fills()['ok']) + self.assertEqual(self.qty(), 0) + self.trade() + self.assertTrue(ls.replay_fills()['ok']) + self.assertEqual(self.qty(), 100) + + def test_partial_ws_then_delayed_remainder(self): + self.trade(qty=73) + ls.replay_fills() + self.assertEqual(self.qty(), 73) + self.trade(seq=2, qty=27) + ls.replay_fills() + self.assertEqual(self.qty(), 100) + + def test_external_fill_still_books(self): + self.trade() + self.fills.append(dict(self.fills[0], order_id='MANUAL-1', qty=200, raw={})) + result = ls.replay_fills() + self.assertTrue(result['ok'], result) + self.assertEqual(self.qty(), 300) + + def test_broker_identity_works_when_table_id_differs(self): + self.fills[0]['order_id'] = 'DOWNSTREAM-1' + self.trade() + self.assertTrue(ls.replay_fills()['ok']) + self.assertEqual(self.qty(), 100) + + def test_lookup_failure_leaves_fill_retryable(self): + with patch.object(q, 'orders_for_fill', side_effect=RuntimeError('unavailable')): + result = ls.replay_fills() + self.assertFalse(result['ok']) + self.assertEqual(self.qty(), 0) + self.assertNotIn(IID, json.loads(self.fake.params[ls.CURSOR_KEY])['seen']) + self.trade() + self.assertTrue(ls.replay_fills()['ok']) + self.assertEqual(self.qty(), 100) + + def test_conflicting_code_does_not_fall_back_to_external(self): + self.fills[0]['ts_code'] = '600001.SH' + result = ls.replay_fills() + self.assertFalse(result['ok']) + self.assertFalse(self.fake.lots) + self.assertNotIn(IID, json.loads(self.fake.params[ls.CURSOR_KEY])['seen']) + + def test_conflicting_broker_identity_is_held(self): + self.fills[0]['raw']['qmt_order_id'] = 'DIFFERENT' + self.assertFalse(ls.replay_fills()['ok']) + self.assertEqual(self.qty(), 0) + + def test_smoke_table_mirror_cannot_reenter_book(self): + self.fake.qmt.orders[IID]['parent_id'] = 'SMOKE_1' + self.trade() + self.assertTrue(ls.replay_fills()['ok']) + self.assertEqual(self.qty(), 0) + + def test_ownership_survives_dispatch_mode_change(self): + self.fake.params['PMS_DISPATCH_MODE'] = 'shadow' + self.trade() + self.assertTrue(ls.replay_fills()['ok']) + self.assertEqual(self.qty(), 100) + + def test_two_owners_fail_closed(self): + self.fake.qmt.orders['OTHER'] = dict(self.fake.qmt.orders[IID], instruction_id='OTHER') + self.assertFalse(ls.replay_fills()['ok']) + self.assertEqual(self.qty(), 0) + + +if __name__ == '__main__': + unittest.main() diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py index c352dd5..bd50e98 100644 --- a/scripts/test_wiring.py +++ b/scripts/test_wiring.py @@ -546,6 +546,11 @@ class FakeQmtRepo: def get_order(self, instruction_id): return self.orders.get(instruction_id) + def orders_for_fill(self, *, order_id, broker_order_id=None): + return [r for r in self.orders.values() + if r["instruction_id"] == str(order_id) + or (broker_order_id and r.get("broker_order_id") == str(broker_order_id))][:2] + # ---- pms_qmt_inbox ---- def put_trade(self, *, seq, instruction_id, ts_code, side, qty, price, fee=0.0, trade_no=None, processed=0): @@ -646,7 +651,7 @@ def install_fakes(prices=None, positions=None, params=None, high5=None, prev_clo for name in ("upsert_many", "delete_day_kind", "list_range", "count_by_kind"): setattr(consensus_stat_repo, name, getattr(fake.consensus_stats, name)) for _n in ("get_state", "inbox_pending_count", "queue_depth", "enqueue_order", - "request_cancel", "get_order", "inbox_pending", "inbox_mark", + "request_cancel", "get_order", "orders_for_fill", "inbox_pending", "inbox_mark", "inbox_orphan_count", "latest_snapshot"): setattr(qmt_repo, _n, getattr(fake.qmt, _n)) fake.params.update(params or {})