From ed15c24db8fc904f9c5b634fbef6140a3376da08 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Wed, 19 Aug 2026 22:32:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(trader):=20B5=E5=AF=B9=E8=B4=A6=E6=81=92?= =?UTF-8?q?=E7=AD=89=E5=BC=8F=E4=BC=98=E5=85=88=E2=80=94=E2=80=9415:10?= =?UTF-8?q?=E6=97=A5=E7=BB=88=E6=8A=A5=E8=A1=A8=E5=8A=A0=E3=80=8E=E5=85=A8?= =?UTF-8?q?=E8=B4=A6=E6=88=B7=3D=CE=A3=E5=AE=9E=E4=BE=8B=E8=B4=A6=E6=9C=AC?= =?UTF-8?q?+=E6=9C=AA=E5=BD=92=E5=9B=A0=E3=80=8F=E5=B1=82(spec=C2=A7B5)?= =?UTF-8?q?=E2=80=94=E2=80=94=E2=91=A0build=5Fidentity=5Freport:=E6=AF=8FQ?= =?UTF-8?q?MT=E8=B4=A6=E5=8F=B7=E4=B8=80=E8=A1=8C,=E5=BF=AB=E7=85=A7?= =?UTF-8?q?=E5=B8=82=E5=80=BCvs=20=CE=A3=E5=AE=9E=E4=BE=8B=E8=B4=A6?= =?UTF-8?q?=E6=9C=AC=E5=B8=82=E5=80=BC(live=5Fbalance=E6=9C=80=E6=96=B0,da?= =?UTF-8?q?e56e2=E8=B5=B7=3D=E5=AE=9E=E4=BE=8B=E8=A7=86=E5=9B=BE),?= =?UTF-8?q?=E6=9C=AA=E5=BD=92=E5=9B=A0MV+=E5=8D=A0=E6=AF=94,=E5=AE=B9?= =?UTF-8?q?=E5=B7=AE0.5%(=E4=BB=B7=E6=A0=BC=E6=97=B6=E7=82=B9=E5=B7=AE);?= =?UTF-8?q?=E9=80=90=E7=A5=A8=E6=9C=AA=E5=BD=92=E5=9B=A0=3D=E5=BF=AB?= =?UTF-8?q?=E7=85=A7=E6=8C=81=E4=BB=93=E2=88=92=CE=A3=E5=AE=9E=E4=BE=8B?= =?UTF-8?q?=E6=8C=81=E4=BB=93(6=E4=BD=8D=E7=A0=81=E5=AF=B9=E9=BD=90)?= =?UTF-8?q?=E5=8D=95=E5=88=97(=E9=87=8D=E5=BB=BA=E5=90=8E=E5=BA=94?= =?UTF-8?q?=E2=89=880,=E5=A4=A7=E6=95=B0=3D=E9=81=97=E7=95=99/=E6=89=8B?= =?UTF-8?q?=E5=8A=A8=E4=BB=93)=E2=91=A1=E7=8A=B6=E6=80=81=E5=9B=9B?= =?UTF-8?q?=E6=80=81:pass/no=5Finstances(=E9=87=8D=E5=BB=BA=E6=9C=9F?= =?UTF-8?q?=E6=97=A0=E5=AE=9E=E4=BE=8B=3D=E6=81=92=E7=AD=89=E5=BC=8F?= =?UTF-8?q?=E6=88=90=E7=AB=8B)/snapshot=5Fmissing(=E5=A6=82=E5=AE=9EFAIL)/?= =?UTF-8?q?unattributed=5Fover=5Ftol(=E6=97=A7=E5=85=A8=E8=B4=A6=E6=88=B7?= =?UTF-8?q?=E8=A1=8C=E5=8F=A0=E5=8A=A0=E6=9C=9F=E5=91=88=E7=8E=B0=E5=A4=A7?= =?UTF-8?q?=E9=A2=9D=E8=B4=9F=E6=9C=AA=E5=BD=92=E5=9B=A0)=E2=91=A2save/loa?= =?UTF-8?q?d=E8=90=BD=E5=BA=93identity=5Freconcile(account+date=E4=B8=BB?= =?UTF-8?q?=E9=94=AE)=E2=91=A315:10=E8=B0=83=E5=BA=A6(=5Fmaybe=5Fdaily=5Fr?= =?UTF-8?q?econcile)=E4=B8=8ECLI=E6=81=92=E7=AD=89=E5=BC=8F=E5=85=88?= =?UTF-8?q?=E8=A1=8C,=E5=86=8D=E9=80=90=E5=AF=B9live=E2=86=94shadow(?= =?UTF-8?q?=E9=85=8D=E5=AF=B9v2=E4=B8=8D=E5=8F=98);+7=E6=B5=8B=E8=AF=95(?= =?UTF-8?q?=E5=AE=B9=E5=B7=AE=E5=86=85=E8=BF=87/=E6=9C=AA=E5=BD=92?= =?UTF-8?q?=E5=9B=A0=E7=A5=A8=E5=8D=95=E5=88=97/=E8=B4=9F=E6=9C=AA?= =?UTF-8?q?=E5=BD=92=E5=9B=A0/=E6=97=A0=E5=BF=AB=E7=85=A7/=E6=97=A0?= =?UTF-8?q?=E5=AE=9E=E4=BE=8B/=E5=A4=9A=E5=AE=9E=E4=BE=8B=CE=A3=E4=B8=8D?= =?UTF-8?q?=E4=B8=B2=E8=B4=A6=E5=8F=B7/=E8=90=BD=E5=BA=93=E5=9B=9E?= =?UTF-8?q?=E8=AF=BB);trader=20248=E7=BB=BF=20[vps]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sanguo_trader/shadow/reconcile_report.py | 144 +++++++++++++++++++ sanguo_trader/shadow/supervisor.py | 10 +- tests/trader/test_shadow_reconcile_report.py | 128 +++++++++++++++++ 3 files changed, 281 insertions(+), 1 deletion(-) 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