""" test_health_deterioration_exit.py [FEATURE 2026-09-11] Regression/coverage for the new force-exit path added after TJGC (2026-09-11) sat open for 3+ hours with a degrading, never-recovering STALE health score while price stayed inside a tight range and never touched the trailing stop -- the pre-existing fade-confirmation path only ever runs AT a stop touch (see position_manager._check_fade_confirmed_exit), so a position that just goes quiet and stale without ever threatening the stop had no exit path at all. Covers: 1. intraday_health.update_health_state()'s new consecutive_deteriorating counter (increments on STALE/UNHEALTHY raw reads, resets on HEALTHY/WATCH). 2. intraday_health.should_force_exit_on_deterioration() threshold / disable-flag behavior. 3. monitor.py's _update_intraday_health() actually triggers position_mgr.exit_position(..., "HEALTH_DETERIORATION") once the threshold is hit for an OPEN-position symbol, and leaves non-open-position symbols and below-threshold symbols alone. Run: python -m pytest tests/test_health_deterioration_exit.py -v """ import sys import os import unittest from unittest.mock import MagicMock, patch from datetime import datetime, timedelta, timezone sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import intraday_health # noqa: E402 import monitor # noqa: E402 def _reading(raw_state): return intraday_health.HealthReading( symbol="TJGC", health_score=50.0, raw_state=raw_state, confirmed_state=raw_state, price_slope_pct=0.0, vwap_slope_pct=0.0, momentum_slope_pct=0.0, price_slope_class="flat", vwap_slope_class="flat", momentum_slope_class="flat", ) class TestConsecutiveDeterioratingCounter(unittest.TestCase): def test_increments_on_stale_reads(self): persisted = {} for _ in range(3): persisted = intraday_health.update_health_state(_reading("STALE"), persisted) self.assertEqual(persisted["consecutive_deteriorating"], 3) def test_increments_on_unhealthy_reads_too(self): persisted = {} for _ in range(3): persisted = intraday_health.update_health_state(_reading("UNHEALTHY"), persisted) self.assertEqual(persisted["consecutive_deteriorating"], 3) def test_resets_on_watch_or_healthy_read(self): persisted = {} for _ in range(2): persisted = intraday_health.update_health_state(_reading("STALE"), persisted) self.assertEqual(persisted["consecutive_deteriorating"], 2) persisted = intraday_health.update_health_state(_reading("WATCH"), persisted) self.assertEqual(persisted["consecutive_deteriorating"], 0, "a recovering WATCH read should reset the counter, not just " "hold it, so a symbol that dips then recovers never gets " "force-exited") def test_mixed_state_reads_all_count_as_deteriorating(self): """A symbol bouncing between STALE and UNHEALTHY without ever touching WATCH/HEALTHY should still accumulate -- this is exactly the TJGC pattern (STALE the whole time, never UNHEALTHY, never recovering).""" persisted = {} for raw in ("STALE", "STALE", "STALE"): persisted = intraday_health.update_health_state(_reading(raw), persisted) self.assertEqual(persisted["consecutive_deteriorating"], 3) class TestShouldForceExitOnDeterioration(unittest.TestCase): def setUp(self): self.cfg = {"exit_on_deterioration_enabled": True, "exit_confirm_reads": 3} def test_false_below_threshold(self): self.assertFalse(intraday_health.should_force_exit_on_deterioration( {"consecutive_deteriorating": 2}, self.cfg)) def test_true_at_threshold(self): self.assertTrue(intraday_health.should_force_exit_on_deterioration( {"consecutive_deteriorating": 3}, self.cfg)) def test_true_above_threshold(self): self.assertTrue(intraday_health.should_force_exit_on_deterioration( {"consecutive_deteriorating": 5}, self.cfg)) def test_disabled_flag_short_circuits(self): cfg = dict(self.cfg, exit_on_deterioration_enabled=False) self.assertFalse(intraday_health.should_force_exit_on_deterioration( {"consecutive_deteriorating": 99}, cfg)) def test_defaults_used_when_cfg_keys_missing(self): # default exit_confirm_reads=3, default enabled=True self.assertTrue(intraday_health.should_force_exit_on_deterioration( {"consecutive_deteriorating": 3}, {})) self.assertFalse(intraday_health.should_force_exit_on_deterioration( {"consecutive_deteriorating": 2}, {})) class FakePositionManager: def __init__(self, open_symbols): self._open = list(open_symbols) self.exit_calls = [] def get_open_symbols(self): return list(self._open) def exit_position(self, symbol, price, reason): self.exit_calls.append((symbol, price, reason)) class FakeStream: def get_bars(self, symbol): now = datetime.now(timezone.utc) return [ {"t": now - timedelta(minutes=i), "o": 10, "h": 10, "l": 10, "c": 10.5, "v": 100} for i in range(5) ] class TestMonitorForcesExitOnDeterioration(unittest.TestCase): def setUp(self): self.orch = monitor.SessionOrchestrator.__new__(monitor.SessionOrchestrator) self.orch.cfg = { "universe": {"min_avg_daily_volume": 500000}, "intraday_health": { "exit_on_deterioration_enabled": True, "exit_confirm_reads": 3, }, } self.orch.stream = FakeStream() self.orch._get_volume_baselines = lambda: {} self.orch.premarket_20 = [{"symbol": "TJGC"}, {"symbol": "OTHER"}] def _run(self, position_mgr, health_state_entries, handle_slot_freed=None): self.orch.position_mgr = position_mgr self.orch._handle_slot_freed = handle_slot_freed or (lambda symbol: None) with patch.object(monitor, "intraday_health") as mock_health, \ patch.object(monitor, "data_store") as mock_ds, \ patch.object(monitor, "top_stocks") as mock_top_stocks: mock_ds.load_health_state.return_value = {} mock_ds.save_health_state.return_value = None def fake_evaluate_symbol(symbol, bars, avg_vol_baseline, health_state): reading = MagicMock() entry = health_state_entries.get(symbol, {"consecutive_deteriorating": 0, "state": "STALE"}) reading.confirmed_state = entry["state"] return reading, entry mock_health.evaluate_symbol.side_effect = fake_evaluate_symbol # Delegate to the REAL threshold function so the test still # exercises actual logic, not just a mock returning True. mock_health.should_force_exit_on_deterioration.side_effect = \ intraday_health.should_force_exit_on_deterioration self.orch._update_intraday_health() return mock_top_stocks def test_open_position_force_exited_at_threshold(self): pm = FakePositionManager(open_symbols=["TJGC"]) freed = [] self._run(pm, {"TJGC": {"consecutive_deteriorating": 3, "state": "STALE"}}, handle_slot_freed=freed.append) self.assertEqual(len(pm.exit_calls), 1) symbol, price, reason = pm.exit_calls[0] self.assertEqual(symbol, "TJGC") self.assertEqual(reason, "HEALTH_DETERIORATION") self.assertEqual(price, 10.5) self.assertEqual(freed, ["TJGC"], "slot should be freed exactly once for the " "force-exited symbol") def test_open_position_not_exited_below_threshold(self): pm = FakePositionManager(open_symbols=["TJGC"]) self._run(pm, {"TJGC": {"consecutive_deteriorating": 2, "state": "STALE"}}) self.assertEqual(pm.exit_calls, [], "2 consecutive reads is below the " "exit_confirm_reads=3 threshold") def test_non_open_position_symbol_never_exited_regardless_of_health(self): """OTHER has no open position -- deteriorating health should only affect its entry eligibility (handled elsewhere), never trigger exit_position for a symbol with no position to exit.""" pm = FakePositionManager(open_symbols=["TJGC"]) self._run(pm, { "TJGC": {"consecutive_deteriorating": 3, "state": "STALE"}, "OTHER": {"consecutive_deteriorating": 10, "state": "UNHEALTHY"}, }) symbols_exited = [c[0] for c in pm.exit_calls] self.assertEqual(symbols_exited, ["TJGC"]) def test_disabled_flag_prevents_force_exit(self): self.orch.cfg["intraday_health"]["exit_on_deterioration_enabled"] = False pm = FakePositionManager(open_symbols=["TJGC"]) self._run(pm, {"TJGC": {"consecutive_deteriorating": 10, "state": "STALE"}}) self.assertEqual(pm.exit_calls, [], "disabling the flag should suppress the " "force-exit even well past threshold") if __name__ == "__main__": unittest.main()