feat(trader): B5对账恒等式优先——15:10日终报表加『全账户=Σ实例账本+未归因』层(spec§B5)——①build_identity_report:每QMT账号一行,快照市值vs Σ实例账本市值(live_balance最新,dae56e2起=实例视图),未归因MV+占比,容差0.5%(价格时点差);逐票未归因=快照持仓−Σ实例持仓(6位码对齐)单列(重建后应≈0,大数=遗留/手动仓)②状态四态:pass/no_instances(重建期无实例=恒等式成立)/snapshot_missing(如实FAIL)/unattributed_over_tol(旧全账户行叠加期呈现大额负未归因)③save/load落库identity_reconcile(account+date主键)④15:10调度(_maybe_daily_reconcile)与CLI恒等式先行,再逐对live↔shadow(配对v2不变);+7测试(容差内过/未归因票单列/负未归因/无快照/无实例/多实例Σ不串账号/落库回读);trader 248绿 [vps]

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)

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)

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