feat(shadow-desk): 双轨日终对账报表(影子P3前半,设计§8.2)—reconcile_report四指标(成交笔数全同/每笔价差vwap对比<10bps带符号供滑点重标/收盘持仓逐只数量/净值月偏差<0.5%)+find_dual_track_pairs自动配对(shadow策略名↔live strategy_class,live#5↔shadow#39实证)+dual_track_reconcile落库upsert;API GET /paper/reconcile(+单配对refresh);CLI python -m;supervisor auto轮询挂15:10后每日一次兜底;14新测试(符号口径SH/XSHG归一/影子拒单不计/部分成交如实报笔数异) [vps]

sanguo_api/routes_paper.py

@@ -141,6 +141,51 @@ def list_papers():
     return {"accounts": out}
 
 
+# ===== 双轨对账（影子柜台 vs 实盘模拟，设计 §8.2 / 影子 P3 前半）=====
+# 注意:须注册在 /paper/{aid} 之前,否则 "reconcile" 被当作 aid → 422
+
+@router.get("/paper/reconcile", dependencies=[Depends(verify_token)])
+def list_reconcile_pairs(date: str | None = None):
+    """双轨配对列表 + 各配对对账报告(按需现算并落库)。
+
+    自动配对:运行中影子账户(mode=shadow)按策略名匹配 live_accounts。
+    date 缺省 = 今天。
+    """
+    from sanguo_trader.shadow.reconcile_report import (
+        build_reconcile_report, find_dual_track_pairs, load_reconcile_report,
+        save_reconcile_report,
+    )
+
+    db = _db_path["path"]
+    out = []
+    for pair in find_dual_track_pairs(db):
+        saved = load_reconcile_report(
+            db, pair["live_account_id"], pair["shadow_account_id"], date or "")
+        report = saved or build_reconcile_report(
+            db, pair["live_account_id"], pair["shadow_account_id"], date)
+        save_reconcile_report(db, report)
+        out.append({**pair, "report": report})
+    return {"pairs": out}
+
+
+@router.get("/paper/reconcile/{live_id}/{shadow_id}", dependencies=[Depends(verify_token)])
+def get_reconcile(live_id: int, shadow_id: int, date: str | None = None,
+                  refresh: bool = False):
+    """单配对对账报告。refresh=true 强制重算(缺省读已存,无则现算)。"""
+    from sanguo_trader.shadow.reconcile_report import (
+        build_reconcile_report, load_reconcile_report, save_reconcile_report,
+    )
+
+    db = _db_path["path"]
+    if not refresh:
+        saved = load_reconcile_report(db, live_id, shadow_id, date or "")
+        if saved is not None:
+            return saved
+    report = build_reconcile_report(db, live_id, shadow_id, date)
+    save_reconcile_report(db, report)
+    return report
+
+
 @router.get("/paper/{aid}", dependencies=[Depends(verify_token)])
 def get_paper(aid: int):
     db = _db_path["path"]

sanguo_trader/shadow/reconcile_report.py

@@ -0,0 +1,320 @@
+"""双轨日终对账报表(影子柜台 vs 实盘模拟,设计 §8.2,影子 P3 前半)。
+
+同一 db 文件(VPS ``backtest_results.db``)里 live_* 实盘侧与 paper_* 影子侧
+同库共存,本模块按日对账四项指标:
+
+| 对比项 | 一致标准 |
+|--------|---------|
+| 成交笔数 | 完全相同(rejected 影子单不计) |
+| 每笔成交价差 | 平均 <10bps(按 symbol+side 聚合 vwap 对比) |
+| 收盘持仓 | 逐只股票+数量相同 |
+| 净值偏差 | 月累计 <0.5%(月内首基线 → 当日,双侧收益率差) |
+
+差异即策略从纸面到真实的真实滑点成本——影子柜台的核心产出之一。
+
+用法:
+    python -m sanguo_trader.shadow.reconcile_report --db <db> [--date YYYY-MM-DD]
+    API: GET /paper/reconcile(见 sanguo_api/routes_paper.py)
+"""
+from __future__ import annotations
+
+import json
+import logging
+import sqlite3
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# §8.2 一致标准
+PRICE_DIFF_BPS_MAX = 10.0      # 每笔成交价差平均上限(bps)
+NAV_MTD_PCT_MAX = 0.5          # 净值月累计偏差上限(%)
+
+
+def _norm_symbol(symbol: str) -> str:
+    """两侧符号口径不同(live=QMT '510300.SH',shadow=jq '510300.XSHG')→ 6 位码。"""
+    return str(symbol or "").split(".", 1)[0].strip()
+
+
+def _norm_side(direction: str) -> str:
+    """live 'buy'/'sell' vs shadow 'long'/'short' → B/S。"""
+    d = str(direction or "").lower()
+    if d in ("buy", "long", "多", "b"):
+        return "B"
+    return "S"
+
+
+def _month_start(date: str) -> str:
+    return f"{date[:7]}-01"
+
+
+def _trade_rows(conn: sqlite3.Connection, table: str, account_id: int,
+                date: str) -> List[Dict[str, Any]]:
+    """按日取成交行。live.traded_at 是 QMT 原样字符串(可能 '2026-08-15 09:35:00'
+    或紧凑格式),用两种 LIKE 兜;shadow 用 bar_date 精确匹配。"""
+    if table == "live_trades":
+        q = ("SELECT symbol, direction, price, volume, traded_at FROM live_trades "
+             "WHERE account_id=? AND (substr(traded_at,1,10)=? OR traded_at LIKE ?)")
+        rows = conn.execute(q, (account_id, date, f"{date.replace('-', '')}%")).fetchall()
+        return [{"symbol": r[0], "direction": r[1], "price": r[2], "volume": r[3]}
+                for r in rows]
+    q = ("SELECT symbol, direction, price, volume FROM paper_trades "
+         "WHERE account_id=? AND bar_date=? AND (rejected IS NULL OR rejected=0)")
+    rows = conn.execute(q, (account_id, date)).fetchall()
+    return [{"symbol": r[0], "direction": r[1], "price": r[2], "volume": r[3]}
+            for r in rows]
+
+
+def _aggregate(trades: List[Dict[str, Any]]) -> Dict[str, Dict[str, float]]:
+    """(norm_symbol, side) → {volume, notional} 聚合(vwap = notional/volume)。"""
+    agg: Dict[str, Dict[str, float]] = {}
+    for t in trades:
+        key = f"{_norm_symbol(t['symbol'])}:{_norm_side(t['direction'])}"
+        a = agg.setdefault(key, {"volume": 0.0, "notional": 0.0})
+        vol = float(t["volume"] or 0)
+        a["volume"] += vol
+        a["notional"] += vol * float(t["price"] or 0)
+    return agg
+
+
+def _reconcile_trades(conn: sqlite3.Connection, live_id: int, shadow_id: int,
+                      date: str) -> Dict[str, Any]:
+    live_rows = _trade_rows(conn, "live_trades", live_id, date)
+    shadow_rows = _trade_rows(conn, "paper_trades", shadow_id, date)
+    live = _aggregate(live_rows)
+    shadow = _aggregate(shadow_rows)
+
+    rows: List[Dict[str, Any]] = []
+    diffs: List[float] = []
+    for key in sorted(set(live) | set(shadow)):
+        symbol, side = key.rsplit(":", 1)
+        lv, sv = live.get(key), shadow.get(key)
+        lv_vwap = lv["notional"] / lv["volume"] if lv and lv["volume"] else None
+        sv_vwap = sv["notional"] / sv["volume"] if sv and sv["volume"] else None
+        bps = None
+        if lv_vwap and sv_vwap:
+            bps = (lv_vwap - sv_vwap) / sv_vwap * 1e4  # 带符号:稳定偏一侧→重标滑点
+            diffs.append(abs(bps))
+        rows.append({
+            "symbol": symbol, "side": side,
+            "live_volume": lv["volume"] if lv else 0,
+            "shadow_volume": sv["volume"] if sv else 0,
+            "live_vwap": round(lv_vwap, 4) if lv_vwap else None,
+            "shadow_vwap": round(sv_vwap, 4) if sv_vwap else None,
+            "price_diff_bps": round(bps, 2) if bps is not None else None,
+        })
+    avg_bps = sum(diffs) / len(diffs) if diffs else 0.0
+    # 笔数=原始成交行数(非聚合桶):实盘部分成交会多行,差异如实呈现待归因
+    live_count = len(live_rows)
+    shadow_count = len(shadow_rows)
+    return {
+        "live_count": live_count,
+        "shadow_count": shadow_count,
+        "count_match": live_count == shadow_count,
+        "rows": rows,
+        "avg_price_diff_bps": round(avg_bps, 2),
+        "pass_price": avg_bps <= PRICE_DIFF_BPS_MAX,
+    }
+
+
+def _reconcile_positions(conn: sqlite3.Connection, live_id: int,
+                         shadow_id: int) -> Dict[str, Any]:
+    lv = {r[0]: r[1] for r in conn.execute(
+        "SELECT symbol, volume FROM live_positions WHERE account_id=?", (live_id,))}
+    sv = {r[0]: r[1] for r in conn.execute(
+        "SELECT symbol, volume FROM paper_positions "
+        "WHERE account_id=? AND scope='account' AND date=("
+        "  SELECT MAX(date) FROM paper_positions WHERE account_id=? AND scope='account')",
+        (shadow_id, shadow_id))}
+    lv_n = {_norm_symbol(s): v for s, v in lv.items()}
+    sv_n = {_norm_symbol(s): v for s, v in sv.items()}
+    rows = []
+    match = True
+    for sym in sorted(set(lv_n) | set(sv_n)):
+        lvol = int(lv_n.get(sym) or 0)
+        svol = int(sv_n.get(sym) or 0)
+        if lvol != svol:
+            match = False
+        rows.append({"symbol": sym, "live_volume": lvol,
+                     "shadow_volume": svol, "volume_diff": lvol - svol})
+    return {"rows": rows, "match": match}
+
+
+def _reconcile_nav(conn: sqlite3.Connection, live_id: int, shadow_id: int,
+                   date: str) -> Dict[str, Any]:
+    """月累计净值偏差:月内(含月前最后一条)首基线 → 当日,双侧收益率差(%)。"""
+
+    def _series(table: str, col: str, aid: int) -> Dict[str, float]:
+        q = (f"SELECT date, {col} FROM {table} WHERE account_id=? "
+             f"AND date<=? ORDER BY date")
+        return {r[0]: float(r[1]) for r in conn.execute(q, (aid, date))}
+
+    live_s = _series("live_balance", "total", live_id)
+    shadow_s = _series("paper_daily_balance", "total_equity", shadow_id)
+    if not live_s or not shadow_s or date not in live_s or date not in shadow_s:
+        return {"live_total": live_s.get(date), "shadow_total": shadow_s.get(date),
+                "mtd_deviation_pct": None, "pass_nav": None, "note": "净值序列不全"}
+
+    ms = _month_start(date)
+
+    def _baseline(series: Dict[str, float]) -> Optional[float]:
+        prior = [v for d, v in series.items() if d < ms]
+        return prior[-1] if prior else next(
+            (v for d, v in sorted(series.items()) if d >= ms), None)
+
+    lb, sb = _baseline(live_s), _baseline(shadow_s)
+    if not lb or not sb:
+        return {"live_total": live_s[date], "shadow_total": shadow_s[date],
+                "mtd_deviation_pct": None, "pass_nav": None, "note": "无月内基线"}
+    live_ret = live_s[date] / lb - 1
+    shadow_ret = shadow_s[date] / sb - 1
+    dev = abs(live_ret - shadow_ret) * 100
+    return {
+        "live_total": live_s[date],
+        "shadow_total": shadow_s[date],
+        "live_mtd_return_pct": round(live_ret * 100, 4),
+        "shadow_mtd_return_pct": round(shadow_ret * 100, 4),
+        "mtd_deviation_pct": round(dev, 4),
+        "pass_nav": dev <= NAV_MTD_PCT_MAX,
+    }
+
+
+def build_reconcile_report(db: str, live_account_id: int, shadow_account_id: int,
+                           date: Optional[str] = None) -> Dict[str, Any]:
+    """构建并返回某日双轨对账报告(纯读,不落库)。"""
+    date = date or datetime.now().strftime("%Y-%m-%d")
+    with sqlite3.connect(db) as conn:
+        trades = _reconcile_trades(conn, live_account_id, shadow_account_id, date)
+        positions = _reconcile_positions(conn, live_account_id, shadow_account_id)
+        nav = _reconcile_nav(conn, live_account_id, shadow_account_id, date)
+    passed = bool(
+        trades["count_match"] and trades["pass_price"]
+        and positions["match"] and nav["pass_nav"] is not False
+    )
+    return {
+        "date": date,
+        "live_account_id": live_account_id,
+        "shadow_account_id": shadow_account_id,
+        "trades": trades,
+        "positions": positions,
+        "nav": nav,
+        "passed": passed,
+    }
+
+
+def find_dual_track_pairs(db: str) -> List[Dict[str, Any]]:
+    """自动配对:运行中影子账户(mode=shadow)按策略名匹配 live_accounts.strategy_class。"""
+    pairs: List[Dict[str, Any]] = []
+    with sqlite3.connect(db) as conn:
+        lives = {r[0]: r[1] for r in conn.execute(
+            "SELECT strategy_class, id FROM live_accounts WHERE status='running'")}
+        for aid, strategies_json in conn.execute(
+                "SELECT id, strategies FROM paper_accounts "
+                "WHERE mode='shadow' AND status='running'"):
+            try:
+                name = (json.loads(strategies_json or "[]") or [{}])[0].get("name", "")
+            except (json.JSONDecodeError, IndexError):
+                continue
+            live_id = lives.get(name)
+            if live_id:
+                pairs.append({"live_account_id": live_id,
+                              "shadow_account_id": aid, "strategy": name})
+    return pairs
+
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS dual_track_reconcile (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    live_account_id INTEGER,
+    shadow_account_id INTEGER,
+    date TEXT,
+    passed INTEGER,
+    report TEXT,
+    created_at TEXT,
+    UNIQUE(live_account_id, shadow_account_id, date)
+)
+"""
+
+
+def save_reconcile_report(db: str, report: Dict[str, Any]) -> None:
+    """落库(upsert 同配对同日覆盖)。"""
+    with sqlite3.connect(db) as conn:
+        conn.execute(_SCHEMA)
+        conn.execute(
+            "INSERT INTO dual_track_reconcile "
+            "(live_account_id, shadow_account_id, date, passed, report, created_at) "
+            "VALUES (?,?,?,?,?,?) "
+            "ON CONFLICT(live_account_id, shadow_account_id, date) "
+            "DO UPDATE SET passed=excluded.passed, report=excluded.report, "
+            "created_at=excluded.created_at",
+            (report["live_account_id"], report["shadow_account_id"], report["date"],
+             1 if report.get("passed") else 0,
+             json.dumps(report, ensure_ascii=False),
+             datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
+        )
+        conn.commit()
+
+
+def load_reconcile_report(db: str, live_account_id: int, shadow_account_id: int,
+                          date: str, *, as_row: bool = False) -> Any:
+    """读已存报告;无 → None。as_row=True 返回表行(含 created_at)而非解析 JSON。"""
+    with sqlite3.connect(db) as conn:
+        conn.execute(_SCHEMA)
+        row = conn.execute(
+            "SELECT live_account_id, shadow_account_id, date, passed, report, "
+            "created_at FROM dual_track_reconcile WHERE live_account_id=? "
+            "AND shadow_account_id=? AND date=?",
+            (live_account_id, shadow_account_id, date),
+        ).fetchone()
+    if row is None:
+        return None
+    if as_row:
+        keys = ("live_account_id", "shadow_account_id", "date", "passed",
+                "report", "created_at")
+        return [dict(zip(keys, row))]
+    return json.loads(row[4])
+
+
+def main() -> None:
+    """CLI:python -m sanguo_trader.shadow.reconcile_report --db <db> [--date ...]
+
+    不传配对 → find_dual_track_pairs 自动配对全部跑一遍并落库。
+    """
+    import argparse
+
+    logging.basicConfig(level=logging.INFO,
+                        format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+    p = argparse.ArgumentParser(description="双轨日终对账报表")
+    p.add_argument("--db", required=True)
+    p.add_argument("--date", default=None)
+    p.add_argument("--live", type=int, default=None, help="显式配对:live 账户 id")
+    p.add_argument("--shadow", type=int, default=None, help="显式配对:影子账户 id")
+    args = p.parse_args()
+
+    if args.live and args.shadow:
+        pairs = [{"live_account_id": args.live, "shadow_account_id": args.shadow,
+                  "strategy": "(explicit)"}]
+    else:
+        pairs = find_dual_track_pairs(args.db)
+        if not pairs:
+            logger.warning("未找到运行中的双轨配对(影子 mode=shadow ↔ live 同策略名)")
+    for pair in pairs:
+        r = build_reconcile_report(args.db, pair["live_account_id"],
+                                   pair["shadow_account_id"], args.date)
+        save_reconcile_report(args.db, r)
+        t = r["trades"]
+        logger.info(
+            "[对账] %s live#%s vs shadow#%s(%s): 笔数 %s/%s=%s 价差%.1fbps=%s "
+            "持仓=%s 净值月偏差=%s%% → %s",
+            r["date"], pair["live_account_id"], pair["shadow_account_id"],
+            pair["strategy"], t["live_count"], t["shadow_count"],
+            "同" if t["count_match"] else "异",
+            t["avg_price_diff_bps"], "过" if t["pass_price"] else "超限",
+            "同" if r["positions"]["match"] else "异",
+            r["nav"]["mtd_deviation_pct"],
+            "PASS" if r["passed"] else "FAIL",
+        )
+
+
+if __name__ == "__main__":
+    main()

sanguo_trader/shadow/supervisor.py

@@ -73,12 +73,44 @@ def spawn_child(acc: dict[str, Any], db_path: str) -> subprocess.Popen:
     return subprocess.Popen(argv, env=env)
 
 
+def _maybe_daily_reconcile(db_path: str, done_dates: set,
+                           now: Optional["datetime.datetime"] = None) -> None:
+    """日终对账(设计 §8.2):收盘后(≥15:10)每日一次全配对跑报表落库。
+
+    API/CLI 也可随时现算;这里只是自动化兜底,done_dates 防当日重复。
+    """
+    import datetime as _dt
+
+    from .reconcile_report import (
+        build_reconcile_report, find_dual_track_pairs, save_reconcile_report,
+    )
+
+    now = now or _dt.datetime.now()
+    if now.hour < 15 or (now.hour == 15 and now.minute < 10):
+        return
+    today = now.strftime("%Y-%m-%d")
+    if today in done_dates:
+        return
+    try:
+        for pair in find_dual_track_pairs(db_path):
+            report = build_reconcile_report(
+                db_path, pair["live_account_id"], pair["shadow_account_id"], today)
+            save_reconcile_report(db_path, report)
+            logger.info("[shadow-supervisor] 日终对账 %s live#%s vs shadow#%s: %s",
+                        today, pair["live_account_id"], pair["shadow_account_id"],
+                        "PASS" if report["passed"] else "FAIL")
+        done_dates.add(today)
+    except Exception as exc:  # noqa: BLE001 - 对账失败不退出主管,下一轮重试
+        logger.warning("[shadow-supervisor] 日终对账失败(下轮重试): %s", exc)
+
+
 def run_auto_supervisor(db_path: Optional[str] = None, poll_sec: float = POLL_SEC) -> None:
     """常驻主循环:同步账户 ↔ 子进程。"""
     db_path = db_path or os.environ.get("SANGUO_SHADOW_DB") \
         or r"C:\sanguo_vnpy_v2\data\backtest_results.db"
     logger.info("[shadow-supervisor] 启动 db=%s", db_path)
     children: Dict[int, subprocess.Popen] = {}
+    reconcile_done: set = set()
 
     while True:
         try:
@@ -109,4 +141,7 @@ def run_auto_supervisor(db_path: Optional[str] = None, poll_sec: float = POLL_SE
             except Exception as exc:  # noqa: BLE001
                 logger.error("[shadow-supervisor] 账户 #%s 拉起失败: %s", aid, exc)
 
+        # 3) 日终双轨对账(收盘后每日一次)
+        _maybe_daily_reconcile(db_path, reconcile_done)
+
         time.sleep(poll_sec)

tests/api/test_paper_routes.py

@@ -191,3 +191,52 @@ def test_pending_endpoint(tmp_path):
     assert data[0]["symbol"] == "600000"
     assert data[0]["side"] == "buy"
     assert data[0]["is_market"] is True
+
+
+# ===== 双轨对账报表(影子 P3 前半)=====
+
+def test_reconcile_routes(tmp_path):
+    """GET /paper/reconcile 自动配对+报告;/paper/reconcile/{l}/{s} 单配对。"""
+    import json as _json
+    import sqlite3
+
+    c, token = _client(tmp_path)
+    db = os.path.join(str(tmp_path), "p.db")
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "INSERT INTO live_accounts (id,name,account,vt_symbol,strategy_class,"
+            "strategy_name,status) VALUES (5,'live','66639661','pool',"
+            "'channel_test','portfolio_channel_test','running')")
+        conn.execute(
+            "INSERT INTO paper_accounts (id,name,strategy_type,mode,status,"
+            "strategies) VALUES (39,'paper','portfolio','shadow','running',"
+            "'[{\"name\": \"channel_test\", \"params\": {}}]')")
+        conn.execute(
+            "INSERT INTO live_trades (account_id,symbol,direction,price,volume,"
+            "traded_at,vt_tradeid) VALUES (5,'510300.SH','buy',4.0,1000,"
+            "'2026-08-15 09:35:00','t1')")
+        conn.execute(
+            "INSERT INTO paper_trades (account_id,strategy_id,datetime,symbol,"
+            "direction,offset,price,volume,rejected,bar_date) VALUES (39,'ct',"
+            "'2026-08-15 09:35:00','510300.XSHG','long','open',4.0,1000,0,"
+            "'2026-08-15')")
+        conn.commit()
+
+    # 自动配对列表(指定 date 保证命中测试数据)
+    r = c.get("/api/v1/paper/reconcile?date=2026-08-15", headers=_auth(token))
+    assert r.status_code == 200
+    pairs = r.json()["pairs"]
+    assert len(pairs) == 1
+    assert pairs[0]["live_account_id"] == 5
+    assert pairs[0]["shadow_account_id"] == 39
+    assert pairs[0]["report"]["trades"]["count_match"] is True
+
+    # 单配对端点
+    r2 = c.get("/api/v1/paper/reconcile/5/39?date=2026-08-15", headers=_auth(token))
+    assert r2.status_code == 200
+    assert r2.json()["trades"]["live_count"] == 1
+
+    # 未配对的 aid 路由不被 reconcile 吞:GET /paper/39 仍走账户详情
+    r3 = c.get("/api/v1/paper/39", headers=_auth(token))
+    assert r3.status_code == 200
+    assert r3.json()["mode"] == "shadow"

tests/trader/test_shadow_reconcile_report.py

@@ -0,0 +1,257 @@
+"""双轨对账报表(影子柜台 vs 实盘模拟,设计 §8.2)单元测试。
+
+纯 DB fixture:同一 db 文件里 live_*/paper_* 两套表(与 VPS backtest_results.db
+同构),验证四项对账指标 + 自动配对 + 报告落库。
+"""
+from __future__ import annotations
+
+import json
+import sqlite3
+
+import pytest
+
+from sanguo_trader.persistence import init_db as init_paper_db
+from sanguo_trader.shadow.reconcile_report import (
+    PRICE_DIFF_BPS_MAX,
+    build_reconcile_report,
+    find_dual_track_pairs,
+    load_reconcile_report,
+    save_reconcile_report,
+)
+
+
+@pytest.fixture()
+def db(tmp_path):
+    db_path = str(tmp_path / "t.db")
+    init_paper_db(db_path)  # paper_* 表
+    from sanguo_live.persistence import init_db as init_live_db
+    init_live_db(db_path)   # live_* 表(同文件共存,与 VPS 一致)
+    return db_path
+
+
+def _add_live_account(db, aid=5, strategy_class="channel_test"):
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "INSERT INTO live_accounts (id,name,account,vt_symbol,strategy_class,"
+            "strategy_name,status) VALUES (?,?,?,?,?,?,?)",
+            (aid, "live-600000", "66639661", "hs300_subset",
+             strategy_class, "portfolio_channel_test", "running"),
+        )
+
+
+def _add_shadow_account(db, aid=39, strategy="channel_test"):
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "INSERT INTO paper_accounts (id,name,strategy_type,mode,status,symbols,"
+            "strategies) VALUES (?,?,?,?,?,?,?)",
+            (aid, "paper", "portfolio", "shadow", "running", '["hs300_subset"]',
+             json.dumps([{"name": strategy, "params": {}}])),
+        )
+
+
+def _add_live_trade(db, aid, symbol, direction, price, volume, traded_at):
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "INSERT INTO live_trades (account_id,strategy_name,symbol,direction,"
+            "offset,price,volume,traded_at,vt_tradeid) VALUES (?,?,?,?,?,?,?,?,?)",
+            (aid, "portfolio_channel_test", symbol, direction, "", price, volume,
+             traded_at, f"t{price}{volume}{symbol}"),
+        )
+
+
+def _add_paper_trade(db, aid, symbol, direction, price, volume, dt, bar_date):
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "INSERT INTO paper_trades (account_id,strategy_id,datetime,symbol,"
+            "direction,offset,match_session,price,volume,commission,stamp_duty,"
+            "transfer_fee,rejected,bar_date) VALUES (?,?,?,?,?,?,?,?,?,?,?,0,0,?)",
+            (aid, "channel_test", dt, symbol, direction, "open",
+             "shadow_realtime", price, volume, 5.0, 0.0, bar_date),
+        )
+
+
+def _add_live_balance(db, aid, date, cash, mv, total):
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "INSERT INTO live_balance (account_id,date,cash,market_value,total) "
+            "VALUES (?,?,?,?,?)", (aid, date, cash, mv, total),
+        )
+
+
+def _add_paper_balance(db, aid, date, cash, mv, total):
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "INSERT INTO paper_daily_balance (account_id,date,cash,market_value,"
+            "total_equity) VALUES (?,?,?,?,?)", (aid, date, cash, mv, total),
+        )
+
+
+def _add_live_position(db, aid, symbol, volume):
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "INSERT INTO live_positions (account_id,symbol,volume,frozen,avg_price,"
+            "updated_at) VALUES (?,?,?,?,?,?)",
+            (aid, symbol, volume, 0.0, 10.0, "2026-08-15 15:00:00"),
+        )
+
+
+def _add_paper_position(db, aid, symbol, volume):
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "INSERT INTO paper_positions (account_id,scope,symbol,date,volume,"
+            "frozen,avg_price,market_value,updated_at) VALUES (?,?,?,?,?,?,?,?,?)",
+            (aid, "account", symbol, "2026-08-15", volume, 0, 10.0, volume * 10.0,
+             "2026-08-15 15:00:00"),
+        )
+
+
+D = "2026-08-15"
+
+
+class TestFindDualTrackPairs:
+    def test_pairs_by_strategy_name(self, db):
+        _add_live_account(db)
+        _add_shadow_account(db)
+        pairs = find_dual_track_pairs(db)
+        assert pairs == [{"live_account_id": 5, "shadow_account_id": 39,
+                          "strategy": "channel_test"}]
+
+    def test_no_shadow_no_pairs(self, db):
+        _add_live_account(db)
+        assert find_dual_track_pairs(db) == []
+
+
+class TestBuildReconcileReport:
+    def test_all_pass_when_both_sides_identical(self, db):
+        _add_live_account(db)
+        _add_shadow_account(db)
+        # 同笔成交(符号口径不同:live 用 600000.SH,shadow 用 600000.XSHG)
+        _add_live_trade(db, 5, "510300.SH", "buy", 4.00, 1000, f"{D} 09:35:00")
+        _add_paper_trade(db, 39, "510300.XSHG", "long", 4.00, 1000,
+                         f"{D} 09:35:00", D)
+        # 持仓一致
+        _add_live_position(db, 5, "510300.SH", 1000)
+        _add_paper_position(db, 39, "510300.XSHG", 1000)
+        # 净值:月初基线同 100 万,当日同 101 万 → 月偏差 0
+        _add_live_balance(db, 5, "2026-08-01", 1_000_000, 0, 1_000_000)
+        _add_live_balance(db, 5, D, 10_000, 1_000_000, 1_010_000)
+        _add_paper_balance(db, 39, "2026-08-01", 1_000_000, 0, 1_000_000)
+        _add_paper_balance(db, 39, D, 10_000, 1_000_000, 1_010_000)
+
+        r = build_reconcile_report(db, 5, 39, D)
+        assert r["trades"]["count_match"] is True
+        assert r["trades"]["live_count"] == 1 and r["trades"]["shadow_count"] == 1
+        assert r["trades"]["rows"][0]["symbol"] == "510300"
+        assert r["trades"]["rows"][0]["price_diff_bps"] == pytest.approx(0, abs=1)
+        assert r["trades"]["pass_price"] is True
+        assert r["positions"]["match"] is True
+        assert r["nav"]["mtd_deviation_pct"] == pytest.approx(0, abs=1e-9)
+        assert r["passed"] is True
+
+    def test_count_mismatch_fails(self, db):
+        _add_live_account(db)
+        _add_shadow_account(db)
+        _add_live_trade(db, 5, "510300.SH", "buy", 4.00, 1000, f"{D} 09:35:00")
+        _add_live_trade(db, 5, "510300.SH", "buy", 4.01, 500, f"{D} 10:00:00")
+        _add_paper_trade(db, 39, "510300.XSHG", "long", 4.00, 1500,
+                         f"{D} 09:35:00", D)
+        r = build_reconcile_report(db, 5, 39, D)
+        assert r["trades"]["count_match"] is False  # 2 vs 1
+
+    def test_price_diff_over_10bps_fails(self, db):
+        _add_live_account(db)
+        _add_shadow_account(db)
+        # 4.004 vs 4.000 = 10bps 边界;4.01 vs 4.00 = 25bps 超限
+        _add_live_trade(db, 5, "510300.SH", "buy", 4.01, 1000, f"{D} 09:35:00")
+        _add_paper_trade(db, 39, "510300.XSHG", "long", 4.00, 1000,
+                         f"{D} 09:35:00", D)
+        r = build_reconcile_report(db, 5, 39, D)
+        assert r["trades"]["rows"][0]["price_diff_bps"] == pytest.approx(25.0, abs=0.5)
+        assert r["trades"]["pass_price"] is False
+        assert PRICE_DIFF_BPS_MAX == 10
+
+    def test_position_volume_mismatch_detected(self, db):
+        _add_live_account(db)
+        _add_shadow_account(db)
+        _add_live_position(db, 5, "510300.SH", 1000)
+        _add_paper_position(db, 39, "510300.XSHG", 800)
+        _add_paper_position(db, 39, "159915.XSHE", 500)  # 影子多出一只
+        r = build_reconcile_report(db, 5, 39, D)
+        assert r["positions"]["match"] is False
+        vols = {row["symbol"]: row for row in r["positions"]["rows"]}
+        assert vols["510300"]["live_volume"] == 1000
+        assert vols["510300"]["shadow_volume"] == 800
+        assert vols["159915"]["live_volume"] == 0
+
+    def test_nav_mtd_deviation_over_threshold_fails(self, db):
+        _add_live_account(db)
+        _add_shadow_account(db)
+        # live 月内 +1.0%,shadow 月内 -0.6% → 偏差 1.6% > 0.5%
+        _add_live_balance(db, 5, "2026-08-01", 1_000_000, 0, 1_000_000)
+        _add_live_balance(db, 5, D, 0, 1_010_000, 1_010_000)
+        _add_paper_balance(db, 39, "2026-08-01", 1_000_000, 0, 1_000_000)
+        _add_paper_balance(db, 39, D, 0, 994_000, 994_000)
+        r = build_reconcile_report(db, 5, 39, D)
+        assert r["nav"]["mtd_deviation_pct"] == pytest.approx(1.6, abs=0.01)
+        assert r["nav"]["pass_nav"] is False
+
+    def test_rejected_shadow_trades_excluded(self, db):
+        _add_live_account(db)
+        _add_shadow_account(db)
+        _add_live_trade(db, 5, "510300.SH", "buy", 4.00, 1000, f"{D} 09:35:00")
+        _add_paper_trade(db, 39, "510300.XSHG", "long", 4.00, 1000,
+                         f"{D} 09:35:00", D)
+        with sqlite3.connect(db) as conn:  # 影子拒单不应计入笔数
+            conn.execute(
+                "INSERT INTO paper_trades (account_id,strategy_id,datetime,symbol,"
+                "direction,price,volume,rejected,reject_reason,bar_date) "
+                "VALUES (39,'channel_test',?,'159915.XSHE','long',2.0,100,1,"
+                "'涨停拒买',?)", (f"{D} 13:45:00", D))
+        r = build_reconcile_report(db, 5, 39, D)
+        assert r["trades"]["shadow_count"] == 1
+
+
+class TestPersistReconcileReport:
+    def test_save_load_roundtrip_and_upsert(self, db):
+        _add_live_account(db)
+        _add_shadow_account(db)
+        _add_paper_trade(db, 39, "510300.XSHG", "long", 4.0, 100, f"{D} 09:35", D)
+        r1 = build_reconcile_report(db, 5, 39, D)
+        save_reconcile_report(db, r1)
+        r1["passed"] = True  # 改一处再存 → upsert 覆盖
+        save_reconcile_report(db, r1)
+        loaded = load_reconcile_report(db, 5, 39, D)
+        assert loaded is not None
+        assert loaded["passed"] is True
+        rows = load_reconcile_report(db, 5, 39, D, as_row=True)
+        assert rows and rows[0]["live_account_id"] == 5
+
+
+class TestDailyReconcileHook:
+    def test_runs_once_after_close_and_skips_before(self, db):
+        """15:10 前不跑;之后跑一次落库,同日第二次跳过。"""
+        from datetime import datetime
+
+        from sanguo_trader.shadow.reconcile_report import load_reconcile_report
+        from sanguo_trader.shadow.supervisor import _maybe_daily_reconcile
+
+        _add_live_account(db)
+        _add_shadow_account(db)
+        _add_live_trade(db, 5, "510300.SH", "buy", 4.0, 1000, f"{D} 09:35:00")
+        _add_paper_trade(db, 39, "510300.XSHG", "long", 4.0, 1000,
+                         f"{D} 09:35:00", D)
+        done: set = set()
+
+        # 盘中 14:00 → 不跑
+        _maybe_daily_reconcile(db, done, now=datetime(2026, 8, 15, 14, 0))
+        assert done == set()
+        assert load_reconcile_report(db, 5, 39, D) is None
+
+        # 收盘后 15:30 → 跑并落库
+        _maybe_daily_reconcile(db, done, now=datetime(2026, 8, 15, 15, 30))
+        assert D in done
+        assert load_reconcile_report(db, 5, 39, D) is not None
+
+        # 同日再触发 → 跳过
+        _maybe_daily_reconcile(db, done, now=datetime(2026, 8, 15, 16, 0))
+        assert done == {D}