diff --git a/sanguo_api/routes_paper.py b/sanguo_api/routes_paper.py index 00cbb50..86145f1 100644 --- a/sanguo_api/routes_paper.py +++ b/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，本机仅空转）。""" diff --git a/sanguo_trader/live_orchestrator.py b/sanguo_trader/live_orchestrator.py index 96ef792..5943fc0 100644 --- a/sanguo_trader/live_orchestrator.py +++ b/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) diff --git a/scripts/verify_live_step.py b/scripts/verify_live_step.py new file mode 100644 index 0000000..f47eb84 --- /dev/null +++ b/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()