diff --git a/sanguo_portfolio/runner_live.py b/sanguo_portfolio/runner_live.py index cedefd4..d4c43b8 100644 --- a/sanguo_portfolio/runner_live.py +++ b/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) diff --git a/tests/api/test_portfolio_live.py b/tests/api/test_portfolio_live.py index 83156ee..3e19a72 100644 --- a/tests/api/test_portfolio_live.py +++ b/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