o yjmp@sdZddlZddlZddlZddlmZmZmZddlmZm Z m Z ddl m Z ddl mZddlmZddlmZed 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 timedelta) get_config) get_logger) get_client)session_open_dtstreamc@seZdZdZd:ddZdZdedefd d Zd d Z d dZ d;defddZ d 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|_ t |_ t|_ d|_t|_dSNr)maxlenrbars latest_quotelatest_trade_price last_update _current_barsub_bucket_seconds sub_maxlenbars_sub_current_bar_subr _trade_logmin_flow_flow_last_side threadingRLock_lock)selfrrrr"!/var/www/screener/trade/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)r!r&bidask_midr"r"r#_classify_tradems   zSymbolBuffer._classify_tradecCsZ|t|jd}|jr'|jdd|kr+|j|jr)|jdd|ksdSdSdSdS)Nsecondsr)r_TRADE_LOG_MAX_AGE_SECONDSrpopleft)r!now_tscutoffr"r"r#_trim_trade_logs (zSymbolBuffer._trim_trade_logcCsd\}}}|j|D]\}}}}|dkr||f}q ||} } d} |r_|dr_|dr_|d|dkr9dkr_nn$|\} } | | d}| | krKdn| | krQdn | |krWdn| |kr]dnd} | dkru|duru| |krmdn| |krsdn|} | | }}|jddd}|j|}|durd d g}|j|<| dkr|d| 7<q | dkr|d| 7<q tt|jd d|_||_WddS1swYdS) a[2026-09-30] Pre-load min_flow from historical trades/quotes (a stock added mid-day by a rescan had only seconds of buy/sell data: SSL 9/30 was bought 16 s after it was added on a '100% buy share'). events: [(ts, "trade"|"quote", a, b)] sorted by time. Same classification as live (quote rule, tick-rule fallback); only min_flow is touched -- bars/volume already come from the official backfill.)NNrquoterr(r*r)Nsecond microsecondi)r replacergetrsorteditemsr)r!eventsqlast_p last_sidetskindabr&sizesider+r,r.mbfr"r"r# backfill_flows6  4 4  "zSymbolBuffer.backfill_flowminutescCsb|jt|j| d}Wdn1swYtdd|Dtdd|DfS)z^(buy_vol, sell_vol) over the last `minutes` minute buckets (incl. the forming minute).Ncs|]}|dVqdS)rNr".0xr"r"r# z(SymbolBuffer.get_flow..csrO)r(Nr"rPr"r"r#rSrT)r listrvaluessum)r!rNr?r"r"r#get_flows$zSymbolBuffer.get_flow4@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)rr0r;)r rrreversed) r!rZr4r5buy_volsell_volrDrIrHtotalr"r"r#get_trade_imbalances$  z SymbolBuffer.get_trade_imbalancebarcCs|j7|d}||j|<|j|t|j|jkr2|jjddt|j|jksWddSWddS1s=wYdSNtFlast)r r move_to_endlenrpopitemr!r`keyr"r"r# _upsert_bars  "zSymbolBuffer._upsert_barcCsT|d}||j|<|j|t|j|jkr(|jjddt|j|jksdSdSra)rrerfrrgrhr"r"r#_upsert_bar_subs   zSymbolBuffer._upsert_bar_subcCs"|j}|j||}|j|ddS)Nrr8)rr9r<)r!rDwidth 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. Nrbr) rbohlcv tick_countupticks downticksrprqrrrsrtr(rurv)rnrrkmaxmin)r!r&rHrD prev_pricebucket is_new_bucketrGr"r"r#_accumulate_bar_subs,    z SymbolBuffer._accumulate_bar_subTcCx|j/t|j}|r"|jdur*|t|jg}Wd|SWd|SWd|S1s5wY|SN)r rUrrVrdictr!include_formingrr"r"r# get_bars_sub   zSymbolBuffer.get_bars_subrHcC:|j||||WddS1swYdSr~)r _on_trade_locked)r!r&rHrDr"r"r#on_trade"zSymbolBuffer.on_tradec Cs(|j}||_||_|||||||||||}|j|||f|||}|dkrD|durD||kr;dn||krAdn|j}||_|j ddd}|j |}|durvddg}|j |<t |j dkrv|j j ddt |j dksh|dkr|d|7<dS|dkr|d|7<dSdS) Nrr(r)r8r;<Frc)rr_accumulate_barr|r/rappendr6rr<rr=rfrg) r!r&rHrDryrIfsiderJrKr"r"r#rs0   zSymbolBuffer._on_trade_lockedr+r,cCrr~)r _on_quote_lockedr!r+r,rDr"r"r#on_quoterzSymbolBuffer.on_quotecCs|||f|_||_dSr~)rrrr"r"r#rs  zSymbolBuffer._on_quote_lockedc Cs@|j|||||||WddS1swYdS)zBDirect minute-bar ingestion (Alpaca also streams aggregated bars).N)r _on_bar_locked)r!rorprqrrrsrDr"r"r#on_bar"s"zSymbolBuffer.on_barc Csdt|dr |jdddn|}|||||||d||_|jdur.|jd|kr0d|_dSdSdS)Nr<rr8rbrorprqrrrsrb)hasattrr<rjrr)r!rorprqrrrsrD minute_bucketr"r"r#r's  zSymbolBuffer._on_bar_lockedcCs|jddd}|jdus|jd|kr*|jdur||j||||||d|_dS|j}t|d||d<t|d||d<||d<|d|7<dS) Nrr8rbrrprqrrrs)r<rrjrwrx)r!r&rHrDrrGr"r"r#r2s    zSymbolBuffer._accumulate_barcCr}r~)r rUrrVrrrr"r"r#get_barsArzSymbolBuffer.get_bars stale_secondscCsH|jdurdSttj}|j}|jdur|jtjd}|||kS)NT)tzinfo)rrnowrutcrr< total_seconds)r!rrrdr"r"r#is_staleHs   zSymbolBuffer.is_staleN)rrrrMrY)T)__name__ __module__ __qualname____doc__r$r2floatintr/r6rLrXr_rrjrkrnr|rUrrrrrrrrrboolrr"r"r"r#r s. F # r c@seZdZddZd-dedefddZdefdd Zdefd d Zd d Z defddZ defddZ ddZ ddZ ddZdefddZdedefddZdedefddZdefd d!Zd.ded#efd$d%Zd-dedefd&d'Zdedefd(d)Zdefd*d+Zd,S)/ StreamManagercstd|_t|_|jdd|jdd}tdt|dtfdd |_t |_ d|_ t |_ d|_ d|_d|_t|_t|_dS) N streamingsub_minute_bucket_secondsrsub_minute_buffer_minutesrMr(rcs tdS)Nrr)r r"rr"r#Zs z(StreamManager.__init__..)rcfgr clientr=rwrrbuffersset _subscribedshadow bars_only_stream_thread_looprEvent _stop_event _connected)r!sub_buffer_minutesr"rr#r$Ss   zStreamManager.__init__rMsymbolsrNc Cs"ttj}ttj}||krdSt||t|d}|D]o}zP|j |||}|j |||}t dd|Ddd|Dddd} |j | | |j ||\} } td |d t|d |d | d d| d  Wqty} ztd|d | WYd} ~ qd} ~ wwdS)z[2026-09-30] Last `minutes` of trades + quotes from Alpaca -> buy/sell per minute (SymbolBuffer.backfill_flow), for symbols subscribed after the open.N)rNcS$g|]}|dd|d|dfqS)rbtradepsr")rQrbr"r"r# w$z0StreamManager._backfill_flow..cSr)rbr7rGrFr"rPr"r"r#rxrcSs|dSrr")er"r"r#rxsz.StreamManager._backfill_flow..)riz[STREAM] Flow backfilled : z trades over z min, buy z.0fz / sell z"[STREAM] Flow backfill failed for )rrrrr astimezonerwrrget_historical_tradesget_historical_quotesr>rrLrXloginforf Exceptionwarning) r!rrNropen_utcstartsymboltradesquotesevbvsvrr"r"r#_backfill_flowks4   "zStreamManager._backfill_flowc s^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|jdurzddlm|j|fd d |DWn tyYnwt 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)rendz[STREAM] Backfill failed for rrr8rBarc s>g|]}|jt|jt|jt|jt|jt|jqSr") timestampropenhighlowclosevolume)rQrG_SBarr"r#rs z+StreamManager._backfill..z[STREAM] Backfilled z with z historical bars (session open $z.4fz) before live subscription)r rrrrrrget_minute_barsrrrrrr<rjrrrrrrr entry_rulesr seed_barsrrf) r!ropen_dtrrrrbufr`rr"rr# _backfillsD       zStreamManager._backfillcsT|fdd|Dtjj|fdd_jjjdddS)Ncg|] }|jvr|qSr"rrPr!r"r#rz'StreamManager.start..T)targetargsdaemon )timeout) rrrThread _run_foreverrrrwait)r!rr"rr#rs  zStreamManager.startcCsN|j|jr#|jr%z t|j|jWdSty"YdSwdSdSr~)rrrrasynciorun_coroutine_threadsafestop_wsrrr"r"r#stops   zStreamManager.stopc sfdd|D}|rOjdurOjt|8_zjjjg|Rjjjg|RWntyN}zt d|d|WYd}~nd}~wwfdd|D}|r_jduradS | |j |z1jjjg|Rjjjg|Rjjjg|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")rrrQrrr"r#rrz-StreamManager.add_symbols..Nz#[STREAM] tick subscribe failed for rcrr"rrrr"r#rrz[STREAM] Subscribed z added symbols: z [STREAM] add_symbols failed for )rrrsubscribe_trades _on_tradesubscribe_quotes _on_quoterrrrrrupdatesubscribe_bars_on_barrrf)r!rpromoternew_symsr"rr# add_symbolss2 "    $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_barsrrrrpoprrf)r!rgonersymr"rr#remove_symbolss     "zStreamManager.remove_symbolscsv|j|j}|t|jt|j|j|jdur9z|j|j|jt|jt|jWdSty8YdSwdSr~) rrrrr&rHrrr)r!rrr"r"r#rs  ( zStreamManager._on_tradec s|j|j}|jrK|jrM|t|jt|j|j|jdurOz |j|j|jt|jt|jt|jp5dt|j p;dWdSt yJYdSwdSdSdSr) rr bid_price ask_pricerrrrbid_sizeask_sizer)r!r7rr"r"r#rs     zStreamManager._on_quotec s|j|j}|t|jt|jt|jt|jt|j|j |j durYz(ddl m }|j |j||j t|jt|jt|jt|jt|jWdSt yXYdSwdS)Nrr)rrrrrrrrrrrrrr)r!r`rrr"r"r#rs     zStreamManager._on_barc 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"rrPrr"r#r+rz.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_streamrrrr>rrrrrrrrrfrrunrrxrtimesleeperror) r!rbackoffs max_attemptsattemptcurrentticksrdelayr"rr#rsJ           zStreamManager._run_foreverrr'cC|j|Sr~)rrr!rr"r"r#rCszStreamManager.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.)rrrr"r"r#rFszStreamManager.get_bars_subcCs |j|jSr~)rrrr"r"r# get_quoteLs zStreamManager.get_quoterYrZcC|j||Sr~)rr_)r!rrZr"r"r#r_Osz!StreamManager.get_trade_imbalancecCr)zF(buy_vol, sell_vol) over the last `minutes` minutes (entry_v2 step 3).)rrX)r!rrNr"r"r#rXRszStreamManager.get_flowcCs|j||jdS)Nstale_data_seconds)rrrrr"r"r#is_symbol_staleVzStreamManager.is_symbol_stalecCs|jo |j Sr~)rrrrr"r"r# is_healthyYr"zStreamManager.is_healthyNrr)rrrr$rUrrrrrrrrrrrstrrrrrr_rXrr!r#r"r"r"r#rRs&6    (r)rrrr collectionsrrrrrr config_loaderr logger_setupr alpaca_clientr market_timer rr rr"r"r"r#s    5