fithia2/tests/unit/orb_trader/test_detection_guards.py

"""Unit tests for ORBTradingEngine guard conditions.

Covers early-exit paths in run_orb_detection (rolling_loss, circuit_breaker)
and the max_simultaneous_entries guard in run_breakout_check.
"""
from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest

from apps.orb_trader.engine import ORBTradingEngine

_DETECTION_PATCHES = (
    "apps.orb_trader.engine.load_universe",
    "apps.orb_trader.engine.enrich_daily_bars",
    "apps.orb_trader.engine.compute_orb_candidates",
)


def _make_engine(
    *,
    initial_equity: float = 10_000.0,
    rolling_loss_days: int | None = None,
    rolling_loss_threshold: float | None = None,
    max_simultaneous_entries: int | None = None,
) -> ORBTradingEngine:
    params = SimpleNamespace(
        daily_budget_reset=True,
        rolling_loss_days=rolling_loss_days,
        rolling_loss_threshold=rolling_loss_threshold,
        max_simultaneous_entries=max_simultaneous_entries,
        drawdown_governor_threshold=None,
        drawdown_governor_min_scale=0.30,
        streak_sizing_win_bonus=None,
        streak_sizing_loss_penalty=None,
        streak_sizing_max=2.5,
        streak_sizing_min=0.5,
        orb_minutes=5,
        market_regime_spy_threshold=None,
        min_candidate_breadth=None,
        atr_stop_multiplier=0.75,
        risk_per_trade_pct=0.05,
        max_position_pct=0.70,
    )
    session = SimpleNamespace(
        session_id="test-session",
        session_name="test",
        initial_equity=initial_equity,
    )
    state = MagicMock()
    state.get_equity.return_value = initial_equity
    state.get_peak_equity.return_value = initial_equity
    state.list_snapshots.return_value = []
    state.list_trades.return_value = []

    engine = object.__new__(ORBTradingEngine)
    engine._session = session
    engine._params = params
    engine._state = state
    engine._broker = MagicMock()
    engine._broker.get_bars.return_value = {}
    engine._broker.get_intraday_bars.return_value = {}
    engine._log_callback = None
    engine._enrichment = {}
    engine._daily_bars = {}
    engine._candidates = []
    engine._pending_cands = []
    engine._pre_screened_tickers = None
    engine._date_str = ""
    return engine


def _run_detection_no_pipeline(eng: ORBTradingEngine, date_str: str) -> dict:
    """Run run_orb_detection with heavy library calls patched out."""
    with patch("apps.orb_trader.engine.load_universe", return_value=[]), \
         patch("apps.orb_trader.engine.enrich_daily_bars", return_value={}), \
         patch("apps.orb_trader.engine.compute_orb_candidates", return_value=[]):
        return eng.run_orb_detection(date_str)


# ── Rolling loss filter ───────────────────────────────────────────────────────

class TestRollingLossFilter:
    def _snapshot(self, date: str, daily_pnl: float) -> dict:
        return {"date": date, "daily_pnl": daily_pnl}

    def test_not_enough_history_no_skip(self):
        # Only 2 snapshots but roll_days=3 → filter does not trigger
        eng = _make_engine(rolling_loss_days=3, rolling_loss_threshold=-0.05)
        eng._state.list_snapshots.return_value = [
            self._snapshot("2026-01-01", -300.0),
            self._snapshot("2026-01-02", -300.0),
        ]
        result = _run_detection_no_pipeline(eng, "2026-01-05")
        assert result.get("skip_reason") != "rolling_loss"

    def test_loss_below_threshold_triggers_skip(self):
        # 3 days of -200 = -600 total; -600/10000 = -6% < -5% threshold → skip
        eng = _make_engine(rolling_loss_days=3, rolling_loss_threshold=-0.05)
        eng._state.list_snapshots.return_value = [
            self._snapshot("2026-01-01", -200.0),
            self._snapshot("2026-01-02", -200.0),
            self._snapshot("2026-01-03", -200.0),
        ]
        result = eng.run_orb_detection("2026-01-05")
        assert result["skip_reason"] == "rolling_loss"
        assert result["orb_candidates"] == 0

    def test_loss_exactly_at_threshold_no_skip(self):
        # -100 + -200 + -200 = -500; -500/10000 = -5.0% = threshold → NOT below → no skip
        eng = _make_engine(rolling_loss_days=3, rolling_loss_threshold=-0.05)
        eng._state.list_snapshots.return_value = [
            self._snapshot("2026-01-01", -100.0),
            self._snapshot("2026-01-02", -200.0),
            self._snapshot("2026-01-03", -200.0),
        ]
        result = _run_detection_no_pipeline(eng, "2026-01-05")
        assert result.get("skip_reason") != "rolling_loss"

    def test_future_snapshots_excluded_from_window(self):
        # Snapshot for date_str itself must not be counted (filter: date < date_str)
        # 2 past snapshots < roll_days=3 → no skip
        eng = _make_engine(rolling_loss_days=3, rolling_loss_threshold=-0.05)
        eng._state.list_snapshots.return_value = [
            self._snapshot("2026-01-01", -400.0),
            self._snapshot("2026-01-02", -400.0),
            self._snapshot("2026-01-05", -400.0),  # same-day: must not count
        ]
        result = _run_detection_no_pipeline(eng, "2026-01-05")
        assert result.get("skip_reason") != "rolling_loss"

    def test_rolling_window_uses_last_n_days(self):
        # 5 past snapshots, roll_days=3 → only last 3 used
        # Last 3 are +100 each → sum=+300 → no trigger
        eng = _make_engine(rolling_loss_days=3, rolling_loss_threshold=-0.05)
        eng._state.list_snapshots.return_value = [
            self._snapshot("2025-12-30", -400.0),
            self._snapshot("2025-12-31", -400.0),
            self._snapshot("2026-01-01", 100.0),
            self._snapshot("2026-01-02", 100.0),
            self._snapshot("2026-01-03", 100.0),
        ]
        result = _run_detection_no_pipeline(eng, "2026-01-05")
        assert result.get("skip_reason") != "rolling_loss"


# ── Circuit breaker ───────────────────────────────────────────────────────────

class TestCircuitBreaker:
    def test_triggers_at_25pct_drawdown(self):
        # Peak=10000, current=7400 → DD=26% ≥ 25% → halt
        eng = _make_engine(initial_equity=10_000)
        eng._state.get_equity.return_value = 7_400.0
        eng._state.get_peak_equity.return_value = 10_000.0

        result = eng.run_orb_detection("2026-01-05")
        assert result["skip_reason"] == "circuit_breaker"
        eng._state.set_session_status.assert_called_once_with("test-session", "paused")

    def test_does_not_trigger_below_threshold(self):
        # Peak=10000, current=7600 → DD=24% < 25% → continues
        eng = _make_engine(initial_equity=10_000)
        eng._state.get_equity.return_value = 7_600.0
        eng._state.get_peak_equity.return_value = 10_000.0

        result = _run_detection_no_pipeline(eng, "2026-01-05")
        assert result.get("skip_reason") != "circuit_breaker"
        eng._state.set_session_status.assert_not_called()


# ── max_simultaneous_entries guard ───────────────────────────────────────────

def _make_cand(ticker: str, price: float = 100.0) -> dict:
    return {
        "ticker": ticker,
        "direction": "bullish",
        "orb_bar": {"high": price, "low": price * 0.98, "open": price * 0.99, "close": price},
        "atr": 1.0,
        "score": 0.8,
        "rvol": 2.0,
        "gap_pct": 0.03,
    }


def _mock_position(ticker: str) -> MagicMock:
    pos = MagicMock()
    pos.ticker = ticker
    return pos


class TestMaxSimultaneousEntries:
    def _make_breakout_engine(self, *, max_sim: int | None, open_positions: list) -> ORBTradingEngine:
        eng = _make_engine(max_simultaneous_entries=max_sim)
        daily_state = MagicMock()
        daily_state.kill_switch = False
        eng._state.get_daily_state.return_value = daily_state
        eng._state.get_open_positions.return_value = open_positions
        return eng

    def test_at_max_candidate_stays_pending(self):
        # 3 positions open, max_sim=3 → new candidate must NOT enter
        eng = self._make_breakout_engine(
            max_sim=3,
            open_positions=[_mock_position("A"), _mock_position("B"), _mock_position("C")],
        )
        eng._pending_cands = [_make_cand("AAPL", price=150.0)]

        snap = MagicMock()
        snap.price = 155.0  # would trigger breakout if guard not active
        with patch("apps.orb_trader.engine.get_snapshots", return_value={"AAPL": snap}):
            result = eng.run_breakout_check("2026-01-05")

        assert result["filled"] == 0
        assert result["remaining"] == 1

    def test_below_max_proceeds_past_guard(self):
        # 2 positions open, max_sim=3 → not blocked by guard; snapshot check runs
        eng = self._make_breakout_engine(
            max_sim=3,
            open_positions=[_mock_position("A"), _mock_position("B")],
        )
        eng._pending_cands = [_make_cand("AAPL", price=150.0)]

        # Return no snapshot → candidate stays pending for snapshot reason (not max_sim)
        with patch("apps.orb_trader.engine.get_snapshots", return_value={}) as mock_snaps:
            result = eng.run_breakout_check("2026-01-05")

        mock_snaps.assert_called_once()  # proceeded past max_sim guard
        assert result["remaining"] == 1   # pending, but due to missing snapshot

    def test_no_limit_proceeds_past_guard(self):
        # max_simultaneous_entries=None → no position cap enforced
        eng = self._make_breakout_engine(
            max_sim=None,
            open_positions=[
                _mock_position("A"), _mock_position("B"),
                _mock_position("C"), _mock_position("D"),
            ],
        )
        eng._pending_cands = [_make_cand("AAPL", price=150.0)]

        with patch("apps.orb_trader.engine.get_snapshots", return_value={}) as mock_snaps:
            result = eng.run_breakout_check("2026-01-05")

        mock_snaps.assert_called_once()
        assert result["remaining"] == 1