diff --git a/tests/api/test_portfolio_live.py b/tests/api/test_portfolio_live.py index fdae217..7585c1a 100644 --- a/tests/api/test_portfolio_live.py +++ b/tests/api/test_portfolio_live.py @@ -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