"""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", ) class TestLiveParamParity: def test_constructor_does_not_disable_strategy_volume_filters(self): params = SimpleNamespace( settlement_days=2, slippage_bps=5.0, min_rvol=3.0, min_premarket_dollar_vol=2_500_000.0, ) session = SimpleNamespace( session_id="test-session", session_name="test", initial_equity=10_000.0, ) eng = ORBTradingEngine( session=session, broker=MagicMock(), state=MagicMock(), params=params, ) assert eng._params is not params assert eng._params.min_rvol == pytest.approx(3.0) assert eng._params.min_premarket_dollar_vol == pytest.approx(2_500_000.0) assert eng._params.settlement_days == 0 assert eng._params.slippage_bps == pytest.approx(0.0) assert params.settlement_days == 2 assert params.slippage_bps == pytest.approx(5.0) 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) class TestCandidateSizeScale: def test_isolated_downside_profiles_apply_live_size_scale(self): eng = _make_engine() params = eng._params params.broad_gapup_continuation_enabled = False params.isolated_downside_loss_cap_min_abs_gap_pct = 0.02 params.isolated_downside_loss_cap_max_ret_5d = 10.0 params.isolated_downside_loss_cap_min_premarket_dollar_vol = 20_000_000 params.isolated_downside_loss_cap_max_premarket_dollar_vol = None params.isolated_downside_loss_cap_min_body_ratio = 0.20 params.isolated_downside_loss_cap_min_close_location = 0.50 params.isolated_downside_loss_cap_min_orb_return = 0.0 params.isolated_downside_loss_cap_max_body_ratio = None params.isolated_downside_loss_cap_max_close_location = None params.isolated_downside_loss_cap_max_score_rank_pct = None params.isolated_downside_loss_cap_allowed_trigger_types = ["orb"] params.isolated_downside_size_scale = 0.35 params.isolated_downside_pressure_min_abs_gap_pct = 0.02 params.isolated_downside_pressure_max_ret_5d = 10.0 params.isolated_downside_pressure_min_premarket_dollar_vol = 300_000_000 params.isolated_downside_pressure_max_body_ratio = 0.05 params.isolated_downside_pressure_max_close_location = 0.25 params.isolated_downside_pressure_allowed_trigger_types = ["orb"] params.isolated_downside_pressure_size_scale = 0.35 thrust_cand = { "ticker": "DJT", "direction": "bullish", "gap_pct": -0.062, "sector_confirmation_active": False, "ret_5d": -0.09, "premarket_dollar_vol": 270_000_000, "body_ratio": 0.62, "close_location": 0.78, "orb_return": 0.039, "trigger_type": "orb", } pressure_cand = { "ticker": "FUTU", "direction": "bullish", "gap_pct": -0.186, "sector_confirmation_active": False, "ret_5d": 0.20, "premarket_dollar_vol": 315_000_000, "body_ratio": 0.0, "close_location": 0.20, "orb_return": -0.005, "trigger_type": "orb", } assert eng._candidate_size_scale(thrust_cand) == pytest.approx(0.35) assert eng._candidate_size_scale(pressure_cand) == pytest.approx(0.35) def test_persisted_candidate_size_scale_is_reused(self): eng = _make_engine() assert eng._candidate_size_scale({"size_scale": 0.35}) == pytest.approx(0.35) def test_overextended_downside_reclaim_applies_live_size_scale(self): eng = _make_engine() params = eng._params params.broad_gapup_continuation_enabled = False params.overextended_downside_reclaim_min_abs_gap_pct = 0.15 params.overextended_downside_reclaim_min_ret_5d = 0.15 params.overextended_downside_reclaim_min_premarket_dollar_vol = None params.overextended_downside_reclaim_allowed_trigger_types = ["orb"] params.overextended_downside_reclaim_size_scale = 0.25 cand = { "ticker": "BULL", "direction": "bullish", "gap_pct": -0.205, "ret_5d": 4.59, "premarket_dollar_vol": 34_000_000, "sector_confirmation_active": True, "trigger_type": "orb", } assert eng._candidate_size_scale(cand) == pytest.approx(0.25) def test_mid_attention_exhaustion_applies_live_size_scale(self): eng = _make_engine() params = eng._params params.broad_gapup_continuation_enabled = False params.mid_attention_exhaustion_min_rvol = 15.0 params.mid_attention_exhaustion_max_rvol = 25.0 params.mid_attention_exhaustion_min_premarket_dollar_vol = 20_000_000 params.mid_attention_exhaustion_max_premarket_dollar_vol = 100_000_000 params.mid_attention_exhaustion_allowed_trigger_types = ["orb"] params.mid_attention_exhaustion_size_scale = 0.25 cand = { "ticker": "OSCR", "direction": "bullish", "gap_pct": -0.134, "rvol": 21.2, "premarket_dollar_vol": 97_000_000, "trigger_type": "orb", } assert eng._candidate_size_scale(cand) == pytest.approx(0.25) def test_orphan_thin_attention_applies_live_size_scale(self): eng = _make_engine() params = eng._params params.broad_gapup_continuation_enabled = False params.orphan_thin_attention_max_premarket_dollar_vol = 2_500_000 params.orphan_thin_attention_allowed_trigger_types = ["orb"] params.orphan_thin_attention_size_scale = 0.25 cand = { "ticker": "THIN", "direction": "bullish", "gap_pct": 0.028, "premarket_dollar_vol": 900_000, "sector_confirmation_active": False, "trigger_type": "orb", } assert eng._candidate_size_scale(cand) == pytest.approx(0.25) def test_orphan_thin_attention_ignores_sector_confirmed_candidates(self): eng = _make_engine() params = eng._params params.broad_gapup_continuation_enabled = False params.orphan_thin_attention_max_premarket_dollar_vol = 2_500_000 params.orphan_thin_attention_allowed_trigger_types = ["orb"] params.orphan_thin_attention_size_scale = 0.25 cand = { "ticker": "GROUP", "direction": "bullish", "gap_pct": 0.028, "premarket_dollar_vol": 900_000, "sector_confirmation_active": True, "trigger_type": "orb", } assert eng._candidate_size_scale(cand) == pytest.approx(1.0) def test_orphan_thin_attention_uses_filter_premarket_for_iex_live_parity(self): eng = _make_engine() params = eng._params params.broad_gapup_continuation_enabled = False params.orphan_thin_attention_max_premarket_dollar_vol = 2_500_000 params.orphan_thin_attention_allowed_trigger_types = ["orb"] params.orphan_thin_attention_size_scale = 0.25 cand = { "ticker": "IEXLIQ", "direction": "bullish", "gap_pct": 0.028, "premarket_dollar_vol": 900_000, "filter_premarket_dollar_vol": 27_000_000, "sector_confirmation_active": False, "trigger_type": "orb", } assert eng._candidate_size_scale(cand) == pytest.approx(1.0) def test_mid_attention_exhaustion_uses_filter_volume_for_iex_live_parity(self): eng = _make_engine() params = eng._params params.broad_gapup_continuation_enabled = False params.mid_attention_exhaustion_min_rvol = 15.0 params.mid_attention_exhaustion_max_rvol = 25.0 params.mid_attention_exhaustion_min_premarket_dollar_vol = 20_000_000 params.mid_attention_exhaustion_max_premarket_dollar_vol = 100_000_000 params.mid_attention_exhaustion_allowed_trigger_types = ["orb"] params.mid_attention_exhaustion_size_scale = 0.25 cand = { "ticker": "IEXMID", "direction": "bullish", "gap_pct": -0.134, "rvol": 0.7, "filter_rvol": 21.2, "premarket_dollar_vol": 3_000_000, "filter_premarket_dollar_vol": 90_000_000, "trigger_type": "orb", } assert eng._candidate_size_scale(cand) == pytest.approx(0.25) def test_unsupported_thin_attention_uses_filter_first_bar_for_iex_live_parity(self): eng = _make_engine() params = eng._params params.broad_gapup_continuation_enabled = False params.unsupported_attention_allowed_trigger_types = ["orb"] params.unsupported_attention_size_scale = 0.25 params.unsupported_attention_thin_max_premarket_dollar_vol = 10_000_000 params.unsupported_attention_thin_min_gap_pct = 0.02 params.unsupported_attention_thin_min_rvol = 15.0 params.unsupported_attention_thin_max_first_bar_dollar_vol = 1_000_000 params.unsupported_attention_ignore_high_conviction = True cand = { "ticker": "IEXFIRST", "direction": "bullish", "gap_pct": 0.028, "rvol": 20.0, "premarket_dollar_vol": 900_000, "first_bar_dollar_vol": 50_000, "filter_first_bar_dollar_vol": 1_500_000, "sector_confirmation_active": False, "trigger_type": "orb", } assert eng._candidate_size_scale(cand) == pytest.approx(1.0) def test_gap_up_fill_trap_applies_live_size_scale(self): eng = _make_engine() params = eng._params params.broad_gapup_continuation_enabled = False params.gap_up_fill_exit_min_gap_pct = 0.02 params.gap_up_fill_exit_max_premarket_dollar_vol = 10_000_000 params.gap_up_fill_exit_min_ret_5d = 0.0 params.gap_up_fill_exit_max_ret_5d = 0.25 params.gap_up_fill_exit_max_close_location = 0.85 params.gap_up_fill_trap_max_orb_return = 0.035 params.gap_up_fill_trap_allowed_trigger_types = ["orb"] params.gap_up_fill_trap_size_scale = 0.0 cand = { "ticker": "TRAP", "direction": "bullish", "gap_pct": 0.028, "ret_5d": 0.04, "premarket_dollar_vol": 4_600_000, "body_ratio": 0.46, "close_location": 0.60, "orb_return": 0.014, "trigger_type": "orb", } assert eng._candidate_size_scale(cand) == pytest.approx(0.0) def test_gap_up_fill_trap_keeps_exceptional_first_bar_expansion(self): eng = _make_engine() params = eng._params params.broad_gapup_continuation_enabled = False params.gap_up_fill_exit_min_gap_pct = 0.02 params.gap_up_fill_exit_max_premarket_dollar_vol = 10_000_000 params.gap_up_fill_exit_min_ret_5d = 0.0 params.gap_up_fill_exit_max_ret_5d = 0.25 params.gap_up_fill_exit_max_close_location = 0.85 params.gap_up_fill_trap_max_orb_return = 0.035 params.gap_up_fill_trap_allowed_trigger_types = ["orb"] params.gap_up_fill_trap_size_scale = 0.0 cand = { "ticker": "LEADER", "direction": "bullish", "gap_pct": 0.028, "ret_5d": 0.04, "premarket_dollar_vol": 4_600_000, "body_ratio": 0.65, "close_location": 0.72, "orb_return": 0.067, "trigger_type": "orb", } assert eng._candidate_size_scale(cand) == pytest.approx(1.0) # ── 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 _make_bar(open_price: float = 100.0) -> dict: return { "timestamp": "2026-05-01T09:30:00-04:00", "open": open_price, "high": open_price * 1.01, "low": open_price * 0.99, "close": open_price, "volume": 100_000, } def _make_bar_at(timestamp: str, open_price: float = 100.0) -> dict: return { "timestamp": timestamp, "open": open_price, "high": open_price * 1.01, "low": open_price * 0.99, "close": open_price, "volume": 100_000, } def _mock_position(ticker: str) -> MagicMock: pos = MagicMock() pos.ticker = ticker return pos class TestLiveRegimeFallback: def test_prescreen_fetches_regime_ticker_and_soft_fallback_scales_day(self): eng = _make_engine() eng._params.market_regime_spy_threshold = 0.0015 eng._params.market_regime_ticker = "QQQ" eng._params.soft_day_fallback_on_regime_skip = True eng._params.soft_day_regime_skip_size_scale = 0.1 eng._params.soft_day_combined_size_scale_floor = 0.35 eng._params.min_candidate_breadth = None eng._pre_screened_tickers = ["ABC"] eng._daily_bars = {"QQQ": [{}], "ABC": [{}]} eng._enrichment = { "QQQ": {"2026-05-01": {"prev_close": 100.0, "today_open": 100.0}}, "ABC": {"2026-05-01": {"prev_close": 10.0, "today_open": 10.0}}, } eng._broker.get_intraday_bars.return_value = { "QQQ": [_make_bar(100.0)], "ABC": [_make_bar(11.0)], } with patch( "apps.orb_trader.engine.compute_orb_candidates", return_value=[_make_cand("QQQ"), _make_cand("ABC")], ): result = eng.run_orb_detection("2026-05-01") fetched = eng._broker.get_intraday_bars.call_args.args[0] assert fetched[0] == "QQQ" assert "ABC" in fetched assert result["orb_candidates"] == 1 assert result.get("skip_reason") is None assert eng._soft_day_reason == "market_regime" assert eng._day_size_scale == pytest.approx(0.35) eng._state.save_candidate.assert_called_once() assert eng._state.save_candidate.call_args.args[0].ticker == "ABC" def test_live_regime_uses_regular_session_open_not_premarket_tick(self): eng = _make_engine() eng._params.market_regime_spy_threshold = 0.0015 eng._params.market_regime_ticker = "QQQ" eng._params.soft_day_fallback_on_regime_skip = False eng._params.min_candidate_breadth = None eng._pre_screened_tickers = ["ABC"] eng._daily_bars = {"QQQ": [{}], "ABC": [{}]} eng._enrichment = { "QQQ": {"2026-05-01": {"prev_close": 100.0, "today_open": 100.0}}, "ABC": {"2026-05-01": {"prev_close": 10.0, "today_open": 10.0}}, } eng._broker.get_intraday_bars.return_value = { "QQQ": [ _make_bar_at("2026-05-01T08:15:00-04:00", 100.0), _make_bar_at("2026-05-01T09:30:00-04:00", 101.0), ], "ABC": [ _make_bar_at("2026-05-01T08:15:00-04:00", 10.0), _make_bar_at("2026-05-01T09:30:00-04:00", 11.0), ], } with patch( "apps.orb_trader.engine.compute_orb_candidates", return_value=[_make_cand("ABC")], ) as compute_mock: result = eng.run_orb_detection("2026-05-01") assert result.get("skip_reason") is None assert eng._enrichment["QQQ"]["2026-05-01"]["today_open"] == pytest.approx(101.0) assert eng._day_size_scale == pytest.approx(1.0) assert compute_mock.call_args.kwargs["bars_by_ticker"]["QQQ"][0]["timestamp"] == ( "2026-05-01T08:15:00-04:00" ) eng._state.save_candidate.assert_called_once() class TestLiveMarketThrustParity: def test_market_thrust_sleeve_requires_override_active(self): eng = _make_engine() eng._params.min_candidate_breadth = 0.90 eng._params.soft_day_fallback_on_breadth_skip = True eng._params.soft_day_breadth_skip_size_scale = 0.10 eng._params.market_thrust_liquid_continuation_enabled = True eng._params.market_thrust_breadth_override_enabled = True eng._params.market_thrust_breadth_override_min_primary_close_location = 0.95 eng._params.market_orb_quality_ticker = "SPY" eng._params.market_orb_quality_secondary_ticker = "QQQ" eng._params.broad_gapup_continuation_enabled = False eng._params.market_thrust_opening_impulse_reclaim_enabled = False eng._pre_screened_tickers = ["ABC"] eng._daily_bars = {"SPY": [{}], "QQQ": [{}], "ABC": [{}]} eng._enrichment = { "SPY": {"2026-05-01": {"prev_close": 100.0, "today_open": 100.0}}, "QQQ": {"2026-05-01": {"prev_close": 100.0, "today_open": 100.0}}, "ABC": {"2026-05-01": {"prev_close": 10.0, "today_open": 10.0}}, } eng._broker.get_intraday_bars.return_value = { "SPY": [_make_bar(100.0)], "QQQ": [_make_bar(100.0)], "ABC": [_make_bar(11.0)], } with patch( "apps.orb_trader.engine.compute_orb_candidates", return_value=[], ) as compute_mock: result = eng.run_orb_detection("2026-05-01") assert result["orb_candidates"] == 0 assert eng._market_thrust_breadth_override_active is False assert compute_mock.call_count == 1 params = compute_mock.call_args.kwargs["params"] assert getattr(params, "market_thrust_liquid_continuation_enabled") is False 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 # ── EOD DB sweep ────────────────────────────────────────────────────────────── class TestEodDbSweep: def test_sweep_runs_with_no_open_positions(self): # Regression: before fix, DB sweep was inside the "has positions" branch; # after a server restart with no open positions, stale pending records leaked. eng = _make_engine() eng._state.get_open_positions.return_value = [] eng._state.list_candidates.return_value = [ {"ticker": "AAPL", "status": "pending"}, {"ticker": "NVDA", "status": "pending"}, {"ticker": "TSLA", "status": "filled"}, # already filled, must not be touched ] result = eng.run_eod_exit("2026-01-05") assert result["closed"] == 0 # Both pending records should be timed out calls = [ call.args for call in eng._state.update_candidate_status.call_args_list ] assert ("test-session", "2026-01-05", "AAPL", "timeout") in calls assert ("test-session", "2026-01-05", "NVDA", "timeout") in calls # Filled record must not be touched assert ("test-session", "2026-01-05", "TSLA", "timeout") not in calls def test_in_memory_pending_swept_with_no_open_positions(self): # In-memory pending candidates must also be swept unconditionally eng = _make_engine() eng._state.get_open_positions.return_value = [] eng._state.list_candidates.return_value = [] eng._pending_cands = [ {"ticker": "MSFT", "direction": "bullish", "orb_bar": {}, "atr": 1.0, "score": 0.5, "rvol": 1.5, "gap_pct": 0.02}, ] eng.run_eod_exit("2026-01-05") assert eng._pending_cands == [] # cleared eng._state.update_candidate_status.assert_called_once_with( "test-session", "2026-01-05", "MSFT", "timeout" )