"""双轨对账报表(影子柜台 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_partial_fill_split_same_total_volume_matches(self, db): """QMT 部分成交拆多行 vs 影子一行,总量一致 → count_match=True。 2026-08-24 修订:旧口径按原始行数比(27 vs 5 恒 False,000887 一笔 3800 股被拆 19 行)纯计数噪音;改按 票+方向 聚合总量比。""" _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 True # 聚合总量 1500 == 1500 assert r["trades"]["live_count"] == 2 and r["trades"]["shadow_count"] == 1 # 原始行数仍如实呈现 def test_aggregated_volume_mismatch_fails(self, db): """聚合总量不一致(真缺一笔/量差)→ count_match=False。""" _add_live_account(db) _add_shadow_account(db) _add_live_trade(db, 5, "510300.SH", "buy", 4.00, 1500, f"{D} 09:35:00") _add_paper_trade(db, 39, "510300.XSHG", "long", 4.00, 1200, f"{D} 09:35:00", D) r = build_reconcile_report(db, 5, 39, D) assert r["trades"]["count_match"] is False # 1500 != 1200 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} class TestFindPairsByInstance: """2026-08-16 v2 配对:同策略多账户按 instance_id 精确配(不再策略名 dict 收敛)。""" def test_same_strategy_multiple_lives_pair_by_instance(self, db): """VPS 实况复刻:live#10/#11 同为 channel_test 不同实例,影子各归各。""" with sqlite3.connect(db) as conn: conn.executemany( "INSERT INTO live_accounts (id,name,account,vt_symbol,strategy_class," "strategy_name,status,instance_id,interval) VALUES (?,?,?,?,?,?,?,?,?)", [(10, "live10", "66639661", "all", "channel_test", "s", "running", 4, "15m"), (11, "live11", "66639661", "all", "channel_test", "s", "running", 3, "d")], ) conn.executemany( "INSERT INTO paper_accounts (id,name,strategy_type,mode,status,symbols," "strategies,instance_id,interval) VALUES (?,?,?,?,?,?,?,?,?)", [(44, "sh44", "portfolio", "shadow", "running", '["all"]', json.dumps([{"name": "channel_test", "params": {}}]), 4, "15m"), (45, "sh45", "portfolio", "shadow", "running", '["all"]', json.dumps([{"name": "channel_test", "params": {}}]), 3, "d")], ) pairs = {(p["live_account_id"], p["shadow_account_id"]): p["strategy"] for p in find_dual_track_pairs(db)} # v1 缺陷:策略名 dict 收敛 → 44/45 都配给 live#11;v2 按 instance 各归各 assert pairs == {(10, 44): "channel_test", (11, 45): "channel_test"} def test_instance_prefers_same_interval(self, db): """同实例不同周期(15m vs d):影子配同周期的 live。""" with sqlite3.connect(db) as conn: conn.executemany( "INSERT INTO live_accounts (id,name,account,vt_symbol,strategy_class," "strategy_name,status,instance_id,interval) VALUES (?,?,?,?,?,?,?,?,?)", [(20, "liveA", "66639661", "all", "all_weather", "s", "running", 1, "15m"), (21, "liveB", "66639661", "all", "all_weather", "s", "running", 1, "d")], ) conn.execute( "INSERT INTO paper_accounts (id,name,strategy_type,mode,status,symbols," "strategies,instance_id,interval) VALUES (?,?,?,?,?,?,?,?,?)", (60, "sh60", "portfolio", "shadow", "running", '["all"]', json.dumps([{"name": "all_weather", "params": {}}]), 1, "d"), ) pairs = find_dual_track_pairs(db) assert pairs == [{"live_account_id": 21, "shadow_account_id": 60, "strategy": "all_weather"}] # ===== B5 恒等式对账(spec §multi-strategy-instance-budget §B5) ===== from sanguo_trader.shadow.reconcile_report import ( # noqa: E402 IDENTITY_TOL_PCT, build_identity_report, load_identity_report, save_identity_report, ) def _seed_snapshot(db, account="66639661", cash=1e6, mv=1e6, positions=None): from sanguo_live.persistence import upsert_account_snapshot upsert_account_snapshot(db, account, cash=cash, market_value=mv, total=cash + mv, positions=positions or []) def _seed_inst_mv(db, aid, mv, positions=None): """实例账本:live_balance 最新市值 + live_positions 视图。""" with sqlite3.connect(db) as conn: conn.execute( "INSERT INTO live_balance (account_id,date,cash,market_value,total) " "VALUES (?,?,?,?,?)", (aid, "2026-08-19 15:00:00", 0, mv, mv)) for sym, vol in (positions or {}).items(): conn.execute( "INSERT INTO live_positions (account_id,symbol,volume,frozen," "avg_price,updated_at) VALUES (?,?,?,?,?,?)", (aid, sym, vol, 0, 10, "x")) def test_identity_pass_within_tolerance(db): """Σ实例市值≈快照市值(容差内)→ pass;逐票对账无未归因。""" _add_live_account(db, aid=5) _seed_snapshot(db, mv=1_000_000, positions=[ {"symbol": "600036.SH", "volume": 1000, "can_use": 1000, "avg_price": 38, "mv": 38000}]) _seed_inst_mv(db, 5, mv=997_000, positions={"600036.XSHG": 1000}) r = build_identity_report(db, "2026-08-19") assert len(r["rows"]) == 1 row = r["rows"][0] assert row["status"] == "pass" # 0.3% < 0.5% 容差 assert row["instance_mv_total"] == 997_000 assert row["unattributed_mv"] == 3_000 assert row["unattributed_positions"] == [] # 逐票对齐 assert r["identity_passed"] is True def test_identity_unattributed_position_listed(db): """快照有实例没有的票(遗留/手动仓)→ 未归因票单列+超容差 FAIL。""" _add_live_account(db, aid=5) _seed_snapshot(db, mv=1_000_000, positions=[ {"symbol": "600036.SH", "volume": 1000, "can_use": 1000, "avg_price": 38, "mv": 38000}, {"symbol": "518880.SH", "volume": 5000, "can_use": 5000, "avg_price": 7, "mv": 35000}]) # 黄金ETF=手动仓 _seed_inst_mv(db, 5, mv=500_000, positions={"600036.XSHG": 1000}) r = build_identity_report(db) row = r["rows"][0] assert row["status"] == "unattributed_over_tol" assert row["unattributed_mv"] == 500_000 unattr = row["unattributed_positions"] assert [p["symbol"] for p in unattr] == ["518880"] assert unattr[0]["diff"] == 5000 assert r["identity_passed"] is False def test_identity_instance_over_snapshot_negative(db): """Σ实例>快照(旧全账户行叠加期)→ 未归因为负,如实呈现 FAIL。""" _add_live_account(db, aid=5) _seed_snapshot(db, mv=1_000_000) _seed_inst_mv(db, 5, mv=8_000_000) # dae56e2 前的全账户行 r = build_identity_report(db) assert r["rows"][0]["unattributed_mv"] == -7_000_000 assert r["rows"][0]["status"] == "unattributed_over_tol" def test_identity_snapshot_missing_and_no_instances(db): """无快照→snapshot_missing;有快照无实例→no_instances(不算 FAIL)。""" _add_live_account(db, aid=5) # 实例无快照 r = build_identity_report(db) assert r["rows"][0]["status"] == "snapshot_missing" assert r["identity_passed"] is False # 反向:快照在、实例删光(重建期) with sqlite3.connect(db) as conn: conn.execute("DELETE FROM live_accounts") _seed_snapshot(db, mv=1_000_000) r2 = build_identity_report(db) assert r2["rows"][0]["status"] == "no_instances" assert r2["identity_passed"] is True # 无实例=无可归因,恒等式成立 def test_identity_multi_instance_sum(db): """多实例共享账户:Σ逐实例市值。""" with sqlite3.connect(db) as conn: conn.executemany( "INSERT INTO live_accounts (id,name,account,vt_symbol,strategy_class," "strategy_name,status) VALUES (?,?,?,?,?,?,?)", [(5, "a", "66639661", "x", "s", "s", "running"), (6, "b", "66639661", "x", "s", "s", "running"), (7, "c", "OTHER", "x", "s", "s", "running")]) _seed_snapshot(db, mv=1_000_000) _seed_snapshot(db, account="OTHER", mv=500_000) _seed_inst_mv(db, 5, 400_000) _seed_inst_mv(db, 6, 595_000) _seed_inst_mv(db, 7, 500_000) r = build_identity_report(db) by_acc = {row["account"]: row for row in r["rows"]} assert by_acc["66639661"]["instance_mv_total"] == 995_000 assert by_acc["66639661"]["status"] == "pass" assert by_acc["OTHER"]["status"] == "pass" assert len(by_acc["66639661"]["instances"]) == 2 # 不串账号 def test_identity_save_load_roundtrip(db): _add_live_account(db, aid=5) _seed_snapshot(db, mv=1_000_000) _seed_inst_mv(db, 5, 1_000_000) r = build_identity_report(db, "2026-08-19") save_identity_report(db, r) loaded = load_identity_report(db, "2026-08-19") assert len(loaded) == 1 assert loaded[0]["status"] == "pass" assert loaded[0]["account"] == "66639661" assert load_identity_report(db, "1999-01-01") == [] def test_identity_tolerance_constant(): """容差 0.5%(spec §B5:价格时点差)。""" assert IDENTITY_TOL_PCT == 0.5