o A%jp_@sdZddlZddlZddlZddlmZmZmZddlmZm Z m Z ddl m Z ddl mZddlmZddlmZed ZGd d d Zd efd dZGdddZdS)aJ 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 monitor.py reads and hands to the rule modules (rules_api.MarketView). 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 timedelta) get_config) get_logger) get_client)session_open_dtstreamc@seZdZdZd4ddZdZdedefd d Zd d Z d5defddZ de fddZ de fddZ ddZddZd6defddZdedefddZdedefd d!Zd"ed#efd$d%Zd"ed#efd&d'Zd(d)Zd*d+Zd,d-Zd6defd.d/Zd0edefd1d2Zd3S)7 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.cCsV||_t|_d|_d|_d|_d|_||_||_t|_ d|_ t |_ t |_dSN)maxlenrbars latest_quotelatest_trade_price last_update _current_barsub_bucket_seconds sub_maxlenbars_sub_current_bar_subr _trade_log threadingRLock_lock)selfrrrr "/var/www/screener/trade1/stream.py__init__%s  zSymbolBuffer.__init__ZpricereturncCsf|jdurdS|j\}}}|r|sdS||krdS||krdS||d}||kr+dS||kr1dSdS)z@+1 buyer-initiated, -1 seller-initiated, 0 unknown/no quote yet.Nr)r)rr$bidask_midr r r!_classify_tradegs   zSymbolBuffer._classify_tradecCsZ|t|jd}|jr'|jdd|kr+|j|jr)|jdd|ksdSdSdSdS)Nsecondsr)r_TRADE_LOG_MAX_AGE_SECONDSrpopleft)rnow_tscutoffr r r!_trim_trade_logys (zSymbolBuffer._trim_trade_log4@window_secondsc Cs|jH|js WddS|jdd}|t|d}d}}t|jD]\}}}||kr2n|dkr;||7}q'|dkrC||7}q'Wdn1sNwY||} | r_||| SdS)z(buy_vol - sell_vol) / (buy_vol + sell_vol) over the last window_seconds of classified trades, or None if there's nothing to compute it from (fails open, same convention as spread_pct when a quote isn't available).Nr'rr.g)rrrreversed) rr6r2r3buy_volsell_voltssidesizetotalr r r!get_trade_imbalance~s$  z SymbolBuffer.get_trade_imbalancebarcCs|j7|d}||j|<|j|t|j|jkr2|jjddt|j|jksWddSWddS1s=wYdSNtF)last)rr move_to_endlenrpopitemrr?keyr r r! _upsert_bars  "zSymbolBuffer._upsert_barcCsT|d}||j|<|j|t|j|jkr(|jjddt|j|jksdSdSr@)rrCrDrrErFr r r!_upsert_bar_subs   zSymbolBuffer._upsert_bar_subcCs"|j}|j||}|j|ddS)Nrsecond microsecond)rrKreplace)rr:width bucket_secondr r r!_sub_bucket_forszSymbolBuffer._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. NrAr) rAohlcv tick_countupticks downticksrRrSrTrUrVr&rWrX)rPrrImaxmin)rr$r<r: prev_pricebucket is_new_bucketbr r r!_accumulate_bar_subs,    z SymbolBuffer._accumulate_bar_subTcCx|j/t|j}|r"|jdur*|t|jg}Wd|SWd|SWd|S1s5wY|Sr)rlistrvaluesrdictrinclude_formingrr r r! get_bars_sub   zSymbolBuffer.get_bars_subr<cC:|j||||WddS1swYdSr)r_on_trade_locked)rr$r<r:r r r!on_trade"zSymbolBuffer.on_tradecCsZ|j}||_||_|||||||||||}|j|||f||dSr)rr_accumulate_barr_r-rappendr4)rr$r<r:r[r;r r r!ris zSymbolBuffer._on_trade_lockedr)r*cCrhr)r_on_quote_lockedrr)r*r:r r r!on_quoterkzSymbolBuffer.on_quotecCs|||f|_||_dSr)rrror r r!rns  zSymbolBuffer._on_quote_lockedc Cs@|j|||||||WddS1swYdS)zBDirect minute-bar ingestion (Alpaca also streams aggregated bars).N)r_on_bar_locked)rrQrRrSrTrUr:r r r!on_bars"zSymbolBuffer.on_barc Csdt|dr |jdddn|}|||||||d||_|jdur.|jd|kr0d|_dSdSdS)NrMrrJrArQrRrSrTrUrA)hasattrrMrHrr)rrQrRrSrTrUr: minute_bucketr r r!rqs  zSymbolBuffer._on_bar_lockedcCs|jddd}|jdus|jd|kr*|jdur||j||||||d|_dS|j}t|d||d<t|d||d<||d<|d|7<dS) NrrJrArsrRrSrTrU)rMrrHrYrZ)rr$r<r:rur^r r r!rls    zSymbolBuffer._accumulate_barcCr`r)rrarrbrrcrdr r r!get_barsrgzSymbolBuffer.get_bars stale_secondscCsH|jdurdSttj}|j}|jdur|jtjd}|||kS)NT)tzinfo)rrnowrutcrxrM total_seconds)rrwryrBr r r!is_stale s   zSymbolBuffer.is_staleN)rrrr5)T)__name__ __module__ __qualname____doc__r"r0floatintr-r4r>rcrHrIrPr_rarfrjrirprnrrrqrlrvboolr|r r r r!r s* @#  r r%cCs2|jt|jt|jt|jt|jt|jdS)Nrs) timestampropenhighlowclosevolume)r^r r r! _bar_dictsrc@seZdZddZdefddZdefddZdd Zdefd d Zdefd d Z ddZ ddZ ddZ de fddZdefddZde defddZde defddZde fdd Zd*de d"efd#d$Zde defd%d&Zdefd'd(Zd)S)+ StreamManagercstd|_t|_|jdd|jdd}tdt|dtfdd |_t |_ g|_ t |_ d|_ d|_d|_t|_t|_dS) N streamingsub_minute_bucket_secondsrsub_minute_buffer_minutesr&<cs tdS)Nrr)r r rr r!$s z(StreamManager.__init__..)rcfgr clientgetrYrrbuffersset _subscribed listeners bars_only_stream_thread_looprEvent _stop_event _connected)rsub_buffer_minutesr rr!r"s   zStreamManager.__init__symbolsc Cs,t}ttj}||tjkrdS|D]~}z |jj|||d}Wnty?}zt d|d|WYd}~qd}~ww|sCq|j |}|D]%}|j j ddd} || t|jt|jt|jt|jt|jdqJ|d|d d |Dt d |d t|d t|djddqdS)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 : rrJrs seed_barscSsg|]}t|qSr )r).0r^r r r! dsz+StreamManager._backfill..z[STREAM] Backfilled z with z historical bars (session open $z.4fz) before live subscription)r rryrrz astimezonerget_minute_bars ExceptionlogwarningrrrMrHrrrrrr_notifyinforD) rropen_dtrysymbolrebufr?rur r r! _backfill9s4     zStreamManager._backfillcCs<||tj|j|fdd|_|j|jjdddS)NT)targetargsdaemon )timeout)rrThread _run_foreverrrrwait)rrr r r!ris  zStreamManager.startcCsN|j|jr#|jr%z t|j|jWdSty"YdSwdSdSr)rrrrasynciorun_coroutine_threadsafestop_wsrrr r r!stopps   zStreamManager.stopc sxfdd|D}|rOjdurOjt|8_zjjjg|Rjjjg|RWntyN}zt d|d|WYd}~nd}~wwfdd|D}|r_jduradS |j |z1jjjg|Rjjjg|Rjj jg|Rtdt|d|WdSty}zt d |d|WYd}~dSd}~ww) a[FIX 2026-09-23] Used to schedule _subscribe_async() onto self._loop -- but self._stream.run() runs on alpaca-py's OWN loop (asyncio.run inside run()), so self._loop never ran and the coroutine never executed: added symbols were silently never subscribed. alpaca-py's subscribe_*/unsubscribe_* are themselves thread-safe (they post onto the stream's running loop and wait), so they're called directly from the caller's thread here.cs$g|]}|jvr|jvr|qSr )rrrsrr r!rs$z-StreamManager.add_symbols..Nz#[STREAM] tick subscribe failed for rcg|] }|jvr|qSr rrrr r!rz[STREAM] Subscribed z added symbols: z [STREAM] add_symbols failed for )rrrsubscribe_trades _on_tradesubscribe_quotes _on_quoterrrrrupdatesubscribe_bars_on_barrrD)rrpromoternew_symsr rr! add_symbolsxs0 "   $zStreamManager.add_symbolsc sfdd|D}|s dSj|jdurJzjj|jj|jj|WntyI}ztd|d|WYd}~nd}~ww|D] }j |dqLt dt |d|dS)zUnsubscribes and drops the buffer for symbols no longer watched, so the subscription set stays ~one scanner list wide instead of growing with every rescan.csg|] }|jvr|qSr rrrr r!rrz0StreamManager.remove_symbols..Nz#[STREAM] remove_symbols failed for rz[STREAM] Unsubscribed z dropped symbols: ) rdifference_updaterunsubscribe_tradesunsubscribe_quotesunsubscribe_barsrrrrpoprrD)rrgonersymr rr!remove_symbolss     "zStreamManager.remove_symbolscsP|j|j}|t|jt|j|j|d|j|jt|jt|jdS)Nrj)rrrjrr$r<rr)rtraderr r r!rs &zStreamManager._on_tradec s||j|j}|jr:|jr<|t|jt|j|j|d|j|jt|jt|jt|jp/dt|j p5ddSdSdS)Nrpr) rr bid_price ask_pricerprrrbid_sizeask_size)rquoterr r r!rs   zStreamManager._on_quotecsZ|j|j}|t|jt|jt|jt|jt|j|j | d|jt |dS)Nrr) rrrrrrrrrrrrr)rr?rr r r!rs  zStreamManager._on_barmethodc Gs|jD]@}t||d}|durqz||WqtyCt|ddsAtdt|jd|dzd|_Wn ty@YnwYqwdS)N_stream_error_loggedFz[STREAM] listener .z failed (ignored)T)rgetattrrr exceptiontyper~r)rrrlstfnr r r!rs"       zStreamManager._notifyc sjd}jd}d}js||krzct_tjj_ j s/t |_ t j }fdd|D}|rUj j jg|Rj jjg|Rj jjg|Rtdt|dj d}j WnEty}z9jrWYd}~n<|t|t|d}td |d |d |dd |d t||d7}WYd}~nd}~wwjs||ks||krtddSdS)Nreconnect_backoff_secondsmax_reconnect_attemptsrcrr )r)rxrr r!rrz.StreamManager._run_forever..z#[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.)rris_setrnew_event_looprset_event_loopr new_streamrrrsortedrrrrrrrrrDrrunrrZrtimesleeperror) rrbackoffs max_attemptsattemptcurrentticksrdelayr rr!rsJ           zStreamManager._run_foreverrr%cC|j|Sr)rrvrrr r r!rvszStreamManager.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.)rrfrr r r!rfszStreamManager.get_bars_subcCs |j|jSr)rrrr r r! get_quotes zStreamManager.get_quoter5r6cCs|j||Sr)rr>)rrr6r r r!r>sz!StreamManager.get_trade_imbalancecCs|j||jdS)Nstale_data_seconds)rr|rrr r r!is_symbol_stalezStreamManager.is_symbol_stalecCs|jo |j Sr)rrrrr r r! is_healthyrzStreamManager.is_healthyNr})r~rrr"rarrrrrrrrstrrrrvrfrrr>rrrr r r r!rs$0(r)rrrr collectionsrrrrrr config_loaderr logger_setupr alpaca_clientr market_timer rr rcrrr r r r!s    y