"""Unit tests for live ORB VWAP-reclaim polling.""" from __future__ import annotations import datetime as dt import json from types import SimpleNamespace from unittest.mock import MagicMock, patch from zoneinfo import ZoneInfo from apps.orb_trader.engine import ORBTradingEngine _DATE = "2026-01-05" _ET = ZoneInfo("America/New_York") def _bar(time_str: str, *, open_: float, high: float, low: float, close: float, volume: int = 1_000) -> dict: return { "timestamp": f"{_DATE}T{time_str}-05:00", "open": open_, "high": high, "low": low, "close": close, "volume": volume, } def _make_engine() -> ORBTradingEngine: params = SimpleNamespace( sim_bar_minutes=5, orb_minutes=5, order_timeout_minutes=25, nofill_vwap_reclaim_enabled=True, nofill_vwap_reclaim_min_score_pct=0.8, nofill_vwap_reclaim_max_trades=1, nofill_vwap_reclaim_size_scale=0.04, soft_day_vwap_reclaim_enabled=False, vwap_reclaim_window_start_min=30, vwap_reclaim_window_end_min=150, vwap_reclaim_require_prior_dip=True, vwap_reclaim_min_clearance_pct=0.0, vwap_reclaim_require_orb_open_retake=False, vwap_reclaim_confirm_rel_vol=None, 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, max_trades_per_day=3, max_simultaneous_entries=3, atr_stop_multiplier=0.75, risk_per_trade_pct=0.05, max_position_pct=0.70, broad_gapup_continuation_enabled=False, ) session = SimpleNamespace( session_id="test-session", session_name="test", initial_equity=10_000.0, ) state = MagicMock() state.get_open_positions.return_value = [] state.list_trades.return_value = [] state.get_equity.return_value = 10_000.0 state.get_peak_equity.return_value = 10_000.0 state.get_daily_state.return_value = SimpleNamespace(kill_switch=False) broker = MagicMock() broker.get_intraday_bars.return_value = { "RECL": [ _bar("09:30:00", open_=100.0, high=101.0, low=99.8, close=100.5), _bar("09:35:00", open_=100.5, high=100.6, low=98.8, close=99.0), _bar("09:40:00", open_=99.0, high=99.8, low=98.7, close=99.5), _bar("10:00:00", open_=99.8, high=102.0, low=99.7, close=101.5), ] } broker.get_account.return_value = SimpleNamespace(buying_power=100_000.0) order = SimpleNamespace(id="buy-1") broker.submit_market_buy.return_value = order broker.get_order.return_value = SimpleNamespace(status="filled", filled_avg_price=103.0) eng = object.__new__(ORBTradingEngine) eng._session = session eng._params = params eng._state = state eng._broker = broker eng._log_callback = None eng._date_str = _DATE eng._day_size_scale = 1.0 eng._soft_day_reason = None eng._market_orb_quality_max_trades = None eng._market_orb_quality_reason = None eng._market_thrust_breadth_override_active = False eng._market_thrust_index_breadth_override_active = False eng._market_thrust_opening_breadth_override_active = False eng._candidates = [ { "ticker": "RECL", "direction": "bullish", "orb_bar": broker.get_intraday_bars.return_value["RECL"][0], "atr": 2.0, "rvol": 5.0, "gap_pct": 0.03, "score": 0.95, "body_ratio": 0.5, "close_location": 0.8, "premarket_dollar_vol": 5_000_000, } ] eng._pending_cands = list(eng._candidates) eng._now_et = lambda: dt.datetime(2026, 1, 5, 10, 5, tzinfo=_ET) # type: ignore[method-assign] return eng @patch("apps.orb_trader.engine.time") @patch("apps.orb_trader.engine.get_snapshots") def test_stop_check_polls_live_nofill_vwap_reclaim(mock_snapshots, mock_time): mock_snapshots.return_value = {"RECL": SimpleNamespace(price=103.0)} eng = _make_engine() result = eng.run_stop_check(_DATE) assert result["reclaim_filled"] == 1 eng._broker.submit_market_buy.assert_called_once() pos = eng._state.save_position.call_args.args[0] assert pos.ticker == "RECL" assert pos.trigger_type == "vwap_reclaim" assert pos.shares > 0 eng._state.update_candidate_status.assert_called_with( "test-session", _DATE, "RECL", "filled" ) @patch("apps.orb_trader.engine.time") @patch("apps.orb_trader.engine.get_snapshots") def test_stop_check_rebuilds_reclaim_candidate_from_persisted_metadata( mock_snapshots, mock_time, ): mock_snapshots.return_value = {"RECL": SimpleNamespace(price=103.0)} eng = _make_engine() saved_cand = dict(eng._candidates[0]) metadata = { "orb_bar": saved_cand["orb_bar"], "body_ratio": saved_cand["body_ratio"], "close_location": saved_cand["close_location"], "premarket_dollar_vol": saved_cand["premarket_dollar_vol"], "live_day_context": { "day_size_scale": 1.0, "soft_day_reason": None, "market_orb_quality_max_trades": None, "market_orb_quality_reason": None, "market_thrust_breadth_override_active": False, "market_thrust_index_breadth_override_active": False, "market_thrust_opening_breadth_override_active": False, }, } eng._state.list_candidates.return_value = [ { "status": "pending", "ticker": "RECL", "direction": "bullish", "orb_high": saved_cand["orb_bar"]["high"], "orb_low": saved_cand["orb_bar"]["low"], "atr": saved_cand["atr"], "rvol": saved_cand["rvol"], "gap_pct": saved_cand["gap_pct"], "composite_score": saved_cand["score"], "size_scale": 1.0, "metadata_json": json.dumps(metadata), } ] eng._candidates = [] eng._pending_cands = [] result = eng.run_stop_check(_DATE) assert result["reclaim_filled"] == 1 assert eng._state.list_candidates.called assert eng._candidates[0]["orb_bar"]["timestamp"] == "2026-01-05T09:30:00-05:00" eng._broker.submit_market_buy.assert_called_once() @patch("apps.orb_trader.engine.time") @patch("apps.orb_trader.engine.get_snapshots") def test_stop_check_polls_market_thrust_impulse_late_breakout( mock_snapshots, mock_time, ): mock_snapshots.return_value = {"RECL": SimpleNamespace(price=103.0)} eng = _make_engine() eng._params.nofill_vwap_reclaim_enabled = False eng._params.soft_day_vwap_reclaim_enabled = False eng._params.market_thrust_opening_impulse_reclaim_enabled = True eng._params.market_thrust_opening_impulse_reclaim_entry_mode = "late_breakout" eng._params.market_thrust_opening_impulse_reclaim_min_score_pct = 0.0 eng._params.market_thrust_opening_impulse_reclaim_max_trades = 1 eng._params.market_thrust_opening_impulse_reclaim_size_scale = 0.10 eng._params.market_thrust_opening_impulse_reclaim_only_when_no_primary_trades = False eng._params.market_thrust_opening_impulse_reclaim_no_thrust_max_trades = None eng._params.market_thrust_opening_impulse_reclaim_no_thrust_size_scale = None eng._params.late_breakout_window_start_min = 30 eng._params.late_breakout_window_end_min = 150 eng._params.late_breakout_min_clearance_pct = 0.0 eng._params.late_breakout_confirm_rel_vol = None eng._params.late_breakout_require_vwap_confirmation = False eng._candidates[0]["market_thrust_opening_impulse_reclaim"] = True eng._pending_cands = list(eng._candidates) result = eng.run_stop_check(_DATE) assert result["reclaim_filled"] == 1 eng._broker.submit_market_buy.assert_called_once() pos = eng._state.save_position.call_args.args[0] assert pos.ticker == "RECL" assert pos.trigger_type == "market_thrust_opening_impulse_reclaim" eng._state.update_candidate_status.assert_called_with( "test-session", _DATE, "RECL", "filled" ) @patch("apps.orb_trader.engine.time") @patch("apps.orb_trader.engine.get_snapshots") def test_market_thrust_impulse_no_primary_allows_auxiliary_prior_trade( mock_snapshots, mock_time, ): mock_snapshots.return_value = {"RECL": SimpleNamespace(price=103.0)} eng = _make_engine() eng._state.list_trades.return_value = [ { "date": _DATE, "ticker": "AUX", "trigger_type": "vwap_reclaim", "pnl": 10.0, "exit_reason": "close", } ] eng._params.nofill_vwap_reclaim_enabled = False eng._params.soft_day_vwap_reclaim_enabled = False eng._params.market_thrust_opening_impulse_reclaim_enabled = True eng._params.market_thrust_opening_impulse_reclaim_entry_mode = "late_breakout" eng._params.market_thrust_opening_impulse_reclaim_min_score_pct = 0.0 eng._params.market_thrust_opening_impulse_reclaim_max_trades = 1 eng._params.market_thrust_opening_impulse_reclaim_size_scale = 0.10 eng._params.market_thrust_opening_impulse_reclaim_only_when_no_primary_trades = True eng._params.market_thrust_opening_impulse_reclaim_no_thrust_max_trades = None eng._params.market_thrust_opening_impulse_reclaim_no_thrust_size_scale = None eng._params.late_breakout_window_start_min = 30 eng._params.late_breakout_window_end_min = 150 eng._params.late_breakout_min_clearance_pct = 0.0 eng._params.late_breakout_confirm_rel_vol = None eng._params.late_breakout_require_vwap_confirmation = False eng._candidates[0]["market_thrust_opening_impulse_reclaim"] = True eng._pending_cands = list(eng._candidates) result = eng.run_stop_check(_DATE) assert result["reclaim_filled"] == 1 pos = eng._state.save_position.call_args.args[0] assert pos.trigger_type == "market_thrust_opening_impulse_reclaim"