signal-platform/tests/unit/test_daily_reentry_matrix.py

from datetime import date, timedelta

import pytest

from scripts.run_daily_reentry_matrix import (
    PrecomputedDailyEngine,
    ReentryPolicy,
    _live_universe_rank_map,
)


RANKING_KEY = "strategy_rank"
ORD = date(2025, 1, 6).toordinal()


def _candidate(day_ord: int, *, stop: float = 91.0, rank: float = 81.0) -> dict:
    return {
        "qualified": True,
        "direction": "long",
        "symbol": "AAA",
        "date": date.fromordinal(day_ord).isoformat(),
        "entry": 100.0,
        "stop": stop,
        "target": 120.0,
        RANKING_KEY: rank,
    }


def _state(sessions: int = 0) -> dict:
    return {
        "sessions_since_stop": sessions,
        "previous_stop": 90.0,
        "previous_rank": 80.0,
        "gate_went_unqualified": False,
    }


def _call(policy: ReentryPolicy, day_ord: int, state: dict, sessions: int):
    state["sessions_since_stop"] = sessions
    return policy("AAA", day_ord, state, object())


def test_next_session_blocks_only_stop_day():
    engine = PrecomputedDailyEngine([_candidate(ORD), _candidate(ORD + 1)])
    policy = ReentryPolicy("next_session", engine, RANKING_KEY)
    state = _state()

    assert _call(policy, ORD, state, 0) is None
    assert _call(policy, ORD + 1, state, 1) is not None


@pytest.mark.parametrize("sessions", [2, 3, 5])
def test_cooldown_unlocks_at_exact_boundary(sessions):
    engine = PrecomputedDailyEngine([
        _candidate(ORD + sessions - 1),
        _candidate(ORD + sessions),
    ])
    policy = ReentryPolicy(f"cooldown_{sessions}", engine, RANKING_KEY)
    state = _state()

    assert _call(policy, ORD + sessions - 1, state, sessions - 1) is None
    assert _call(policy, ORD + sessions, state, sessions) is not None


def test_gate_reset_requires_failure_before_requalification():
    engine = PrecomputedDailyEngine([_candidate(ORD), _candidate(ORD + 2)])
    policy = ReentryPolicy("gate_reset", engine, RANKING_KEY)
    state = _state()

    assert _call(policy, ORD, state, 0) is None
    assert _call(policy, ORD + 1, state, 1) is None
    emitted = _call(policy, ORD + 2, state, 2)
    assert emitted is not None
    assert emitted["_reentry_reason"] == "gate_failed_then_requalified"


def test_strict_gate_reset_ignores_stop_day_failure():
    engine = PrecomputedDailyEngine([
        _candidate(ORD + 1),
        _candidate(ORD + 3),
    ])
    policy = ReentryPolicy("strict_gate_reset", engine, RANKING_KEY)
    state = _state()

    # An unqualified stop-day close alone does not reset the strict policy.
    assert _call(policy, ORD, state, 0) is None
    assert state["gate_went_unqualified"] is False
    assert _call(policy, ORD + 1, state, 1) is None

    # A later unqualified close establishes the reset; only then may the next
    # qualified setup re-enter.
    assert _call(policy, ORD + 2, state, 2) is None
    assert state["gate_went_unqualified"] is True
    emitted = _call(policy, ORD + 3, state, 3)
    assert emitted is not None
    assert emitted["_reentry_reason"] == (
        "post_stop_gate_failed_then_requalified"
    )


def test_improved_gate_reset_requires_better_stop_and_non_weaker_rank():
    engine = PrecomputedDailyEngine([
        _candidate(ORD + 1, stop=89.0, rank=82.0),
        _candidate(ORD + 2, stop=92.0, rank=79.0),
        _candidate(ORD + 3, stop=92.0, rank=81.0),
    ])
    policy = ReentryPolicy("gate_reset_improved", engine, RANKING_KEY)
    state = _state()

    assert _call(policy, ORD, state, 0) is None
    assert _call(policy, ORD + 1, state, 1) is None
    assert _call(policy, ORD + 2, state, 2) is None
    emitted = _call(policy, ORD + 3, state, 3)
    assert emitted is not None
    assert emitted["_reentry_reason"] == "gate_reset_with_improved_stop_and_rank"


def test_two_session_confirmation_excludes_stop_day_close():
    engine = PrecomputedDailyEngine([
        _candidate(ORD),
        _candidate(ORD + 1),
        _candidate(ORD + 2),
    ])
    policy = ReentryPolicy("two_session_confirmation", engine, RANKING_KEY)
    state = _state()

    assert _call(policy, ORD, state, 0) is None
    assert _call(policy, ORD + 1, state, 1) is None
    emitted = _call(policy, ORD + 2, state, 2)
    assert emitted is not None
    assert emitted["_reentry_reason"] == "two_qualified_post_stop_closes"


def _rank_observation(
    symbol: str,
    *,
    raw: float,
    residual: float,
    volatility: float,
) -> dict:
    return {
        "symbol": symbol,
        "date": date.fromordinal(ORD).isoformat(),
        "ranking_period": ("date", ORD),
        "momentum": raw,
        "residual_momentum": residual,
        "vol_6m": volatility,
    }


def test_live_universe_rank_uses_each_ticker_once_and_residual_when_available():
    observations = [
        _rank_observation("AAA", raw=0.1, residual=0.3, volatility=0.1),
        _rank_observation("BBB", raw=0.3, residual=0.1, volatility=0.2),
        _rank_observation("CCC", raw=0.2, residual=0.2, volatility=0.3),
    ]
    first_benchmark_day = date.fromordinal(ORD) - timedelta(days=300)
    benchmark = {
        first_benchmark_day + timedelta(days=offset): 100.0
        for offset in range(252)
    }

    ranks = _live_universe_rank_map(observations, benchmark, 0.8)

    assert ranks[("AAA", date.fromordinal(ORD).isoformat())] == {
        "momentum_percentile": 100.0,
        "volatility_percentile": 0.0,
        "strategy_rank": 80.0,
    }
    assert ranks[("BBB", date.fromordinal(ORD).isoformat())][
        "momentum_percentile"
    ] == 0.0
    assert ranks[("CCC", date.fromordinal(ORD).isoformat())][
        "strategy_rank"
    ] == 60.0


def test_live_universe_rank_uses_raw_fallback_before_benchmark_is_ready():
    observations = [
        _rank_observation("AAA", raw=0.1, residual=0.3, volatility=0.1),
        _rank_observation("BBB", raw=0.3, residual=0.1, volatility=0.2),
    ]

    ranks = _live_universe_rank_map(observations, {}, 0.8)

    assert ranks[("AAA", date.fromordinal(ORD).isoformat())][
        "momentum_percentile"
    ] == 0.0
    assert ranks[("BBB", date.fromordinal(ORD).isoformat())][
        "momentum_percentile"
    ] == 100.0


def test_live_universe_rank_rejects_duplicate_ticker_date():
    observation = _rank_observation(
        "AAA", raw=0.1, residual=0.2, volatility=0.1
    )

    with pytest.raises(ValueError, match="one observation"):
        _live_universe_rank_map([observation, dict(observation)], {}, 0.8)