test(api): 修channel_test随启随验过时断言——实盘=单probe_all定时(启动+90s随墙钟),四时点改由无env直测钉死 [nas]
策略session 08-24移交项#2: test_facade_injects_run_daily_for_channel_test 在HEAD恒红,非flaky——原断言的9:35/10:45/13:45/14:30四时点是回测/无env路径;实盘(SANGUO_LIVE_STRATEGY置位)自随启随验(channel_test.py:78-84)只注册一个「启动+90s」的probe_all,时:分随墙钟(13:33/13:34/13:35三跑三变之源)。 修=断言随启随验设计:单probe_all+时刻落在[注册前+90s,注册后+90s]取分窗口(纯时刻比较防1900年日期陷阱,容忍跨午夜);新增无env直测(ChannelTestConfig(on_demand=False))钉死四时点路径,原断言价值不丢。 3跑稳定;tests/api 165全绿。
This commit is contained in:
@@ -233,23 +233,61 @@ def test_setup_idempotent_across_both_hooks(monkeypatch):
|
||||
|
||||
|
||||
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 挂上。"""
|
||||
"""channel_test 自注册依赖 facade.run_daily(缺注入时静默 no-op 永不开仓)。
|
||||
实盘(env 置位)走随启随验:只注册一个「启动后约90秒」的 probe_all 定时
|
||||
(时间随墙钟=原断言四时点恒红的原因);四时点行为由下方无 env 直测钉死。"""
|
||||
import bullet_trade.core as bt_core
|
||||
from datetime import datetime, timedelta
|
||||
from sanguo_portfolio import live_strategy
|
||||
|
||||
registered = _patch_live_wiring(monkeypatch)
|
||||
monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "channel_test")
|
||||
_reset_live_state()
|
||||
|
||||
before = datetime.now()
|
||||
live_strategy.process_initialize(object())
|
||||
after = datetime.now()
|
||||
|
||||
daily_times = sorted(r[2] for r in registered if r[0] == "daily")
|
||||
assert daily_times == ["10:45", "13:45", "14:30", "9:35"]
|
||||
daily = [r for r in registered if r[0] == "daily"]
|
||||
assert [r[1] for r in daily] == ["probe_all"]
|
||||
# 注册时刻在 before~after 之间 → 探针时:分必落在 [floor(before+90s), floor(after+90s)]
|
||||
# (纯时刻比较;strptime 是 1900 年日期不能与 now 直接比。跨午夜时两者恰为首尾)
|
||||
probe_t = datetime.strptime(daily[0][2], "%H:%M").time()
|
||||
lo_t = (before + timedelta(seconds=90)).replace(second=0, microsecond=0).time()
|
||||
hi_t = (after + timedelta(seconds=90)).replace(second=0, microsecond=0).time()
|
||||
assert lo_t <= probe_t <= hi_t or probe_t in (lo_t, hi_t)
|
||||
# facade 拿到的是 bullet_trade 顶层 run_daily 本尊,不是默认空 lambda
|
||||
assert live_strategy._STATE["strategy"].broker.run_daily is bt_core.run_daily
|
||||
|
||||
|
||||
def test_channel_test_backtest_mode_registers_four_fixed_times():
|
||||
"""回测/无 env 路径:9:35/10:45/13:45/14:30 四时点保留(随启随验只影响实盘)。"""
|
||||
from sanguo_portfolio.strategies.channel_test import (
|
||||
ChannelTestConfig, ChannelTestStrategy,
|
||||
)
|
||||
|
||||
class _FakeBroker:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def set_benchmark(self, *a):
|
||||
pass
|
||||
|
||||
def set_option(self, *a):
|
||||
pass
|
||||
|
||||
def run_daily(self, f, t):
|
||||
self.calls.append((getattr(f, "__name__", str(f)), t))
|
||||
|
||||
b = _FakeBroker()
|
||||
s = ChannelTestStrategy(
|
||||
provider=object(), broker=b, config=ChannelTestConfig(on_demand=False))
|
||||
s.initialize(object())
|
||||
|
||||
times = sorted(t for _, t in b.calls)
|
||||
assert times == ["10:45", "13:45", "14:30", "9:35"]
|
||||
|
||||
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user