You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
379 lines
14 KiB
Python
379 lines
14 KiB
Python
"""Unit tests for the Breakout52w engine.
|
|
|
|
Mirror of test_low_vol_anomaly.py: same lookahead defense pattern, same
|
|
rebalance-day semantics, same provider Protocol. The signal differs —
|
|
the engine emits symbols that printed a NEW 52-week-high close on T-1,
|
|
ranked DESCENDING by 20d volume ratio.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from libs.backtest.domain import LookaheadViolationError, StrategyEngineConfig
|
|
from libs.backtest.breakout_52w import (
|
|
BREAKOUT_52W_EVENT_TYPE,
|
|
_SnapshotStoreBarAdapter,
|
|
build_candidates,
|
|
compute_52w_high_breakout,
|
|
compute_volume_ratio_20d,
|
|
is_rebalance_day,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_engine(**overrides: Any) -> StrategyEngineConfig:
|
|
base: dict[str, Any] = dict(
|
|
engine_id="breakout_52w_silo",
|
|
event_types=[BREAKOUT_52W_EVENT_TYPE],
|
|
direction="long_only",
|
|
timing_class="after_close",
|
|
entry_timing_policy="next_open",
|
|
max_holding_days=21,
|
|
breakout_52w_enabled=True,
|
|
breakout_52w_lookback_days=30, # smaller for tests
|
|
breakout_52w_top_n=3,
|
|
breakout_52w_holding_days=21,
|
|
breakout_52w_min_avg_dollar_volume=1_000_000.0,
|
|
breakout_52w_min_price=5.0,
|
|
breakout_52w_stop_pct=0.10,
|
|
breakout_52w_target_pct=0.30,
|
|
)
|
|
base.update(overrides)
|
|
return StrategyEngineConfig(**base)
|
|
|
|
|
|
def _business_days(start: dt.date, count: int) -> list[dt.date]:
|
|
out: list[dt.date] = []
|
|
cursor = start
|
|
while len(out) < count:
|
|
if cursor.weekday() < 5:
|
|
out.append(cursor)
|
|
cursor = cursor + dt.timedelta(days=1)
|
|
return out
|
|
|
|
|
|
def _flat_then_breakout(
|
|
days: list[dt.date],
|
|
*,
|
|
flat_close: float = 50.0,
|
|
breakout_close: float = 60.0,
|
|
flat_volume: float = 1_000_000.0,
|
|
breakout_volume: float = 5_000_000.0,
|
|
) -> dict[dt.date, dict[str, Any]]:
|
|
"""All bars at flat_close (so flat_close is the max-high) until the LAST bar
|
|
breaks out to breakout_close on elevated volume."""
|
|
out: dict[dt.date, dict[str, Any]] = {}
|
|
last_idx = len(days) - 1
|
|
for i, d in enumerate(days):
|
|
if i == last_idx:
|
|
c = breakout_close
|
|
v = breakout_volume
|
|
else:
|
|
c = flat_close
|
|
v = flat_volume
|
|
out[d] = {
|
|
"open": c,
|
|
"high": c + 0.1,
|
|
"low": c - 0.1,
|
|
"close": c,
|
|
"volume": v,
|
|
}
|
|
return out
|
|
|
|
|
|
def _flat_only(
|
|
days: list[dt.date],
|
|
*,
|
|
close: float = 50.0,
|
|
volume: float = 1_000_000.0,
|
|
) -> dict[dt.date, dict[str, Any]]:
|
|
"""All bars flat — NEVER a 52w breakout (close == prior max high)."""
|
|
out: dict[dt.date, dict[str, Any]] = {}
|
|
for d in days:
|
|
out[d] = {
|
|
"open": close,
|
|
"high": close + 0.1,
|
|
"low": close - 0.1,
|
|
"close": close,
|
|
"volume": volume,
|
|
}
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pure trigger computations — 52w-high logic on synthetic data
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_52w_high_breakout_excludes_t_minus_1_from_max_window() -> None:
|
|
"""The prior-max-high window MUST be bars[-(lookback+1):-1], NOT including T-1.
|
|
|
|
Setup: all 60 bars at high=50, then T-1 closes at 60 (so its high=60.1).
|
|
If the window incorrectly INCLUDED T-1, the prior max would be 60.1 and
|
|
the breakout check would (incorrectly) compare 60 > 60.1 = False.
|
|
With the correct exclusion, prior max = 50.1 and breakout = 60 > 50.1 = True.
|
|
"""
|
|
days = _business_days(dt.date(2024, 1, 2), 60)
|
|
bars_dict = _flat_then_breakout(days, flat_close=50.0, breakout_close=60.0)
|
|
bars = sorted(bars_dict.items(), key=lambda kv: kv[0])
|
|
is_brk, last_close, prior_max, used = compute_52w_high_breakout(
|
|
bars, lookback_days=30
|
|
)
|
|
assert is_brk is True
|
|
assert last_close == 60.0
|
|
assert abs(prior_max - 50.1) < 1e-9
|
|
# 30 bars in the window, all strictly before T-1
|
|
assert len(used) == 30
|
|
assert all(d < days[-1] for d in used)
|
|
|
|
|
|
def test_52w_high_breakout_no_break_on_flat_series() -> None:
|
|
"""Constant close + constant high → close == prior max high, not strictly greater."""
|
|
days = _business_days(dt.date(2024, 1, 2), 60)
|
|
bars_dict = _flat_only(days, close=50.0)
|
|
bars = sorted(bars_dict.items(), key=lambda kv: kv[0])
|
|
is_brk, _, _, _ = compute_52w_high_breakout(bars, lookback_days=30)
|
|
assert is_brk is False
|
|
|
|
|
|
def test_52w_high_breakout_insufficient_history() -> None:
|
|
days = _business_days(dt.date(2024, 1, 2), 5)
|
|
bars_dict = _flat_only(days)
|
|
bars = sorted(bars_dict.items(), key=lambda kv: kv[0])
|
|
is_brk, _, _, used = compute_52w_high_breakout(bars, lookback_days=30)
|
|
assert is_brk is False
|
|
assert used == []
|
|
|
|
|
|
def test_volume_ratio_20d_basic() -> None:
|
|
"""volume_ratio = vol_T-1 / median(vol over 20 bars ending T-2)."""
|
|
days = _business_days(dt.date(2024, 1, 2), 30)
|
|
bars_dict = _flat_then_breakout(
|
|
days, flat_volume=1_000_000.0, breakout_volume=5_000_000.0
|
|
)
|
|
bars = sorted(bars_dict.items(), key=lambda kv: kv[0])
|
|
ratio = compute_volume_ratio_20d(bars)
|
|
assert ratio is not None
|
|
# Last bar vol 5M, prior 20-bar median = 1M → ratio = 5.0
|
|
assert abs(ratio - 5.0) < 1e-9
|
|
|
|
|
|
def test_volume_ratio_20d_insufficient_history() -> None:
|
|
days = _business_days(dt.date(2024, 1, 2), 10)
|
|
bars_dict = _flat_only(days)
|
|
bars = sorted(bars_dict.items(), key=lambda kv: kv[0])
|
|
ratio = compute_volume_ratio_20d(bars)
|
|
assert ratio is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rebalance-day semantics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_is_rebalance_day_first_trading_day_of_new_month() -> None:
|
|
bars = [(dt.date(2024, 1, 31), {"close": 100.0})]
|
|
assert is_rebalance_day(bars, dt.date(2024, 2, 1)) is True
|
|
|
|
|
|
def test_is_rebalance_day_mid_month() -> None:
|
|
bars = [(dt.date(2024, 1, 15), {"close": 100.0})]
|
|
assert is_rebalance_day(bars, dt.date(2024, 1, 16)) is False
|
|
|
|
|
|
def test_is_rebalance_day_empty_bars_returns_false() -> None:
|
|
assert is_rebalance_day([], dt.date(2024, 1, 2)) is False
|
|
|
|
|
|
def test_is_rebalance_day_lookahead_raises() -> None:
|
|
bars = [(dt.date(2024, 2, 1), {"close": 100.0})]
|
|
with pytest.raises(LookaheadViolationError):
|
|
is_rebalance_day(bars, dt.date(2024, 2, 1))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lookahead defense: build_candidates rejects T+0 bars
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _LeakyProvider:
|
|
"""Returns bars INCLUDING the decision_date — must be caught by build_candidates."""
|
|
|
|
def __init__(self, bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]]) -> None:
|
|
self.bars_by_symbol = bars_by_symbol
|
|
|
|
def get_bars_before(
|
|
self, symbol: str, as_of_date: dt.date, lookback_days: int
|
|
) -> list[tuple[dt.date, dict[str, Any]]]:
|
|
sym_bars = self.bars_by_symbol.get(symbol.upper(), {})
|
|
ordered = sorted(sym_bars.items(), key=lambda kv: kv[0])
|
|
return [(d, b) for d, b in ordered if d <= as_of_date][-lookback_days:]
|
|
|
|
|
|
def test_build_raises_lookahead_when_provider_returns_t0_bar() -> None:
|
|
"""Proof-by-contradiction: a leaky provider that returns T+0 must trip the assertion."""
|
|
days = _business_days(dt.date(2024, 1, 2), 80)
|
|
decision_date = days[-1]
|
|
next_date = decision_date + dt.timedelta(days=1)
|
|
bars_dict = _flat_then_breakout(days)
|
|
leaky = _LeakyProvider({"FOO": bars_dict})
|
|
engine = _make_engine()
|
|
with pytest.raises(LookaheadViolationError):
|
|
build_candidates(
|
|
decision_date=decision_date,
|
|
next_trading_date=next_date,
|
|
universe_symbols=["FOO"],
|
|
engine=engine,
|
|
bar_provider=leaky,
|
|
)
|
|
|
|
|
|
def test_build_returns_empty_on_non_rebalance_day() -> None:
|
|
"""Mid-month decision_date must yield 0 candidates (no rebalance)."""
|
|
days = _business_days(dt.date(2024, 1, 2), 80)
|
|
bars_dict = _flat_then_breakout(days)
|
|
bars_by_symbol = {f"SYM{i}": bars_dict for i in range(5)}
|
|
adapter = _SnapshotStoreBarAdapter(bars_by_symbol=bars_by_symbol)
|
|
engine = _make_engine()
|
|
|
|
# Use a mid-month date (last_bar.month == decision_date.month).
|
|
decision_date = days[40]
|
|
next_date = decision_date + dt.timedelta(days=1)
|
|
candidates = build_candidates(
|
|
decision_date=decision_date,
|
|
next_trading_date=next_date,
|
|
universe_symbols=list(bars_by_symbol.keys()),
|
|
engine=engine,
|
|
bar_provider=adapter,
|
|
)
|
|
assert candidates == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Build candidates — happy path with multiple symbols + top-N selection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_emits_top_n_breakouts_ranked_by_volume_ratio() -> None:
|
|
"""Construct symbols all breaking out with distinct volume ratios; highest ratio
|
|
must rank first."""
|
|
days = _business_days(dt.date(2024, 1, 2), 200)
|
|
rebalance_idx = None
|
|
for i in range(40, len(days)):
|
|
if days[i].month != days[i - 1].month:
|
|
rebalance_idx = i
|
|
break
|
|
assert rebalance_idx is not None
|
|
decision_date = days[rebalance_idx]
|
|
next_date = decision_date + dt.timedelta(days=1)
|
|
# Include all days strictly BEFORE the decision_date so bars[-1] is T-1.
|
|
used_days = days[:rebalance_idx]
|
|
|
|
bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]] = {}
|
|
# All three break out (close 60 > prior max high ~50.1) with distinct volume spikes.
|
|
bars_by_symbol["LOWVOL_BREAK"] = _flat_then_breakout(
|
|
used_days, flat_close=50.0, breakout_close=60.0,
|
|
flat_volume=1_000_000.0, breakout_volume=2_000_000.0,
|
|
)
|
|
bars_by_symbol["MIDVOL_BREAK"] = _flat_then_breakout(
|
|
used_days, flat_close=50.0, breakout_close=60.0,
|
|
flat_volume=1_000_000.0, breakout_volume=4_000_000.0,
|
|
)
|
|
bars_by_symbol["HIVOL_BREAK"] = _flat_then_breakout(
|
|
used_days, flat_close=50.0, breakout_close=60.0,
|
|
flat_volume=1_000_000.0, breakout_volume=8_000_000.0,
|
|
)
|
|
# Flat-only — should NEVER appear (no 52w breakout).
|
|
bars_by_symbol["NO_BREAK"] = _flat_only(used_days, close=50.0)
|
|
|
|
adapter = _SnapshotStoreBarAdapter(bars_by_symbol=bars_by_symbol)
|
|
engine = _make_engine(breakout_52w_top_n=3)
|
|
candidates = build_candidates(
|
|
decision_date=decision_date,
|
|
next_trading_date=next_date,
|
|
universe_symbols=list(bars_by_symbol.keys()),
|
|
engine=engine,
|
|
bar_provider=adapter,
|
|
)
|
|
assert len(candidates) == 3
|
|
syms_emitted = [c.symbol for c in candidates]
|
|
# Highest volume ratio MUST be ranked #1
|
|
assert syms_emitted[0] == "HIVOL_BREAK"
|
|
# All three breakouts emitted; NO_BREAK absent
|
|
assert set(syms_emitted) == {"HIVOL_BREAK", "MIDVOL_BREAK", "LOWVOL_BREAK"}
|
|
|
|
|
|
def test_candidate_has_no_progress_and_trailing_disabled() -> None:
|
|
"""The synthetic candidates must override v9.x base-config no-progress and
|
|
trailing gates so the engine actually holds for its full window."""
|
|
days = _business_days(dt.date(2024, 1, 2), 200)
|
|
rebalance_idx = None
|
|
for i in range(40, len(days)):
|
|
if days[i].month != days[i - 1].month:
|
|
rebalance_idx = i
|
|
break
|
|
assert rebalance_idx is not None
|
|
decision_date = days[rebalance_idx]
|
|
next_date = decision_date + dt.timedelta(days=1)
|
|
# Include all days strictly BEFORE the decision_date so bars[-1] is T-1.
|
|
used_days = days[:rebalance_idx]
|
|
|
|
bars_by_symbol = {
|
|
"BREAKOUT": _flat_then_breakout(
|
|
used_days, flat_close=50.0, breakout_close=60.0,
|
|
flat_volume=1_000_000.0, breakout_volume=5_000_000.0,
|
|
)
|
|
}
|
|
adapter = _SnapshotStoreBarAdapter(bars_by_symbol=bars_by_symbol)
|
|
engine = _make_engine(breakout_52w_top_n=1)
|
|
cands = build_candidates(
|
|
decision_date=decision_date,
|
|
next_trading_date=next_date,
|
|
universe_symbols=list(bars_by_symbol.keys()),
|
|
engine=engine,
|
|
bar_provider=adapter,
|
|
)
|
|
assert len(cands) == 1
|
|
c = cands[0]
|
|
assert c.engine_early_failure_no_progress_days == 0
|
|
assert c.engine_early_failure_no_progress_r == 0.0
|
|
assert c.engine_trailing_model == "none"
|
|
assert c.event_type == BREAKOUT_52W_EVENT_TYPE
|
|
assert c.trade_direction == "long"
|
|
|
|
|
|
def test_no_break_no_candidates() -> None:
|
|
"""If no symbol prints a 52w breakout, return []."""
|
|
days = _business_days(dt.date(2024, 1, 2), 200)
|
|
rebalance_idx = None
|
|
for i in range(40, len(days)):
|
|
if days[i].month != days[i - 1].month:
|
|
rebalance_idx = i
|
|
break
|
|
assert rebalance_idx is not None
|
|
decision_date = days[rebalance_idx]
|
|
next_date = decision_date + dt.timedelta(days=1)
|
|
# Include all days strictly BEFORE the decision_date so bars[-1] is T-1.
|
|
used_days = days[:rebalance_idx]
|
|
|
|
bars_by_symbol = {
|
|
f"FLAT{i}": _flat_only(used_days, close=50.0) for i in range(5)
|
|
}
|
|
adapter = _SnapshotStoreBarAdapter(bars_by_symbol=bars_by_symbol)
|
|
engine = _make_engine()
|
|
cands = build_candidates(
|
|
decision_date=decision_date,
|
|
next_trading_date=next_date,
|
|
universe_symbols=list(bars_by_symbol.keys()),
|
|
engine=engine,
|
|
bar_provider=adapter,
|
|
)
|
|
assert cands == []
|