"""Replay bounds for a symbol carrying a research terminal mark. ``_replay_ticker`` normally stops ``HORIZON`` bars before the end so every setup has a full resolution window. For a delisted name that is exactly wrong: the collapse happens in those final bars, so the signals that would have been caught by it are the ones the bound hides. ``terminal_date`` opts a single symbol into walking through its last real bar. The fixtures build a controlled setup rather than replaying a real ticker, so the tests do not depend on whether SIVB happened to qualify on a given day. """ from __future__ import annotations from datetime import date, timedelta from types import SimpleNamespace import pytest from app.services import backtest_service as bt def _bar(day: date, close: float, *, high: float | None = None, low: float | None = None): return SimpleNamespace( date=day, open=close, high=close * 1.01 if high is None else high, low=close * 0.99 if low is None else low, close=close, volume=1_000_000, ) def _series(n: int, *, start: date = date(2021, 1, 4)) -> list: """A steadily rising series long enough to clear MIN_LOOKBACK.""" bars = [] day = start for i in range(n): while day.weekday() >= 5: day += timedelta(days=1) bars.append(_bar(day, 100.0 + i * 0.25)) day += timedelta(days=1) return bars def _terminal_bar(previous: date) -> SimpleNamespace: day = previous + timedelta(days=1) while day.weekday() >= 5: day += timedelta(days=1) return SimpleNamespace( date=day, open=0.01, high=0.01, low=0.01, close=0.01, volume=0 ) @pytest.fixture() def replay(monkeypatch): """Force one long setup on every evaluated bar so signal dates are countable.""" def _fake_setups(window, config, activation, *, target_model): close = float(window[-1].close) return [{ "direction": "long", "entry": close, "stop": close * 0.95, # Reachable within HORIZON on the rising fixture, but not within the # four bars left before a terminal event. "target": close * 1.02, "rr": 2.0, "confidence": 80, "momentum": 0.1, "best_prob": 0.5, "primary_prob": 0.5, "meets_core": True, "action": "BUY", "risk_level": "medium", "target_model": target_model, "primary_sources": [], "primary_strength": 1.0, "primary_rejection_count": 0, "primary_last_rejection_age": None, "primary_distance_atr": 1.0, "raw_level_count": 1, "gate_level_count": 1, }] monkeypatch.setattr(bt, "_window_setups", _fake_setups) def _run(records, terminal_date=None): return bt._replay_ticker( "TEST", records, {}, {}, None, bt.PRODUCTION_GTL_TARGET_MODEL, "daily", terminal_date, ) return _run def test_setup_near_a_terminal_event_is_emitted_and_resolves_at_the_mark(replay): """A setup five sessions before the halt is replayed, and if it is still open at the terminal bar the near-zero price resolves it as a loss.""" real = _series(bt.MIN_LOOKBACK + 40) records = real + [_terminal_bar(real[-1].date)] without = replay(records) with_mark = replay(records, terminal_date=records[-1].date) last_real = real[-1].date fifth_last = real[-5].date without_dates = {c["date"] for c in without} with_dates = {c["date"] for c in with_mark} assert fifth_last.isoformat() not in without_dates, ( "baseline should hide signals inside the final HORIZON bars" ) assert fifth_last.isoformat() in with_dates assert last_real.isoformat() in with_dates late = next(c for c in with_mark if c["date"] == fifth_last.isoformat()) assert late["outcome"] == bt.OUTCOME_STOP_HIT assert late["realized_r"] < -1.0, "a gap to 0.01 must fill below the stop, not at it" def test_no_setup_is_generated_on_the_terminal_bar(replay): real = _series(bt.MIN_LOOKBACK + 40) terminal = _terminal_bar(real[-1].date) records = real + [terminal] candidates = replay(records, terminal_date=terminal.date) assert terminal.date.isoformat() not in {c["date"] for c in candidates} assert max(c["date"] for c in candidates) == real[-1].date.isoformat() def test_a_position_that_exits_before_the_event_keeps_its_own_exit(replay): """An early setup resolves on its own bars; the terminal mark must not retroactively turn it into a total loss.""" real = _series(bt.MIN_LOOKBACK + 40) records = real + [_terminal_bar(real[-1].date)] with_mark = replay(records, terminal_date=records[-1].date) early_date = real[bt.MIN_LOOKBACK - 1].date.isoformat() early = next(c for c in with_mark if c["date"] == early_date) assert early["outcome"] == bt.OUTCOME_TARGET_HIT assert early["realized_r"] == pytest.approx(2.0) def test_unmapped_symbols_and_runs_without_the_option_are_unchanged(replay): """The ordinary replay bounds are untouched when no terminal date is given.""" records = _series(bt.MIN_LOOKBACK + 40) baseline = replay(records) again = replay(records, terminal_date=None) assert [c["date"] for c in baseline] == [c["date"] for c in again] assert max(c["date"] for c in baseline) < records[-1].date.isoformat(), ( "without a terminal mark the walk must still stop HORIZON bars early" ) def test_a_mismatched_terminal_date_is_rejected(replay): """The date is validated against the final bar rather than trusted, so a stale events file fails loudly instead of shifting the replay window.""" records = _series(bt.MIN_LOOKBACK + 40) with pytest.raises(ValueError, match="does not match the final bar"): replay(records, terminal_date=date(1999, 1, 1))