o Tcj@@sdZddlZddlZddlZddlmZmZmZddlmZm Z ddl m Z ddl m Z ddlmZddlmZe d ZGd d d ZGd d d ZdS)a7 stream.py Manages the Alpaca SIP real-time stream for a set of symbols. Buffers incoming trades/quotes/bars into an in-memory per-symbol rolling window that fast_pipeline.py and position_manager.py read from. Runs the stream in a background thread so monitor.py's main loop stays simple synchronous polling logic (poll the buffer, not the network). Handles: - reconnect with backoff (config: streaming.reconnect_backoff_seconds) - stale data detection (config: streaming.stale_data_seconds) - symbol subscribe/unsubscribe as the watchlist changes through the day N) defaultdictdeque OrderedDict)datetimetimezone) get_config) get_logger) get_client)session_open_dtstreamc@seZdZdZd%ddZdefddZdefd d Zd d Zd dZ d&de fddZ de de fddZ de de fddZddZddZd&de fdd Zd!edefd"d#Zd$S)' SymbolBuffera Rolling window of 1-min bars + latest quote for one symbol, plus a second, parallel sub-minute rolling window (bars_sub / get_bars_sub()) at a configurable bucket width for finer-grained stream-feature work. See config.json's streaming._note_sub_minute.cCsD||_t|_d|_d|_d|_d|_||_||_t|_ d|_ dSN) maxlenrbars latest_quotelatest_trade_price last_update _current_barsub_bucket_seconds sub_maxlenbars_sub_current_bar_sub)selfrrrr stream.py__init__%s  zSymbolBuffer.__init__barcCT|d}||j|<|j|t|j|jkr(|jjddt|j|jksdSdSNtF)last)r move_to_endlenrpopitemrrkeyrrr _upsert_barR   zSymbolBuffer._upsert_barcCrr )rr#r$rr%r&rrr_upsert_bar_subYr)zSymbolBuffer._upsert_bar_subcCs"|j}|j||}|j|ddS)Nrsecond microsecond)rr,replace)rtswidth bucket_secondrrr_sub_bucket_for`szSymbolBuffer._sub_bucket_forc Cs||}|jdup|jd|k}|r+|jdur||j||||||dddd |_|j}|sNt|d||d<t|d||d<||d<|d|7<|d d 7<|durv||krh|d d 7<dS||krx|d d 7<dSdSdS) a [SIMULATION-ONLY 2026-09-05] prev_price is the trade price immediately before this one (captured by on_trade() BEFORE it overwrites self.latest_trade_price) -- compared against the new price for tick direction (upticks/downticks), deliberately continuous across bucket boundaries rather than resetting direction tracking to "unknown" at the start of every new bucket, so a bucket's uptick_ratio isn't artificially diluted by treating its first tick as directionless. tick_count/upticks/downticks are plain O(1) counters -- no raw ticks are stored, so this adds no unbounded memory regardless of how many trades print per bucket. Nr!r) r!ohlcv tick_countupticks downticksr4r5r6r7r8r9r:)r2rr*maxmin)rpricesizer/ prev_pricebucket is_new_bucketbrrr_accumulate_bar_subes,    z SymbolBuffer._accumulate_bar_subTreturncC,t|j}|r|jdur||jg}|Sr)listrvaluesrrinclude_formingrrrr get_bars_sub zSymbolBuffer.get_bars_subr>r?cCs4|j}||_||_|||||||||dSr)rr_accumulate_barrD)rr>r?r/r@rrron_trades zSymbolBuffer.on_tradebidaskcCs|||f|_||_dSr)rr)rrOrPr/rrron_quotes  zSymbolBuffer.on_quotec Csdt|dr |jdddn|}|||||||d||_|jdur.|jd|kr0d|_dSdSdS)zBDirect minute-bar ingestion (Alpaca also streams aggregated bars).r.rr+r!r3r4r5r6r7Nr!)hasattrr.r(rr)rr3r4r5r6r7r/ minute_bucketrrron_bars  zSymbolBuffer.on_barcCs|jddd}|jdus|jd|kr*|jdur||j||||||d|_dS|j}t|d||d<t|d||d<||d<|d|7<dS) Nrr+r!rRr4r5r6r7)r.rr(r<r=)rr>r?r/rTrCrrrrMs    zSymbolBuffer._accumulate_barcCrFr)rGrrHrrIrrrget_barsrLzSymbolBuffer.get_bars stale_secondscCsH|jdurdSttj}|j}|jdur|jtjd}|||kS)NT)tzinfo)rrnowrutcrXr. total_seconds)rrWrYr"rrris_stales   zSymbolBuffer.is_staleN)r rr)T)__name__ __module__ __qualname____doc__rdictr(r*r2rDrGrKfloatrNrQrUrMrVintboolr\rrrrr s -# r c@seZdZddZdefddZdefddZdd Zdefd d Zd d Z ddZ ddZ ddZ defddZ dedefddZdedefddZdefddZdedefddZdefd d!Zd"S)# StreamManagercstd|_t|_|jdd|jdd}tdt|dtfdd |_t |_ d|_ d|_ d|_ t|_t|_dS) N streamingsub_minute_bucket_secondsrsub_minute_buffer_minutesr;<cs tdS)Nrr)r rrkrrs z(StreamManager.__init__..)rcfgr clientgetr<rcrbuffersset _subscribed_stream_thread_loop threadingEvent _stop_event _connected)rsub_buffer_minutesrrkrrs   zStreamManager.__init__symbolsc Cst}ttj}||tjkrdS|D]r}z |jj|||d}Wnty?}zt d|d|WYd}~qd}~ww|sCq|j |}|D]%}|j j ddd} || t|jt|jt|jt|jt|jdqJt d|d t|d t|djd d qdS) a [BUGFIX 2026-09-09] stream_features.compute_features() treats bars[0] as "the session's first 1-min bar" (session_open in its extension_from_open_pct calc) on the assumption a symbol's buffer always spans the whole session from 09:30. That's only true if the symbol has been subscribed since before the open -- false for (a) any symbol added mid-day via add_symbols() (no history, subscribe_trades/bars only deliver ticks going forward), and (b) EVERY symbol after any intraday process restart, since SymbolBuffer is in-memory only and starts empty again. Confirmed root cause of a real miss: SRAD, added to the pool at 12:13 on 2026-09-09 (well after its 11:55 reversal low and post a 12:27 restart besides), had its "session open" read as ~$12.65 (the price when the buffer happened to start) instead of the real $12.69 09:30 open -- silently understating how extended it actually was, undermining the whole extension gate. Fix: before subscribing a symbol to the live stream, pull its real 1-min bars since market_open_time via REST and pre-load the buffer, so bars[0] is always genuinely the session's first bar regardless of when/why the subscription started. No-op before the open (nothing to backfill yet -- the live stream will deliver the real bar 0 itself, same as today). N)startendz[STREAM] Backfill failed for z: rr+rRz[STREAM] Backfilled z with z historical bars (session open $z.4fz) before live subscription)r rrYrrZ astimezonernget_minute_bars Exceptionlogwarningrp timestampr.r(rbopenhighlowclosevolumeinfor$) rr{open_dtrYsymbolrebufrrTrrr _backfills2     zStreamManager._backfillcCs<||tj|j|fdd|_|j|jjdddS)NT)targetargsdaemon )timeout)rrvThread _run_foreverrtr|rywaitrr{rrrr|s  zStreamManager.startcCsN|j|jr#|jr%z t|j|jWdSty"YdSwdSdSr)rxrqrursasynciorun_coroutine_threadsafestop_wsrrrrrstop s   zStreamManager.stopc sfdd|D}|rjdusjdurdS|j|z t|jWdStyJ}zt d|WYd}~dSd}~ww)Ncsg|] }|jvr|qSr)rr).0srrr sz-StreamManager.add_symbols..zadd_symbols failed: ) rsrurrrupdaterr_subscribe_asyncrrr)rr{new_symsrrrr add_symbolss    zStreamManager.add_symbolscsH|jj|jg|R|jj|jg|R|jj|jg|RdSr)rssubscribe_trades _on_tradesubscribe_quotes _on_quotesubscribe_bars_on_barrrrrr!szStreamManager._subscribe_asynccs.|j|j}|t|jt|j|jdSr)rprrNrbr>r?r)rtraderrrrr's  zStreamManager._on_tradecsB|j|j}|jr|jr|t|jt|j|jdSdSdSr)rpr bid_price ask_pricerQrbr)rquoterrrrr+s    zStreamManager._on_quotecsF|j|j}|t|jt|jt|jt|jt|j|j dSr) rprrUrbrrrrrr)rrrrrrr0s  zStreamManager._on_barcCs|jd}|jd}d}|js||krzPt|_t|j|j|_ t ||_ |j j |j g|R|j j|jg|R|j j|jg|Rtdt|d|j d}|j WnEty}z9|jrxWYd}~n<|t|t|d}td|d|d |dd |d t||d7}WYd}~nd}~ww|js||ks||krtd dSdS) Nreconnect_backoff_secondsmax_reconnect_attemptsrz#[STREAM] Connecting SIP stream for z symbolsr;z[STREAM] Disconnected (z); reconnecting in z s (attempt /)zf[STREAM] Max reconnect attempts exceeded. Stream is DOWN. monitor.py should fall back to REST polling.)rmrxis_setrnew_event_loopruset_event_looprn new_streamrsrqrrrrrrrrrrr$ryrunrr=rtimesleeperror)rr{backoffs max_attemptsattemptrdelayrrrr6sB          zStreamManager._run_foreverrrEcC|j|Sr)rprVrrrrrrVWszStreamManager.get_barscCr)zSub-minute (configurable bucket width, default 30s) rolling bars -- see SymbolBuffer.get_bars_sub() / config.json's streaming._note_sub_minute. Not consumed by anything yet.)rprKrrrrrKZszStreamManager.get_bars_subcCs |j|jSr)rprrrrr get_quote`s zStreamManager.get_quotecCs|j||jdS)Nstale_data_seconds)rpr\rmrrrris_symbol_staleczStreamManager.is_symbol_stalecCs|jo |j Sr)ryrrxrrrr is_healthyfrzStreamManager.is_healthyN)r]r^r_rrGrr|rrrrrrrstrrVrKrrdrrrrrrres / !re)r`rrvr collectionsrrrrr config_loaderr logger_setupr alpaca_clientr market_timer rr rerrrrs    &