""" test_resistance_freeze_bugfix.py Regression test for the "frozen resistance" bug found in production logs on 2026-08-19: CLSK entered 5 times between 11:27 and 13:24 citing the identical "breakout of resistance $11.70" every time, despite trading as high as $12.20 in between. Root cause: monitor.py's _run_intraday_full_rescan() re-inserted a symbol's OLD dict from self.premarket_20 (with a stale pm_high/resistance) into the new pool whenever that symbol had an open position but fell outside the top full_rescan_pool_size health cut on that pass -- instead of using the symbol's freshly-rescored dict (which the same rescan pass had already computed, with a current pm_high). This test simulates two consecutive full rescans where the held symbol's health score dips low enough to fall out of the top-N cut both times, while its price genuinely makes new highs between rescans, and asserts that self.premarket_20's entry for that symbol reflects the NEW price high after the fix (it did NOT, before the fix). Run: python -m pytest tests/test_resistance_freeze_bugfix.py -v or: python tests/test_resistance_freeze_bugfix.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 def _scored(symbol, pm_high, total_score, health_score=None, health_state="WATCH"): """Build a minimal scored-candidate dict shaped like scorer.score_premarket_candidate()'s output plus the fields premarket_scanner.scan()/monitor.py bolt on.""" return { "symbol": symbol, "total_score": total_score, "pm_high": pm_high, "pm_low": pm_high - 1.0, "health_score": health_score if health_score is not None else total_score, "health_state": health_state, "flags": {"meets_min_volume": True, "spread_ok": True}, } class FakePositionManager: def __init__(self, open_symbols): self._open = list(open_symbols) def get_open_symbols(self): return list(self._open) class FakeStream: def add_symbols(self, symbols): pass def get_bars(self, symbol): # Enough bars to satisfy the >=3 check inside the rescan loop. 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) ] class TestResistanceFreezeBugfix(unittest.TestCase): def setUp(self): self.orch = monitor.SessionOrchestrator.__new__(monitor.SessionOrchestrator) self.orch.cfg = { "intraday_health": { "full_rescan_pool_size": 2, # deliberately tiny so CLSK gets cut "full_rescan_lookback_hours": 1, "full_rescan_interval_minutes": 30, }, "universe": {"min_avg_daily_volume": 500000}, } self.orch._prefiltered_universe = ["CLSK", "AAA", "BBB"] self.orch.position_mgr = FakePositionManager(open_symbols=["CLSK"]) self.orch.stream = FakeStream() self.orch._sync_opening_volume_baseline = lambda pool: None # CLSK currently sits in premarket_20 with an OLD, stale pm_high -- # this stands in for "the value set at some earlier rescan before # CLSK's price ran higher." self.orch.premarket_20 = [ _scored("CLSK", pm_high=11.70, total_score=40, health_score=40, health_state="WATCH"), _scored("AAA", pm_high=20.00, total_score=90, health_score=90, health_state="HEALTHY"), _scored("BBB", pm_high=30.00, total_score=85, health_score=85, health_state="HEALTHY"), ] def _run_rescan_with(self, clsk_fresh_pm_high, clsk_health_score): """Simulate one full rescan pass where the whole universe gets rescored fresh (CLSK's price has moved to clsk_fresh_pm_high), but CLSK's health_score is deliberately kept low so it does NOT make the top-2 (full_rescan_pool_size) cut -- forcing the 'keep open-position symbol' fallback path to fire.""" rescored_universe = [ _scored("CLSK", pm_high=clsk_fresh_pm_high, total_score=clsk_health_score), _scored("AAA", pm_high=20.50, total_score=91), _scored("BBB", pm_high=30.50, total_score=86), ] with patch.object(monitor, "premarket_scanner") as mock_pms, \ patch.object(monitor, "intraday_health") as mock_health, \ patch.object(monitor, "data_store") as mock_ds: mock_pms.scan.return_value = rescored_universe # Health engine: CLSK reads WATCH but with the deliberately low # score so it loses the top-2 cut to AAA/BBB every time. Still # eligible for entry (WATCH is eligible) -- the point of this # bug was never about eligibility, only about whether pm_high # gets refreshed. def fake_evaluate_symbol(symbol, bars, avg_vol_baseline, health_state): reading = MagicMock() reading.health_score = clsk_health_score if symbol == "CLSK" else 95 reading.confirmed_state = "WATCH" return reading, {"state": "WATCH"} mock_health.evaluate_symbol.side_effect = fake_evaluate_symbol mock_health.is_eligible_for_entry.return_value = True mock_ds.load_health_state.return_value = {} mock_ds.save_health_state.return_value = None mock_ds.load_watchlist.return_value = {} mock_ds.save_watchlist.return_value = None mock_ds.write_top_stocks.return_value = None self.orch._run_intraday_full_rescan() def test_open_position_symbol_gets_fresh_pm_high_after_fix(self): # First rescan: CLSK's price has already moved up to 12.20, but its # health score (35) keeps it out of the top-2 cut (AAA=91, BBB=86). self._run_rescan_with(clsk_fresh_pm_high=12.20, clsk_health_score=35) clsk_entry = next(r for r in self.orch.premarket_20 if r["symbol"] == "CLSK") self.assertEqual( clsk_entry["pm_high"], 12.20, "CLSK's pm_high/resistance should reflect this rescan's fresh price " "high (12.20), not the old stale value (11.70) -- this is exactly " "the bug: resistance froze at $11.70 across 5 real CLSK entries " "spanning 11:27-13:24 today despite price trading up to $12.20." ) # Second rescan, later: price has moved again to 12.35, health score # still keeps CLSK out of the cut. This proves the fix isn't a # one-time refresh -- it keeps tracking fresh data on every pass, # not just re-freezing at whatever value it got the first time it # fell out of the cut. self._run_rescan_with(clsk_fresh_pm_high=12.35, clsk_health_score=38) clsk_entry = next(r for r in self.orch.premarket_20 if r["symbol"] == "CLSK") self.assertEqual( clsk_entry["pm_high"], 12.35, "pm_high should keep tracking fresh price action on subsequent " "rescans too, not just refresh once and then re-freeze." ) def test_slot_freed_upserts_stale_entry(self): """_handle_slot_freed should replace (not skip) an existing stale entry for a symbol chosen as a fresh replacement candidate.""" self.orch.stream = FakeStream() fresh_candidate = _scored("CLSK", pm_high=12.50, total_score=70, health_score=70, health_state="HEALTHY") with patch.object(monitor, "top_stocks") as mock_top_stocks: mock_top_stocks.find_replacement.return_value = fresh_candidate self.orch._handle_slot_freed("SOME_EXITED_SYMBOL") clsk_entries = [r for r in self.orch.premarket_20 if r["symbol"] == "CLSK"] self.assertEqual(len(clsk_entries), 1, "should not end up with duplicate CLSK entries") self.assertEqual( clsk_entries[0]["pm_high"], 12.50, "the fresh replacement candidate's pm_high should overwrite the " "stale entry, not be silently discarded because CLSK was already " "present in premarket_20." ) if __name__ == "__main__": unittest.main()