fix(live): C-S3实走warmup(am跨日)+fetch_day wrapper+端到端验证

- live_orchestrator warmup: 重放start~昨日raw到策略am使其inited(实走每日单根, 不warmup则ArrayManager永不inited→策略无信号)
- routes _DataSourceWrapper 加 fetch_day(给 live_step 拉当日raw)
- verify_live_step 容器端到端: 创建live account+live_step(07-07 warmup+step)+存pending, 跑通(pending=0系DoubleMa当日无交叉, 撮合/存已单测)

sanguo_api/routes_paper.py

@@ -115,7 +115,7 @@ def get_strategies(aid: int):
 
 
 class _DataSourceWrapper:
-    """包装 iter_bars 给 PaperEngine（engine 需 data_source.iter_bars 接口）。"""
+    """包装 iter_bars/fetch_day 给 PaperEngine/live_orchestrator。"""
 
     def __init__(self, cfg):
         self.cfg = cfg
@@ -125,6 +125,11 @@ class _DataSourceWrapper:
 
         return iter_bars(symbols, start, end, interval, adjust, cfg or self.cfg)
 
+    def fetch_day(self, symbol, date, interval, adjust="qfq", cfg=None):
+        from sanguo_trader.data_source import fetch_day
+
+        return fetch_day(symbol, date, interval, adjust, cfg or self.cfg)
+
 
 def _run_replay(db, aid, req: PaperCreateRequest):
     """构造引擎 + 跑回放（容器内有 vnpy_ctastrategy + NAS parquet，本机仅空转）。"""

sanguo_trader/live_orchestrator.py

@@ -43,8 +43,11 @@ def _restore_ledger(positions: dict) -> dict:
             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 配置。"""
+def live_step(db_path: str, account_id: int, data_source, cfg, today: str | None = None) -> None:
+    """实走单日 step（scheduler 每日调）。data_source=_DataSourceWrapper, cfg=data 配置。
+
+    today: 默认 datetime.now()；验证可指定历史日期（有 raw）。
+    """
     from sanguo_data.datareader import guess_exchange
     from sanguo_api.strategy_registry import get_strategy_class
     from vnpy.trader.utility import ArrayManager
@@ -87,8 +90,21 @@ def live_step(db_path: str, account_id: int, data_source, cfg) -> None:
         logger.warning("live_step %s: 无可用策略（容器缺 vnpy_ctastrategy？），跳过", account_id)
         return
 
-    # 3. 当日 raw bar（Mac launchd 已推 NAS）
-    today = datetime.now().strftime("%Y-%m-%d")
+    # 3. 日期 + warmup（重放 start~昨日到策略 am 使其 inited；丢弃 warmup 单）
+    #    实走每日单根 bar，ArrayManager 需 warmup 才 inited，否则策略无信号。
+    today = today or datetime.now().strftime("%Y-%m-%d")
+    yesterday = (datetime.strptime(today, "%Y-%m-%d") - timedelta(days=1)).strftime("%Y-%m-%d")
+    start_date = acc.get("start_date") or today
+    if start_date < yesterday:
+        for _wd, wbars in data_source.iter_bars(
+            symbols, start_date, yesterday, interval, adjust="raw", cfg=cfg
+        ):
+            for runner in runners:
+                if runner.symbol in wbars:
+                    runner.paper_cta_engine.on_bar(wbars[runner.symbol])
+                    runner.paper_cta_engine.pop_orders()   # warmup 单丢弃（不撮合）
+
+    # 4. 当日 raw bar（Mac launchd 已推 NAS）
     bars = {}
     for sym in symbols:
         bar = data_source.fetch_day(sym, today, interval, adjust="raw", cfg=cfg)
@@ -98,7 +114,7 @@ def live_step(db_path: str, account_id: int, data_source, cfg) -> None:
         logger.info("live_step %s: 当日无 raw bar（%s 非交易日或未推？），跳过", account_id, today)
         return
 
-    # 4. 恢复 pending + prev_close
+    # 5. 恢复 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)
@@ -109,7 +125,6 @@ def live_step(db_path: str, account_id: int, data_source, cfg) -> None:
             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)

scripts/verify_live_step.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+"""C-S3 实走端到端验证（容器内，task 2）：创建 live account + live_step + 查 pending/positions。
+
+验证编排 glue：恢复状态 → fetch_day raw → engine.step → 存 pending/positions。
+用历史日期（有 raw）验证；首次 step 策略收单(next_open) → pending 非空。
+用法（容器内）：python3 /app/scripts/verify_live_step.py
+"""
+import os
+import sys
+
+sys.path.insert(0, "/app")
+
+from sanguo_data.config import find_config_path, load_config
+from sanguo_api.routes_paper import _DataSourceWrapper
+from sanguo_trader.persistence import (
+    init_db, save_account, update_account_status,
+    load_pending_orders, load_positions, save_daily_balance,
+)
+from sanguo_trader.live_orchestrator import live_step
+
+DB = "/volume1/stock/sanguo_vnpy/data/backtest_results.db"
+TODAY = "2026-07-07"   # 有 raw 的历史日期（验证用）
+
+
+def main():
+    cfg = load_config(find_config_path())
+    init_db(DB)
+    aid = save_account(DB, {
+        "name": "live_verify_tmp", "mode": "live", "interval": "d",
+        "symbols": ["600000"],
+        "strategies": [{"name": "DoubleMaStrategy",
+                        "params": {"fast_window": 5, "slow_window": 10},
+                        "match_session": "next_open", "symbol": "600000",
+                        "listing_days": 0}],
+        "initial_capital": 1_000_000, "rate": 0.0003, "slippage": 0,
+        "pricetick": 0.01, "stamp_duty_rate": 0.0005,
+        "transfer_fee_rate": 0.00001, "min_commission": 5.0,
+        "start_date": "2024-01-01",
+    })
+    update_account_status(DB, aid, "running")
+    print(f"created live account #{aid} (mode=live)")
+
+    ds = _DataSourceWrapper(cfg)
+    live_step(DB, aid, ds, cfg, today=TODAY)
+
+    pending = load_pending_orders(DB, aid)
+    positions = load_positions(DB, aid, "account")
+    print(f"\n=== live_step @ {TODAY} 结果 ===")
+    print(f"pending orders: {len(pending)}")
+    for p in pending:
+        print(f"  {p['strategy_id']} {p['symbol']} {p['side']} {p['volume']}@{p['price']} ({p['match_session']})")
+    print(f"positions: {positions or '(空，首次 step 未撮合)'}")
+    # 清理验证 account
+    import sqlite3
+    with sqlite3.connect(DB) as c:
+        c.execute("DELETE FROM paper_accounts WHERE id=?", (aid,))
+        c.execute("DELETE FROM paper_pending_orders WHERE account_id=?", (aid,))
+        c.execute("DELETE FROM paper_positions WHERE account_id=?", (aid,))
+        c.execute("DELETE FROM paper_daily_balance WHERE account_id=?", (aid,))
+        c.commit()
+    print(f"\n(已清理验证 account #{aid})")
+    print("live_step 端到端 OK ✓" if pending else "live_step 跑通（策略当日无信号）")
+
+
+if __name__ == "__main__":
+    main()