diff --git a/sanguo_trader/shadow/reconcile_report.py b/sanguo_trader/shadow/reconcile_report.py index 095d9d0..1fd228b 100644 --- a/sanguo_trader/shadow/reconcile_report.py +++ b/sanguo_trader/shadow/reconcile_report.py @@ -202,6 +202,138 @@ def build_reconcile_report(db: str, live_account_id: int, shadow_account_id: int } +# ===== B5 恒等式对账(spec §multi-strategy-instance-budget §B5) ===== +# 全账户 = Σ实例账本 + 未归因遗留仓;先过恒等式,再做逐对 live↔shadow 行为对比。 + +IDENTITY_TOL_PCT = 0.5 # 未归因占账户市值容忍上限(%;价格时点差) + + +def build_identity_report(db: str, date: Optional[str] = None) -> Dict[str, Any]: + """账户恒等式报告:每 QMT 账号一行,snapshot 市值 vs Σ实例账本市值。 + + - Σ实例市值 = 同账号各实盘实例 live_balance 最新一条 market_value 之和 + (dae56e2 起为实例账本视图;旧全账户行会如实呈现为大额负未归因)。 + - 未归因仓 = 按 6 位码逐票对账:snapshot 持仓 − Σ实例持仓(live_positions)。 + - 无快照/无实例的账号如实标注(snapshot_missing / no_instances)。 + """ + date = date or datetime.now().strftime("%Y-%m-%d") + rows: List[Dict[str, Any]] = [] + with sqlite3.connect(db) as conn: + conn.row_factory = sqlite3.Row + snapshots = {r["account"]: dict(r) for r in conn.execute( + "SELECT * FROM qmt_account_snapshot")} + # 同 QMT 账号分组:实例最新账本 + 实例持仓视图 + inst_rows = [dict(r) for r in conn.execute( + "SELECT id, name, account FROM live_accounts")] + by_acc: Dict[str, List[Dict[str, Any]]] = {} + for r in inst_rows: + by_acc.setdefault((r.get("account") or "").strip(), []).append(r) + accounts = sorted(set(snapshots) | {a for a in by_acc if a}) + for acc in accounts: + snap = snapshots.get(acc) + insts = by_acc.get(acc, []) + mv_total = 0.0 + for inst in insts: + last = conn.execute( + "SELECT market_value FROM live_balance " + "WHERE account_id=? ORDER BY id DESC LIMIT 1", + (inst["id"],)).fetchone() + inst["mv"] = float(last[0] or 0) if last else 0.0 + mv_total += inst["mv"] + snap_mv = float(snap["market_value"]) if snap else None + unattr_mv = (snap_mv - mv_total) if snap_mv is not None else None + unattr_pct = (abs(unattr_mv) / snap_mv * 100 + if snap_mv not in (None, 0) and unattr_mv is not None + else None) + # 逐票未归因:快照持仓 − Σ实例持仓(6 位码对齐) + snap_pos: Dict[str, float] = {} + if snap: + try: + for p in json.loads(snap.get("positions") or "[]"): + snap_pos[_norm_symbol(p.get("symbol"))] = float( + p.get("volume") or 0) + except (ValueError, TypeError): + snap_pos = {} + inst_pos: Dict[str, float] = {} + for inst in insts: + for r in conn.execute( + "SELECT symbol, volume FROM live_positions " + "WHERE account_id=?", (inst["id"],)): + inst_pos[_norm_symbol(r[0])] = inst_pos.get( + _norm_symbol(r[0]), 0.0) + float(r[1] or 0) + unattr_positions = [ + {"symbol": s, "snapshot_volume": v, + "instance_volume": inst_pos.get(s, 0.0), + "diff": v - inst_pos.get(s, 0.0)} + for s, v in sorted(snap_pos.items()) + if abs(v - inst_pos.get(s, 0.0)) > 0.5 + ] + if snap is None: + status = "snapshot_missing" + elif not insts: + status = "no_instances" + elif unattr_pct is not None and unattr_pct <= IDENTITY_TOL_PCT: + status = "pass" + else: + status = "unattributed_over_tol" + rows.append({ + "account": acc, "date": date, "status": status, + "snapshot_mv": snap_mv, "instance_mv_total": mv_total, + "unattributed_mv": unattr_mv, + "unattributed_pct": round(unattr_pct, 4) if unattr_pct is not None else None, + "tolerance_pct": IDENTITY_TOL_PCT, + "instances": [{"id": i["id"], "name": i["name"], "mv": i["mv"]} + for i in insts], + "unattributed_positions": unattr_positions, + }) + passed = all(r["status"] in ("pass", "no_instances") for r in rows) \ + and bool(rows) + return {"date": date, "rows": rows, "identity_passed": passed} + + +_IDENTITY_SCHEMA = """ +CREATE TABLE IF NOT EXISTS identity_reconcile ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + account TEXT, + date TEXT, + passed INTEGER, + report TEXT, + created_at TEXT, + UNIQUE(account, date) +) +""" + + +def save_identity_report(db: str, report: Dict[str, Any]) -> None: + """恒等式报告落库(同账号同日覆盖)。""" + with sqlite3.connect(db) as conn: + conn.execute(_IDENTITY_SCHEMA) + for row in report["rows"]: + conn.execute( + "INSERT INTO identity_reconcile (account, date, passed, report, created_at) " + "VALUES (?,?,?,?,?) ON CONFLICT(account, date) DO UPDATE SET " + "passed=excluded.passed, report=excluded.report, " + "created_at=excluded.created_at", + (row["account"], report["date"], + 1 if row["status"] in ("pass", "no_instances") else 0, + json.dumps(row, ensure_ascii=False), + datetime.now().strftime("%Y-%m-%d %H:%M:%S")), + ) + conn.commit() + + +def load_identity_report(db: str, date: str) -> List[Dict[str, Any]]: + """读已存恒等式行(无 → 空表)。""" + with sqlite3.connect(db) as conn: + try: + rows = conn.execute( + "SELECT report FROM identity_reconcile WHERE date=? ORDER BY account", + (date,)).fetchall() + except sqlite3.OperationalError: + return [] + return [json.loads(r[0]) for r in rows] + + def find_dual_track_pairs(db: str) -> List[Dict[str, Any]]: """自动配对(2026-08-16 v2):优先 instance_id+周期精确配对,无 instance 回退策略名。 @@ -322,6 +454,18 @@ def main() -> None: pairs = find_dual_track_pairs(args.db) if not pairs: logger.warning("未找到运行中的双轨配对(影子 mode=shadow ↔ live 同策略名)") + # B5 恒等式先行:全账户 = Σ实例账本 + 未归因,再做逐对行为对比 + identity = build_identity_report(args.db, args.date) + save_identity_report(args.db, identity) + for row in identity["rows"]: + logger.info( + "[恒等式] %s %s: 快照市值=%.0f Σ实例=%.0f 未归因=%.0f(%s%%) " + "未归因票%d只 → %s", + identity["date"], row["account"], + row["snapshot_mv"] or 0, row["instance_mv_total"], + row["unattributed_mv"] or 0, row["unattributed_pct"], + len(row["unattributed_positions"]), row["status"], + ) for pair in pairs: r = build_reconcile_report(args.db, pair["live_account_id"], pair["shadow_account_id"], args.date) diff --git a/sanguo_trader/shadow/supervisor.py b/sanguo_trader/shadow/supervisor.py index 288c2d7..ab06247 100644 --- a/sanguo_trader/shadow/supervisor.py +++ b/sanguo_trader/shadow/supervisor.py @@ -97,7 +97,8 @@ def _maybe_daily_reconcile(db_path: str, done_dates: set, import datetime as _dt from .reconcile_report import ( - build_reconcile_report, find_dual_track_pairs, save_reconcile_report, + build_identity_report, build_reconcile_report, find_dual_track_pairs, + save_identity_report, save_reconcile_report, ) now = now or _dt.datetime.now() @@ -107,6 +108,13 @@ def _maybe_daily_reconcile(db_path: str, done_dates: set, if today in done_dates: return try: + # B5 恒等式先行:全账户 = Σ实例账本 + 未归因;再逐对 live↔shadow 行为对比 + identity = build_identity_report(db_path, today) + save_identity_report(db_path, identity) + for row in identity["rows"]: + logger.info("[shadow-supervisor] 恒等式 %s %s: 未归因=%.0f → %s", + today, row["account"], + row["unattributed_mv"] or 0, row["status"]) for pair in find_dual_track_pairs(db_path): report = build_reconcile_report( db_path, pair["live_account_id"], pair["shadow_account_id"], today) diff --git a/tests/trader/test_shadow_reconcile_report.py b/tests/trader/test_shadow_reconcile_report.py index 950c2ad..cb62b61 100644 --- a/tests/trader/test_shadow_reconcile_report.py +++ b/tests/trader/test_shadow_reconcile_report.py @@ -300,3 +300,131 @@ class TestFindPairsByInstance: 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