fix(strategy): channel_test探针盘后下单双守卫——09-01 18:48轮换重启实锤:随启随验run_daily锚定now+90s=18:50,探针16笔真单盘后全挂柜台(~107万买单冻结排队次日集合竞价);且run_daily为每日重复锚定,08-30周日13:07重启即埋下「每天13:09盘后风格漂移」隐患,jq_strategy调度恢复失败回退注册当前时刻是第三路径;修=①注册侧:锚点落非交易时段(工作日09:25-11:35/12:55-15:05)不注册+大声告警②执行侧:probe_all首行守卫,盘外触发整轮跳过零下单(兜调度错锚/回退路径);墙钟抽_now()可注入,测试autouse钉盘中防CI时段漂移;+3守卫测试+11边界参数化;25/25 [vps]

tests/portfolio/test_channel_test.py

@@ -6,6 +6,11 @@
 """
 from __future__ import annotations
 
+from datetime import datetime
+
+import pytest
+
+import sanguo_portfolio.strategies.channel_test as ct
 from sanguo_portfolio.strategies import ChannelTestConfig, ChannelTestStrategy
 from sanguo_portfolio.strategies.all_weather import BrokerFacade
 from sanguo_portfolio.strategies.channel_test import UNIVERSE_BY_TYPE
@@ -154,7 +159,7 @@ def test_initialize_on_demand_registers_single_probe(monkeypatch):
     name, t = registered[0]
     assert name == "probe_all"
     hh, mm = t.split(":")
-    now = datetime.now()
+    now = ct._now()  # 墙钟可注入(盘后防护 _now 缝),断言跟注入钟走
     delta = (int(hh) * 60 + int(mm)) - (now.hour * 60 + now.minute)
     assert 0 <= delta <= 3  # 未来 ~90 秒(±跨分钟容差)
 
@@ -192,3 +197,78 @@ def test_probe_all_runs_sizing_sell_t1_swap_sequence():
     assert "510300.XSHG" in sells    # T+1 探针卖 own 第一只(或 rotate 换仓卖出)
     swap_buy = [c for m, c, v in broker.calls if m == "ov" and v > 0]
     assert swap_buy                  # 资金复用:卖后另买未持有票
+
+
+# ======================== 盘后下单防护(2026-09-01 实锤) ========================
+@pytest.fixture(autouse=True)
+def _pin_now_in_session(monkeypatch):
+    """钉住守卫墙钟=盘中(周二 10:00),既有探针测试不随 CI 运行时段漂移;
+    守卫自身的界外用例各自显式 monkeypatch 覆盖。"""
+    ct._now = lambda: datetime(2026, 9, 1, 10, 0)  # type: ignore[assignment]
+
+
+@pytest.mark.parametrize("ts,expected", [
+    ("2026-09-01 09:24", False),   # 集合竞价前
+    ("2026-09-01 09:26", True),    # 竞价缓冲内
+    ("2026-09-01 11:34", True),    # 午前收盘缓冲
+    ("2026-09-01 11:36", False),   # 午休
+    ("2026-09-01 12:54", False),
+    ("2026-09-01 12:56", True),    # 午后
+    ("2026-09-01 15:04", True),    # 收盘缓冲
+    ("2026-09-01 15:06", False),
+    ("2026-09-01 18:50", False),   # 09-01 事故时刻(盘后轮换锚点)
+    ("2026-09-05 10:00", False),   # 周六
+    ("2026-09-06 10:00", False),   # 周日
+])
+def test_in_trading_session_boundaries(ts, expected):
+    assert ct._in_trading_session(datetime.fromisoformat(ts)) is expected
+
+
+def test_probe_all_skips_out_of_session(monkeypatch, caplog):
+    """盘外触发(调度错锚/恢复回退)整轮跳过:零下单 + 告警留痕。"""
+    ct._now = lambda: datetime(2026, 9, 1, 18, 50)  # type: ignore[assignment]
+    broker = _MockBroker()
+    s = ChannelTestStrategy(provider=None, broker=broker,
+                            config=ChannelTestConfig(hold_n=2, on_demand=True))
+    ctx = _Ctx({"510300.XSHG": _Pos(5000)}, cash=100000)
+    with caplog.at_level("WARNING"):
+        s.probe_all(ctx)
+    assert broker.calls == []       # 一笔单都不许挂
+    assert any("非交易时段" in r.message for r in caplog.records)
+
+
+def test_initialize_on_demand_after_hours_skips_registration(monkeypatch, caplog):
+    """盘后重启(live env)不注册随启随验——run_daily 锚点落在时段外必跳过,
+    否则 16 笔真单挂柜台排队次日集合竞价(09-01 18:48 轮换实锤)。"""
+    monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "channel_test")
+    ct._now = lambda: datetime(2026, 9, 1, 18, 48)  # type: ignore[assignment]
+    registered: list = []
+
+    class _RegBroker(_MockBroker):
+        def __init__(self) -> None:
+            super().__init__()  # dataclass 字段会遮蔽子类方法,注入须在 super 后
+            self.run_daily = lambda fn, t: registered.append(
+                (getattr(fn, "__name__", str(fn)), t))
+
+    s = ChannelTestStrategy(provider=None, broker=_RegBroker(),
+                            config=ChannelTestConfig(on_demand=True))
+    with caplog.at_level("WARNING"):
+        s.initialize(object())
+    assert registered == []         # 不注册
+    assert any("非交易时段" in r.message for r in caplog.records)
+
+
+def test_initialize_on_demand_in_session_still_registers():
+    """盘中重启注册行为不变(回归锚定时间=now+90s 落在盘中)。"""
+    registered: list = []
+
+    class _RegBroker2(_MockBroker):
+        def __init__(self) -> None:
+            super().__init__()  # dataclass 字段会遮蔽子类方法,注入须在 super 后
+            self.run_daily = lambda fn, t: registered.append(
+                (getattr(fn, "__name__", str(fn)), t))
+
+    s = ChannelTestStrategy(provider=None, broker=_RegBroker2(),
+                            config=ChannelTestConfig(on_demand=True))
+    s.initialize(object())
+    assert registered == [("probe_all", "10:01")]  # 10:00 + 90s