diff --git a/sanguo_portfolio/strategies/channel_test.py b/sanguo_portfolio/strategies/channel_test.py index e5bcc14..a6e09b0 100644 --- a/sanguo_portfolio/strategies/channel_test.py +++ b/sanguo_portfolio/strategies/channel_test.py @@ -23,12 +23,33 @@ import logging import os from dataclasses import dataclass, field from datetime import datetime, timedelta -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Optional from .all_weather import BrokerFacade, _available_cash, _get_positions logger = logging.getLogger(__name__) + +# ---------------- 盘后下单防护(2026-09-01 实锤) ---------------- +def _now() -> datetime: + """当前时刻(独立函数便于测试 monkeypatch;测试禁依赖墙钟)。""" + return datetime.now() + + +def _in_trading_session(now: datetime) -> bool: + """A股交易时段(含集合竞价/收盘缓冲):工作日 09:25-11:35 / 12:55-15:05。 + + 09-01 实锤:盘后 18:48 轮换重启 → 随启随验 run_daily 锚定 18:50 → 探针 + 16 笔真单盘后全挂柜台(~107 万排队次日集合竞价)。live 探针的**注册**与 + **执行**双处守卫;jq_strategy 调度恢复失败回退注册到当前时刻的路径由 + 执行侧守卫兜底。影子/回测撮合 24h 安全不受影响。 + """ + if now.weekday() >= 5: + return False + hm = now.hour * 60 + now.minute + return (9 * 60 + 25 <= hm <= 11 * 60 + 35 + or 12 * 60 + 55 <= hm <= 15 * 60 + 5) + # universe 按类型分组；轮换时从每组轮流取 → 每天的持仓组合跨类型 UNIVERSE_BY_TYPE: Dict[str, List[str]] = { "宽基ETF": ["510300.XSHG", "510050.XSHG", "510500.XSHG", "159915.XSHE"], @@ -79,8 +100,16 @@ class ChannelTestStrategy: # 随启随验:只注册一个"启动后约90秒"的探针任务。必须经 facade.run_daily # 注册而非直接调用——回调才会收到 B2 包装的代理 context(实例预算视图), # 绕过它探针就测不到定寸虚拟化。重启实例(停止→启动)即再验一轮。 - h, m = self._probe_time() - b.run_daily(self.probe_all, f"{h}:{m:02d}") + # 盘前/盘后/周末重启不注册(锚点在时段外=盘后真单挂柜台,09-01 实锤)。 + probe_at = _now() + timedelta(seconds=90) + if not _in_trading_session(probe_at): + logger.warning( + "[PROBE] 重启于非交易时段(探针锚定 %s),跳过随启随验注册" + "——防盘后挂单排队次日集合竞价;下次盘中重启再验", + probe_at.strftime("%H:%M")) + return + b.run_daily(self.probe_all, + probe_at.strftime("%H:%M")) return b.run_daily(self.rotate, "9:35") # 主调仓:卖旧买新 if self.config.intraday_partial: @@ -97,15 +126,13 @@ class ChannelTestStrategy: return self.config.on_demand return bool(os.environ.get("SANGUO_LIVE_STRATEGY")) - @staticmethod - def _probe_time() -> Tuple[int, int]: - """探针触发时刻=启动后约 90 秒(留足引擎 bootstrap;时:分对齐 "9:35" 风格)。""" - t = datetime.now() + timedelta(seconds=90) - return int(t.strftime("%H")), int(t.strftime("%M")) - def probe_all(self, context: Any) -> None: """一轮打满全场景,每步 [PROBE-n] 留痕(日志即验收凭据)。 + 非交易时段整轮跳过(执行侧守卫):调度错锚/jq_strategy 恢复失败回退 + 注册到当前时刻等路径把探针拖到盘外时,绝不让真单挂柜台排队次日 + 集合竞价(2026-09-01 18:50 实锤,16 笔 ~107 万)。 + - PROBE-0 实例视图快照:cash/持仓应为**本实例预算切片**(B2 生效的标志, ≠全账户 990 万;实盘闭市时下单被 QMT 拒不阻断,决策层输出已在日志可验); - PROBE-1 主调仓(rotate):定寸=实例视图等权 → sizing 证明; @@ -115,6 +142,11 @@ class ChannelTestStrategy: 重启第二轮起有票可试); - PROBE-4 资金复用(swap_one):卖昨日仓→即刻买新(空仓时跳过)。 """ + if not _in_trading_session(_now()): + logger.warning( + "[PROBE] 非交易时段触发(调度错锚/恢复回退),整轮跳过" + "——防盘后挂单排队次日集合竞价") + return cash = _available_cash(context) positions = _get_positions(context, self.broker) logger.info("[PROBE-0] 实例视图 cash=%.2f 持仓=%s (应=本实例预算切片,非全账户)", diff --git a/tests/portfolio/test_channel_test.py b/tests/portfolio/test_channel_test.py index a57cec7..368f199 100644 --- a/tests/portfolio/test_channel_test.py +++ b/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