""" monitor.py Orchestrator for this project -- the piece that ties scanner, stream, smart_engine, exit, and data_store together into a runnable daily lifecycle: START -> wait for schedule.premarket_scan_time -> run universe.py + scanner.py (writes today's candidate list) -> wait for market open -> subscribe stream.py to the candidate list [BUGFIX 2026-09-17: was subscribed right after the scan, before the open -- stream.py's live tick handlers have no premarket/regular- session boundary filter (unlike its own REST _backfill(), which correctly starts from session_open_dt()), so any premarket prints between the scan and 09:30 silently sat in the same buffer smart_engine.py reads as "session bars." Let several 2026-09-17 entries (MARA, NVAX) pass Stage 1/2 within 25-70s of the open on premarket-contaminated VWAP/trend/bar- count, chasing the opening print instead of a real move. Subscribing only after the open removes the contamination window outright rather than trying to filter it out.] -> main loop, every schedule.poll_interval_seconds: for each OPEN position: pull bars from the stream, call exit.evaluate() (hard stop at the entry stop_price + confirmed multi-signal deterioration), close on should_exit for each WATCHED (non-position) candidate, while a slot is free and before no_new_entries_after: pull bars/quote/resistance from the stream, call smart_engine.evaluate(), enter on BUY every cycle: log the full indicator dict for every symbol evaluated (data_store.append_decision_record), whether or not it ever traded -> force_liquidate_time -> close everything, write the day's trade summary -> shutdown Run directly: python monitor.py Designed to be started once per day via cron shortly before the open, and to exit cleanly after end-of-day liquidation. A singleton file lock (data_store.acquire_singleton_lock) prevents two instances ever running against this account at once. """ import os import sys import time import signal import threading from datetime import datetime, timezone from config_loader import get_config, get_env from logger_setup import get_logger import market_time import data_store import universe import scanner import smart_engine import exit as exit_engine import exit_giveback_room import exit_resistance_stall import reentry from position_manager import PositionManager from stream import StreamManager from alpaca_client import get_client log = get_logger("monitor") BASE_DIR = os.path.dirname(os.path.abspath(__file__)) PID_PATH = os.path.join(BASE_DIR, "monitor.pid") class SessionOrchestrator: def __init__(self): self.cfg = get_config() get_env().validate() self.client = get_client() self.position_mgr = PositionManager() self.stream = StreamManager() self.candidates = [] # today's ranked scanner output self._resistance_levels = {} # symbol -> {label: level} self._avg_vol_baseline = {} # symbol -> real prior-day volume # Caller-held per-symbol state, same persistence/confirmation # contract smart_engine.py / exit.py document -- intentionally # in-memory only (not persisted), same convention as sip_bot's # own monitor.py holds _fast_engine_state. self._entry_persistence = {} self._exit_confirmation = {} self._giveback_room_state = {} self._resistance_stall_state = {} self._last_plan_key = {} self._shutdown = False # [2026-09-23] Intraday rescans: the scan runs on a background # thread (it's ~15-40s of REST calls -- blocking the poll loop that # long would stall exits), and its result is applied on the main # thread at the top of the next poll. self._last_scan_started = None self._rescan_thread = None self._pending_scan = None self._pending_lock = threading.Lock() signal.signal(signal.SIGTERM, self._handle_signal) signal.signal(signal.SIGINT, self._handle_signal) # ------------------------------------------------------------------ def _handle_signal(self, signum, frame): log.info(f"[SHUTDOWN] Received signal {signum}, shutting down gracefully") self._shutdown = True # ------------------------------------------------------------------ def run(self): log.info(f"[START] monitor.py starting | mode={self.cfg['mode']['execution_mode']}") if not market_time.is_weekday(): log.info("[START] Not a weekday, exiting") return self._wait_for_premarket_scan() if self._shutdown: return self._run_scan() self._wait_for_market_open() self._start_streaming() self._main_trading_loop() self._end_of_day_liquidation() log.info("[SHUTDOWN] Session complete") # ------------------------------------------------------------------ def _wait_for_premarket_scan(self): while not market_time.is_premarket_scan_time() and not self._shutdown: time.sleep(5) def _compute_scan(self, candidates_file_suffix: str = "") -> dict: """Pure scan -- returns results without touching live state, so it can run on the rescan thread.""" all_symbols = universe.get_universe_symbols(self.client) log.info(f"[SCAN] Universe size after asset filtering: {len(all_symbols)}") baselines, prev_day_highs, prev_closes = {}, {}, {} prefiltered = universe.prefilter_by_snapshot( self.client, all_symbols, baseline_out=baselines, prev_day_high_out=prev_day_highs, prev_close_out=prev_closes, ) log.info(f"[SCAN] Prefiltered to {len(prefiltered)} symbols") daily_atr = {} range_20d_highs = scanner.compute_range_20d_high(self.client, prefiltered, daily_atr_out=daily_atr) candidates = scanner.scan( self.client, prefiltered=prefiltered, volume_baselines=baselines, prev_day_highs=prev_day_highs, prev_closes=prev_closes, range_20d_highs=range_20d_highs, candidates_file_suffix=candidates_file_suffix, ) return {"candidates": candidates, "baselines": baselines, "daily_atr": daily_atr} def _apply_scan(self, result: dict): """Main-thread only. Swaps in the new candidate list; keeps each already-known symbol's resistance levels from the first scan it appeared in (a mid-session scan's 'premarket_high' is really the last-6h high, not the premarket's).""" self._avg_vol_baseline.update(result["baselines"]) for c in result["candidates"]: m = c["metrics"] self._resistance_levels.setdefault(c["symbol"], { "premarket_high": m.get("premarket_high"), "prev_day_high": m.get("previous_day_high"), "range_20d_high": m.get("range_20d_high"), }) # A price DISTANCE for volatility.py's stop floor, not a level -- # smart_engine/setup_analyzer skip it when reading levels. self._resistance_levels[c["symbol"]]["daily_atr"] = result.get("daily_atr", {}).get(c["symbol"]) old_syms = [c["symbol"] for c in self.candidates] self.candidates = result["candidates"] log.info(f"[SCAN] {len(self.candidates)} candidates selected: " f"{[c['symbol'] for c in self.candidates]}") return old_syms def _run_scan(self): self._last_scan_started = time.time() self._apply_scan(self._compute_scan()) def _maybe_rescan(self): """[2026-09-23] Refresh the 30-candidate list every scanner.rescan_interval_minutes (0 = off) until no_new_entries_after. Symbols that drop off the list are unsubscribed unless a position is open in them; new ones are backfilled and subscribed.""" interval = self.cfg["scanner"].get("rescan_interval_minutes", 0) if not interval: return with self._pending_lock: result, self._pending_scan = self._pending_scan, None if result is not None: old_syms = self._apply_scan(result) new_syms = [c["symbol"] for c in self.candidates] open_syms = set(self.position_mgr.get_open_symbols()) added = [s for s in new_syms if s not in old_syms] dropped = [s for s in self.stream._subscribed if s not in new_syms and s not in open_syms] log.info(f"[RESCAN] applied: +{len(added)} {added} / -{len(dropped)} {dropped}") self.stream.add_symbols(added) self.stream.remove_symbols(dropped) for s in dropped: self._entry_persistence.pop(s, None) if market_time.is_new_entries_cutoff(): return if self._rescan_thread is not None and self._rescan_thread.is_alive(): return # [2026-09-24] Optional faster rescans early in the session (DNA only # started moving after the open and wasn't picked up until the 9:58 # rescan, already +9%): first_rescan_after_open_minutes forces one # rescan that many minutes after the open; rescan_interval_first_hour_minutes # replaces the normal interval until an hour after the open. 0 = off. sc = self.cfg["scanner"] mins_open = market_time.minutes_since_open() early = sc.get("first_rescan_after_open_minutes", 0) first_hour = sc.get("rescan_interval_first_hour_minutes", 0) if first_hour and 0 <= mins_open < 60: interval = min(interval, first_hour) due_early = (early and not getattr(self, "_early_rescan_done", False) and mins_open >= early) if due_early: self._early_rescan_done = True elif self._last_scan_started and time.time() - self._last_scan_started < interval * 60: return self._last_scan_started = time.time() suffix = "_" + market_time.now_et().strftime("%H%M") def worker(): try: res = self._compute_scan(candidates_file_suffix=suffix) except Exception: log.exception("[RESCAN] scan failed; keeping current candidate list") return with self._pending_lock: self._pending_scan = res log.info(f"[RESCAN] starting background rescan ({interval}-min interval)") self._rescan_thread = threading.Thread(target=worker, daemon=True) self._rescan_thread.start() def _start_streaming(self): symbols = [c["symbol"] for c in self.candidates] if symbols: self.stream.start(symbols) def _wait_for_market_open(self): while not market_time.is_past_market_open() and not self._shutdown: time.sleep(2) # ------------------------------------------------------------------ def _main_trading_loop(self): interval = self.cfg["schedule"]["poll_interval_seconds"] while not self._shutdown and not market_time.is_force_liquidate_time(): # [BUGFIX 2026-09-22] One bad poll cycle used to propagate up # to main(), which exits on RuntimeError -- the bot died at # 10:49 ET with positions open and never reached EOD # liquidation. Log it and keep polling instead. try: self._maybe_rescan() self._update_open_positions() if self.position_mgr.has_available_slot() and not market_time.is_new_entries_cutoff(): self._scan_for_entries() except Exception: log.exception("[LOOP] poll cycle failed; continuing") time.sleep(interval) # ------------------------------------------------------------------ def _resistance_for(self, symbol: str, session_high: float) -> dict: levels = dict(self._resistance_levels.get(symbol, {})) levels["session_high"] = session_high return levels def _update_open_positions(self): for symbol in self.position_mgr.get_open_symbols(): bars = self.stream.get_bars(symbol) if not bars: continue p = self.position_mgr.positions[symbol] # Checked first, ahead of exit.py's three layers -- reacts to a # peak-referenced giveback before trend/imbalance/hard-stop # would ever fire on absolute/session-wide readings. giveback_state = self._giveback_room_state.get(symbol, {}) giveback_decision = exit_giveback_room.evaluate( symbol, bars, p["entry_price"], giveback_state) self._giveback_room_state[symbol] = giveback_decision.state_out data_store.append_decision_record({ "symbol": symbol, "kind": "exit_giveback_room", "state": giveback_decision.state, "reason": giveback_decision.reason, "metrics": giveback_decision.metrics, }) if giveback_decision.should_exit: current_price = bars[-1]["c"] self.position_mgr.exit_position(symbol, current_price, giveback_decision.reason) self._exit_confirmation.pop(symbol, None) self._giveback_room_state.pop(symbol, None) self._resistance_stall_state.pop(symbol, None) continue # [2026-09-23] Stalled-at-resistance layer (exit_resistance_stall.py): # after ~60 min of failed tests of a ceiling with no breakout, # sell the next time price reaches the bottom of that ceiling. if self.cfg.get("resistance_stall", {}).get("enabled"): stall_decision = exit_resistance_stall.evaluate( symbol, bars, p["entry_time"], datetime.now(timezone.utc), self._resistance_stall_state.get(symbol, {})) self._resistance_stall_state[symbol] = stall_decision.state_out data_store.append_decision_record({ "symbol": symbol, "kind": "exit_resistance_stall", "state": stall_decision.state, "reason": stall_decision.reason, "metrics": stall_decision.metrics, }) if stall_decision.should_exit: self.position_mgr.exit_position(symbol, bars[-1]["c"], stall_decision.reason) self._exit_confirmation.pop(symbol, None) self._giveback_room_state.pop(symbol, None) self._resistance_stall_state.pop(symbol, None) continue confirmation_state = self._exit_confirmation.get(symbol, {}) exit_imbalance_window = self.cfg.get("exit", {}).get("imbalance_window_seconds", 20) trade_imbalance = self.stream.get_trade_imbalance(symbol, window_seconds=exit_imbalance_window) decision = exit_engine.evaluate(symbol, bars, p["entry_price"], confirmation_state, trade_imbalance=trade_imbalance, stop_price=p.get("stop_price")) self._exit_confirmation[symbol] = decision.confirmation data_store.append_decision_record({ "symbol": symbol, "kind": "exit", "state": decision.state, "reason": decision.reason, "metrics": decision.metrics, }) if decision.should_exit: current_price = bars[-1]["c"] self.position_mgr.exit_position(symbol, current_price, decision.reason) self._exit_confirmation.pop(symbol, None) self._giveback_room_state.pop(symbol, None) self._resistance_stall_state.pop(symbol, None) def _scan_for_entries(self): reentry_cfg = self.cfg["trading"].get("reentry_after_loss", {}) last_trade = {} if reentry_cfg.get("mode", "off") != "off": for t in data_store.load_today_trades(): if t.get("status") == "closed": last_trade[t["symbol"]] = t for c in self.candidates: symbol = c["symbol"] if self.position_mgr.is_symbol_open(symbol): continue if not self.position_mgr.has_available_slot(): break bars = self.stream.get_bars(symbol) if not bars: continue allowed, why = reentry.check(symbol, bars[-1]["c"], last_trade.get(symbol), reentry_cfg) if not allowed: self._entry_persistence.pop(symbol, None) data_store.append_decision_record({ "symbol": symbol, "kind": "entry", "state": "REJECT", "reasons_for": [], "reasons_against": [why], "metrics": {"price": bars[-1]["c"]}, }) continue quote_tuple = self.stream.get_quote(symbol) quote = {"bid": quote_tuple[0], "ask": quote_tuple[1]} if quote_tuple else None session_high = max(b["h"] for b in bars) resistance = self._resistance_for(symbol, session_high) cumulative_volume = sum(b["v"] for b in bars) avg_vol_baseline = self._avg_vol_baseline.get(symbol) persistence_state = self._entry_persistence.get(symbol, {}) imbalance_window = self.cfg.get("smart_engine", {}).get("imbalance_window_seconds", 20) trade_imbalance = self.stream.get_trade_imbalance(symbol, window_seconds=imbalance_window) decision = smart_engine.evaluate( symbol, bars, cumulative_volume, avg_vol_baseline, resistance, persistence_state, quote=quote, trade_imbalance=trade_imbalance) self._entry_persistence[symbol] = decision.persistence # [2026-09-23] Plain-English plan log, only when a symbol's plan # changes state/setup/trigger type (numbers move every tick). plan = decision.metrics.get("plan") if plan: key = (plan["state"], plan["setup"], tuple(t.split(":")[0] for t in plan["triggers"])) if self._last_plan_key.get(symbol) != key: self._last_plan_key[symbol] = key log.info(f"[PLAN] {symbol} @ {decision.metrics.get('price')}: {plan['state']}" + (f" ({plan['setup']})" if plan["setup"] else "") + f" | stop {plan['stop']}, target {plan['target']} ({plan['target_label']}), " f"R:R {plan['reward_risk']} | " + "; ".join(plan["reasons"] + plan["triggers"])) data_store.append_decision_record({ "symbol": symbol, "kind": "entry", "state": decision.state, "reasons_for": decision.reasons_for, "reasons_against": decision.reasons_against, "metrics": decision.metrics, }) if decision.should_enter: stop = decision.metrics.get("stage3", {}).get("stop") if stop is None: continue entered = self.position_mgr.enter_position( symbol, decision.metrics["price"], stop, reasons=decision.reasons_for) if entered: self._entry_persistence.pop(symbol, None) # ------------------------------------------------------------------ def _end_of_day_liquidation(self): log.info("[EOD] Force-liquidating all open positions") max_wait = self.cfg["schedule"]["eod_liquidation_max_wait_seconds"] poll_interval = self.cfg["schedule"]["poll_interval_seconds"] deadline = time.time() + max_wait while time.time() < deadline: open_symbols = self.position_mgr.get_open_symbols() if not open_symbols: break for symbol in open_symbols: bars = self.stream.get_bars(symbol) current_price = bars[-1]["c"] if bars else self.position_mgr.positions[symbol]["entry_price"] self.position_mgr.exit_position(symbol, current_price, "END_OF_DAY") time.sleep(poll_interval) else: still_open = self.position_mgr.get_open_symbols() if still_open: log.warning(f"[EOD] {len(still_open)} position(s) still not resolved after " f"{max_wait}s: {sorted(still_open)}") self.stream.stop() trades = data_store.load_today_trades() data_store.write_trades_summary(trades) total_pl = sum(t.get("pl_dollars", 0) for t in trades if t.get("status") == "closed") wins = sum(1 for t in trades if t.get("status") == "closed" and t.get("pl_dollars", 0) > 0) losses = sum(1 for t in trades if t.get("status") == "closed" and t.get("pl_dollars", 0) <= 0) log.info(f"[EOD] Session summary: trades={len(trades)} wins={wins} losses={losses} " f"total_P/L=${total_pl:.2f}") def main(): try: with data_store.acquire_singleton_lock(): data_store.ensure_dirs() with open(PID_PATH, "w") as f: f.write(str(os.getpid())) try: orchestrator = SessionOrchestrator() orchestrator.run() finally: try: os.remove(PID_PATH) except OSError: pass except RuntimeError as e: log.error(str(e)) sys.exit(1) if __name__ == "__main__": main()