fix(portfolio): 实盘/影子引擎重启后定时任务全丢根治——process_initialize+facade注入run_daily [vps]

P0 根因(2026-08-17 VPS 16 引擎空转零成交事故):bullet_trade LiveEngine 重启时
恢复 g(live_state.json/g.pkl)则跳过 initialize 走断点续跑,持久化旧任务按
module+func 反射恢复,而我们的任务是策略实例 bound method,恢复必失败
('无法恢复调度任务')→进程活着、分钟心跳正常、调度任务列表为空,开盘零成交零日志
(shadow_43/47/49 日志三段实锤:首启'已注册定时任务'→重启'无法恢复'→末代零任务)。
修法:live_strategy 加 process_initialize(引擎每次进程启动必调,resume 含),
装配抽 _setup 幂等(每进程一次);+3 回归测试(resume 只调 process_initialize
仍注册/双钩子不重复/facade 注入)。

P1 顺带根治:BrokerFacade 补注入 bullet_trade 顶层 run_daily/run_monthly——
此前 live facade 缺注入,策略自身 initialize 里的 b.run_daily 全部静默 no-op
(channel_test 无 _register_schedule 分支,4 账户连首启都不可能开仓);注入后
定时注册回归策略自身 initialize 单一事实源,_setup 不再调 _register_schedule
代注册(回测路径不变,runner_backtest._register_schedule 保留并补
ChannelTestStrategy 自注册分支消误导告警)。

829 测试绿

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)

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")

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