diff --git a/sanguo_portfolio/live_trade_receipt.py b/sanguo_portfolio/live_trade_receipt.py new file mode 100644 index 0000000..5c4cf81 --- /dev/null +++ b/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 diff --git a/sanguo_portfolio/runner_live.py b/sanguo_portfolio/runner_live.py index 7981861..8354904 100644 --- a/sanguo_portfolio/runner_live.py +++ b/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) diff --git a/tests/portfolio/test_live_trade_receipt.py b/tests/portfolio/test_live_trade_receipt.py new file mode 100644 index 0000000..bb30b8b --- /dev/null +++ b/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()