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]

sanguo_portfolio/live_reconcile.py

@@ -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}
+               "foreign": len(trades_all) - ours,
-    if backfilled:
+               "conservation_gaps": conservation_gaps}
+    if backfilled or conservation_gaps:
         logger.warning(
             "[live-reconcile] EOD对账补插 %d 笔 (account=%s QMT全量%d 本实例%d "
-            "别家%d)——存在引擎未报形态,查当日名单/日志", backfilled,
+            "别家%d) 守恒缺口=%s——存在引擎未报形态,查当日名单/日志",
-            account_id, summary["qmt_trades"], ours, summary["foreign"])
+            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)

sanguo_portfolio/providers/sanguo_fundamentals.py

@@ -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,

sanguo_trader/shadow/runner.py

@@ -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

tests/portfolio/test_live_reconcile.py

@@ -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 ------------------

tests/portfolio/test_provider.py

@@ -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

tests/portfolio/test_shadow_price_getter.py

@@ -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