""" test_simulation_e2e.py Exercises the full decision pipeline (scoring -> entry confirmation -> trailing stop -> exit -> replacement search) against synthetic bar data, with NO Alpaca API calls at all. This lets you validate strategy logic end-to-end any time, not just during market hours. Run: python tests/test_simulation_e2e.py """ import sys import os import json import shutil import tempfile import unittest from datetime import datetime, timedelta, timezone sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) def make_bars(prices, volumes, start=None): start = start or datetime(2026, 8, 17, 13, 30, tzinfo=timezone.utc) bars = [] for i, (p, v) in enumerate(zip(prices, volumes)): bars.append({ "t": start + timedelta(minutes=i), "o": p - 0.01, "h": p + 0.03, "l": p - 0.03, "c": p, "v": v, }) return bars class TestSimulationEndToEnd(unittest.TestCase): def setUp(self): # Redirect data_store's working dir to a temp dir so tests never # touch real state/logs. self.tmpdir = tempfile.mkdtemp() self._orig_cwd = os.getcwd() os.chdir(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import config_loader config_loader.get_config.cache_clear() global data_store, position_manager, entry_engine, scorer, risk_manager, intraday_health import data_store import position_manager import entry_engine import scorer import risk_manager import intraday_health data_store.BASE_DIR = self.tmpdir if False else data_store.BASE_DIR # keep real BASE_DIR for config # Instead, monkeypatch the dirs() function output paths to tmpdir self._orig_dirs = data_store._dirs def fake_dirs(): base = self.tmpdir return { "premarket": os_path(base, "premarket"), "intraday": os_path(base, "intraday"), "trades": os_path(base, "trades"), "state": os_path(base, "state"), } def os_path(base, name): from pathlib import Path p = Path(base) / name p.mkdir(parents=True, exist_ok=True) return p data_store._dirs = fake_dirs data_store.ensure_dirs() def tearDown(self): shutil.rmtree(self.tmpdir, ignore_errors=True) data_store._dirs = self._orig_dirs # ------------------------------------------------------------------ def test_full_pipeline_developing_stock_enters_and_trails(self): """ Simulates a stock that: 1. develops healthy premarket momentum (scores well) 2. confirms at the open (breakout holds, VWAP rising, volume expansion) 3. gets entered 4. runs up, trailing stop follows 5. reverses and gets stopped out with a profit """ # --- Premarket phase --- pm_bars = make_bars( [8.42, 8.51, 8.58, 8.61], [450_000, 620_000, 890_000, 1_020_000], start=datetime(2026, 8, 17, 9, 0, tzinfo=timezone.utc), ) pm_result = scorer.score_premarket_candidate( "DEVX", pm_bars, pm_high=8.65, pm_low=8.40, avg_vol_baseline=2_000_000, bid=8.60, ask=8.62, ) pm_result["pm_high"] = 8.65 self.assertGreater(pm_result["total_score"], 0) # --- Opening confirmation phase --- # Price breaks the premarket high (8.65) and holds, VWAP rising, # volume expansion vs premarket baseline, higher lows. open_bars = make_bars( [8.60, 8.66, 8.70, 8.75, 8.78], [100_000, 150_000, 180_000, 200_000, 210_000], start=datetime(2026, 8, 17, 13, 30, tzinfo=timezone.utc), ) quote = (8.77, 8.78, datetime.now(timezone.utc)) # [UPDATED 2026-08-20] entry_engine.evaluate_entry() now requires a # fresh health_reading, computed by the caller from these same # open_bars, and hard-disqualifies entry without one (RCAT staleness # fix, 2026-08-18 -- see entry_engine.py's module docstring). Compute # it the same way monitor.py's _scan_for_entries() does in # production: intraday_health.compute_health() on the live bars, # right before the entry decision, not from a cached state file. health_reading = intraday_health.compute_health("DEVX", open_bars, avg_vol_baseline=500_000) self.assertIn(health_reading.raw_state, ("HEALTHY", "WATCH"), f"Expected DEVX's rising, volume-expanding open to read HEALTHY/WATCH, " f"got {health_reading.raw_state} (score={health_reading.health_score})") decision = entry_engine.evaluate_entry( "DEVX", pm_result, open_bars, quote, opening_baseline_volume=1_020_000, health_reading=health_reading, ) self.assertTrue(decision.should_enter, f"Expected entry, got reasons_against={decision.reasons_against}") # --- Enter position --- pm = position_manager.PositionManager(simulation=True) entry_price = open_bars[-1]["c"] pos = pm.enter_position("DEVX", entry_price, account_equity=100_000, bars=open_bars, reason="; ".join(decision.reasons_for)) self.assertIsNotNone(pos) self.assertEqual(pos["status"], "open") initial_stop = pos["current_stop"] self.assertLess(initial_stop, entry_price) # --- Price runs up, stop should trail upward only --- run_prices = [8.85, 8.95, 9.05, 9.15] last_stop = initial_stop for p in run_prices: exit_signal = pm.update_position("DEVX", p, bars=None) self.assertIsNone(exit_signal) # still running, no stop hit current_stop = pm.positions["DEVX"]["current_stop"] self.assertGreaterEqual(current_stop, last_stop) last_stop = current_stop # --- Price reverses and hits trailing stop --- highest = pm.positions["DEVX"]["highest_price"] stop_price = pm.positions["DEVX"]["current_stop"] exit_signal = pm.update_position("DEVX", stop_price - 0.01, bars=None) self.assertEqual(exit_signal, "TRAILING_STOP") closed = pm.exit_position("DEVX", stop_price - 0.01, exit_signal) self.assertEqual(closed["status"], "closed") # Should have exited with a profit since price ran well above entry self.assertGreater(closed["current_pl"], 0) def test_fading_stock_does_not_confirm_entry(self): """A stock that fades immediately after open should NOT enter, even if it looked fine in premarket.""" pm_bars = make_bars( [8.80, 8.76, 8.65, 8.60], [1_200_000, 1_250_000, 1_270_000, 1_270_000], start=datetime(2026, 8, 17, 9, 0, tzinfo=timezone.utc), ) pm_result = scorer.score_premarket_candidate( "FADEX", pm_bars, pm_high=8.85, pm_low=8.55, avg_vol_baseline=2_000_000, bid=8.58, ask=8.62, ) pm_result["pm_high"] = 8.85 # Opens weak: fades immediately, volume collapses, VWAP declining open_bars = make_bars( [8.55, 8.50, 8.44, 8.38, 8.30], [80_000, 60_000, 40_000, 30_000, 20_000], start=datetime(2026, 8, 17, 13, 30, tzinfo=timezone.utc), ) quote = (8.29, 8.31, datetime.now(timezone.utc)) # [UPDATED 2026-08-20] Same fresh-health requirement as the DEVX # test above. FADEX's declining prices/volume here should # legitimately read UNHEALTHY on their own merits (negative price # slope, price below VWAP, lower-highs/lower-lows structure) -- # exercise that real path rather than relying on the "no reading # supplied" disqualifier, so this test still catches a regression # in compute_health()'s fade detection itself, not just in # evaluate_entry()'s handling of a missing reading. health_reading = intraday_health.compute_health("FADEX", open_bars, avg_vol_baseline=2_000_000) self.assertEqual(health_reading.raw_state, "UNHEALTHY", f"Expected FADEX's fading open to read UNHEALTHY, " f"got {health_reading.raw_state} (score={health_reading.health_score})") decision = entry_engine.evaluate_entry( "FADEX", pm_result, open_bars, quote, opening_baseline_volume=1_270_000, health_reading=health_reading, ) self.assertFalse(decision.should_enter) def test_slot_replacement_excludes_open_and_cooldown_symbols(self): import top_stocks data_store.save_watchlist({ "premarket_20": [], "final_10": [{"symbol": "AAA"}, {"symbol": "BBB"}], }) data_store.save_cooldowns({ "AAA": datetime.now(timezone.utc).isoformat(), }) cooldowns = data_store.load_cooldowns() self.assertTrue(top_stocks._is_on_cooldown("AAA", cooldowns)) self.assertFalse(top_stocks._is_on_cooldown("BBB", cooldowns)) def test_max_positions_enforced(self): import config_loader cfg = config_loader.get_config() max_positions = cfg["trading"]["max_positions"] pm = position_manager.PositionManager(simulation=True) for i in range(max_positions): symbol = f"SYM{i}" bars = make_bars([10.0, 10.0, 10.0], [1000, 1000, 1000]) pos = pm.enter_position(symbol, 10.0, account_equity=100_000, bars=bars, reason="test") self.assertIsNotNone(pos) self.assertFalse(pm.has_available_slot()) # One more attempt should be rejected overflow_bars = make_bars([10.0, 10.0, 10.0], [1000, 1000, 1000]) overflow = pm.enter_position("OVERFLOW", 10.0, account_equity=100_000, bars=overflow_bars, reason="test") self.assertIsNone(overflow) if __name__ == "__main__": unittest.main(verbosity=2)