feat(live): 成交回报回执②回执↔账本闭环比对器——用户钦点「再出现时根因可查」:09-04实证柜台有成交/台账无行/引擎会话视图无行且时间退化,「推送到达→回调处理→落库」三跳全静默无法定位故障点;观测契约三层=①引擎侧trade-receipt指纹行(策略session实现,tag/字段名以本模块解析器为准)+②本模块parse+compare:有回执无账行=引擎处理丢(直接证据),柜台有成交无回执=柜台没推(EOD恒等式层),③账本行带柜台成交号硬键(替换会退化的时间五元组);原则宁缺勿错:tid=NONE不做模糊匹配硬凑,如实分列;ledger_only为信息项(回执上线前旧行/repair行)不判脏;TDD 8测先红(模块不存在)后绿 [vps]

2026-09-06 10:53:24 +08:00
parent 694a0d2238
commit 67a2e49f21
2 changed files with 213 additions and 0 deletions
@@ -0,0 +1,101 @@
+# -*- coding: utf-8 -*-
+"""成交回报回执:②解析 [trade-receipt] 指纹行 + 回执↔账本闭环比对。
+
+背景(2026-09-04 实证):柜台有成交、台账无行、引擎会话成交视图也无行且
+时间字段退化——「推送到达→回调处理→落库」三跳全静默,故障点无法定位。
+观测契约:
+  ① 引擎侧(策略 session 实现):成交回调任何处理之前打一行指纹,永不抛:
+     [trade-receipt] tid={柜台成交号} order={订单号} acc={账号} sym={代码}
+     vol={量} px={价} t_raw={回报自带时间,无则NONE} recv={本地时刻}
+  ② 本模块:解析该行,与 live_trades 当日行闭环——
+     有回执无账行=引擎处理丢(②③之间);柜台有成交且无回据=柜台没推(由
+     EOD 恒等式/守恒缺口负责,不在此层)。
+  ③ 账本行带柜台成交号(硬键),判重/补插不再依赖会退化的时间字段。
+
+设计原则:宁缺勿错——tid=NONE 的行不做模糊匹配硬凑,如实分列。
+"""
+
+import re
+from typing import Any, Dict, List
+
+# 回执行锚:tag 全文匹配,字段名与①契约严格一致(勿改拼写)
+_RECEIPT_RE = re.compile(
+    r"\[trade-receipt\]\s+tid=(\S+)\s+order=(\S+)\s+acc=(\S+)\s+sym=(\S+)\s+"
+    r"vol=(\S+)\s+px=(\S+)\s+t_raw=(\S+)\s+recv=(\S+)"
+)
+
+_NONE = {"NONE", "None", "none", ""}
+
+
+def _or_none(v: str) -> Any:
+    return None if v in _NONE else v
+
+
+def _to_float(v: str) -> Any:
+    try:
+        return float(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def parse_trade_receipts(log_text: str) -> List[Dict[str, Any]]:
+    """从引擎日志文本解析全部回执行;同 tid 重复回执去重并计 dup_count。"""
+    seen: Dict[str, Dict[str, Any]] = {}
+    order_keys: List[str] = []
+    for m in _RECEIPT_RE.finditer(log_text):
+        tid = _or_none(m.group(1))
+        key = tid if tid is not None else "_orphan_%d" % len(order_keys)
+        if key in seen:
+            seen[key]["dup_count"] += 1
+            continue
+        row = {
+            "tid": tid,
+            "order": _or_none(m.group(2)),
+            "acc": _or_none(m.group(3)),
+            "sym": _or_none(m.group(4)),
+            "vol": _to_float(m.group(5)),
+            "px": _to_float(m.group(6)),
+            "t_raw": _or_none(m.group(7)),
+            "recv": _or_none(m.group(8)),
+            "dup_count": 1,
+        }
+        seen[key] = row
+        order_keys.append(key)
+    return [seen[k] for k in order_keys]
+
+
+def compare_receipts_with_ledger(receipts: List[Dict[str, Any]],
+                                 ledger: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """回执(①的产物)与账本行(live_trades 映射,带 tid)闭环比对。
+
+    ledger 行键约定:tid/symbol/volume/price(由调用方从 DB 行映射)。
+    matched 只认 tid 硬键;tid 缺失双方不猜,各自入 receipt_only/ledger_only。
+    ledger_only 为信息项(回执上线前的旧数据/repair 行),不判脏;
+    receipt_only 才是警报——「到了没记」即引擎处理跳丢失的直接证据。
+    """
+    ledger_by_tid: Dict[str, Dict[str, Any]] = {
+        r["tid"]: r for r in ledger if r.get("tid") is not None
+    }
+    receipt_only: List[Dict[str, Any]] = []
+    matched = 0
+    for r in receipts:
+        tid = r.get("tid")
+        if tid is not None and tid in ledger_by_tid:
+            matched += 1
+        else:
+            receipt_only.append(r)
+    matched_tids = set()
+    for r in receipts:
+        if r.get("tid") is not None and r.get("tid") in ledger_by_tid:
+            matched_tids.add(r["tid"])
+    ledger_only = [r for r in ledger
+                   if r.get("tid") is not None and r["tid"] not in matched_tids]
+    ledger_only += [r for r in ledger if r.get("tid") is None]
+    return {
+        "receipts_n": len(receipts),
+        "ledger_n": len(ledger),
+        "matched_n": matched,
+        "receipt_only": receipt_only,
+        "ledger_only": ledger_only,
+        "clean": not receipt_only,
+    }
@@ -0,0 +1,112 @@
+# -*- coding: utf-8 -*-
+"""trade_receipt:成交回报回执解析+回执↔账本闭环比对。
+
+背景(2026-09-04 实证):柜台有成交、台账无行、引擎会话视图也无行——
+故障点在「推送到达→回调处理→落库」三跳上且全静默。①(引擎侧)每笔回调
+到达即打 [trade-receipt] 指纹行;本模块②解析该行并与 live_trades 闭环:
+有回执无账行=引擎处理丢,柜台有成交无回执=柜台没推——一分为二定位。
+"""
+
+import pytest
+
+from sanguo_trader.shadow.trade_receipt import (
+    compare_receipts_with_ledger,
+    parse_trade_receipts,
+)
+
+
+class TestParseTradeReceipts:
+    def test_parse_full_line(self):
+        text = ("2026-09-04 09:34:23,101 INFO __main__: [trade-receipt] "
+                "tid=b9765f order=1098910971 acc=20 sym=002633.XSHE "
+                "vol=100 px=14.84 t_raw=09:34:23 recv=09:34:23.101")
+        rs = parse_trade_receipts(text)
+        assert len(rs) == 1
+        r = rs[0]
+        assert r["tid"] == "b9765f"
+        assert r["order"] == "1098910971"
+        assert r["acc"] == "20"
+        assert r["sym"] == "002633.XSHE"
+        assert r["vol"] == 100
+        assert r["px"] == 14.84
+        assert r["t_raw"] == "09:34:23"
+
+    def test_parse_none_t_raw_still_counted(self):
+        """时间字段退化(2026-09-04 实证形态)也必须留痕可解析。"""
+        text = ("2026-09-04 13:37:38,975 INFO __main__: [trade-receipt] "
+                "tid=5d9496 order=1098910973 acc=20 sym=002633.XSHE "
+                "vol=400 px=14.84 t_raw=NONE recv=13:37:38.975")
+        rs = parse_trade_receipts(text)
+        assert len(rs) == 1
+        assert rs[0]["t_raw"] is None
+        assert rs[0]["vol"] == 400
+
+    def test_parse_ignores_noise_and_duplicates(self):
+        """噪线不误吞;同 tid 重复回执去重计数(重复本身值得知道)。"""
+        text = "\n".join([
+            "2026-09-04 09:30:00 INFO jq_strategy: 开盘调度 start",
+            ("2026-09-04 09:34:23 INFO __main__: [trade-receipt] tid=aaa "
+             "order=1 acc=20 sym=600036.XSHG vol=100 px=41.4 t_raw=09:34:23 recv=09:34:23.5"),
+            "2026-09-04 09:34:24 INFO __main__: [instance-ledger] 恢复 1 笔",
+            ("2026-09-04 09:34:25 INFO __main__: [trade-receipt] tid=aaa "
+             "order=1 acc=20 sym=600036.XSHG vol=100 px=41.4 t_raw=09:34:23 recv=09:34:25.0"),
+        ])
+        rs = parse_trade_receipts(text)
+        assert len(rs) == 1
+        assert rs[0]["dup_count"] == 2
+
+    def test_parse_tid_missing_kept_with_none(self):
+        """tid 缺失(异常形态)不丢弃行——留痕优先。"""
+        text = ("2026-09-04 09:34:23 INFO __main__: [trade-receipt] "
+                "tid=NONE order=2 acc=20 sym=510500.XSHG vol=100 px=7.85 t_raw=NONE recv=x")
+        rs = parse_trade_receipts(text)
+        assert len(rs) == 1
+        assert rs[0]["tid"] is None
+
+
+class TestCompareReceiptsWithLedger:
+    def test_all_matched_clean(self):
+        receipts = [
+            {"tid": "a", "order": "1", "sym": "002633.XSHE", "vol": 100, "px": 14.84},
+            {"tid": "b", "order": "1", "sym": "002633.XSHE", "vol": 400, "px": 14.84},
+        ]
+        ledger = [
+            {"tid": "a", "symbol": "002633.XSHE", "volume": 100, "price": 14.84},
+            {"tid": "b", "symbol": "002633.XSHE", "volume": 400, "price": 14.84},
+        ]
+        rep = compare_receipts_with_ledger(receipts, ledger)
+        assert rep["receipts_n"] == 2
+        assert rep["ledger_n"] == 2
+        assert rep["matched_n"] == 2
+        assert rep["receipt_only"] == []
+        assert rep["ledger_only"] == []
+        assert rep["clean"] is True
+
+    def test_receipt_only_is_the_smoking_gun(self):
+        """有回执无账行=引擎处理丢(②③之间)——2026-09-04 的未定位跳。"""
+        receipts = [
+            {"tid": "a", "order": "1", "sym": "002633.XSHE", "vol": 100, "px": 14.84},
+            {"tid": "b", "order": "1", "sym": "002633.XSHE", "vol": 100, "px": 14.84},
+        ]
+        ledger = [{"tid": "a", "symbol": "002633.XSHE", "volume": 100, "price": 14.84}]
+        rep = compare_receipts_with_ledger(receipts, ledger)
+        assert rep["clean"] is False
+        assert [r["tid"] for r in rep["receipt_only"]] == ["b"]
+
+    def test_ledger_only_informational(self):
+        """有账行无回执=回执日志后上线前的旧行/估算行,信息项不算脏。"""
+        receipts = []
+        ledger = [{"tid": "repair:x", "symbol": "002633.XSHE", "volume": 100, "price": 14.84}]
+        rep = compare_receipts_with_ledger(receipts, ledger)
+        assert rep["ledger_only"] == ledger
+        assert rep["clean"] is True  # ledger_only 不判脏:只盯「到了没记」
+
+    def test_tid_none_rows_never_match_by_guess(self):
+        """tid=NONE 的行不做模糊匹配硬凑——宁缺勿错。"""
+        receipts = [{"tid": None, "order": "1", "sym": "002633.XSHE", "vol": 100, "px": 14.84}]
+        ledger = [{"tid": None, "symbol": "002633.XSHE", "volume": 100, "price": 14.84}]
+        rep = compare_receipts_with_ledger(receipts, ledger)
+        # tid 缺失双方都不猜:receipt_only 与 ledger_only 各自如实列出
+        assert rep["matched_n"] == 0
+        assert len(rep["receipt_only"]) == 1
+        assert len(rep["ledger_only"]) == 1