成交回放：按委托归属隔离逐笔与汇总来源，避免重复入账（台账 019）
一、修复汇总回放的缺口 155 模拟仓八个委托同时由 ws 逐笔与 trading_order 汇总入账，形成三项持仓差异。出口身份相同的汇总镜像只推进游标，账本由原逐笔通道负责。 二、失败守成与兼容 查询失败、代码方向冲突、券商号冲突及多记录命中保留待重试。部分成交、逐笔迟到和模式切换不再重复记账；外部人工成交保持原回放。不改历史账本、交易参数和策略。 三、验证与部署边界 新增十二例先复现再通过，配置隔离及禁止联网的四十三批回归全部通过。155 部署及自然成交判收单独记录，不以离线测试冒充历史差异已修复。
This commit is contained in:
baobao
parent
6fa028527b
commit
e9afc7a3a3
10
DEVLOG.md
10
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。
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 与过往双记相互叠加；冻结证据不足时不能伪造无误差历史。任何后续修复须另存可逆备份与调整依据。
|
||||
|
|
@ -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。
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -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 {})
|
||||
|
|
|
|||
Loading…
Reference in New Issue