diff --git a/sanguo_portfolio/strategies/channel_test.py b/sanguo_portfolio/strategies/channel_test.py index 0976802..e5bcc14 100644 --- a/sanguo_portfolio/strategies/channel_test.py +++ b/sanguo_portfolio/strategies/channel_test.py @@ -13,12 +13,17 @@ universe 按资产类型分组（每类多只，轮换时跨类型取样 → 各 - 13:45 卖后买：清掉 1 只持仓换买另 1 只(当日资金复用;清的是昨日仓,T+1 可卖) - 14:30 T+1 探针：当日买入立即试卖 → 两端都应被 T+1 拒(验证拒单通路) - 涨跌停/停牌/部分成交：靠真实行情自然出现，差异进双轨对账报告 +随启随验(2026-08-20,on_demand 自动于 LiveEngine 进程生效):不再等盘中固定 +时点——实例每次启动/重启后 ~90 秒经 B2 代理 context 打一轮全场景探针 +(probe_all,[PROBE-0..4] 日志留痕),影子柜台 24h 可验,重启实例=再验一轮。 """ from __future__ import annotations import logging +import os from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional +from datetime import datetime, timedelta +from typing import Any, Dict, List, Optional, Tuple from .all_weather import BrokerFacade, _available_cash, _get_positions @@ -44,6 +49,11 @@ class ChannelTestConfig: intraday_partial: bool = True # 10:45 部分调仓(加减仓) intraday_swap: bool = True # 13:45 卖后买(资金复用) benchmark: str = "000300.XSHG" + # 随启随验(2026-08-20 用户拍板):None=自动——LiveEngine 进程(supervisor 注入 + # SANGUO_LIVE_STRATEGY)下 True,启动/重启后 ~90s 打一轮全场景探针,不绑盘中 + # 固定时点(开盘窗口不再是验证瓶颈:影子柜台 24h 可验,实盘闭市下单被拒也是 + # 验证项);回测/单测(无 env)False,维持四时点旧行为。显式 True/False 覆盖。 + on_demand: Optional[bool] = None class ChannelTestStrategy: @@ -65,6 +75,13 @@ class ChannelTestStrategy: b.set_benchmark(self.config.benchmark) b.set_option("use_real_price", True) b.set_option("avoid_future_data", True) + if self._on_demand(): + # 随启随验:只注册一个"启动后约90秒"的探针任务。必须经 facade.run_daily + # 注册而非直接调用——回调才会收到 B2 包装的代理 context(实例预算视图), + # 绕过它探针就测不到定寸虚拟化。重启实例(停止→启动)即再验一轮。 + h, m = self._probe_time() + b.run_daily(self.probe_all, f"{h}:{m:02d}") + return b.run_daily(self.rotate, "9:35") # 主调仓:卖旧买新 if self.config.intraday_partial: b.run_daily(self.partial_adjust, "10:45") # 部分加减仓 @@ -73,6 +90,46 @@ class ChannelTestStrategy: if self.config.probe_t1: b.run_daily(self.t1_probe, "14:30") # T+1 拒单探针 + # ---------------- 随启随验 ---------------- + def _on_demand(self) -> bool: + """随启随验开关(见 ChannelTestConfig.on_demand 注释)。""" + if self.config.on_demand is not None: + 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] 留痕(日志即验收凭据)。 + + - PROBE-0 实例视图快照:cash/持仓应为**本实例预算切片**(B2 生效的标志, + ≠全账户 990 万;实盘闭市时下单被 QMT 拒不阻断,决策层输出已在日志可验); + - PROBE-1 主调仓(rotate):定寸=实例视图等权 → sizing 证明; + - PROBE-2 只卖自己:rotate 的卖出对象全部来自实例账本持仓(A1 读隔离, + 对照前端"账户实况"实例分解即可确认没碰别家的票); + - PROBE-3 T+1 拒单探针:当日买立即卖,两端应拒(冷启动首轮空仓会跳过, + 重启第二轮起有票可试); + - PROBE-4 资金复用(swap_one):卖昨日仓→即刻买新(空仓时跳过)。 + """ + cash = _available_cash(context) + positions = _get_positions(context, self.broker) + logger.info("[PROBE-0] 实例视图 cash=%.2f 持仓=%s (应=本实例预算切片,非全账户)", + cash, sorted(positions.keys())) + logger.info("[PROBE-1] 主调仓:定寸=实例视图等权") + self.rotate(context) + logger.info("[PROBE-2] 只卖自己:卖出对象全部来自实例账本持仓") + if self.config.probe_t1: + logger.info("[PROBE-3] T+1 探针:当日买立即卖,预期被拒(空仓跳过)") + self.t1_probe(context) + if self.config.intraday_swap: + logger.info("[PROBE-4] 资金复用:卖后买(空仓跳过)") + self.swap_one(context) + logger.info("[PROBE] 本轮探针序列完成;成交归因见 live_trades,前端三层+账户实况") + # ---------------- 目标组合 ---------------- def _target_set(self) -> List[str]: """跨类型取样 hold_n 只:每个类型组按 day 偏移轮流供一只,凑满 hold_n。""" diff --git a/tests/portfolio/test_channel_test.py b/tests/portfolio/test_channel_test.py index 21e29d1..a57cec7 100644 --- a/tests/portfolio/test_channel_test.py +++ b/tests/portfolio/test_channel_test.py @@ -115,7 +115,8 @@ def test_t1_probe_fires(): assert broker.calls == [("otv", "510300.XSHG", 0)] -def test_initialize_registers_intraday_schedules(): +def test_initialize_registers_intraday_schedules(monkeypatch): + monkeypatch.delenv("SANGUO_LIVE_STRATEGY", raising=False) # 防环境串扰翻到随启随验 registered: list[tuple[str, str]] = [] class _RegBroker(_MockBroker): @@ -129,3 +130,65 @@ def test_initialize_registers_intraday_schedules(): s.initialize(object()) times = {t for _, t in registered} assert {"9:35", "10:45", "13:45", "14:30"} <= times # 四个盘中时点全注册 + + +# ---------------- 随启随验(2026-08-20) ---------------- + +def test_initialize_on_demand_registers_single_probe(monkeypatch): + """LiveEngine env 下只注册一个"启动后约90秒"的 probe_all,不再挂固定时点。""" + from datetime import datetime + + monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "channel_test") + registered: list[tuple[str, str]] = [] + + class _RegBroker(_MockBroker): + def __init__(self) -> None: + super().__init__() + self.run_daily = lambda fn, t: registered.append( + (getattr(fn, "__name__", str(fn)), t)) + + s = ChannelTestStrategy(provider=None, broker=_RegBroker(), + config=ChannelTestConfig()) + s.initialize(object()) + assert len(registered) == 1 + name, t = registered[0] + assert name == "probe_all" + hh, mm = t.split(":") + now = datetime.now() + delta = (int(hh) * 60 + int(mm)) - (now.hour * 60 + now.minute) + assert 0 <= delta <= 3 # 未来 ~90 秒(±跨分钟容差) + + +def test_initialize_on_demand_explicit_off_keeps_schedule(monkeypatch): + """on_demand=False 显式覆盖:即使 LiveEngine env 在,也维持四时点旧行为。""" + monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "channel_test") + registered: list[tuple[str, str]] = [] + + class _RegBroker(_MockBroker): + def __init__(self) -> None: + super().__init__() + self.run_daily = lambda fn, t: registered.append( + (getattr(fn, "__name__", str(fn)), t)) + + s = ChannelTestStrategy(provider=None, broker=_RegBroker(), + config=ChannelTestConfig(on_demand=False)) + s.initialize(object()) + assert {"9:35", "10:45", "13:45", "14:30"} <= {t for _, t in registered} + + +def test_probe_all_runs_sizing_sell_t1_swap_sequence(): + """一轮探针打满:等权买入(定寸) + 卖出(只卖自己源=own 持仓) + T+1 + 换仓。""" + broker = _MockBroker() + s = ChannelTestStrategy(provider=None, broker=broker, + config=ChannelTestConfig(hold_n=2, probe_t1=True, + intraday_partial=False, + intraday_swap=True, + on_demand=True)) + ctx = _Ctx({"510300.XSHG": _Pos(5000), "600519.XSHG": _Pos(5000)}, cash=4000) + s.probe_all(ctx) + buys = [c for m, c, v in broker.calls if v > 0] + sells = [c for m, c, v in broker.calls if v == 0] + assert buys # 定寸买入发生(实例视图等权) + 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 # 资金复用:卖后另买未持有票