""" test_entry_shortlist_starvation.py Regression test for the shortlist-starvation bug found in the 2026-09-01 session review: monitor.py's _scan_for_entries() ranks premarket_20 by health_score and only evaluate_entry()-checks the top entry_shortlist_size symbols each poll cycle. Real log evidence that day showed F occupying a shortlist slot continuously from 09:32 to 09:56 (~24 minutes) failing entry_engine's confirmation on the same unchanging "momentum fading" hard disqualifier every single time, while CRML -- health-eligible and scoring competitively -- was rotated into premarket_20 at 10:01 but never once reached evaluate_entry() before rotating back out at the next rescan, because F kept winning the health-score ranking and occupying the only available shortlist slot. _EntryAttemptTracker (see monitor.py) fixes this by benching a symbol after max_consecutive_confirmation_failures straight not-confirmed results, freeing its shortlist slot for the next-best candidate. This test covers both the tracker in isolation and the full wiring through _scan_for_entries(). Run: python -m pytest tests/test_entry_shortlist_starvation.py -v or: python tests/test_entry_shortlist_starvation.py """ import sys import os import unittest from unittest.mock import patch, MagicMock from datetime import datetime, timedelta, timezone sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import monitor # noqa: E402 from fast_pipeline import FastPipelineDecision # noqa: E402 from fast_prediction_engine import FastPredictionReading # noqa: E402 from stream_features import StreamFeatures # noqa: E402 class TestEntryAttemptTrackerUnit(unittest.TestCase): """Pure unit tests for the tracker, no monitor.py wiring involved.""" def setUp(self): self.now = datetime(2026, 9, 1, 14, 0, tzinfo=timezone.utc) self.tracker = monitor._EntryAttemptTracker(max_failures=3, cooldown_seconds=180) def test_not_benched_before_threshold(self): self.tracker.record_result("F", confirmed=False, now=self.now) self.tracker.record_result("F", confirmed=False, now=self.now) self.assertFalse(self.tracker.is_benched("F", self.now)) def test_benched_after_max_consecutive_failures(self): for _ in range(3): self.tracker.record_result("F", confirmed=False, now=self.now) self.assertTrue(self.tracker.is_benched("F", self.now)) def test_bench_clears_after_cooldown_elapses(self): for _ in range(3): self.tracker.record_result("F", confirmed=False, now=self.now) later = self.now + timedelta(seconds=181) self.assertFalse(self.tracker.is_benched("F", later)) def test_still_benched_before_cooldown_elapses(self): for _ in range(3): self.tracker.record_result("F", confirmed=False, now=self.now) soon = self.now + timedelta(seconds=60) self.assertTrue(self.tracker.is_benched("F", soon)) def test_a_confirmed_result_resets_the_failure_streak(self): self.tracker.record_result("F", confirmed=False, now=self.now) self.tracker.record_result("F", confirmed=False, now=self.now) self.tracker.record_result("F", confirmed=True, now=self.now) self.tracker.record_result("F", confirmed=False, now=self.now) self.tracker.record_result("F", confirmed=False, now=self.now) # Only 2 consecutive failures since the reset -- not benched yet. self.assertFalse(self.tracker.is_benched("F", self.now)) def test_a_confirmed_result_lifts_an_existing_bench(self): for _ in range(3): self.tracker.record_result("F", confirmed=False, now=self.now) self.assertTrue(self.tracker.is_benched("F", self.now)) self.tracker.record_result("F", confirmed=True, now=self.now) self.assertFalse(self.tracker.is_benched("F", self.now)) def test_symbols_are_tracked_independently(self): for _ in range(3): self.tracker.record_result("F", confirmed=False, now=self.now) self.assertTrue(self.tracker.is_benched("F", self.now)) self.assertFalse(self.tracker.is_benched("CRML", self.now)) class TestScanForEntriesSkipsBenchedSymbols(unittest.TestCase): """ End-to-end through _scan_for_entries(): with entry_shortlist_size=1 and F always outranking CRML on fast_prediction confidence (the ranking key since [FEATURE 2026-09-11 v2] -- "the confidence value should be the value used for selecting the top_5") but never confirming, F should monopolize the only shortlist slot for max_consecutive_confirmation_failures cycles -- then get benched, finally letting CRML (which DOES confirm) get evaluated and enter. """ def setUp(self): self.orch = monitor.SessionOrchestrator.__new__(monitor.SessionOrchestrator) self.orch.cfg = { "intraday_health": { "entry_shortlist_size": 1, "max_consecutive_confirmation_failures": 3, "confirmation_failure_cooldown_seconds": 180, }, "universe": {"min_avg_daily_volume": 500000}, "trading": {"simulated_equity_default": 100000.0}, "streaming": {"sub_minute_bucket_seconds": 30}, } self.orch._entry_tracker = monitor._EntryAttemptTracker( max_failures=3, cooldown_seconds=180) self.orch._fast_engine_state = {} self.orch.simulation = True self.orch.client = None self.orch._opening_volume_baseline = {"F": 1000.0, "CRML": 1000.0} self.orch.premarket_20 = [ {"symbol": "F", "total_score": 75.73}, {"symbol": "CRML", "total_score": 70.22}, ] self.entered = [] class FakePositionManager: def is_symbol_open(_self, symbol): return False def has_available_slot(_self): return True def has_closed_position_today(_self, symbol): return False def enter_position(_self, symbol, entry_price, equity, bars, reason): self.entered.append(symbol) class FakeStream: def is_symbol_stale(_self, symbol): return False def get_bars(_self, symbol): now = datetime.now(timezone.utc) return [{"t": now - timedelta(minutes=i), "o": 1, "h": 1, "l": 1, "c": 1, "v": 100} for i in range(5)] def get_bars_sub(_self, symbol): return [] def get_quote(_self, symbol): return None self.orch.position_mgr = FakePositionManager() self.orch.stream = FakeStream() # health_score is no longer the ranking key ([FEATURE 2026-09-11 v2]) # -- both WATCH here just to pass the eligibility filter. F still # always outranks CRML, but now on fast_prediction confidence (set # up in fake_compute_fast_ranking below). self.health_state = { "F": {"state": "WATCH", "health_score": 90.0}, "CRML": {"state": "WATCH", "health_score": 70.0}, } def _run_one_cycle(self, evaluate_entry_calls): def fake_compute_fast_ranking(symbol, bars, bars_sub, quote, baseline, state=None, sub_bucket_seconds=30, as_of=None, premarket_result=None): # F always outranks CRML on confidence, matching the real # 2026-09-01 case (F was the top-ranked pool member). confidence = 90.0 if symbol == "F" else 70.0 prediction = FastPredictionReading(symbol=symbol, confidence=confidence) features = StreamFeatures(symbol=symbol, computed_at=datetime.now(timezone.utc)) return prediction, features def fake_evaluate_fast_entry_only(symbol, prediction, features, state=None, as_of=None): evaluate_entry_calls.append(symbol) # F never confirms (mirrors real 09-01 "momentum fading" hard # disqualifier that didn't change bar to bar); CRML does. return FastPipelineDecision(symbol=symbol, should_enter=(symbol == "CRML"), confirmation_score=0.0) with patch.object(monitor, "data_store") as mock_ds, \ patch.object(monitor, "intraday_health") as mock_health, \ patch.object(monitor, "compute_fast_ranking", side_effect=fake_compute_fast_ranking), \ patch.object(monitor, "evaluate_fast_entry_only", side_effect=fake_evaluate_fast_entry_only): mock_ds.load_health_state.return_value = self.health_state mock_health.is_eligible_for_entry.return_value = True mock_health.compute_health.return_value = MagicMock(raw_state="WATCH") self.orch._scan_for_entries() def test_f_monopolizes_shortlist_until_benched_then_crml_gets_a_turn(self): calls = [] for _ in range(3): self._run_one_cycle(calls) self.assertEqual(calls, ["F", "F", "F"], "F should occupy the only shortlist slot on every " "cycle before it's been benched -- CRML never gets " "a turn while F still outranks it on health_score") self.assertEqual(self.entered, [], "F never confirms, so nothing should have entered yet") self.assertTrue(self.orch._entry_tracker.is_benched("F", datetime.now(timezone.utc))) # 4th cycle: F is now benched, so CRML should finally be evaluated # and (since it confirms) entered. self._run_one_cycle(calls) self.assertEqual(calls, ["F", "F", "F", "CRML"]) self.assertEqual(self.entered, ["CRML"], "once F is benched, the next-best health-eligible " "candidate should get evaluated and, since it " "confirms, entered -- this is the actual fix for " "the CRML-starvation bug") if __name__ == "__main__": unittest.main()