fix(live): 孪生成交判重+订单守恒双修(08-27 001234实锤)——①EOD对账五元组判重set→Counter计数感知:同分钟同价同量仅成交编号不同的孪生成交,第二笔原被吞成'已覆盖'→台账短100股幻影(事件路径vt_tradeid=哈希不携带成交编号,tid匹配永远落空全靠元组,集合式判重天生分不开孪生);计数后超出部分补插,001234场景在15:05即自愈②守恒校验新增:本实例当日终态订单filled合计vs台账当日成交量合计(前缀6位码归一/跨日订单不计),不平即告警——连QMT流水都缺行的极端形态最后防线③影子撮合价差对齐(用户拍板费率已齐只留撮合差):新增SanguoFundamentals.get_realtime_price(get_full_tick单次RPC+thread超时保护,裸价),build_price_getter实时tick优先/回退昨收fq=pre→raw(除权日不再跳分红因子,08-27 600988实锤47.12 vs 46.85);+3 provider测试+4定价测试+2守恒测试,全量792绿 [vps]
CI/CD / test (push) Successful in 13s
CI/CD / nas-deploy (push) Successful in 37s
CI/CD / nas-verify (push) Successful in 12s

This commit is contained in:
2026-08-28 00:30:13 +08:00
parent c0dd91e8d2
commit 9c33af5d45
6 changed files with 270 additions and 13 deletions
+53 -10
View File
@@ -205,8 +205,16 @@ def eod_reconcile(engine: Any, ledger: Any, db: str, account_id: int,
"""收盘对账:QMT 当日全账户成交 vs live_trades 已落库行,缺失补插。
已覆盖判定(任一即覆盖):① trade_id(vt_tradeid/账本已见)一致
② 时间(分钟)+代码+方向+价+量 五元组一致。别家实例/手动单只统计。
② 时间(分钟)+代码+方向+价+量 五元组一致——**计数感知**:同五元组的
第 N 笔 QMT 成交消耗一条台账记录,超出部分视为缺失补插(08-27 001234
实锤:同分钟同价同量的孪生成交仅成交编号不同,集合式判重把第二笔吞成
"已覆盖"→账本短 100 股幻影)。别家实例/手动单只统计。
守恒校验:本实例当日终态订单 filled 合计 vs 台账当日成交量合计,
不平即告警(连 QMT 流水都缺行的极端形态,补插无从下手时的最后防线)。
"""
from collections import Counter
from sanguo_live.persistence import list_trades
orders = engine.get_orders() or {}
@@ -223,17 +231,21 @@ def eod_reconcile(engine: Any, ledger: Any, db: str, account_id: int,
today = datetime.now().strftime("%Y-%m-%d")
seen_ids = ledger.seen_trade_ids()
tuples = set()
tuple_counts: Counter = Counter()
ledger_vol: Dict[tuple, int] = {}
for r in (list_trades(db, account_id) if db else []):
traded_at = str(r.get("traded_at") or "")
if not traded_at.startswith(today):
continue
seen_ids.add(str(r.get("vt_tradeid") or ""))
tuples.add((
tuple_counts[(
traded_at[:16], str(r.get("symbol") or ""),
str(r.get("direction") or ""), round(float(r.get("price") or 0), 4),
int(float(r.get("volume") or 0)),
))
)] += 1
vkey = (str(r.get("symbol") or "")[:6],
str(r.get("direction") or "") == "buy")
ledger_vol[vkey] = ledger_vol.get(vkey, 0) + int(float(r.get("volume") or 0))
ours, backfilled = 0, 0
for r in trades_all:
@@ -250,22 +262,53 @@ def eod_reconcile(engine: Any, ledger: Any, db: str, account_id: int,
str(r.get("security") or ""), "buy" if is_buy else "sell",
round(float(r.get("price") or 0), 4), int(r.get("amount") or 0),
)
if (tid and tid in seen_ids) or tkey in tuples:
if tid and tid in seen_ids:
continue
if tuple_counts.get(tkey, 0) > 0:
tuple_counts[tkey] -= 1 # 消费一条同名记录;孪生第二笔不再被吞
continue
n = _apply_rows([r], ledger, db, account_id, strategy_name, is_buy,
trade_id_prefix="eod:")
backfilled += n
if n:
seen_ids.add(f"eod:{tid}")
tuples.add(tkey)
tuple_counts[tkey] += 1
vkey = (str(r.get("security") or "")[:6], is_buy)
ledger_vol[vkey] = ledger_vol.get(vkey, 0) + int(r.get("amount") or 0)
# 守恒校验:订单成交合计 vs 台账记录合计(前缀 6 位码归一,跨日订单不计)
order_vol: Dict[tuple, int] = {}
for o in own_by_broker.values():
if _status_str(getattr(o, "status", None)) not in _TERMINAL:
continue
o_dt = getattr(o, "datetime", None)
if o_dt is not None and not str(o_dt).startswith(today):
continue
filled = int(getattr(o, "filled", 0) or 0)
if filled <= 0:
continue
okey = (str(getattr(o, "security", "") or "")[:6],
bool(getattr(o, "is_buy", True)))
order_vol[okey] = order_vol.get(okey, 0) + filled
conservation_gaps = []
for (sym, is_buy), ov in sorted(order_vol.items()):
lv = ledger_vol.get((sym, is_buy), 0)
if ov != lv:
conservation_gaps.append({
"symbol": sym, "side": "buy" if is_buy else "sell",
"order_volume": ov, "ledger_volume": lv, "gap": ov - lv,
})
summary = {"qmt_trades": len(trades_all), "ours": ours,
"backfilled": backfilled,
"foreign": len(trades_all) - ours}
if backfilled:
"foreign": len(trades_all) - ours,
"conservation_gaps": conservation_gaps}
if backfilled or conservation_gaps:
logger.warning(
"[live-reconcile] EOD对账补插 %d 笔 (account=%s QMT全量%d 本实例%d "
"别家%d)——存在引擎未报形态,查当日名单/日志", backfilled,
account_id, summary["qmt_trades"], ours, summary["foreign"])
"别家%d) 守恒缺口=%s——存在引擎未报形态,查当日名单/日志",
backfilled, account_id, summary["qmt_trades"], ours,
summary["foreign"], conservation_gaps)
else:
logger.info("[live-reconcile] EOD对账干净 (account=%s QMT全量%d 本实例%d)",
account_id, summary["qmt_trades"], ours)
@@ -339,6 +339,53 @@ class SanguoMiniQmtProvider(MiniQMTProvider): # type: ignore[misc]
out[jq] = status
return out
def get_realtime_price(
self, security: str, timeout: float = 5.0,
) -> Optional[float]:
"""单标的实时最新价(miniQMT tick lastPrice,裸价)。
2026-08-28 影子撮合价差对齐:影子原按「昨收(fq=pre)」记账,与实盘
盘中真实成交价每笔天然偏差(低开日 ±200bps+,除权日跳一个分红因子
——08-27 600988 实锤 47.12 vs 46.00)。影子撮合改用与实盘同源同刻
的实时价,本方法即取价入口。``get_full_tick`` 单次 RPC,thread+join
(timeout) 保护(开盘 QMT 忙时 RPC 可拖死策略,2197fea 同款教训);
超时/异常/无价 → None(调用方回退最新收盘)。
"""
import threading
try:
xt = self._ensure_xtdata()
except Exception as exc: # noqa: BLE001
logger.warning("xtdata 不可用,实时价返 None: %s", exc)
return None
try:
qmt = self._normalize_security_code(security)
except Exception: # noqa: BLE001 - 归一失败原样透传
qmt = security
box: Dict[str, Any] = {}
def _worker() -> None:
try:
box.update(xt.get_full_tick([qmt]) or {})
except Exception as exc: # noqa: BLE001
logger.warning("get_full_tick 失败,实时价返 None: %s", exc)
t = threading.Thread(target=_worker, daemon=True)
t.start()
t.join(timeout)
if t.is_alive():
logger.warning("get_full_tick 超时 %.0fs,实时价返 None(%s)",
timeout, security)
return None
tick = box.get(qmt)
if not isinstance(tick, dict):
return None
last = _to_float(_get_multi(tick, ["lastPrice", "last_price", "price"]))
if last and last > 0:
return float(last)
return None
def get_value_metrics(
self,
stock: str,
+17 -2
View File
@@ -49,17 +49,32 @@ def shadow_env() -> Dict[str, str]:
def build_price_getter(provider: Any) -> Any:
"""从数据 provider 取标的最新价(实时/最新收盘)。返回闭包给 ShadowBroker。"""
"""从数据 provider 取标的最新价(与实盘同源同刻优先)。返回闭包给 ShadowBroker。
2026-08-28 价差对齐(用户拍板「费率一致后只留撮合方法差」):
- 优先 ``get_realtime_price``(miniQMT tick lastPrice,裸价)——原实现恒用
「昨收 fq=pre」记账,与实盘盘中成交价每笔天然偏差,低开日 ±200bps+;
- 回退最新收盘改**不复权**(fq=raw):实盘成交价为裸价,前复权昨收在除权
日与实盘价跳空一个分红因子(08-27 600988 实锤 47.12 vs 46.85)。
"""
rt_fn = getattr(provider, "get_realtime_price", None)
def get_price(security: str) -> Optional[float]:
from datetime import datetime, timedelta
if rt_fn is not None:
try:
p = rt_fn(security)
if p and float(p) > 0:
return float(p)
except Exception: # noqa: BLE001 - 实时失败回退最新收盘
pass
end = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
start = (datetime.now() - timedelta(days=10)).strftime("%Y-%m-%d")
try:
df = provider.get_price(
security=security, start_date=start, end_date=end,
frequency="daily", fields=["close"], fq="pre",
frequency="daily", fields=["close"], fq="raw",
)
if df is None or len(df) == 0:
return None
+72 -1
View File
@@ -243,6 +243,76 @@ class TestEodReconcile:
eod_reconcile(SimpleNamespace(get_orders=lambda: {}, broker=None),
LiveInstanceLedger(), "", 1, "s") # 不抛
def test_twin_fill_backfilled_not_swallowed(self, db):
"""孪生成交回补(08-27 001234 实锤):同分钟+同价+同量的两笔成交仅
成交编号不同,集合式五元组判重把第二笔吞成"已覆盖"账本短 100
幻影持仓计数感知后第二笔必须补上"""
from sanguo_live.persistence import save_trade
led = LiveInstanceLedger(initial_cash=1_000_000)
led.apply_trade(True, "001234.XSHE", 22.17, 200, "seed_buy", _TODAY)
save_trade(db, 20, {
"strategy_name": "s", "symbol": "001234.XSHE",
"direction": "buy", "offset": "open", "price": 22.17,
"volume": 200, "traded_at": f"{_TODAY} 09:20:00",
"vt_tradeid": "seed_buy"})
# 事件路径只记到孪生之一(vt_tradeid=哈希,非成交编号)
led.apply_trade(False, "001234.XSHE", 22.35, 100, "hash_a", _TODAY)
save_trade(db, 20, {
"strategy_name": "s", "symbol": "001234.XSHE",
"direction": "sell", "offset": "close", "price": 22.35,
"volume": 100, "traded_at": f"{_TODAY} 09:31:57",
"vt_tradeid": "hash_a"})
eng = _engine(
[_order(status="filled", amount=200, filled=200, is_buy=False,
security="001234.XSHE", broker_oid="235")],
[_qmt_trade(order_id="235", security="001234.XSHE", amount=100,
price=22.35, trade_id="1010000032376071",
time=f"{_TODAY} 09:31:57"),
_qmt_trade(order_id="235", security="001234.XSHE", amount=100,
price=22.35, trade_id="1010000032376079",
time=f"{_TODAY} 09:31:57")])
summary = eod_reconcile(eng, led, db, 20, "small_cap_ex")
assert summary["backfilled"] == 1
# 买200-卖200:清零后台账不再持有该键(volume 视为 0)
assert led.positions.get("001234.XSHE", {}).get("volume", 0) == 0
def test_conservation_gap_alarm_when_journal_also_missing(self, db, caplog):
"""守恒校验:订单 filled=1100 但台账只记 1000(连 QMT 流水都缺这行
的极端形态,补插无从下手) 计数守恒不平时大声告警,不再静默幻影"""
from sanguo_live.persistence import save_trade
led = LiveInstanceLedger(initial_cash=1_000_000)
led.apply_trade(False, "001234.XSHE", 22.35, 1000, "h1", _TODAY)
save_trade(db, 20, {
"strategy_name": "s", "symbol": "001234.XSHE",
"direction": "sell", "offset": "close", "price": 22.35,
"volume": 1000, "traded_at": f"{_TODAY} 09:31:57",
"vt_tradeid": "h1"})
eng = _engine(
[_order(status="filled", amount=1100, filled=1100, is_buy=False,
security="001234.XSHE", broker_oid="235")],
[_qmt_trade(order_id="235", security="001234.XSHE", amount=1000,
price=22.35, trade_id="known1",
time=f"{_TODAY} 09:31:57")])
with caplog.at_level("WARNING"):
summary = eod_reconcile(eng, led, db, 20, "small_cap_ex")
assert summary["backfilled"] == 0
assert summary["conservation_gaps"] == [{
"symbol": "001234", "side": "sell",
"order_volume": 1100, "ledger_volume": 1000, "gap": 100}]
assert any("守恒缺口" in r.message for r in caplog.records)
def test_conservation_balanced_no_alarm(self, db, caplog):
"""守恒平时零告警(订单合计=台账合计,含 EOD 回填后的量)。"""
led = LiveInstanceLedger(initial_cash=1_000_000)
eng = _engine(
[_order(status="filled", amount=500, filled=500, is_buy=False,
security="000049.XSHE", broker_oid="1001")],
[_qmt_trade(amount=500, price=15.0)])
with caplog.at_level("WARNING"):
summary = eod_reconcile(eng, led, db, 20, "s")
assert summary["backfilled"] == 1 # 先回填
assert summary["conservation_gaps"] == [] # 回填后守恒平,零缺口
class TestMaybeEodReconcile:
def test_before_window_is_noop(self):
@@ -276,7 +346,8 @@ class TestMaybeEodReconcile:
ok = _engine([], [])
assert maybe_eod_reconcile(
ok, led, db, 20, "s", now=datetime(2026, 8, 25, 15, 8)) == {
"qmt_trades": 0, "ours": 0, "backfilled": 0, "foreign": 0}
"qmt_trades": 0, "ours": 0, "backfilled": 0, "foreign": 0,
"conservation_gaps": []}
# ------------------ 事故重放 + seen_trade_ids ------------------
+24
View File
@@ -710,3 +710,27 @@ class TestFullMethodSurfaceAlignment:
# 方法面钉死测试已迁至 test_provider_method_surface.py(单一真相,
# fail-not-skip 牙齿)——本模块的 requires_bullet_trade mark 会让它在
# bullet_trade 缺失时被 conftest 自动 skip(钉死虚标,issue #35),故移出。
class TestGetRealtimePrice:
"""单标的实时最新价(2026-08-28 影子撮合价差对齐:get_full_tick 裸价)。"""
def test_returns_last_price_from_tick(self, delegate_provider, mock_xtquant):
mock_xtquant["xtdata"].get_full_tick.return_value = {
"600519.SH": {"lastPrice": 1701.5, "lastClose": 1690.0},
}
assert delegate_provider.get_realtime_price("600519.XSHG") == 1701.5
mock_xtquant["xtdata"].get_full_tick.assert_called_once()
def test_none_when_tick_missing_or_priceless(
self, delegate_provider, mock_xtquant):
mock_xtquant["xtdata"].get_full_tick.return_value = {}
assert delegate_provider.get_realtime_price("600519.XSHG") is None
mock_xtquant["xtdata"].get_full_tick.return_value = {
"600519.SH": {"lastPrice": 0, "lastClose": 1690.0},
}
assert delegate_provider.get_realtime_price("600519.XSHG") is None
def test_none_when_xt_raises(self, delegate_provider, mock_xtquant):
mock_xtquant["xtdata"].get_full_tick.side_effect = RuntimeError("断连")
assert delegate_provider.get_realtime_price("600519.XSHG") is None
@@ -0,0 +1,57 @@
# -*- coding: utf-8 -*-
"""影子撮合定价(2026-08-28 价差对齐,用户拍板「费率一致后只留撮合方法差」)。
08-27 双轨 19/60 实锤:影子按昨收 fq=pre记账 vs 实盘盘中真实成交,每笔
天然偏差(600988:47.12 vs 46.00,低开日 -237bps;除权日另跳一个分红因子)
对齐后契约:实时 tick 优先(与实盘同源同刻的裸价),失败回退**不复权**昨收
"""
from __future__ import annotations
from types import SimpleNamespace
from sanguo_trader.shadow.runner import build_price_getter
class _StubProvider:
"""provider 替身:实时价/日线收盘可控,记录 get_price 的 fq 实参。"""
def __init__(self, realtime=None, close=10.0):
self._realtime = realtime
self._close = close
self.fq_seen = []
def get_realtime_price(self, security):
if isinstance(self._realtime, Exception):
raise self._realtime
return self._realtime
def get_price(self, security=None, start_date=None, end_date=None,
frequency=None, fields=None, fq=None, **_):
self.fq_seen.append(fq)
import pandas as pd
return pd.DataFrame({"close": [self._close]})
class TestBuildPriceGetter:
def test_realtime_tick_preferred(self):
"""有实时价 → 直接用(与实盘同源同刻),不碰日线。"""
p = _StubProvider(realtime=12.34, close=10.0)
assert build_price_getter(p)("600519.XSHG") == 12.34
assert p.fq_seen == []
def test_fallback_to_raw_close_when_realtime_none(self):
"""实时不可得 → 回退最新收盘,且 fq=raw(不复权,实盘成交价为裸价)。"""
p = _StubProvider(realtime=None, close=46.85)
assert build_price_getter(p)("600988.XSHG") == 46.85
assert p.fq_seen == ["raw"]
def test_fallback_when_realtime_raises(self):
"""实时抛异常 → 静默回退收盘(撮合不因实时源故障停摆)。"""
p = _StubProvider(realtime=RuntimeError("断连"), close=9.9)
assert build_price_getter(p)("000001.XSHE") == 9.9
def test_provider_without_realtime_method_uses_raw_close(self):
"""旧/回测 provider 无实时方法 → 原路径可用,fq 仍为 raw。"""
p = SimpleNamespace(get_price=_StubProvider(close=22.35).get_price)
getter = build_price_getter(p)
assert getter("001234.XSHE") == 22.35