Add run_stop_check unit tests covering stop_loss, breakeven, and tighten
Tests the stop evaluation loop that mirrors orb_simulator.py:477-580: - Long/short stop_loss hit - Breakeven promotion (stop moves to entry at 1R) - trailing_tighten_at_r: verifies tight ATR multiplier fires at 2R vs normal multiplier, exercising the tighten logic added in V23 port Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>main
parent
d4daf7a951
commit
c5fa1394ed
@ -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)
|
||||||
Loading…
Reference in New Issue