yirugi
/
fithia2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
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.
main
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '956cc78f1b'
${ noResults }
fithia2/tests/unit/backtest/test_vol_breakout_52w.py

901 lines
34 KiB
Python
Raw Blame History

"""Unit tests for the VolBreakout52w engine.
Heavy emphasis on look-ahead defenses — this engine is the honest descendant of
the retired topgainer v1-v54 lineage which collapsed +267% / Sharpe 13.73 →
-4.3% / Sharpe -1.04 once Phase-1's daily_high look-ahead was removed
(memory: project_topgainer_phase1_lookahead_2026-05-05.md). The defense MUST be
airtight.
"""
from __future__ import annotations
import datetime as dt
import random
from typing import Any
import pytest
from libs.backtest.domain import LookaheadViolationError, StrategyEngineConfig
from libs.backtest.vol_breakout_52w import (
VOL_BREAKOUT_52W_EVENT_TYPE,
FrozenT1Features,
VolBreakout52wTriggerInputs,
_SnapshotStoreBarAdapter,
_assert_features_strictly_before_decision_open,
build_candidates,
compute_52w_high_breakout,
compute_atr_normalized,
compute_volume_ratio,
evaluate_trigger,
)
# ---------------------------------------------------------------------------
# Test helpers
# ---------------------------------------------------------------------------
def _make_engine(**overrides: Any) -> StrategyEngineConfig:
base: dict[str, Any] = dict(
engine_id="vol_breakout_52w_long",
event_types=[VOL_BREAKOUT_52W_EVENT_TYPE],
direction="long_only",
timing_class="after_close",
entry_timing_policy="next_open",
max_holding_days=2,
vol_breakout_52w_enabled=True,
vol_breakout_52w_lookback_days=60, # smaller for tests
vol_breakout_52w_volume_ratio_min=2.0,
vol_breakout_52w_volume_median_window=20,
vol_breakout_52w_atr_normalized_min=0.015,
vol_breakout_52w_atr_normalized_max=0.06,
vol_breakout_52w_pre_open_gap_max=0.04,
vol_breakout_52w_skip_if_no_gap_data=True,
vol_breakout_52w_min_avg_dollar_volume=10_000_000.0,
vol_breakout_52w_min_price=5.0,
vol_breakout_52w_stop_pct=0.03,
vol_breakout_52w_target_pct=0.05,
vol_breakout_52w_max_holding_days=2,
)
base.update(overrides)
return StrategyEngineConfig(**base)
def _generate_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 _build_bars(
symbol: str,
trading_days: list[dt.date],
*,
base_close: float = 100.0,
base_high: float = 100.5,
base_low: float = 99.5,
base_volume: float = 5_000_000.0,
last_close: float | None = None,
last_high: float | None = None,
last_low: float | None = None,
last_volume: float | None = None,
atr_jitter: float = 0.5,
) -> dict[str, dict[dt.date, dict[str, Any]]]:
"""Construct a dict-of-dicts bars store for the single symbol.
Default: flat history at base_close, with a small atr_jitter on H-L.
Customize the FINAL bar (T-1 in tests) via the ``last_*`` arguments.
"""
inner: dict[dt.date, dict[str, Any]] = {}
n = len(trading_days)
for i, d in enumerate(trading_days):
is_last = (i == n - 1)
if is_last and last_close is not None:
close_v = last_close
high_v = last_high if last_high is not None else last_close + 0.5
low_v = last_low if last_low is not None else last_close - 0.5
volume_v = last_volume if last_volume is not None else base_volume
else:
close_v = base_close + (i % 7) * 0.05 # tiny drift, never crosses base_high
high_v = base_high + (i % 5) * 0.1 * atr_jitter
low_v = base_low - (i % 5) * 0.1 * atr_jitter
volume_v = base_volume * (1.0 + 0.02 * ((i % 5) - 2))
inner[d] = {
"open": close_v,
"high": high_v,
"low": low_v,
"close": close_v,
"volume": volume_v,
}
return {symbol.upper(): inner}
# ---------------------------------------------------------------------------
# Pure feature computations
# ---------------------------------------------------------------------------
def test_compute_52w_high_breakout_fires_when_close_above_window_max():
days = _generate_business_days(dt.date(2026, 1, 5), 70)
bars = _build_bars("AAPL", days, base_close=100.0, base_high=110.0,
last_close=120.0, last_high=121.0, last_low=119.0)["AAPL"]
series = sorted(bars.items())
is_b, last_close, prior_max, used = compute_52w_high_breakout(series, lookback_days=60)
assert is_b is True
assert last_close == 120.0
assert prior_max <= 110.5 # base_high + small jitter
assert len(used) >= 20
def test_compute_52w_high_breakout_does_not_fire_when_close_at_or_below_max():
days = _generate_business_days(dt.date(2026, 1, 5), 70)
bars = _build_bars("AAPL", days, base_close=100.0, base_high=110.0,
last_close=109.0, last_high=109.5, last_low=108.5)["AAPL"]
series = sorted(bars.items())
is_b, _last_close, prior_max, _used = compute_52w_high_breakout(series, lookback_days=60)
assert is_b is False
assert prior_max >= 109.0
def test_compute_volume_ratio_and_atr_normalized_basic():
days = _generate_business_days(dt.date(2026, 1, 5), 35)
bars = _build_bars("AAPL", days, base_volume=1_000_000.0, last_volume=4_000_000.0,
last_close=100.0, last_high=102.0, last_low=98.0)["AAPL"]
series = sorted(bars.items())
vol_t1, median_t2 = compute_volume_ratio(series, median_window=20)
assert vol_t1 == 4_000_000.0
assert median_t2 == pytest.approx(1_000_000.0, rel=0.05)
atr_norm = compute_atr_normalized(series, window=14)
assert atr_norm is not None
assert 0.005 < atr_norm < 0.06 # synthetic data should lie in a sane band
# ---------------------------------------------------------------------------
# evaluate_trigger — happy path + 4 negative cases
# ---------------------------------------------------------------------------
def _trigger_inputs(**overrides: Any) -> VolBreakout52wTriggerInputs:
base: dict[str, Any] = dict(
symbol="AAPL",
decision_date=dt.date(2026, 4, 13),
next_trading_date=dt.date(2026, 4, 14),
last_bar_date=dt.date(2026, 4, 10),
last_bar_timestamp=dt.datetime(2026, 4, 10, 21, 0, tzinfo=dt.timezone.utc),
last_close=120.0,
prior_252d_max_high=110.0,
is_52w_breakout=True,
volume_t_minus_1=4_000_000.0,
median_volume_20d_t_minus_2=1_000_000.0,
atr_normalized_t_minus_1=0.030,
avg_dollar_volume_20d=200_000_000.0,
pre_open_gap_pct=0.01,
)
base.update(overrides)
return VolBreakout52wTriggerInputs(**base)
def test_trigger_fires_when_all_three_conditions_met():
engine = _make_engine()
passes, reason = evaluate_trigger(_trigger_inputs(), engine)
assert passes is True, reason
assert reason is None
def test_trigger_blocks_when_not_a_breakout():
engine = _make_engine()
passes, reason = evaluate_trigger(_trigger_inputs(is_52w_breakout=False), engine)
assert passes is False
assert "prior 252d max high" in (reason or "")
def test_trigger_blocks_when_volume_ratio_below_min():
engine = _make_engine()
passes, reason = evaluate_trigger(_trigger_inputs(volume_t_minus_1=1_500_000.0), engine)
assert passes is False
assert "volume_ratio" in (reason or "")
def test_trigger_blocks_when_atr_below_band():
engine = _make_engine()
passes, reason = evaluate_trigger(_trigger_inputs(atr_normalized_t_minus_1=0.010), engine)
assert passes is False
assert "atr_normalized" in (reason or "")
def test_trigger_blocks_when_atr_above_band_parabolic():
engine = _make_engine()
passes, reason = evaluate_trigger(_trigger_inputs(atr_normalized_t_minus_1=0.080), engine)
assert passes is False
assert "atr_normalized" in (reason or "")
def test_trigger_blocks_when_price_below_min():
engine = _make_engine(vol_breakout_52w_min_price=10.0)
passes, reason = evaluate_trigger(_trigger_inputs(last_close=4.0), engine)
assert passes is False
assert "min price" in (reason or "")
def test_trigger_blocks_when_adv_below_min():
engine = _make_engine(vol_breakout_52w_min_avg_dollar_volume=50_000_000.0)
passes, reason = evaluate_trigger(_trigger_inputs(avg_dollar_volume_20d=10_000_000.0), engine)
assert passes is False
assert "avg_dollar_volume" in (reason or "")
# ---------------------------------------------------------------------------
# Pre-open gap fade guard
# ---------------------------------------------------------------------------
def test_trigger_blocks_when_pre_open_gap_exceeds_max():
engine = _make_engine()
passes, reason = evaluate_trigger(_trigger_inputs(pre_open_gap_pct=0.05), engine)
assert passes is False
assert "pre_open_gap" in (reason or "")
def test_trigger_passes_when_pre_open_gap_within_max():
engine = _make_engine()
passes, reason = evaluate_trigger(_trigger_inputs(pre_open_gap_pct=0.03), engine)
assert passes is True
assert reason is None
def test_trigger_passes_when_pre_open_gap_data_missing_and_skip_flag_true():
"""Missing gap data + flag=True → no enforcement (skip-with-warning path)."""
engine = _make_engine(vol_breakout_52w_skip_if_no_gap_data=True)
passes, reason = evaluate_trigger(_trigger_inputs(pre_open_gap_pct=None), engine)
assert passes is True
assert reason is None
def test_trigger_passes_when_pre_open_gap_data_missing_and_skip_flag_false():
"""Missing gap data + flag=False → also no enforcement (we cannot enforce a
guard with no data; the warning is logged at the build level)."""
engine = _make_engine(vol_breakout_52w_skip_if_no_gap_data=False)
passes, reason = evaluate_trigger(_trigger_inputs(pre_open_gap_pct=None), engine)
assert passes is True
assert reason is None
# ---------------------------------------------------------------------------
# Look-ahead defenses — the load-bearing tests for this engine
# ---------------------------------------------------------------------------
def test_assert_no_lookahead_rejects_t0_intraday_timestamp():
"""A feature timestamp at 10:30 ET on decision_date is a categorical look-ahead."""
decision_date = dt.date(2026, 4, 13)
leaky_ts = dt.datetime(2026, 4, 13, 14, 30, tzinfo=dt.timezone.utc) # 10:30 ET
with pytest.raises(LookaheadViolationError):
_assert_features_strictly_before_decision_open(
"AAPL", decision_date, [leaky_ts]
)
def test_assert_no_lookahead_rejects_naive_timestamp():
decision_date = dt.date(2026, 4, 13)
with pytest.raises(LookaheadViolationError):
_assert_features_strictly_before_decision_open(
"AAPL", decision_date, [dt.datetime(2026, 4, 10, 21, 0)]
)
def test_assert_no_lookahead_accepts_strictly_prior_timestamp():
decision_date = dt.date(2026, 4, 13)
safe_ts = dt.datetime(2026, 4, 10, 21, 0, tzinfo=dt.timezone.utc)
_assert_features_strictly_before_decision_open(
"AAPL", decision_date, [safe_ts]
) # must NOT raise
def test_evaluate_trigger_re_asserts_last_bar_strictly_before_decision_date():
"""Defence-in-depth: even if a leaky provider snuck through, evaluate_trigger
must trip on ``last_bar_date >= decision_date``. This is the categorical
catch for the topgainer v1-v54 bug."""
engine = _make_engine()
inputs = _trigger_inputs(
decision_date=dt.date(2026, 4, 13),
last_bar_date=dt.date(2026, 4, 13), # SAME DAY — look-ahead
)
with pytest.raises(LookaheadViolationError):
evaluate_trigger(inputs, engine)
def test_frozen_t1_features_blocks_forbidden_field_substring():
"""FrozenT1Features must refuse extras whose names encode T+0 data."""
decision_date = dt.date(2026, 4, 13)
last_bar_date = dt.date(2026, 4, 10)
with pytest.raises(LookaheadViolationError) as excinfo:
FrozenT1Features(
symbol="AAPL",
decision_date=decision_date,
last_bar_date=last_bar_date,
last_close=120.0,
high_252d_max=110.0,
high_252d_max_window=[],
extra={"daily_high": 121.0}, # forbidden — encodes T+0 data
)
assert "daily_high" in str(excinfo.value)
def test_frozen_t1_features_blocks_daily_close_extra():
decision_date = dt.date(2026, 4, 13)
last_bar_date = dt.date(2026, 4, 10)
with pytest.raises(LookaheadViolationError):
FrozenT1Features(
symbol="AAPL",
decision_date=decision_date,
last_bar_date=last_bar_date,
last_close=120.0,
high_252d_max=110.0,
high_252d_max_window=[],
extra={"reaction_daily_close": 121.0},
)
def test_frozen_t1_features_blocks_t0_window_date():
decision_date = dt.date(2026, 4, 13)
last_bar_date = dt.date(2026, 4, 10)
with pytest.raises(LookaheadViolationError):
FrozenT1Features(
symbol="AAPL",
decision_date=decision_date,
last_bar_date=last_bar_date,
last_close=120.0,
high_252d_max=110.0,
high_252d_max_window=[decision_date], # T+0 — forbidden
)
def test_frozen_t1_features_blocks_last_bar_at_or_after_decision_date():
decision_date = dt.date(2026, 4, 13)
with pytest.raises(LookaheadViolationError):
FrozenT1Features(
symbol="AAPL",
decision_date=decision_date,
last_bar_date=decision_date, # same-day — forbidden
last_close=120.0,
high_252d_max=110.0,
high_252d_max_window=[],
)
def test_frozen_t1_features_accepts_strictly_prior_data():
decision_date = dt.date(2026, 4, 13)
last_bar_date = dt.date(2026, 4, 10)
fts = FrozenT1Features(
symbol="AAPL",
decision_date=decision_date,
last_bar_date=last_bar_date,
last_close=120.0,
high_252d_max=110.0,
high_252d_max_window=[dt.date(2026, 1, 5), dt.date(2026, 4, 9)],
extra={"vol_breakout_52w_volume_ratio": 4.0},
)
assert fts.last_bar_date == last_bar_date
# ---------------------------------------------------------------------------
# build_candidates — leaky-provider proof-by-contradiction (the test that
# would have caught the topgainer v1-v54 bug)
# ---------------------------------------------------------------------------
def _build_full_setup(*, days_of_history: int = 80, breakout: bool = True,
vol_spike: float = 4.0):
days = _generate_business_days(dt.date(2026, 1, 5), days_of_history + 2)
decision_date = days[days_of_history]
next_trading_date = days[days_of_history + 1]
prior_days = days[:days_of_history]
last_close = 103.0 if breakout else 101.0
# Tight base H/L (100 +/- 1.5) keeps ATR/close ~0.02 — within the [0.015, 0.06] band.
bars = _build_bars(
"AAPL",
prior_days,
base_close=100.0,
base_high=101.5,
base_low=98.5,
base_volume=1_000_000.0,
last_close=last_close,
last_high=last_close + 1.0,
last_low=last_close - 1.0,
last_volume=int(1_000_000.0 * vol_spike),
atr_jitter=0.3,
)
return {
"symbol": "AAPL",
"decision_date": decision_date,
"next_trading_date": next_trading_date,
"trading_days": days,
"bars": bars,
"bar_provider": _SnapshotStoreBarAdapter(bars_by_symbol=bars),
}
def test_build_emits_candidate_for_eligible_symbol():
setup = _build_full_setup()
engine = _make_engine()
cands = build_candidates(
decision_date=setup["decision_date"],
next_trading_date=setup["next_trading_date"],
universe_symbols=[setup["symbol"]],
engine=engine,
bar_provider=setup["bar_provider"],
pre_open_gap_provider=None,
)
assert len(cands) == 1
cand = cands[0]
assert cand.event_type == VOL_BREAKOUT_52W_EVENT_TYPE
assert cand.symbol == "AAPL"
assert cand.execution_date == setup["next_trading_date"]
assert cand.engine_max_holding_days == 2
assert cand.features["vol_breakout_52w_stop_pct"] == 0.03
assert cand.features["vol_breakout_52w_target_pct"] == 0.05
# Defence-in-depth: candidate's event_timestamp must be strictly before
# 09:30 ET on the decision_date.
assert cand.event_timestamp.date() < setup["decision_date"]
def test_build_does_not_fire_when_not_a_breakout():
setup = _build_full_setup(breakout=False)
engine = _make_engine()
cands = build_candidates(
decision_date=setup["decision_date"],
next_trading_date=setup["next_trading_date"],
universe_symbols=[setup["symbol"]],
engine=engine,
bar_provider=setup["bar_provider"],
pre_open_gap_provider=None,
)
assert cands == []
def test_build_does_not_fire_when_volume_ratio_below_min():
setup = _build_full_setup(vol_spike=1.2)
engine = _make_engine()
cands = build_candidates(
decision_date=setup["decision_date"],
next_trading_date=setup["next_trading_date"],
universe_symbols=[setup["symbol"]],
engine=engine,
bar_provider=setup["bar_provider"],
pre_open_gap_provider=None,
)
assert cands == []
def test_build_raises_lookahead_when_provider_returns_t0_bar():
"""Inject a deliberately leaky provider that returns a bar dated == decision_date.
The engine MUST raise LookaheadViolationError. This is the proof-by-
contradiction test against the topgainer v1-v54 class of bug.
"""
setup = _build_full_setup()
decision_date = setup["decision_date"]
bars = setup["bars"]
# Inject a bar dated ON decision_date.
bars["AAPL"][decision_date] = {
"open": 122.0, "high": 130.0, "low": 121.0, "close": 129.0,
"volume": 9_000_000.0,
}
class LeakyAdapter:
"""Leaks T+0 bar into the screener — a topgainer-style bug."""
def get_bars_before(self, sym, as_of, lookback_days):
inner = bars[sym.upper()]
# Deliberately INCLUDE the bar dated == as_of_date.
ordered = sorted([(d, b) for d, b in inner.items() if d <= as_of])
return ordered[-lookback_days:]
engine = _make_engine()
with pytest.raises(LookaheadViolationError):
build_candidates(
decision_date=decision_date,
next_trading_date=setup["next_trading_date"],
universe_symbols=[setup["symbol"]],
engine=engine,
bar_provider=LeakyAdapter(),
pre_open_gap_provider=None,
)
def test_build_raises_when_next_trading_date_not_strictly_after_decision_date():
setup = _build_full_setup()
engine = _make_engine()
with pytest.raises(LookaheadViolationError):
build_candidates(
decision_date=setup["decision_date"],
next_trading_date=setup["decision_date"], # same day — forbidden
universe_symbols=[setup["symbol"]],
engine=engine,
bar_provider=setup["bar_provider"],
pre_open_gap_provider=None,
)
# ---------------------------------------------------------------------------
# Honest-replay test — clean vs leaky provider on identical data must produce
# either identical (clean ↔ clean) or raise (leaky). Deliberately leaky data
# must NOT silently produce different (better) candidates.
# ---------------------------------------------------------------------------
def test_honest_replay_clean_provider_is_deterministic():
setup = _build_full_setup()
engine = _make_engine()
cands_a = build_candidates(
decision_date=setup["decision_date"],
next_trading_date=setup["next_trading_date"],
universe_symbols=[setup["symbol"]],
engine=engine,
bar_provider=setup["bar_provider"],
pre_open_gap_provider=None,
)
cands_b = build_candidates(
decision_date=setup["decision_date"],
next_trading_date=setup["next_trading_date"],
universe_symbols=[setup["symbol"]],
engine=engine,
bar_provider=setup["bar_provider"],
pre_open_gap_provider=None,
)
# Identical inputs → identical outputs (modulo event_id which encodes inputs).
assert len(cands_a) == len(cands_b) == 1
assert cands_a[0].symbol == cands_b[0].symbol
assert cands_a[0].entry_price_est == cands_b[0].entry_price_est
assert cands_a[0].features == cands_b[0].features
def test_honest_replay_zero_shift_vs_minus1_shift_produces_identical_results():
"""Shift the source bars by 0 vs -1 day. With strict-before discipline,
both views generate the same trigger because the engine never reads T+0.
Specifically: take a setup whose decision_date is D. Then:
- Clean view: bars dated < D.
- Shifted-by-(-1) view: bars dated <= D-1 (== bars < D). SAME SET.
The key invariant is that 0-shift (no extra bar) and explicit -1 shift
yield identical candidates because we honor strict-before T.
"""
setup = _build_full_setup()
engine = _make_engine()
# View A: standard (strict-before T).
cands_a = build_candidates(
decision_date=setup["decision_date"],
next_trading_date=setup["next_trading_date"],
universe_symbols=[setup["symbol"]],
engine=engine,
bar_provider=setup["bar_provider"],
pre_open_gap_provider=None,
)
# View B: explicitly truncate to bars dated <= decision_date - 1 day.
truncated_bars: dict[str, dict[dt.date, dict[str, Any]]] = {}
for sym, sym_bars in setup["bars"].items():
truncated_bars[sym] = {
d: b for d, b in sym_bars.items()
if d < setup["decision_date"] # explicit -1 shift floor
}
cands_b = build_candidates(
decision_date=setup["decision_date"],
next_trading_date=setup["next_trading_date"],
universe_symbols=[setup["symbol"]],
engine=engine,
bar_provider=_SnapshotStoreBarAdapter(bars_by_symbol=truncated_bars),
pre_open_gap_provider=None,
)
assert len(cands_a) == len(cands_b)
if cands_a:
assert cands_a[0].entry_price_est == cands_b[0].entry_price_est
# The breakout-determining stats must agree.
assert cands_a[0].features["vol_breakout_52w_last_close"] == \
cands_b[0].features["vol_breakout_52w_last_close"]
assert cands_a[0].features["vol_breakout_52w_prior_252d_max_high"] == \
cands_b[0].features["vol_breakout_52w_prior_252d_max_high"]
# ---------------------------------------------------------------------------
# Bootstrap permutation test — the engine has no notion of label permutation;
# the test we CAN run is: shuffle decision_date assignments across symbols and
# assert that each symbol's trigger output is unchanged because each candidate
# is computed only from THAT symbol's bars (no cross-symbol leakage).
# ---------------------------------------------------------------------------
def test_bootstrap_permutation_per_symbol_independence():
"""Per-symbol independence: scrambling the order of universe_symbols must
not change the set of emitted candidates. If it does, there is hidden
cross-symbol state leaking into the trigger.
"""
days = _generate_business_days(dt.date(2026, 1, 5), 82)
decision_date = days[80]
next_trading_date = days[81]
prior_days = days[:80]
bars: dict[str, dict[dt.date, dict[str, Any]]] = {}
for sym in ("AAPL", "MSFT", "GOOG"):
bars.update(_build_bars(
sym, prior_days,
base_close=100.0, base_high=102.0, base_low=98.0,
base_volume=1_000_000.0,
last_close=120.0, # all break out
last_high=121.0, last_low=119.0,
last_volume=4_000_000.0,
))
bar_provider = _SnapshotStoreBarAdapter(bars_by_symbol=bars)
engine = _make_engine()
rng = random.Random(12345)
base_order = ["AAPL", "MSFT", "GOOG"]
base = build_candidates(
decision_date=decision_date,
next_trading_date=next_trading_date,
universe_symbols=base_order,
engine=engine,
bar_provider=bar_provider,
pre_open_gap_provider=None,
)
base_symbols = sorted(c.symbol for c in base)
assert base_symbols == ["AAPL", "GOOG", "MSFT"]
for _ in range(8):
order = list(base_order)
rng.shuffle(order)
shuffled = build_candidates(
decision_date=decision_date,
next_trading_date=next_trading_date,
universe_symbols=order,
engine=engine,
bar_provider=bar_provider,
pre_open_gap_provider=None,
)
assert sorted(c.symbol for c in shuffled) == base_symbols
def test_bootstrap_permutation_breaks_edge_when_signal_is_destroyed():
"""If we permute the LAST-bar values across symbols (so the breakout flag
no longer corresponds to the symbol's own history), a symbol's eligibility
must depend ONLY on its own bars. Permuting the universe order alone does
NOT change candidates — that's the test above. Here we instead verify that
forcing one symbol's last-close to a NON-breakout level removes ONLY that
symbol from the candidate list, leaving the others intact.
"""
days = _generate_business_days(dt.date(2026, 1, 5), 82)
decision_date = days[80]
next_trading_date = days[81]
prior_days = days[:80]
bars: dict[str, dict[dt.date, dict[str, Any]]] = {}
for sym, last_close in (("AAPL", 120.0), ("MSFT", 120.0), ("GOOG", 120.0)):
bars.update(_build_bars(
sym, prior_days,
base_close=100.0, base_high=102.0, base_low=98.0,
base_volume=1_000_000.0,
last_close=last_close,
last_high=last_close + 1.0, last_low=last_close - 1.0,
last_volume=4_000_000.0,
))
engine = _make_engine()
base = build_candidates(
decision_date=decision_date,
next_trading_date=next_trading_date,
universe_symbols=["AAPL", "MSFT", "GOOG"],
engine=engine,
bar_provider=_SnapshotStoreBarAdapter(bars_by_symbol=bars),
pre_open_gap_provider=None,
)
assert sorted(c.symbol for c in base) == ["AAPL", "GOOG", "MSFT"]
# Now: kill MSFT's breakout by lowering its last close BELOW prior 252d max high
# (base_high=102 plus jitter ~ 102.2). last_close=100 is clearly not a breakout.
bars2: dict[str, dict[dt.date, dict[str, Any]]] = {}
for sym, last_close in (("AAPL", 120.0), ("MSFT", 100.0), ("GOOG", 120.0)):
bars2.update(_build_bars(
sym, prior_days,
base_close=100.0, base_high=102.0, base_low=98.0,
base_volume=1_000_000.0,
last_close=last_close,
last_high=last_close + 1.0, last_low=last_close - 1.0,
last_volume=4_000_000.0,
))
after = build_candidates(
decision_date=decision_date,
next_trading_date=next_trading_date,
universe_symbols=["AAPL", "MSFT", "GOOG"],
engine=engine,
bar_provider=_SnapshotStoreBarAdapter(bars_by_symbol=bars2),
pre_open_gap_provider=None,
)
assert sorted(c.symbol for c in after) == ["AAPL", "GOOG"]
# ---------------------------------------------------------------------------
# Universe filter
# ---------------------------------------------------------------------------
def test_universe_filter_skips_low_adv_symbol():
days = _generate_business_days(dt.date(2026, 1, 5), 82)
decision_date = days[80]
next_trading_date = days[81]
prior_days = days[:80]
# Tiny ADV: low_volume * close = small.
low_adv = _build_bars("TINY", prior_days,
base_close=100.0, base_high=102.0, base_low=98.0,
base_volume=1000.0, # ~$100k ADV
last_close=120.0, last_high=121.0, last_low=119.0,
last_volume=4000.0)
engine = _make_engine(vol_breakout_52w_min_avg_dollar_volume=10_000_000.0)
cands = build_candidates(
decision_date=decision_date,
next_trading_date=next_trading_date,
universe_symbols=["TINY"],
engine=engine,
bar_provider=_SnapshotStoreBarAdapter(bars_by_symbol=low_adv),
pre_open_gap_provider=None,
)
assert cands == []
def test_universe_filter_skips_low_price_symbol():
days = _generate_business_days(dt.date(2026, 1, 5), 82)
decision_date = days[80]
next_trading_date = days[81]
prior_days = days[:80]
bars = _build_bars("PENNY", prior_days,
base_close=2.0, base_high=2.2, base_low=1.8,
base_volume=10_000_000.0,
last_close=3.0, # below $5 floor
last_high=3.1, last_low=2.9,
last_volume=40_000_000.0)
engine = _make_engine(vol_breakout_52w_min_price=5.0)
cands = build_candidates(
decision_date=decision_date,
next_trading_date=next_trading_date,
universe_symbols=["PENNY"],
engine=engine,
bar_provider=_SnapshotStoreBarAdapter(bars_by_symbol=bars),
pre_open_gap_provider=None,
)
assert cands == []
# ---------------------------------------------------------------------------
# Behavioral exit tests — drive a synthetic position through simulate_exit
# ---------------------------------------------------------------------------
def _build_position_for_breakout(
*,
entry_price: float = 100.0,
stop_pct: float = 0.03,
target_pct: float = 0.05,
days_held: int = 0,
):
from libs.backtest.domain import Candidate, OpenPosition, PlannedOrder
stop_mult = stop_pct / 0.02
target_r = target_pct / stop_pct
synthetic_atr = entry_price * 0.02
cand = Candidate(
event_id="evt_volb_exit",
symbol="AAPL",
score=0.7,
sector="UNKNOWN",
event_type=VOL_BREAKOUT_52W_EVENT_TYPE,
event_timestamp=dt.datetime(2026, 4, 10, 21, 0, tzinfo=dt.timezone.utc),
event_date=dt.date(2026, 4, 13),
filing_time_bucket="post_market",
reaction_date=dt.date(2026, 4, 13),
execution_date=dt.date(2026, 4, 14),
entry_price_est=entry_price,
avg_dollar_volume=200_000_000.0,
atr_14=synthetic_atr,
score_bucket="medium_high",
engine_id="vol_breakout_52w_long",
entry_timing_policy="next_open",
trade_direction="long",
engine_stop_atr_multiplier=stop_mult,
engine_target_1_r=target_r,
engine_target_1_fraction=1.0,
engine_max_holding_days=2,
)
stop_price = entry_price * (1.0 - stop_pct)
target_price = entry_price * (1.0 + target_pct)
plan = PlannedOrder(
candidate=cand,
shares=100,
entry_price_limit=entry_price,
stop_price=stop_price,
target_price=target_price,
risk_dollars=stop_pct * entry_price * 100,
event_date=cand.event_date,
timing_class="after_close",
engine_id=cand.engine_id,
entry_timing_policy="next_open",
shadow_only=False,
)
return OpenPosition(
position_id="pos_volb",
plan=plan,
entry_date=cand.execution_date,
entry_price=entry_price,
entry_fill_slippage_bps=10.0,
current_stop=stop_price,
target_price=target_price,
peak_price=entry_price,
shares_open=100,
shares_total=100,
days_held=days_held,
)
def _exec_config_for_exit_test(max_hold: int = 2):
from libs.backtest.domain import ExecutionConfig
return ExecutionConfig(
entry_fill_model="next_open",
exit_fill_model="daily_bar_approximation",
slippage_bps_base=10.0,
commission_per_share=0.005,
same_bar_priority="stop_first_conservative",
max_holding_days=max_hold,
)
def test_exit_stop_at_minus_3pct():
from libs.backtest.domain import ExitReason
from libs.backtest.execution import simulate_exit
pos = _build_position_for_breakout(entry_price=100.0, stop_pct=0.03)
bar = {"date": dt.date(2026, 4, 15), "open": 99.0, "high": 99.5, "low": 96.5,
"close": 97.0, "volume": 1_000_000}
trade = simulate_exit(pos, bar, _exec_config_for_exit_test(), dt.date(2026, 4, 15))
assert trade is not None
assert trade.exit_reason == ExitReason.STOP
def test_exit_target_at_plus_5pct():
from libs.backtest.domain import ExitReason
from libs.backtest.execution import simulate_exit
pos = _build_position_for_breakout(entry_price=100.0, target_pct=0.05)
bar = {"date": dt.date(2026, 4, 15), "open": 102.0, "high": 105.5, "low": 101.0,
"close": 104.0, "volume": 1_000_000}
trade = simulate_exit(pos, bar, _exec_config_for_exit_test(), dt.date(2026, 4, 15))
assert trade is not None
assert trade.exit_reason == ExitReason.TARGET
def test_exit_forced_max_hold_at_day_2_moc():
"""When days_held >= max_holding_days=2 and no stop/target hit, exit reason is TIME."""
from libs.backtest.domain import ExitReason
from libs.backtest.execution import simulate_exit
pos = _build_position_for_breakout(entry_price=100.0, days_held=2)
cfg = _exec_config_for_exit_test(max_hold=2)
bar = {"date": dt.date(2026, 4, 15), "open": 102.0, "high": 103.0, "low": 99.0,
"close": 102.5, "volume": 1_000_000}
trade = simulate_exit(pos, bar, cfg, dt.date(2026, 4, 15))
assert trade is not None
assert trade.exit_reason == ExitReason.TIME
def test_exit_intraday_priority_stop_before_target_when_both_touched():
"""Same-bar priority: stop_first_conservative — stop wins when both lines touched."""
from libs.backtest.domain import ExitReason
from libs.backtest.execution import simulate_exit
pos = _build_position_for_breakout(entry_price=100.0, stop_pct=0.03, target_pct=0.05)
# Wide bar that touches both 97.0 (stop) AND 105.0 (target).
bar = {"date": dt.date(2026, 4, 15), "open": 99.5, "high": 105.5, "low": 96.5,
"close": 100.0, "volume": 1_000_000}
trade = simulate_exit(pos, bar, _exec_config_for_exit_test(), dt.date(2026, 4, 15))
assert trade is not None
assert trade.exit_reason == ExitReason.STOP
Reference in New Issue View Git Blame Copy Permalink