diff --git a/sanguo_portfolio/live_strategy.py b/sanguo_portfolio/live_strategy.py index f39e37c..23324d9 100644 --- a/sanguo_portfolio/live_strategy.py +++ b/sanguo_portfolio/live_strategy.py @@ -10,6 +10,14 @@ sanguo_portfolio 的 StrategyTemplate 策略挂到 run_daily/run_monthly 定时 SANGUO_LIVE_MAX_POOL 选股池上限(默认 30) 数据 provider 由 runner_live ``set_data_provider`` 先行注入(miniQMT live 模式)。 + +⚠️ **process_initialize 是定时任务的生命线**(2026-08-17 VPS 16 引擎空转事故): +LiveEngine 重启时若 runtime 里恢复了 g(live_state.json/g.pkl),会**跳过 initialize** +走"断点续跑"路径,并把持久化的旧任务按 ``module+func`` 反射恢复——而我们的任务 +是策略实例的 bound method,恢复必然失败(``无法恢复调度任务``)。结果:进程活着、 +分钟心跳正常,但调度任务列表为空,开盘后零成交零日志。 +聚宽语义的正解是 ``process_initialize``(每次进程启动必调,含 resume):任务注册 +放这里,resume 重启后才能补挂。initialize 钩子保留(新策略首启由引擎调用)。 """ from __future__ import annotations @@ -18,6 +26,9 @@ import os logger = logging.getLogger(__name__) +# 进程内单例:initialize/process_initialize 双钩子共用一份策略实例与装配状态 +_STATE: dict = {"strategy": None, "wired": False} + def _build_live_strategy(provider): """env 配置 → StrategyTemplate 实例(对齐 runner_backtest._build_strategy)。""" @@ -65,28 +76,42 @@ def _build_live_strategy(provider): return factories[name]() -def initialize(context): - """LiveEngine 启动时回调:挂策略 + 定时器 + 费用滑点。""" +def _setup(context): + """装配策略(幂等,每进程一次):建实例 + broker facade + 定时任务。 + + initialize(新策略首启)与 process_initialize(每次进程启动)共用;第二个 + 钩子进来时直接返回,避免重复装配。各策略的 initialize 均为幂等配置,且定时 + 任务由策略自身经注入的 facade.run_daily/run_monthly 注册(单一事实源,无需 + 再调 runner_backtest._register_schedule 代注册)。 + """ from bullet_trade.core.api import ( # type: ignore order_target_value as bt_otv, order_value as bt_ov, set_order_cost, set_slippage, ) + from bullet_trade.core import run_daily as bt_run_daily # type: ignore + from bullet_trade.core import run_monthly as bt_run_monthly # type: ignore from bullet_trade.core.settings import ( # type: ignore OrderCost, FixedSlippage, set_option as bt_set_option, ) from bullet_trade.data.api import get_data_provider # type: ignore - from sanguo_portfolio.runner_backtest import _register_schedule from sanguo_portfolio.strategies.all_weather import BrokerFacade + if _STATE["wired"]: + return + strategy = _build_live_strategy(get_data_provider()) - _register_schedule(strategy) - # broker 注入(与回测同构):下单委托 bullet_trade 顶层 API,live 下路由 engine + _STATE["strategy"] = strategy + # broker 注入(与回测同构):下单委托 bullet_trade 顶层 API,live 下路由 engine; + # run_daily/run_monthly 同步注入——策略在自身 initialize 里经 facade 自挂定时 + # 任务,facade 缺注入时静默 no-op(2026-08-17 前 channel_test 等永不开仓的根因之一) strategy.broker = BrokerFacade( order_target_value=lambda c, v: bt_otv(c, v), order_value=lambda c, v: bt_ov(c, v), set_option=lambda k, v: bt_set_option(k, v), + run_daily=bt_run_daily, + run_monthly=bt_run_monthly, ) # A 股费用 + 滑点(与回测默认一致) set_order_cost( @@ -99,4 +124,15 @@ def initialize(context): ) set_slippage(FixedSlippage(value=0.001)) strategy.initialize(context) + _STATE["wired"] = True logger.info("live strategy 已挂载: %s", type(strategy).__name__) + + +def initialize(context): + """LiveEngine 新策略首启回调(策略文件级,g 未恢复时才被调)。""" + _setup(context) + + +def process_initialize(context): + """LiveEngine 每次进程启动必调(resume 重启含)——定时任务在这里补挂。""" + _setup(context) diff --git a/sanguo_portfolio/runner_backtest.py b/sanguo_portfolio/runner_backtest.py index 94e30bc..d5f992b 100644 --- a/sanguo_portfolio/runner_backtest.py +++ b/sanguo_portfolio/runner_backtest.py @@ -226,6 +226,7 @@ def _register_schedule(strategy: Any) -> None: from .strategies import ( AllWeatherStrategy, AllWeatherExStrategy, + ChannelTestStrategy, MomentumTimingStrategy, MomentumTimingExStrategy, SmallCapStrategy, @@ -233,6 +234,10 @@ def _register_schedule(strategy: Any) -> None: ValueSelectionStrategy, ValueSelectionExStrategy, ) + if isinstance(strategy, ChannelTestStrategy): + # 自带调度:initialize 里经 facade.run_daily 挂 9:35/10:45/13:45/14:30 + # 四时点(live 适配层已注入 run_daily),这里不代注册 + return if isinstance(strategy, (AllWeatherStrategy, AllWeatherExStrategy)): run_daily(strategy.prepare_stock_list, "9:05") run_monthly(strategy.monthly_adjustment, 1, "9:30") diff --git a/tests/api/test_portfolio_live.py b/tests/api/test_portfolio_live.py index dbd07df..69963bd 100644 --- a/tests/api/test_portfolio_live.py +++ b/tests/api/test_portfolio_live.py @@ -142,6 +142,7 @@ def test_live_strategy_adapter_builds_all_strategies(monkeypatch): ("momentum_timing", "MomentumTimingStrategy"), ("value_selection", "ValueSelectionStrategy"), ("small_cap", "SmallCapStrategy"), + ("channel_test", "ChannelTestStrategy"), # TET Phase2 副本:影子/实盘同样可发起(2026-08-16 VPS shadow#42 因缺项拉起即崩) ("all_weather_ex", "AllWeatherExStrategy"), ("momentum_timing_ex", "MomentumTimingExStrategy"), @@ -156,6 +157,88 @@ def test_live_strategy_adapter_builds_all_strategies(monkeypatch): live_strategy._build_live_strategy(_FakeProvider()) +def _patch_live_wiring(monkeypatch): + """打桩 bullet_trade 顶层 API:记录定时注册,provider 给假件。""" + import bullet_trade.core as bt_core + import bullet_trade.data.api as bt_data_api + + registered = [] + monkeypatch.setattr( + bt_core, "run_daily", + lambda f, t, **kw: registered.append( + ("daily", getattr(f, "__name__", str(f)), t))) + monkeypatch.setattr( + bt_core, "run_monthly", + lambda f, d, t, **kw: registered.append( + ("monthly", getattr(f, "__name__", str(f)), d, t))) + + class _FakeProvider: + pass + + monkeypatch.setattr(bt_data_api, "get_data_provider", lambda: _FakeProvider()) + return registered + + +def _reset_live_state(): + from sanguo_portfolio import live_strategy + live_strategy._STATE["strategy"] = None + live_strategy._STATE["wired"] = False + + +def test_process_initialize_registers_tasks_on_resume_boot(monkeypatch): + """P0 回归:resume 重启(g 已恢复,引擎跳过 initialize 只调 process_initialize) + 定时任务仍要注册——2026-08-17 VPS 16 引擎重启后零任务空转、开盘零成交。""" + from sanguo_portfolio import live_strategy + + registered = _patch_live_wiring(monkeypatch) + monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "all_weather") + _reset_live_state() + + # 只调 process_initialize,不调 initialize(复刻 resume 路径) + live_strategy.process_initialize(object()) + + names = {r[1] for r in registered} + assert "prepare_stock_list" in names + assert "monthly_adjustment" in names + assert "stop_loss" in names + assert live_strategy._STATE["wired"] is True + + +def test_setup_idempotent_across_both_hooks(monkeypatch): + """新策略首启:引擎先调 initialize 再调 process_initialize——任务不重复注册, + 策略实例复用同一份。""" + from sanguo_portfolio import live_strategy + + registered = _patch_live_wiring(monkeypatch) + monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "momentum_timing") + _reset_live_state() + + live_strategy.initialize(object()) + first = live_strategy._STATE["strategy"] + live_strategy.process_initialize(object()) + + assert live_strategy._STATE["strategy"] is first + assert len([r for r in registered if r[1] == "handle_data"]) == 1 + + +def test_facade_injects_run_daily_for_channel_test(monkeypatch): + """channel_test 自注册依赖 facade.run_daily(缺注入时静默 no-op 永不开仓): + process_initialize 后 9:35/10:45/13:45/14:30 四时点应经注入的 run_daily 挂上。""" + import bullet_trade.core as bt_core + from sanguo_portfolio import live_strategy + + registered = _patch_live_wiring(monkeypatch) + monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "channel_test") + _reset_live_state() + + live_strategy.process_initialize(object()) + + daily_times = sorted(r[2] for r in registered if r[0] == "daily") + assert daily_times == ["10:45", "13:45", "14:30", "9:35"] + # facade 拿到的是 bullet_trade 顶层 run_daily 本尊,不是默认空 lambda + assert live_strategy._STATE["strategy"].broker.run_daily is bt_core.run_daily + + def test_normalize_vt_symbol_bare_code_gets_exchange_suffix(): """CTA 实盘标的:裸 6 位码自动补交易所后缀(6→SSE,0/3→SZSE)。""" from sanguo_api.routes_live import _normalize_vt_symbol