From 6931a7b541d81247558397f0a0f7a4f4c6730028 Mon Sep 17 00:00:00 2001
From: claude_dev <claude-dev@sanguo.local>
Date: Wed, 8 Jul 2026 06:49:31 +0800
Subject: [PATCH] =?UTF-8?q?feat(trader):=20C-S3=E5=AE=9E=E8=B5=B0=E5=90=8E?=
 =?UTF-8?q?=E7=AB=AF=E9=AA=A8=E6=9E=B6=E2=80=94live=5Forchestrator+?=
 =?UTF-8?q?=E5=85=A8=E5=B1=80scheduler=20job+routes=20live?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

架构(简化,避per-account闭包注入):
- live_orchestrator live_step(account_id)自包含: 恢复cash/positions/pending→fetch_day raw当日→engine.step→存状态
- run_live_step(db)遍历live accounts调live_step; scheduler register_live_step_job全局20:30 job
- app startup注册全局job; routes create mode=live存account running(不跑回放)
- TODO(分期项): prev_close昨日raw/listing_days IPO算/realized_pnl恢复
- 113/113通过, live_orchestrator import OK
---
 sanguo_api/app.py                  |  10 ++
 sanguo_api/routes_paper.py         |   4 +
 sanguo_trader/live_orchestrator.py | 169 +++++++++++++++++++++++++++++
 sanguo_trader/scheduler.py         |  20 +++-
 4 files changed, 202 insertions(+), 1 deletion(-)
 create mode 100644 sanguo_trader/live_orchestrator.py

diff --git a/sanguo_api/app.py b/sanguo_api/app.py
index 0c52b5d..1bfd654 100644
--- a/sanguo_api/app.py
+++ b/sanguo_api/app.py
@@ -37,4 +37,14 @@ def create_app(db_path: str, file_dir=None, auth_config=None, max_workers: int =
     app.include_router(paper_router, prefix="/api/v1")
     set_db_path(db_path)
 
+    @app.on_event("startup")
+    def _register_live_step():
+        """容器启动：注册每日 20:30 全局实走 step job（C-S3，遍历 live accounts）。"""
+        try:
+            from sanguo_trader.scheduler import register_live_step_job
+            register_live_step_job(db_path)
+        except Exception as e:  # noqa: BLE001  本机无 apscheduler 时跳过
+            import logging
+            logging.getLogger(__name__).warning("register_live_step_job 失败: %s", e)
+
     return app
\ No newline at end of file
diff --git a/sanguo_api/routes_paper.py b/sanguo_api/routes_paper.py
index 84eb50e..00cbb50 100644
--- a/sanguo_api/routes_paper.py
+++ b/sanguo_api/routes_paper.py
@@ -72,6 +72,10 @@ def create_paper(req: PaperCreateRequest):
                 update_account_status(db, aid, "failed", str(e))
         threading.Thread(target=_bg, daemon=True).start()
         status = "running"
+    elif req.mode == "live":   # 实走：全局 job 每日 20:30 遍历 step（不跑回放）
+        from sanguo_trader.persistence import update_account_status
+        update_account_status(db, aid, "running")
+        status = "running"
     return {"account_id": aid, "status": status}
 
 
diff --git a/sanguo_trader/live_orchestrator.py b/sanguo_trader/live_orchestrator.py
new file mode 100644
index 0000000..96ef792
--- /dev/null
+++ b/sanguo_trader/live_orchestrator.py
@@ -0,0 +1,169 @@
+"""C-S3 实走编排（task 2）：每日 live_step 单根推进。
+
+架构（简化，避开 APScheduler per-account 闭包注入）：
+- 容器 APScheduler 每日 20:30 调 live_runner.run_live_step（全局 job，遍历 live accounts）
+- live_step(account_id) 自包含：恢复状态 → 读当日 raw bar → engine.step → 存状态
+- 当日 raw 由 Mac launchd 增量推 NAS（run_daily_update.sh 加 raw 增量）
+
+状态恢复：cash=最后余额 / positions=paper_positions / pending=paper_pending_orders
+TODO（分期项）：prev_close 从昨日 raw close 读（首版用当日 open 兜底）；
+listing_days 从 IPO 日算（首版 stub 0）；realized_pnl 未恢复（归因次日重置）。
+"""
+import json
+import logging
+import sqlite3
+from datetime import datetime, timedelta
+
+from .account import Account
+from .cta_adapter import PaperCtaEngine
+from .engine import PaperEngine
+from .limit import lot_size_for
+from .models import AccountConfig, MatchSession, OrderSide, PaperOrder
+from .position_ledger import PositionLedger
+from .persistence import (
+    load_last_balance, load_positions, save_positions,
+    load_pending_orders, save_pending_orders,
+)
+from .strategy_runner import StrategyRunner
+
+logger = logging.getLogger(__name__)
+
+
+def _get_account(db_path: str, account_id: int) -> dict:
+    with sqlite3.connect(db_path) as conn:
+        conn.row_factory = sqlite3.Row
+        return dict(conn.execute(
+            "SELECT * FROM paper_accounts WHERE id=?", (account_id,)
+        ).fetchone())
+
+
+def _restore_ledger(positions: dict) -> dict:
+    """{symbol:{volume,frozen,avg_price}} → {symbol: PositionLedger}。"""
+    return {sym: PositionLedger(sym, p["volume"], p.get("frozen", 0), p["avg_price"])
+            for sym, p in positions.items()}
+
+
+def live_step(db_path: str, account_id: int, data_source, cfg) -> None:
+    """实走单日 step（scheduler 每日调）。data_source=_DataSourceWrapper, cfg=data 配置。"""
+    from sanguo_data.datareader import guess_exchange
+    from sanguo_api.strategy_registry import get_strategy_class
+    from vnpy.trader.utility import ArrayManager
+
+    acc = _get_account(db_path, account_id)
+    symbols = json.loads(acc["symbols"] or "[]")
+    strategies = json.loads(acc["strategies"] or "[]")
+    interval = acc.get("interval") or "d"
+    initial = acc["initial_capital"]
+
+    # 1. 恢复 account（cash + positions）
+    account = Account(initial)
+    last_bal = load_last_balance(db_path, account_id)
+    if last_bal:
+        account.cash = last_bal["cash"]
+    account.positions = _restore_ledger(load_positions(db_path, account_id, "account"))
+
+    # 2. 构造 runners（恢复 positions）
+    runners: list = []
+    for s in strategies:
+        cls = get_strategy_class(s["name"])
+        if cls is None:
+            continue
+        cta = PaperCtaEngine(s["name"], match_session=s.get("match_session", "next_open"),
+                             listing_days=s.get("listing_days", 0),
+                             size=lot_size_for(s["symbol"]))
+        vt = f"{s['symbol']}.{guess_exchange(s['symbol']).value}"
+        strat = cls(cta, s["name"], vt, s.get("params", {}))
+        strat.trading = True
+        if not hasattr(strat, "am"):
+            strat.am = ArrayManager(20)
+        cta.set_strategy(strat)
+        runner = StrategyRunner(s["name"], strategy=strat, paper_cta_engine=cta,
+                                symbol=s["symbol"])
+        runner.positions = _restore_ledger(
+            load_positions(db_path, account_id, f"strategy:{s['name']}"))
+        runners.append(runner)
+
+    if not runners:
+        logger.warning("live_step %s: 无可用策略（容器缺 vnpy_ctastrategy？），跳过", account_id)
+        return
+
+    # 3. 当日 raw bar（Mac launchd 已推 NAS）
+    today = datetime.now().strftime("%Y-%m-%d")
+    bars = {}
+    for sym in symbols:
+        bar = data_source.fetch_day(sym, today, interval, adjust="raw", cfg=cfg)
+        if bar is not None:
+            bars[sym] = bar
+    if not bars:
+        logger.info("live_step %s: 当日无 raw bar（%s 非交易日或未推？），跳过", account_id, today)
+        return
+
+    # 4. 恢复 pending + prev_close
+    pending = []
+    for o in load_pending_orders(db_path, account_id):
+        runner = next((r for r in runners if r.strategy_id == o["strategy_id"]), None)
+        if runner is None:
+            continue
+        pending.append((PaperOrder(
+            o["strategy_id"], o["symbol"], OrderSide(o["side"]), o["price"], o["volume"],
+            o["is_market"], MatchSession(o["match_session"]), o["listing_days"],
+        ), runner))
+    # prev_close：昨日 raw close（fetch_day 昨日）；失败兜底用当日 open
+    yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
+    prev_close = {}
+    for sym in symbols:
+        ybar = data_source.fetch_day(sym, yesterday, interval, adjust="raw", cfg=cfg)
+        prev_close[sym] = ybar.close_price if ybar else bars[sym].open_price
+
+    # 5. engine.step（单根当日）
+    acc_cfg = AccountConfig(
+        initial_capital=initial, rate=acc["rate"], slippage=acc["slippage"],
+        pricetick=acc["pricetick"], stamp_duty_rate=acc["stamp_duty_rate"],
+        transfer_fee_rate=acc["transfer_fee_rate"], min_commission=acc["min_commission"],
+    )
+    pe = PaperEngine(account, runners, data_source, acc_cfg, db_path, account_id,
+                     symbols, acc.get("start_date") or today, today, interval, adjust="raw")
+    pending_new, _closes = pe.step(today, bars, prev_close, pending)
+
+    # 6. 存状态（pending + positions）
+    save_pending_orders(db_path, account_id, [
+        {"strategy_id": o.strategy_id, "symbol": o.symbol, "side": o.side.value,
+         "price": o.price, "volume": o.volume, "is_market": o.is_market,
+         "match_session": o.match_session.value, "listing_days": o.listing_days}
+        for o, _r in pending_new
+    ])
+    save_positions(db_path, account_id, "account", {
+        sym: {"volume": p.volume, "frozen": p.frozen, "avg_price": p.avg_price}
+        for sym, p in account.positions.items()}, today)
+    for r in runners:
+        save_positions(db_path, account_id, f"strategy:{r.strategy_id}", {
+            sym: {"volume": p.volume, "frozen": p.frozen, "avg_price": p.avg_price}
+            for sym, p in r.positions.items()}, today)
+    logger.info("live_step %s 完成 @%s，pending=%d positions=%d",
+                account_id, today, len(pending_new), len(account.positions))
+
+
+def list_live_accounts(db_path: str) -> list[int]:
+    """所有 mode=live & status=running 的 account_id（live_runner 遍历用）。"""
+    with sqlite3.connect(db_path) as conn:
+        rows = conn.execute(
+            "SELECT id FROM paper_accounts WHERE mode='live' AND status='running'"
+        ).fetchall()
+    return [r[0] for r in rows]
+
+
+def run_live_step(db_path: str) -> None:
+    """每日全局 step：遍历所有 live accounts 调 live_step（scheduler 20:30 调）。
+
+    lazy import _DataSourceWrapper 避免与 routes_paper 循环 import。
+    """
+    from sanguo_data.config import find_config_path, load_config
+    from sanguo_api.routes_paper import _DataSourceWrapper
+
+    cfg = load_config(find_config_path())
+    data_source = _DataSourceWrapper(cfg)
+    for aid in list_live_accounts(db_path):
+        try:
+            live_step(db_path, aid, data_source, cfg)
+        except Exception as e:  # noqa: BLE001
+            logger.error("live_step account %s 失败: %s", aid, e)
diff --git a/sanguo_trader/scheduler.py b/sanguo_trader/scheduler.py
index ac43206..21385b5 100644
--- a/sanguo_trader/scheduler.py
+++ b/sanguo_trader/scheduler.py
@@ -54,7 +54,10 @@ def _default_step(account_id: int) -> None:
 
 
 def restore_live_jobs(db_path: str, job_func=None) -> int:
-    """容器启动：遍历 status=running & mode=live 重建 job（spec §9.2 H-3）。"""
+    """容器启动：遍历 status=running & mode=live 重建 job（spec §9.2 H-3）。
+
+    简化架构：实走用全局 job（register_live_step_job），此函数保留兼容老 per-account 设计。
+    """
     with sqlite3.connect(db_path) as conn:
         rows = conn.execute(
             "SELECT id FROM paper_accounts WHERE status='running' AND mode='live'"
@@ -62,3 +65,18 @@ def restore_live_jobs(db_path: str, job_func=None) -> int:
     for (aid,) in rows:
         register_paper_job(aid, job_func=job_func)
     return len(rows)
+
+
+def register_live_step_job(db_path: str, hour: int = 20, minute: int = 30) -> str:
+    """全局每日 step job：20:30 遍历所有 live accounts 调 run_live_step（C-S3 简化架构）。
+
+    job_func 用 lambda 闭包 db_path（add_job 只传无参 callable）。
+    """
+    from .live_orchestrator import run_live_step  # lazy 避循环 import
+    s = start_scheduler()
+    s.add_job(
+        lambda: run_live_step(db_path),
+        "cron", hour=hour, minute=minute,
+        id="live_step_global", replace_existing=True,
+    )
+    return "live_step_global"