feat(live): 成交回报回执日志①——09-04守恒缺口观测件 [vps]——根因件:002633/002836缺口复盘发现引擎会话当日成交视图行缺+时间字段中途退化None,QMT行dump全打查询时刻即铁证,事后无从对「哪笔什么时候以什么形态被看到」;用户钦点观测机制「再出现时根因可查」,dev契约:任何处理/匹配/去重之前打[trade-receipt] tid/order/acc/sym/vol/px/t_raw/recv八字段(②解析此格式,拼写冻结);实现=with_trade_receipt包装QmtBroker.get_trades(bullet_trade轮询制无成交推送回调,_Callback仅桥接断连,broker.get_trades是柜台行进本进程唯一咽喉,引擎同步/a窄修/b宽修全经它),与with_connect_retry同款实例级包装源码零改动;铁律=t_raw原样照记None也记/日志永不抛逐行try-except包死/get_trades原异常照抛行为不变;量控=观测层变化检测非处理层去组:同tid签名(t_raw+量+价+单号)不变不重打,退化即重打(捕获09-04形态),行照常流向消费方;③结论免改:vt_tradeid三条归因路径已承载柜台成交号(即时=引擎trades键/a窄修=trade_id/b宽修=eod:前缀),既有测试已钉(test_live_reconcile.py:123,206+test_live_instance_ledger.py:236),②按tid比回执↔live_trades即闭环;TDD 7新测先红后绿(caplog须set_level(INFO)坑)+598全绿

sanguo_portfolio/live_trade_receipt.py

@@ -0,0 +1,96 @@
+# -*- coding: utf-8 -*-
+"""成交回报回执日志(①,2026-09-04 守恒缺口观测件)。
+
+挂点=包装 QmtBroker.get_trades:bullet_trade 是轮询制(无成交推送回调,
+qmt._Callback 仅桥接断连),``broker.get_trades()`` 是柜台成交行进入本进程
+的唯一咽喉——引擎同步(_sync_trades_from_broker)、a窄修(reconcile_pending)、
+b宽修(eod_reconcile) 全部经它取行。回执在**任何处理/匹配/去重之前**打。
+
+行格式(dev 2026-09-06 契约,② 解析——字段名与顺序不得改):
+    [trade-receipt] tid={柜台成交号} order={订单号} acc={账号} sym={代码}
+    vol={量} px={价} t_raw={回报自带时间,无则NONE} recv={本地时刻}
+
+铁律:
+- t_raw 原样照记,None/退化也记——09-04 事故根因即时间字段中途退化,
+  补插五元组无从对齐,回执即为此留证;
+- 日志永不抛、永不影响交易路径(逐行 try/except 包死;get_trades 本身的
+  异常原样上抛,不改变消费方行为);
+- 量控=观测层变化检测,非处理层去重:同 tid 状态签名(t_raw+量+价+单号)
+  不变不重打,退化/漂移即重打;行照常流向所有消费方。签名表进程内
+  数百行量级(当日成交数),引擎轮换即清,无需淘汰。
+
+③ 注(同批观测契约):live_trades.vt_tradeid 三条归因路径已承载柜台成交号
+(即时归因=引擎 trades 键、a窄修=trade_id、b宽修=eod: 前缀),测试已钉死,
+无 schema 改动;② 用本回执与 live_trades 行按 tid 比对即闭环。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# tid -> 状态签名(观测层变化检测;线程安全=py dict 原子操作,竞争最坏多打一行)
+_RECEIPT_SEEN: Dict[str, str] = {}
+
+
+def _first_present(row: Dict[str, Any], *names: str) -> Any:
+    for n in names:
+        v = row.get(n)
+        if v is not None:
+            return v
+    return None
+
+
+def _t_raw_str(v: Any) -> str:
+    return "NONE" if v is None else str(v)
+
+
+def log_trade_receipts(rows: Optional[List[Any]], account: str) -> None:
+    """逐行打回执(纯日志副作用,永不抛)。"""
+    if not rows:
+        return
+    recv = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    for r in rows:
+        try:
+            if not isinstance(r, dict):
+                continue
+            tid = str(_first_present(r, "trade_id") or "")
+            if not tid:
+                continue
+            t_raw = _first_present(r, "time", "trade_time")
+            vol = _first_present(r, "amount", "volume", "trade_volume")
+            px = _first_present(r, "price", "traded_price")
+            order = _first_present(r, "order_id", "entrust_id")
+            sig = "|".join((_t_raw_str(t_raw), str(vol), str(px), str(order)))
+            if _RECEIPT_SEEN.get(tid) == sig:
+                continue
+            _RECEIPT_SEEN[tid] = sig
+            logger.info(
+                "[trade-receipt] tid=%s order=%s acc=%s sym=%s vol=%s px=%s "
+                "t_raw=%s recv=%s",
+                tid, order, account, _first_present(r, "security"),
+                vol, px, _t_raw_str(t_raw), recv)
+        except Exception:  # noqa: BLE001 - 回执绝不影响交易路径
+            pass
+
+
+def with_trade_receipt(broker: Any, account: str) -> Any:
+    """包一层 broker.get_trades:返回行交给消费方之前先打回执。
+
+    与 ``with_connect_retry`` 同款实例级包装(bullet_trade 源码零改动);
+    get_trades 原异常照抛,回执打点失败吞掉不声张。
+    """
+    orig = broker.get_trades
+
+    def _get_trades() -> Any:
+        rows = orig()
+        try:
+            log_trade_receipts(rows, account)
+        except Exception:  # noqa: BLE001 - 双保险,契约=永不影响交易路径
+            pass
+        return rows
+
+    broker.get_trades = _get_trades  # type: ignore[method-assign]
+    return broker

sanguo_portfolio/runner_live.py

@@ -30,6 +30,8 @@ import time
 from pathlib import Path
 from typing import Any, Dict
 
+from .live_trade_receipt import with_trade_receipt
+
 logger = logging.getLogger(__name__)
 
 ADAPTER_FILE = Path(__file__).resolve().parent / "live_strategy.py"
@@ -297,6 +299,10 @@ def run_live(provider_config: Dict[str, Any] | None = None) -> None:
 
     broker = with_connect_retry(
         QmtBroker(account_id=cfg["account"], data_path=cfg["mini_path"]))
+    # ① 成交回报回执(2026-09-04 守恒缺口观测契约):柜台行进本进程的第一站
+    # 留痕(tid/order/t_raw 原样,t_raw 退化形态可查);与 with_connect_retry
+    # 同款实例级包装,bullet_trade 源码零改动
+    broker = with_trade_receipt(broker, cfg["account"])
     logger.info("QmtBroker 装配 account=%s data_path=%s(connect 失败退避重试 %d×%.0fs)",
                 cfg["account"], cfg["mini_path"],
                 CONNECT_RETRY_ATTEMPTS, CONNECT_RETRY_WAIT_SEC)

tests/portfolio/test_live_trade_receipt.py

@@ -0,0 +1,111 @@
+# -*- coding: utf-8 -*-
+"""成交回报回执日志单测(①,2026-09-04 守恒缺口观测件)。
+
+契约(dev 2026-09-06 钦定,② 解析本格式——字段名与顺序不得改):
+    [trade-receipt] tid={柜台成交号} order={订单号} acc={账号} sym={代码}
+    vol={量} px={价} t_raw={回报自带时间,无则NONE} recv={本地时刻}
+
+铁律:
+- 任何处理/匹配/去重**之前**打(挂点=broker.get_trades 包装,bullet_trade
+  轮询制下它是柜台行进本进程的唯一咽喉);
+- t_raw 原样照记,None/退化也记(09-04 事故根因=时间字段中途退化);
+- 日志永不抛、永不影响交易路径(get_trades 原异常照抛,行为不变);
+- 观测层变化检测:同 tid 状态签名不变不重打,退化/漂移即重打——行照常
+  流向消费方,这不是处理层去重。
+"""
+import logging
+
+import pytest
+
+from sanguo_portfolio.live_trade_receipt import (
+    log_trade_receipts,
+    with_trade_receipt,
+)
+
+_ROW = {
+    "trade_id": "ecf0fbda55b9765f", "order_id": "1098910841",
+    "security": "002633.SZ", "amount": 100, "price": 14.84,
+    "time": "20260904133738",
+}
+
+
+@pytest.fixture(autouse=True)
+def _clear_seen(caplog):
+    from sanguo_portfolio import live_trade_receipt
+    live_trade_receipt._RECEIPT_SEEN.clear()
+    caplog.set_level(logging.INFO, logger="sanguo_portfolio.live_trade_receipt")
+    yield
+    live_trade_receipt._RECEIPT_SEEN.clear()
+
+
+def test_receipt_line_format_pins_contract(caplog):
+    """格式铁契:tag+八字段顺序,recv 为本地时刻。"""
+    log_trade_receipts([_ROW], "66639661")
+    lines = [r.getMessage() for r in caplog.records
+             if r.getMessage().startswith("[trade-receipt]")]
+    assert len(lines) == 1
+    assert lines[0].startswith(
+        "[trade-receipt] tid=ecf0fbda55b9765f order=1098910841 "
+        "acc=66639661 sym=002633.SZ vol=100 px=14.84 "
+        "t_raw=20260904133738 recv=")
+    recv = lines[0].rsplit("recv=", 1)[1]
+    assert len(recv) == 19 and recv[4] == "-" and recv[13] == ":"  # YYYY-MM-DD HH:MM:SS
+
+
+def test_receipt_t_raw_none_logged_as_none(caplog):
+    """t_raw=None → 字面 NONE(09-04 退化形态照记)。"""
+    log_trade_receipts([dict(_ROW, time=None)], "a1")
+    assert "t_raw=NONE" in caplog.records[-1].getMessage()
+
+
+def test_repeat_same_state_suppressed(caplog):
+    """同 tid 状态签名不变 → 只打一次(观测层量控)。"""
+    for _ in range(3):
+        log_trade_receipts([_ROW], "a1")
+    lines = [r.getMessage() for r in caplog.records
+             if r.getMessage().startswith("[trade-receipt]")]
+    assert len(lines) == 1
+
+
+def test_degraded_t_raw_relogged(caplog):
+    """t_raw 中途退化(有值→None)→ 签名变化即重打——捕获 09-04 形态。"""
+    log_trade_receipts([_ROW], "a1")
+    log_trade_receipts([dict(_ROW, time=None)], "a1")
+    lines = [r.getMessage() for r in caplog.records
+             if r.getMessage().startswith("[trade-receipt]")]
+    assert len(lines) == 2
+    assert "t_raw=NONE" in lines[1]
+
+
+def test_bad_rows_skipped_never_raise(caplog):
+    """非 dict 行 / 缺 tid 行:静默跳过,绝不抛。"""
+    log_trade_receipts([None, "junk", {"order_id": "x"}, 42], "a1")
+    assert not [r for r in caplog.records
+                if r.getMessage().startswith("[trade-receipt]")]
+
+
+class _Broker:
+    def __init__(self, rows=None, exc=None):
+        self._rows, self._exc = rows or [], exc
+
+    def get_trades(self):
+        if self._exc:
+            raise self._exc
+        return self._rows
+
+
+def test_wrapper_returns_rows_unchanged(caplog):
+    """包装层透传:同一列表原样返回,回执即打。"""
+    rows = [dict(_ROW)]
+    b = with_trade_receipt(_Broker(rows), "66639661")
+    out = b.get_trades()
+    assert out is rows
+    assert any(r.getMessage().startswith("[trade-receipt]")
+               for r in caplog.records)
+
+
+def test_wrapper_preserves_original_exception():
+    """get_trades 原异常照抛(不吞不改消费方行为)。"""
+    b = with_trade_receipt(_Broker(exc=RuntimeError("qmt down")), "a1")
+    with pytest.raises(RuntimeError, match="qmt down"):
+        b.get_trades()