signal-platform/tests/unit/test_backtest_terminal_events.py

"""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))