diff --git a/tests/unit/orb_trader/test_detection_guards.py b/tests/unit/orb_trader/test_detection_guards.py new file mode 100644 index 0000000..b3b4404 --- /dev/null +++ b/tests/unit/orb_trader/test_detection_guards.py @@ -0,0 +1,248 @@ +"""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