From 534a06aad7945734605504d04433cf04dc575939 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Fri, 14 Aug 2026 23:18:07 +0800 Subject: [PATCH] =?UTF-8?q?feat(shadow-desk):=20=E5=8F=8C=E8=BD=A8?= =?UTF-8?q?=E6=97=A5=E7=BB=88=E5=AF=B9=E8=B4=A6=E6=8A=A5=E8=A1=A8(?= =?UTF-8?q?=E5=BD=B1=E5=AD=90P3=E5=89=8D=E5=8D=8A,=E8=AE=BE=E8=AE=A1=C2=A7?= =?UTF-8?q?8.2)=E2=80=94reconcile=5Freport=E5=9B=9B=E6=8C=87=E6=A0=87(?= =?UTF-8?q?=E6=88=90=E4=BA=A4=E7=AC=94=E6=95=B0=E5=85=A8=E5=90=8C/?= =?UTF-8?q?=E6=AF=8F=E7=AC=94=E4=BB=B7=E5=B7=AEvwap=E5=AF=B9=E6=AF=94<10bp?= =?UTF-8?q?s=E5=B8=A6=E7=AC=A6=E5=8F=B7=E4=BE=9B=E6=BB=91=E7=82=B9?= =?UTF-8?q?=E9=87=8D=E6=A0=87/=E6=94=B6=E7=9B=98=E6=8C=81=E4=BB=93?= =?UTF-8?q?=E9=80=90=E5=8F=AA=E6=95=B0=E9=87=8F/=E5=87=80=E5=80=BC?= =?UTF-8?q?=E6=9C=88=E5=81=8F=E5=B7=AE<0.5%)+find=5Fdual=5Ftrack=5Fpairs?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=E9=85=8D=E5=AF=B9(shadow=E7=AD=96=E7=95=A5?= =?UTF-8?q?=E5=90=8D=E2=86=94live=20strategy=5Fclass,live#5=E2=86=94shadow?= =?UTF-8?q?#39=E5=AE=9E=E8=AF=81)+dual=5Ftrack=5Freconcile=E8=90=BD?= =?UTF-8?q?=E5=BA=93upsert;API=20GET=20/paper/reconcile(+=E5=8D=95?= =?UTF-8?q?=E9=85=8D=E5=AF=B9refresh);CLI=20python=20-m;supervisor=20auto?= =?UTF-8?q?=E8=BD=AE=E8=AF=A2=E6=8C=8215:10=E5=90=8E=E6=AF=8F=E6=97=A5?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=E5=85=9C=E5=BA=95;14=E6=96=B0=E6=B5=8B?= =?UTF-8?q?=E8=AF=95(=E7=AC=A6=E5=8F=B7=E5=8F=A3=E5=BE=84SH/XSHG=E5=BD=92?= =?UTF-8?q?=E4=B8=80/=E5=BD=B1=E5=AD=90=E6=8B=92=E5=8D=95=E4=B8=8D?= =?UTF-8?q?=E8=AE=A1/=E9=83=A8=E5=88=86=E6=88=90=E4=BA=A4=E5=A6=82?= =?UTF-8?q?=E5=AE=9E=E6=8A=A5=E7=AC=94=E6=95=B0=E5=BC=82)=20[vps]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sanguo_api/routes_paper.py | 45 +++ sanguo_trader/shadow/reconcile_report.py | 320 +++++++++++++++++++ sanguo_trader/shadow/supervisor.py | 35 ++ tests/api/test_paper_routes.py | 49 +++ tests/trader/test_shadow_reconcile_report.py | 257 +++++++++++++++ 5 files changed, 706 insertions(+) create mode 100644 sanguo_trader/shadow/reconcile_report.py create mode 100644 tests/trader/test_shadow_reconcile_report.py diff --git a/sanguo_api/routes_paper.py b/sanguo_api/routes_paper.py index 1712984..85acb3f 100644 --- a/sanguo_api/routes_paper.py +++ b/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"] diff --git a/sanguo_trader/shadow/reconcile_report.py b/sanguo_trader/shadow/reconcile_report.py new file mode 100644 index 0000000..1353cf0 --- /dev/null +++ b/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 [--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 [--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() diff --git a/sanguo_trader/shadow/supervisor.py b/sanguo_trader/shadow/supervisor.py index 8bf546e..1c9ff1f 100644 --- a/sanguo_trader/shadow/supervisor.py +++ b/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) diff --git a/tests/api/test_paper_routes.py b/tests/api/test_paper_routes.py index 2358be4..618906a 100644 --- a/tests/api/test_paper_routes.py +++ b/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" diff --git a/tests/trader/test_shadow_reconcile_report.py b/tests/trader/test_shadow_reconcile_report.py new file mode 100644 index 0000000..261cd1d --- /dev/null +++ b/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}