From 20bdd689af3d76b6a6cd70cda238e3d25699f4d2 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Wed, 8 Jul 2026 06:46:57 +0800 Subject: [PATCH] =?UTF-8?q?feat(persistence):=20C-S3=E5=AE=9E=E8=B5=B0?= =?UTF-8?q?=E8=B7=A8=E6=97=A5=E7=8A=B6=E6=80=81=E2=80=94paper=5Fpending=5F?= =?UTF-8?q?orders+positions/last=5Fbalance=E5=AD=98=E5=8F=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sanguo_trader/persistence.py | 98 +++++++++++++++++++++++++++++++- tests/trader/test_persistence.py | 32 +++++++++++ 2 files changed, 129 insertions(+), 1 deletion(-) diff --git a/sanguo_trader/persistence.py b/sanguo_trader/persistence.py index 141b2ad..cf02f2c 100644 --- a/sanguo_trader/persistence.py +++ b/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 diff --git a/tests/trader/test_persistence.py b/tests/trader/test_persistence.py index 22b8888..235fcc4 100644 --- a/tests/trader/test_persistence.py +++ b/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)