feat(persistence): C-S3实走跨日状态—paper_pending_orders+positions/last_balance存取

sanguo_trader/persistence.py

@@ -1,6 +1,7 @@
-"""模拟盘 SQLite 持久化（4 表 + checkpoint + WAL，spec §8 / §9.1）。
+"""模拟盘 SQLite 持久化（4 表 + pending + checkpoint + WAL，spec §8 / §9.1）。
 
 表：paper_accounts / paper_trades / paper_positions / paper_daily_balance
+    / paper_pending_orders（C-S3 实走跨日 pending）
 WAL 模式支持 worker 进程写 + 主进程读（spec §9.1 共享 DB 进度）。
 """
 import json
@@ -43,6 +44,13 @@ CREATE TABLE IF NOT EXISTS paper_daily_balance (
     cash REAL, market_value REAL, total_equity REAL,
     per_strategy_pnl TEXT, is_checkpoint INTEGER
 );
+CREATE TABLE IF NOT EXISTS paper_pending_orders (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    account_id INTEGER, strategy_id TEXT, symbol TEXT,
+    side TEXT, price REAL, volume INTEGER,
+    is_market INTEGER, match_session TEXT, listing_days INTEGER,
+    created_at TEXT
+);
 """
 
 
@@ -192,3 +200,91 @@ def list_strategy_summary(db_path: str, account_id: int) -> list[dict]:
             (account_id,),
         )
         return [dict(r) for r in cur.fetchall()]
+
+
+# ---- C-S3 实走跨日状态（pending 订单 + positions 恢复）----
+
+def save_pending_orders(db_path: str, account_id: int, orders: list[dict]) -> None:
+    """持久化跨日 pending 订单（NEXT_OPEN 昨日发今日撮合）。覆盖式（每根 step 后重写）。"""
+    with sqlite3.connect(db_path) as conn:
+        conn.execute(
+            "DELETE FROM paper_pending_orders WHERE account_id=?", (account_id,)
+        )
+        conn.executemany(
+            """INSERT INTO paper_pending_orders
+               (account_id, strategy_id, symbol, side, price, volume,
+                is_market, match_session, listing_days, created_at)
+               VALUES (?,?,?,?,?,?,?,?,?,?)""",
+            [
+                (account_id, o.get("strategy_id", ""), o.get("symbol", ""),
+                 o.get("side", ""), o.get("price", 0), o.get("volume", 0),
+                 1 if o.get("is_market") else 0, o.get("match_session", ""),
+                 o.get("listing_days", 0), _now())
+                for o in orders
+            ],
+        )
+        conn.commit()
+
+
+def load_pending_orders(db_path: str, account_id: int) -> list[dict]:
+    """恢复跨日 pending 订单（实走 step 前读）。"""
+    with sqlite3.connect(db_path) as conn:
+        conn.row_factory = sqlite3.Row
+        cur = conn.execute(
+            "SELECT strategy_id, symbol, side, price, volume, "
+            "is_market, match_session, listing_days "
+            "FROM paper_pending_orders WHERE account_id=? ORDER BY id",
+            (account_id,),
+        )
+        rows = [dict(r) for r in cur.fetchall()]
+        for r in rows:
+            r["is_market"] = bool(r["is_market"])
+        return rows
+
+
+def save_positions(db_path: str, account_id: int, scope: str,
+                   positions: dict, date: str) -> None:
+    """持久化持仓快照（实走跨日恢复）。positions={symbol:{volume,frozen,avg_price}}。"""
+    with sqlite3.connect(db_path) as conn:
+        conn.execute(
+            "DELETE FROM paper_positions WHERE account_id=? AND scope=?",
+            (account_id, scope),
+        )
+        conn.executemany(
+            """INSERT INTO paper_positions
+               (account_id, scope, symbol, date, volume, frozen, avg_price, updated_at)
+               VALUES (?,?,?,?,?,?,?,?)""",
+            [
+                (account_id, scope, sym, date, p["volume"], p.get("frozen", 0),
+                 p["avg_price"], _now())
+                for sym, p in positions.items() if p.get("volume", 0) > 0
+            ],
+        )
+        conn.commit()
+
+
+def load_positions(db_path: str, account_id: int, scope: str) -> dict:
+    """恢复持仓快照 → {symbol:{volume,frozen,avg_price}}。"""
+    with sqlite3.connect(db_path) as conn:
+        conn.row_factory = sqlite3.Row
+        cur = conn.execute(
+            "SELECT symbol, volume, frozen, avg_price FROM paper_positions "
+            "WHERE account_id=? AND scope=?",
+            (account_id, scope),
+        )
+        return {r["symbol"]: {"volume": r["volume"], "frozen": r["frozen"],
+                              "avg_price": r["avg_price"]}
+                for r in cur.fetchall()}
+
+
+def load_last_balance(db_path: str, account_id: int) -> dict | None:
+    """实走恢复：最后一条余额（cash 用于恢复 account.cash）。"""
+    with sqlite3.connect(db_path) as conn:
+        conn.row_factory = sqlite3.Row
+        cur = conn.execute(
+            "SELECT cash, market_value, total_equity, date FROM paper_daily_balance "
+            "WHERE account_id=? ORDER BY date DESC LIMIT 1",
+            (account_id,),
+        )
+        row = cur.fetchone()
+        return dict(row) if row else None

tests/trader/test_persistence.py

@@ -3,6 +3,8 @@ from sanguo_trader.persistence import (
     init_db, save_account, save_trade, save_daily_balance,
     update_account_status, update_checkpoint, load_checkpoint,
     list_trades, list_daily_balance,
+    save_pending_orders, load_pending_orders,
+    save_positions, load_positions, load_last_balance,
 )
 
 
@@ -13,6 +15,36 @@ def test_init_and_save_account(tmp_path):
     assert aid > 0
 
 
+def test_pending_orders_roundtrip(tmp_path):
+    """C-S3 实走跨日 pending 订单持久化（覆盖式）。"""
+    db = str(tmp_path / "p.db")
+    init_db(db)
+    aid = save_account(db, {"name": "t"})
+    save_pending_orders(db, aid, [
+        {"strategy_id": "s1", "symbol": "600000", "side": "buy", "price": 10.5,
+         "volume": 100, "is_market": True, "match_session": "next_open", "listing_days": 0},
+    ])
+    loaded = load_pending_orders(db, aid)
+    assert len(loaded) == 1 and loaded[0]["symbol"] == "600000"
+    assert loaded[0]["is_market"] is True and loaded[0]["volume"] == 100
+    save_pending_orders(db, aid, [])   # 覆盖式清空
+    assert load_pending_orders(db, aid) == []
+
+
+def test_positions_and_last_balance_roundtrip(tmp_path):
+    """C-S3 实走 positions 快照 + 最后余额恢复。"""
+    db = str(tmp_path / "p.db")
+    init_db(db)
+    aid = save_account(db, {"name": "t", "initial_capital": 1_000_000})
+    save_positions(db, aid, "account",
+                   {"600000": {"volume": 200, "frozen": 0, "avg_price": 10.5}}, "2024-01-02")
+    pos = load_positions(db, aid, "account")
+    assert pos["600000"]["volume"] == 200 and pos["600000"]["avg_price"] == 10.5
+    save_daily_balance(db, aid, "2024-01-02", 99795.0, 2100.0, 101895.0)
+    bal = load_last_balance(db, aid)
+    assert bal["cash"] == 99795.0 and bal["date"] == "2024-01-02"
+
+
 def test_save_trade_and_list(tmp_path):
     db = str(tmp_path / "p.db")
     init_db(db)