"""Unit tests for ORBTradingEngine.run_stop_check stop management logic. Tests the stop evaluation loop that mirrors orb_simulator.py:477-580. Key behaviors verified: stop_loss hit, breakeven promotion, trailing activation, and trailing_tighten_at_r (tight ATR multiplier at 2R). """ from __future__ import annotations from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest from apps.orb_trader.engine import ORBTradingEngine from apps.orb_trader.models import ORBPositionRow _DATE = "2026-01-05" _ET_OFFSET = "-05:00" # January is EST def _ts(time_str: str) -> str: """Return a market-hours timestamp for the test date in ET ISO format.""" return f"{_DATE}T{time_str}{_ET_OFFSET}" def _bar(*, high: float, low: float, time_str: str = "09:35:00") -> dict: return { "timestamp": _ts(time_str), "open": (high + low) / 2, "high": high, "low": low, "close": (high + low) / 2, "volume": 1000, } def _make_position( *, ticker: str = "AAPL", direction: str = "long", entry_price: float = 100.0, shares: int = 10, stop_distance: float = 2.0, current_stop: float | None = None, trailing_active: bool = False, ) -> ORBPositionRow: if current_stop is None: current_stop = (entry_price - stop_distance) if direction == "long" else (entry_price + stop_distance) return ORBPositionRow( session_id="test-session", date=_DATE, ticker=ticker, direction=direction, entry_price=entry_price, entry_time=_ts("09:32:00"), # before all test bars shares=shares, orb_high=entry_price + 1, orb_low=entry_price - 1, atr_at_entry=stop_distance / 0.75, stop_distance=stop_distance, current_stop=current_stop, peak_price=entry_price, trailing_active=trailing_active, rvol=2.0, composite_score=0.7, order_id="order-1", ) def _make_engine(pos: ORBPositionRow, bars: list[dict]) -> ORBTradingEngine: params = SimpleNamespace( sim_bar_minutes=5, daily_budget_reset=True, 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, # Stop params (V23 values) breakeven_at_r=1.0, trailing_at_r=1.0, trailing_stop_atr_multiplier=0.8, trailing_tighten_at_r=2.0, trailing_stop_atr_multiplier_tight=0.3, # Kill switches daily_max_loss_pct=0.05, max_stops_per_day=5, ) session = SimpleNamespace( session_id="test-session", session_name="test", initial_equity=10_000.0, ) state = MagicMock() state.get_open_positions.return_value = [pos] state.get_equity.return_value = 10_000.0 state.get_peak_equity.return_value = 10_000.0 state.list_trades.return_value = [] daily_state_mock = MagicMock() daily_state_mock.kill_switch = False daily_state_mock.cumulative_loss = 0.0 daily_state_mock.stops_hit = 0 state.get_daily_state.return_value = daily_state_mock broker = MagicMock() broker.get_intraday_bars.return_value = {pos.ticker: bars} close_order = MagicMock() close_order.id = "close-order-1" broker.close_position.return_value = close_order fill_order = MagicMock() fill_order.filled_avg_price = pos.current_stop # fill at stop price fill_order.status = "filled" broker.get_order.return_value = fill_order engine = object.__new__(ORBTradingEngine) engine._session = session engine._params = params engine._state = state engine._broker = broker engine._log_callback = None engine._date_str = _DATE return engine # ── Stop loss ───────────────────────────────────────────────────────────────── class TestStopLossHit: @patch("apps.orb_trader.engine.time") def test_long_stop_hit_below_stop(self, mock_time): # bar_low=96.5 < stop=98.0 → stop_loss triggered pos = _make_position(entry_price=100.0, stop_distance=2.0, current_stop=98.0) bars = [_bar(high=100.5, low=96.5, time_str="09:40:00")] eng = _make_engine(pos, bars) result = eng.run_stop_check(_DATE) assert result["stops_hit"] == 1 eng._broker.close_position.assert_called_once_with("AAPL", qty=10) trade = eng._state.save_trade.call_args[0][0] assert trade.exit_reason == "stop_loss" @patch("apps.orb_trader.engine.time") def test_long_no_stop_hit_above_stop(self, mock_time): # bar_high=101.5 → R=0.75 < breakeven_at_r=1.0 → stop stays at 98.0 # bar_low=99.0 > 98.0 → no stop hit pos = _make_position(entry_price=100.0, stop_distance=2.0, current_stop=98.0) bars = [_bar(high=101.5, low=99.0, time_str="09:40:00")] eng = _make_engine(pos, bars) result = eng.run_stop_check(_DATE) assert result["stops_hit"] == 0 eng._broker.close_position.assert_not_called() @patch("apps.orb_trader.engine.time") def test_short_stop_hit_above_stop(self, mock_time): # Short: bar_high >= stop → stop triggered pos = _make_position( direction="short", entry_price=100.0, stop_distance=2.0, current_stop=102.0, ) bars = [_bar(high=103.0, low=99.0, time_str="09:40:00")] eng = _make_engine(pos, bars) result = eng.run_stop_check(_DATE) assert result["stops_hit"] == 1 trade = eng._state.save_trade.call_args[0][0] assert trade.exit_reason == "stop_loss" # ── Breakeven promotion ─────────────────────────────────────────────────────── class TestBreakevenPromotion: @patch("apps.orb_trader.engine.time") def test_stop_moves_to_entry_at_1r(self, mock_time): # entry=100, stop_distance=2, breakeven_at_r=1.0 # bar_high=102.5 → current_r = (102.5-100)/2 = 1.25 ≥ 1.0 → stop moves to 100 pos = _make_position(entry_price=100.0, stop_distance=2.0, current_stop=98.0) bars = [ _bar(high=102.5, low=101.0, time_str="09:40:00"), _bar(high=102.0, low=100.5, time_str="09:45:00"), # second bar: no stop hit ] eng = _make_engine(pos, bars) eng.run_stop_check(_DATE) # DB should be updated with stop >= entry price (100.0) upd_calls = eng._state.update_position_stop.call_args_list assert len(upd_calls) == 1 updated_stop = upd_calls[0].args[3] # positional: session_id, date, ticker, stop, ... assert updated_stop >= 100.0 # ── Trailing tighten at 2R ──────────────────────────────────────────────────── class TestTrailingTighten: @patch("apps.orb_trader.engine.time") def test_tighten_uses_tight_multiplier_at_2r(self, mock_time): # entry=100, stop_distance=2, atr_at_entry=2/0.75≈2.667 # trailing_at_r=1.0, tighten_at_r=2.0, tight_mult=0.3, normal_mult=0.8 # Bar 1: high=104.1 → R=(104.1-100)/2=2.05 ≥ 2.0 → should use tight_mult=0.3 # Expected trail = peak - atr*0.3 = 104.1 - 2.667*0.3 ≈ 103.3 pos = _make_position( entry_price=100.0, stop_distance=2.0, current_stop=98.0, trailing_active=True, # already trailing ) bars = [_bar(high=104.1, low=101.0, time_str="09:40:00")] eng = _make_engine(pos, bars) # Pre-warm peak_price to 104.1 (so the tighten kicks in) pos = eng._state.get_open_positions.return_value[0] pos.peak_price = 104.1 pos.current_stop = 100.0 # at breakeven eng.run_stop_check(_DATE) eng._broker.close_position.assert_not_called() upd_calls = eng._state.update_position_stop.call_args_list assert upd_calls updated_stop = upd_calls[-1].args[3] atr = 2.0 / 0.75 expected_tight = 104.1 - atr * 0.3 expected_normal = 104.1 - atr * 0.8 # Tight stop should be higher (tighter) than normal stop assert updated_stop > expected_normal - 0.01 assert updated_stop == pytest.approx(expected_tight, abs=0.05)