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