feat(strategy): channel_test随启随验探针模式(用户拍板2026-08-20:开盘窗口不再是验证瓶颈,随时启动随时验全场景)——on_demand=None自动:LiveEngine进程(supervisor注入SANGUO_LIVE_STRATEGY)→启动/重启后~90s经facade.run_daily注册probe_all打一轮全场景;必须走facade注册而非直接调用:回调才收到B2包装的代理context(实例预算视图),绕过它探针测不到定寸虚拟化;影子柜台24h可验(收盘价撮合),实盘闭市下单被QMT拒亦属验证项(决策层输出已在日志);重启实例=再验一轮;[PROBE-0]实例视图快照(cash=预算切片非全账户=B2生效标志)/[PROBE-1]rotate主调仓定寸=实例视图等权/[PROBE-2]只卖自己(卖出对象全部来自own账本,对照前端账户实况实例分解)/[PROBE-3]T+1当日买立即卖应拒(冷启动空仓跳过,第二轮起有票)/[PROBE-4]swap_one资金复用;回测/单测无env维持四时点旧行为,显式True/False覆盖;+3测试(env下仅注册1个未来~90s的probe_all/显式off保四时点/probe_all全序列下单断言);旧四时点测试补delenv防环境串扰 [vps]
This commit is contained in:
@@ -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。"""
|
||||
|
||||
@@ -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 # 资金复用:卖后另买未持有票
|
||||
|
||||
Reference in New Issue
Block a user