o Mj@s dZddlZddlZddlZddlZddlZddlmZmZmZddl m Z m Z m Z ddl mZddlZddlZddlZddlZddlZddlZddlmZmZmZddlmZddlmZdd lmZed ZGd d d Z Gd ddZ!ddZ"e#dkre"dSdS)ax monitor.py Main orchestrator. Runs the full daily lifecycle: START -> premarket_scanner.py (09:29, one full-universe scan) -> top candidates (premarket_20) -> start/maintain SIP stream for premarket_20 -> opening confirmation -> entry (evaluated across premarket_20) -> continuous intraday health scoring of premarket_20 (every intraday_health.eval_interval_seconds) -> full intraday rescan + premarket_20 rotation (every intraday_health.full_rescan_interval_minutes, until the no-new-entries cutoff) -> position management + trailing stops -> on exit -> top_stocks.py -> find replacement (from premarket_20) -> confirm -> enter -> end-of-day liquidation -> shutdown [UPDATED 2026-09-01] Per explicit request, the single scan moved from 09:00 to schedule.premarket_scan_time (09:29) and the 09:00-09:25 development-monitoring window plus the 09:25 final-scoring review were eliminated entirely -- there's no time left for either between a 09:29 scan and the 09:30 open. final_10 is now populated directly from premarket_20's top candidates.total_score ranking right after the scan (see _run_premarket_scan()), with no development-trend adjustment applied (there's no multi-snapshot history to compute one from anymore). [FEATURE 2026-08-17] premarket_20 is now the primary intraday candidate universe (previously this was final_10, which restricted the bot to only 10 of the original 20 candidates for the entire day). final_10 is still written for the dashboard's own tab and for backward compatibility, but no longer gates entries or replacement search. Run directly: python monitor.py Designed to be started once per day via cron shortly before 09:00 ET, and to exit cleanly after end-of-day liquidation. A singleton file lock prevents two instances from ever running against the same account simultaneously. N)datetimetimezone timedelta) get_configget_envvalidate_config) get_logger)compute_fast_rankingevaluate_fast_entry_only snapshot_dict)PositionManager) StreamManager) get_clientmonitorc@sNeZdZdZdedefddZdedede fd d Z ded e defd d Z dS)_EntryAttemptTrackera [FEATURE 2026-09-01] Fixes the shortlist-starvation bug found in the 2026-09-01 session review -- see config.json's intraday_health._note_confirmation_failure_cooldown for the full real-world case (F monopolizing a shortlist slot for ~24 minutes while CRML, health-eligible and competitively scored, never got a single evaluate_entry() call before rotating back out). _scan_for_entries() ranks premarket_20 by health_score and only evaluates the top entry_shortlist_size symbols each poll cycle. A symbol that ranks highly but keeps failing entry_engine's confirmation on a condition that doesn't change bar-to-bar (e.g. a hard momentum disqualifier) can occupy one of those slots forever, since nothing here ever demoted it. This class tracks consecutive not-confirmed results per symbol and temporarily benches one after max_failures in a row, freeing its shortlist slot for the next-best candidate for cooldown_seconds. Deliberately does NOT change entry_engine's actual confirmation criteria in any way -- a benched symbol that comes off the bench is evaluated against the exact same rules as before. This only changes WHICH candidates get a turn, never what it takes to pass. Pure in-memory, per-session state (like self.premarket_20 itself) -- intentionally not persisted, since a fresh session should start every symbol unbenched. max_failurescooldown_secondscCs||_||_i|_i|_dSN)rr_consecutive_failures_benched_until)selfrrr monitor.py__init__bs z_EntryAttemptTracker.__init__symbolnowreturncCs|j|}|duo ||kSr)rget)rrruntilrrr is_benchedhs z_EntryAttemptTracker.is_benched confirmedc Cs|r|j|d|j|ddS|j|dd}||jkrC|t|jd|j|<d|j|<td|d|jdd|ddS||j|<dS) Nr)seconds[ENTRY] z& benched from the entry shortlist for z.0fzs after zE consecutive not-confirmed attempts -- giving other candidates a turn) rpoprrrrrloginfo)rrr rcountrrr record_resultls   z"_EntryAttemptTracker.record_resultN) __name__ __module__ __qualname____doc__intfloatrstrrboolrr(rrrrrEs rc@seZdZddZddZddZddZd d Zd d Zd dZ de fddZ de fddZ de fddZ defddZddZddZddZdd Zd!efd"d#Zd$d%Zd&d'Zd(d)Zd*d+Zd,S)-SessionOrchestratorcCst|_t|jt|jdddk|_t|_t|jd|_ t |_ t |jdd|jddd|_ tt}t|g|_g|_g|_i|_i|_i|_i|_i|_d|_d|_d |_i|_ttj |j!ttj"|j!t#$|j%dS) Nmodeexecution_mode simulation)r4intraday_health%max_consecutive_confirmation_failures%confirmation_failure_cooldown_seconds)rrF)&rcfgrrvalidater4rclientr position_mgrr streamr_entry_trackerr5prune_stale_health_state data_storeload_health_statesave_health_state premarket_20final_10_prefiltered_universe_opening_volume_baseline_volume_baseline_prev_day_high _prev_close_volatility_baseline_last_health_eval_last_full_rescan _shutdown_fast_engine_statesignalSIGTERM_handle_signalSIGINTatexitregister_cleanup)rpruned_health_staterrrr~s8     zSessionOrchestrator.__init__cCstd|dd|_dS)NzHUTDOWN] Received signal z, shutting down gracefullyT)r%r&rL)rsignumframerrrrPs z"SessionOrchestrator._handle_signalcCs0z|jWn tyYnwtddS)NzHUTDOWN] Cleanup complete)r<stop Exceptionr%r&rrrrrTs  zSessionOrchestrator._cleanupcCstd|jddtstddS||jr!dS||| | | | tddS)Nz#[START] monitor.py starting | mode=r2r3z[START] Not a weekday, exitingzHUTDOWN] Session complete) r%r&r8 market_time is_weekday_wait_for_premarket_scanrL_run_premarket_scan_run_breakout_scanner_shadow_start_streaming_wait_for_market_open_main_trading_loop_end_of_day_liquidationrZrrrruns zSessionOrchestrator.runcC6ts|jstdts|jrdSdSdSdS)N)r[is_premarket_scan_timerLtimesleeprZrrrr] z,SessionOrchestrator._wait_for_premarket_scancCst|j}tdt|tj|j||j|j|j d|_ tdt|j dt |j|j |_ tj |jdd|j ||j|d|_|jd|jdd|_td d |jDd d |jDd t|j||jdS) Nz1[PREMARKET] Universe size after asset filtering: ) baseline_outprev_day_high_outprev_close_outzREMARKET] Prefiltered to z symbols in price/volume band candidatespremarket_candidate_count) prefilteredvolume_baselinesprev_day_highsvolatility_baselinesfinal_candidate_countcSg|] }|d|ddqSr total_score)rscorer.0rrrr z;SessionOrchestrator._run_premarket_scan..cSrurvrryrrrr|r})rBrC)premarket_scannerget_universe_symbolsr:r%r&lenprefilter_by_snapshotrFrGrHrDcompute_volatility_baselinerIscanr8_get_volume_baselines_get_volatility_baselinesrBrCr?save_watchlistwrite_top_stocks_sync_opening_volume_baseline)runiverserrrr^s0    z'SessionOrchestrator._run_premarket_scancspjdidds dStjttjtjfdd}tj |ddd dS) a [FEATURE 2026-09-16] Runs breakout_scanner.py's candidate-scoring model (docs/breakout_bot_design_conversation.pdf's Part II design) once, right after the real premarket scan above, against the SAME cached prefiltered universe/baselines -- purely for side-by-side comparison (writes data/breakout_scanner/{date}_ scanner.json, a completely separate file from sip_bot's own candidate output). Never touches self.premarket_20/self.final_10 or anything else the live trading pipeline reads, and unlike _run_premarket_scan() above, nothing calls this again during the day -- it's a one-shot morning comparison, not part of the 30-minute intraday rescan cadence. Runs on a background thread rather than inline: measured 2026-09-16 at ~17s for a 307-symbol prefiltered pool (scan + compute_range_20d_high's extra bulk daily-bars call) -- scaled to a real ~540-symbol 9:29 premarket pool that's roughly another 30s stacked sequentially on top of the real scan's own ~15-25s, which this project's own schedule.premarket_scan_time note already flags as having very little buffer before market_open_ time. A background thread means this can take as long as it needs without ever pushing _start_streaming()/market open later. Reads (self.client / prefiltered/baseline data) are snapshotted onto plain copies before the thread starts, since self._volume_ baseline/_prev_day_high keep getting mutated by the live pipeline as the day goes on (e.g. replacement candidates) -- the thread works off its own copies rather than racing those live reads/ writes. self.client itself IS shared with the main thread for the rest of the session (concurrent read-only GET calls against alpaca-py's underlying requests.Session/urllib3 connection pool, which pools connections per-thread safely for this kind of simple concurrent usage) -- any failure here is caught and logged, never raised into the main thread. Wrapped in try/except inside the worker: a bug in this shadow scanner must never delay market open or crash the real session -- see breakout_scanner.py's own module docstring for the full design and why it's an intentionally separate, differently- scored implementation rather than a variant of premarket_ scanner.py. breakout_scannerenabledTNcsLztj}tjj|dWdSty%tdYdSw)N)rprqrr prev_closesrange_20d_highszN[BREAKOUT-SCANNER] Shadow scan failed -- continuing sip_bot session unaffected)rcompute_range_20d_highr:rrYr% exception)rprefiltered_snapshotprev_closes_snapshotprev_day_highs_snapshotrvolume_baselines_snapshotrr_worker;s zASessionOrchestrator._run_breakout_scanner_shadow.._workerzbreakout-scanner-shadow)targetnamedaemon) r8rlistrDdictrrGrH threadingThreadstart)rrrrrr_ s*    z0SessionOrchestrator._run_breakout_scanner_shadowrcC t|diS)a^[Phase 5 -- 2026-09-08] Defensive accessor for self._volume_ baseline -- some of this project's existing tests construct a SessionOrchestrator via __new__() (bypassing __init__ entirely) and set only the specific attributes each test needs, per this project's established lightweight-test-double pattern (see tests/test_resistance_freeze_bugfix.py). getattr with a default keeps this new attribute from breaking those tests, and behaves identically to a plain self._volume_baseline read for every real session (where __init__ always sets it).rFgetattrrZrrrrLs z)SessionOrchestrator._get_volume_baselinescCr)Same defensive-accessor pattern as _get_volume_baselines() above, for the same reason (test doubles built via __new__() that don't set every __init__ attribute) -- see that method's docstring.rGrrZrrr_get_prev_day_highsX z'SessionOrchestrator._get_prev_day_highscCr)rrIrrZrrrr^rz-SessionOrchestrator._get_volatility_baselinesrncCs6|D]}|di}t|ddd|j|d<qdS)aYKeeps self._opening_volume_baseline current as premarket_20 gains symbols (initial scan, slot-replacement additions, and full-rescan rotations) -- entry_engine's opening-volume-expansion check needs a real baseline per symbol, not the accidental "no baseline -> ratio always looks huge" behavior of a missing key.metricspremarket_volumer!rN)rmaxrE)rrnr{rrrrrds z1SessionOrchestrator._sync_opening_volume_baselinecCsTdd|jD}|stddS|j||jr#tddStddS)NcSg|]}|dqSrrryrrrr|sz8SessionOrchestrator._start_streaming..z5[OPEN] No candidates to stream; skipping stream startz[OPEN] SIP stream activezF[OPEN] SIP stream failed to become healthy; will rely on REST fallback)rBr%warningr<r is_healthyr&error)rsymbolsrrrr`os   z$SessionOrchestrator._start_streamingcCre)N)r[is_past_market_openrLrhrirZrrrra}rjz)SessionOrchestrator._wait_for_market_opencCs|jdd}|jd}ttj}||_||_|jst s|j | |j r5ts5||j ttj}|ddrZ||j}||dkrZ|||_tst||j}||ddkrt|||_t||jst rdSdSdSdS) Nschedulepoll_interval_seconds_intradayr5rTeval_interval_secondsfull_rescan_interval_minutes<)r8rrrutcrJrKrLr[is_force_liquidate_timer;poll_pending_exits_update_open_positionshas_available_slotis_new_entries_cutoff_scan_for_entriesreconcile_with_brokerr total_seconds_update_intraday_health_run_intraday_full_rescanrhri)rinterval health_cfgrelapsedrrrrbs0        z&SessionOrchestrator._main_trading_loopcCs|jD]<}|j|}|sq|dd}|j|}|j|}|jj|||||d}|rA|j|||t || |qdS)Nc)bars_subquote) r;get_open_symbolsr<get_bars get_bars_sub get_quoteupdate_position exit_position top_stocksregister_cooldown_handle_slot_freed)rrbars current_pricerr exit_signalrrrrs        z*SessionOrchestrator._update_open_positions freed_symbolcsxtd|j}tj|||dsdS|j dgfdd|j D|_ |j | gdS)Nz [SLOT] 1 position slot available)exclude_symbolsrqrsrcs g|] }|ddkr|qSrrry candidaterrr|s z:SessionOrchestrator._handle_slot_freed..) r%r&r;rrfind_replacementrrr< add_symbolsrBappendr)rr open_symbolsrrrrs   z&SessionOrchestrator._handle_slot_freedc Cst}|jdd}ttj}|jddd}g}|jD]P}|d}|j |r,q||i}|dd} t | sIt d |d | q|j|rYt d |d q|j||rjt d |d q||qg} |D]]}|d}|j|} | rt| d krt d |dqt|j|} |j|} |j|d}|||}|j|i}t|| | | |||||d \}}| |j||| | |||fqt| jdddd| d|}dd|D}|rt dddd|Dt | ddD]S\}\}}}} } }}}||vrqt!id|ddd|d|ddd dd!|j"d"|j#d#|j$d$|j%d%|j&d&|j'd'|j(d(dd)gd*gt)||q|jd+d,}t |ddD]\}\}}}} } }}}|j *svdSt +|| |}||idd-}|j,|krt d.|d/|d0|j,d1|j-d2d3 t.|||||d4}t /d5|d6|j0d7|jd8|j"d9|j#d:|j$d;|j%d<|j&d=|j'd>|j(d?|j1d2d@dA|j2p|j3pdBt!id|ddd|d|j0d |j4d|jd!|j"d"|j#d#|j$d$|j%d%|j&d&|j'd'|j(d(|j1d)|j2d*|j3t)||t5|dCdsE|j6||j4ttj|j4r{| dDdE}|j7sX|j89nd}|jdFdG}|rit:|j;n|}dA|j2}|j <|||| |q`dS)HNr5entry_shortlist_size streamingsub_minute_bucket_secondsrstateWATCHz[ENTRY] Skipping z : health=z: stale stream dataz.: benched after repeated confirmation failuresz1: insufficient live bars for a fresh ranking readr!)rsub_bucket_secondsas_ofpremarket_resultcSs|dS)Nrr)trrr1sz7SessionOrchestrator._scan_for_entries..TkeyreversecSsh|]^}}}|qSrr)rz_symrrr 3z8SessionOrchestrator._scan_for_entries..z[ENTRY] Shortlist this cycle: z, css(|]^}}}|d|ddVqdS)z (confidence=.1f)Nr)rzconfrrrrr 7s&z8SessionOrchestrator._scan_for_entries..)r shortlistedFconfidence_rank confidence should_enter directionmomentum volume_state vwap_state ema9_state resistance extensionpersistence_seconds_elapsed reasons_forreasons_againstrmin_avg_daily_volume?r#z: cached health=z but fresh read=z (score=rz0) -- using the fresh read for the entry decision)rrz[FAST]  z (confidence=z , direction=z , momentum=z , volume=z, vwap=z, ema9=z , resistance=z , extension=z, held=zs): z; zno specific reason recordedinsufficient_datarrtradingsimulated_equity_default)=r?r@r8rrrrrrBr;is_symbol_openr5is_eligible_for_entryr%debugr<is_symbol_staler=rrrrrrrErrM setdefaultr rsortjoin enumerateappend_fast_engine_decisionrrrrrrrr rcompute_health raw_state health_scorer r&rrrrrrr(r4r: get_accountr.equityenter_position) r health_stateshortlist_sizerreligibler{rentryconfirmed_staterankedrrrbaseline real_baseline fast_state predictionfeatures shortlistshortlisted_symbolsrankravg_vol_baseline fresh_reading cached_statedecision entry_priceaccount sim_equityrreasonrrrrsB              $  &        z%SessionOrchestrator._scan_for_entriesc Cs|jdd}t}t|j}|jD]c}|d}|j|}|r)t |dkr*q| ||}t ||||\}} | ||<||vrxt | |jdrx|dd} td|d | d d | d d |j|| dt|||qt|dS)a [FEATURE 2026-08-17] Evaluates every symbol currently in premarket_20 through intraday_health.evaluate_symbol(), using the SIP stream's already-buffered bars (no extra API calls). Persists the hysteresis-confirmed state to state/intraday_health.json, which _scan_for_entries() and top_stocks.find_replacement() both read. This function otherwise never touches position_manager.py or triggers an exit -- an existing open position's trailing stop is left alone regardless of what health state its symbol reports here. [FEATURE 2026-09-11] One narrow exception: if the symbol has an open position AND intraday_health. should_force_exit_on_deterioration() confirms its health has read STALE-or-worse for exit_confirm_reads consecutive evaluations in a row, force-close it here -- see intraday_health.py's module docstring for why. rrrrr5rrz[HEALTH-EXIT] z$ health confirmed deteriorating for consecutive_deterioratingz consecutive reads (state=rz); forcing exitHEALTH_DETERIORATIONN)r8r?r@setr;rrBr<rrrrr5evaluate_symbol"should_force_exit_on_deteriorationr%r&rrrrrA) rfallback_baselinerrr{rrr#reading new_entryrrrrrs:       z+SessionOrchestrator._update_intraday_healthc Cs*|js tddStdt|jd|jd}|jdd}t}tj t|j|j|d| | | d }g}|D][}|d }| ||}|d d |jDvr_|j|nd} | rit| d kry|d|d<d|d<||qAt|| ||\} } | ||<| j|d<| j|d<t| jr||qAt||stddS|jdddd|d|d} dd|D} t|j}dd| D}|jD]P}|d }||vr$||vr$| |}|dur | |||td|d| ddd qtd|d!| ddd d"| |||q| |_|| |jd#d | D|j d$i d%d&}| d|}t }d'd | D|d(<d)d |D|d*<t!|t"| | D]}td|d d+|dd,| dqktd-t| d.t|dS)/a\ [FEATURE 2026-08-17] Every intraday_health.full_rescan_interval_ minutes (default 30), re-scores the ENTIRE cached prefiltered universe (not just the current premarket_20) against live, current bars -- using the same scoring function and the same health-eligibility filter as the rest of the intraday system -- and rotates premarket_20 to the resulting top N. This is what lets the bot discover a genuinely new candidate that wasn't part of the original 09:00 list, not just rotate among the original 20 all day. Reuses self._prefiltered_universe (captured once at 09:00) rather than re-pulling the full tradable-asset list and a fresh bulk snapshot every 30 minutes. Any symbol with an open position is always kept in the pool regardless of whether the fresh rescan re-selects it -- dropping it here would only affect future entry/replacement eligibility bookkeeping, never the open position itself, but keeping it visible avoids losing track of it in logs/dashboard. zD[INTRADAY-RESCAN] No cached prefiltered universe available; skippingNz*[INTRADAY-RESCAN] Starting full rescan of z prefiltered symbolsr5rrfull_rescan_lookback_hours)candidate_countrplookback_hoursrqrrrsrcSrrr)rzprrrr|rzASessionOrchestrator._run_intraday_full_rescan..rrwrrrzb[INTRADAY-RESCAN] No health-eligible candidates found; keeping the existing premarket_20 unchangedcSs|d|dfS)Nrrwr)r{rrrrsz?SessionOrchestrator._run_intraday_full_rescan..Trfull_rescan_pool_sizecSsi|]}|d|qSrrryrrr @rzASessionOrchestrator._run_intraday_full_rescan..cSsh|]}|dqSrrryrrrrBrz@SessionOrchestrator._run_intraday_full_rescan..z[INTRADAY-RESCAN] z6 kept in pool (open position) with refreshed pm_high=$pm_highr.2fz` kept in pool (open position) but missing from this rescan's universe -- reusing stale pm_high=$zj; resistance will not reflect current price action until this symbol reappears in the prefiltered universecSrrrryrrrr|WrrnrtcS&g|]}|d|d|ddqSrrwr)rrxhealthrryrrrr|jrBcSr<r=r?ryrrrr|nr@rCz score=z health=z*[INTRADAY-RESCAN] Rotated premarket_20 -> z candidates, final_10 -> top )#rDr%rr&rr8r?r@r~rrrrrrBr<rrr5r.rrrrAr r-r;raddrrrload_watchlistrr)rr8r0rrescoredrr{rr#rr1r2new_poolrescored_by_symbolr pool_symbolsfresh final_count final_slice watchlistrrrrs   $        #                  z-SessionOrchestrator._run_intraday_full_rescanc CsVtd|jjdd|jdidd}|jdd}t|}t|krN|j|j|j }|sSessionOrchestrator._end_of_day_liquidation..css2|]}|ddkr|dddkrdVqdSrNrOrPrr!Nr?rQrrrr0css2|]}|ddkr|dddkrdVqdSrRr?rQrrrrrSz[EOD] Session summary: trades=z wins=z losses=z total_P/L=$r:N)r%r&r; liquidate_allr8rrhrrget_closing_symbolsrirrsortedr<rXr?load_today_tradeswrite_trades_summarysum) rmax_wait poll_intervaldeadline still_opentradestotal_plwinslossesrrrrc|s:          z+SessionOrchestrator._end_of_day_liquidationN)r)r*r+rrPrTrdr]r^r_rrrrrrr`rarbrr/rrrrrcrrrrr1}s.<*C  -D1 )r1c Csz$ttt}|WdWdS1swYWdStyC}ztt|t dWYd}~dSd}~ww)Nr!) r?acquire_singleton_lock ensure_dirsr1rd RuntimeErrorr%rr/sysexit) orchestratorerrrmains  &ri__main__)$r,rerhrNrRrrrr config_loaderrrr logger_setuprr[r?r~rrr5 fast_pipeliner r r position_managerr r<r alpaca_clientrr%rr1rir)rrrrs@-    8L