fix(live): 组合实盘快照守卫——cash<=0(持仓先到资金未同步)不落balance;2026-08-14实况=首条total=2931成收益率基线→前端341080%假收益率;抽出_snapshot_once可测+回归测试 [vps]

sanguo_portfolio/runner_live.py

@@ -60,38 +60,53 @@ def live_env() -> Dict[str, str]:
     }
 
 
-def _snapshot_loop(engine: Any, db: str, account_id: int,
-                   interval_sec: float = 60.0) -> None:
-    """后台线程:把 engine 组合快照落 live_positions/live_balance(供 API 读)。
+def _snapshot_once(engine: Any, db: str, account_id: int) -> None:
+    """单次快照:portfolio → live_positions/live_balance。
 
-    LiveEngine 的账户/持仓由 broker 同步进 context.portfolio(LivePortfolioProxy),
-    这里只读转储;任何异常只 warning 不中断(engine 主循环不受影响)。
+    现金<=0 视为「broker 账户尚未同步完成」跳过 balance 落库:
+    QMT 持仓先到、资金后到时 total=持仓市值(无现金),写库会成为前端
+    收益率的基线 → 假收益率 341080%(2026-08-14 实况)。满仓账户的
+    cash 本就≈0,此情形少牺牲(balance 少几条,positions 照落)。
     """
     from datetime import datetime
 
     from sanguo_live.persistence import save_balance, save_positions
 
+    portfolio = engine.context.portfolio
+    positions: Dict[str, Dict[str, Any]] = {}
+    for sym, pos in (getattr(portfolio, "positions", None) or {}).items():
+        vol = int(getattr(pos, "total_amount", 0) or 0)
+        if vol <= 0:
+            continue
+        positions[str(sym)] = {
+            "volume": float(vol),
+            "frozen": float(vol - int(getattr(pos, "closeable_amount", vol) or 0)),
+            "avg_price": float(getattr(pos, "avg_cost", 0) or 0),
+        }
+    save_positions(db, account_id, positions)
+    cash = float(getattr(portfolio, "available_cash", 0) or 0)
+    total = float(getattr(portfolio, "total_value", 0) or 0)
+    if cash <= 0:
+        logger.info("[live-snapshot] cash=%s(账户未同步完成?),跳过 balance "
+                    "(account=%s total=%s)", cash, account_id, total)
+        return
+    save_balance(
+        db, account_id, datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+        cash, market_value=max(total - cash, 0.0), total=total,
+    )
+
+
+def _snapshot_loop(engine: Any, db: str, account_id: int,
+                   interval_sec: float = 60.0) -> None:
+    """后台线程:定时把 engine 组合快照落库(供 API 读)。
+
+    LiveEngine 的账户/持仓由 broker 同步进 context.portfolio(LivePortfolioProxy),
+    这里只读转储;任何异常只 warning 不中断(engine 主循环不受影响)。
+    """
     while True:
         time.sleep(interval_sec)
         try:
-            portfolio = engine.context.portfolio
-            positions: Dict[str, Dict[str, Any]] = {}
-            for sym, pos in (getattr(portfolio, "positions", None) or {}).items():
-                vol = int(getattr(pos, "total_amount", 0) or 0)
-                if vol <= 0:
-                    continue
-                positions[str(sym)] = {
-                    "volume": float(vol),
-                    "frozen": float(vol - int(getattr(pos, "closeable_amount", vol) or 0)),
-                    "avg_price": float(getattr(pos, "avg_cost", 0) or 0),
-                }
-            save_positions(db, account_id, positions)
-            cash = float(getattr(portfolio, "available_cash", 0) or 0)
-            total = float(getattr(portfolio, "total_value", 0) or 0)
-            save_balance(
-                db, account_id, datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-                cash, market_value=max(total - cash, 0.0), total=total,
-            )
+            _snapshot_once(engine, db, account_id)
         except Exception as e:  # noqa: BLE001
             logger.warning("[live-snapshot] 落库失败 (account=%s): %s", account_id, e)
 

tests/api/test_portfolio_live.py

@@ -188,3 +188,31 @@ def test_update_live_normalizes_vt_symbol(tmp_path, monkeypatch):
     assert r.status_code == 200
     acc = c.get(f"/api/v1/live/{aid}", headers=h).json()
     assert acc["vt_symbol"] == "300024.SZSE"
+
+
+def test_snapshot_once_skips_unsynced_cash():
+    """cash<=0(账户未同步完成)不落 balance——治假收益率(2026-08-14 实况 341080%)。"""
+    import os
+    import tempfile
+    import types
+
+    from sanguo_live.persistence import init_db, list_balance
+    from sanguo_portfolio.runner_live import _snapshot_once
+
+    db = os.path.join(tempfile.mkdtemp(), "l.db")
+    init_db(db)
+
+    def _mk_portfolio(cash, total):
+        p = types.SimpleNamespace(
+            available_cash=cash, total_value=total, positions={})
+        ctx = types.SimpleNamespace(portfolio=p)
+        return types.SimpleNamespace(context=ctx)
+
+    # 未同步完成: cash=0, total=持仓市值 → 不落 balance
+    _snapshot_once(_mk_portfolio(0, 2931.0), db, 3)
+    assert list_balance(db, 3) == []
+    # 正常: cash>0 → 落库
+    _snapshot_once(_mk_portfolio(9_997_077.51, 10_000_008.51), db, 3)
+    rows = list_balance(db, 3)
+    assert len(rows) == 1
+    assert rows[0]["total"] == 10_000_008.51