""" stream_features.py [SIMULATION-ONLY 2026-09-05] Pure computation layer over stream.py's rolling buffers (1-min bars, 30s-default sub-bars with tick counters, latest quote) producing one normalized StreamFeatures snapshot per symbol. NOT consumed anywhere in the live bot yet -- see config.json's stream_features._note. This is the foundation trend_engine.py / regime_engine.py / prediction_engine.py will build on next. No state, no file I/O, no network calls -- a pure function of whatever bars/bars_sub/quote/prior are handed to it, same contract as intraday_health.compute_health(). Nearly every calculation here is a thin wrapper over indicators.py functions that already exist and are already used elsewhere in this project (vwap, ema_series, atr, rsi, normalized_slope_pct, consolidation_tightness, spread_pct) -- this module's job is assembling them into the richer feature set the prediction engine needs, not inventing new math. TWO DIFFERENT KINDS OF "ACCELERATION" -- deliberately kept separate: 1. Cross-horizon comparison (slope.* below): slope_30s vs slope_1m vs slope_3m vs slope_5m computed from the SAME snapshot of bars, no history required. If shorter horizons show a progressively stronger move in the same direction than longer horizons, that's "accelerating"; the reverse pattern is "decelerating." This is what the project's original design brief's worked examples (5m +0.20%, 3m +0.35%, 1m +0.50%, 30s +0.70% => accelerating) actually describe -- a shape comparison at one instant, not a time derivative. See acceleration.momentum_acceleration. 2. True time-derivative (acceleration.velocity_acceleration / slope_acceleration): this reading's velocity/slope minus the PRIOR reading's, normalized by elapsed time. Needs the caller to hold and pass in the previous StreamFeatures for this symbol (`prior` param) -- this module stays stateless; state ownership lives with the caller, same pattern as intraday_health.compute_health()/update_health_state()'s `persisted` dict. """ from dataclasses import dataclass, field from datetime import datetime, timezone from config_loader import get_config from indicators import ( vwap, vwap_series, vwap_slope, extension_from_vwap_pct, ema_series, atr, rsi, normalized_slope_pct, consolidation_tightness, spread_pct, classify_slope, ) DEFAULT_CONFIG = { "min_bars_1m": 3, "min_bars_sub": 2, "slope_sub_lookback_buckets": 4, "vwap_time_window_bars": 20, "ema9_period": 9, "ema20_period": 20, "ema_slope_lookback_bars": 5, "rsi_period": 14, "atr_period": 14, "range_expansion_lookback_bars": 6, "pressure_velocity_scale_pct_per_sec": 0.05, "pressure_weights": {"uptick": 0.40, "velocity": 0.35, "volume": 0.25}, # [Phase 3] VWAP intelligence -- see _vwap_hold_and_rejections()/ # _classify_vwap() below. "vwap_min_hold_bars_for_support": 5, "vwap_min_rejections_for_failure": 2, "vwap_max_extension_atr": 3.0, "vwap_flat_slope_threshold_pct": 0.02, } @dataclass class StreamFeatures: symbol: str computed_at: datetime insufficient_data: bool = False missing: list = field(default_factory=list) meta: dict = field(default_factory=dict) # e.g. sub_bucket_seconds actually used price: dict = field(default_factory=dict) velocity: dict = field(default_factory=dict) slope: dict = field(default_factory=dict) acceleration: dict = field(default_factory=dict) volume: dict = field(default_factory=dict) vwap: dict = field(default_factory=dict) ema: dict = field(default_factory=dict) volatility: dict = field(default_factory=dict) rsi: dict = field(default_factory=dict) quote: dict = field(default_factory=dict) trade_flow: dict = field(default_factory=dict) pressure: dict = field(default_factory=dict) def _insufficient(symbol: str, missing: list, computed_at: datetime = None) -> StreamFeatures: return StreamFeatures( symbol=symbol, computed_at=computed_at or datetime.now(timezone.utc), insufficient_data=True, missing=missing, ) def _closes(bars: list) -> list: return [b["c"] for b in bars] def _window(bars: list, n_bars: int) -> list: return bars[-n_bars:] if len(bars) >= n_bars else bars def _pct_change(now: float, then: float) -> float: if then == 0: return 0.0 return (now - then) / then * 100.0 def _horizon_slope_pct(bars: list, n_bars: int, bar_seconds: float = 60.0) -> float: """normalized_slope_pct() over the last n_bars closes, converted to a %-per-MINUTE basis via (60 / bar_seconds) -- 'slope over an N-bar window,' the same standard interpretation intraday_health.py already uses for its own slope_lookback_bars (documented there as roughly an N-minute window at 1-min bars). [BUGFIX 2026-09-05] normalized_slope_pct() returns %-change PER BAR, not per unit time -- comparing that raw value between a 60s-bar series and a 30s-bar series is comparing different units (a "slope of 8 per bar" on 30s bars is a FASTER underlying rate than "8 per bar" on 60s bars, since each 30s bar covers half the time). Confirmed live in this module's own testing: an explicitly accelerating synthetic price series produced a NEGATIVE momentum_acceleration (the cross-horizon waterfall read as decelerating) purely because slope_sub (30s bars) was being compared unscaled against slope_1m/ 3m/5m (60s bars). Every horizon must be expressed on the same per-minute basis before any cross-horizon comparison (this function's output, and everything downstream: acceleration.momentum_acceleration here and trend_engine.py's flat-threshold classification) is meaningful. Floors the window at 2 bars: a regression line through a single point is undefined and normalized_slope_pct() correctly returns 0.0 for it per its own guard -- silently reading that 0.0 as "flat" would be indistinguishable from an actual flat reading. n_bars=1 ("slope over the last 1 minute") is inherently meaningless as a fitted line; the last 2 bars is the smallest window that has an actual slope, and for exactly 2 points a least-squares fit reduces to the plain two-point slope anyway, so this doesn't quietly change behavior for any n_bars >= 2 call site.""" window = _window(bars, max(n_bars, 2)) if len(window) < 2: return 0.0 per_bar = normalized_slope_pct(_closes(window)) return per_bar * (60.0 / bar_seconds) def _bucket_seconds_between(bar_a: dict, bar_b: dict) -> float: return max((bar_b["t"] - bar_a["t"]).total_seconds(), 1.0) # --------------------------------------------------------------------------- # [Phase 3] VWAP intelligence -- hold-duration and rejection-count are both # derived purely from the bars window itself (via vwap_series' per-bar # running VWAP), NOT from any new persistent counter. This deliberately # preserves this module's existing stateless contract (the only cross-call # state anywhere in this file is the single `prior` reading) rather than # adding a second, differently-shaped kind of caller-held history just for # these two fields. # --------------------------------------------------------------------------- def _vwap_hold_and_rejections(bars: list, window_bars: int) -> tuple: """ hold_duration_bars: current consecutive streak of closes above the running VWAP, counted backward from the latest bar (0 if the latest close is at/below VWAP right now). rejection_count: within the last `window_bars`, how many bars had their HIGH reach or cross the running VWAP but still CLOSED below it -- an intrabar rejection, distinct from simply trading below VWAP the whole bar. """ series = vwap_series(bars) hold_duration = 0 for i in range(len(bars) - 1, -1, -1): if bars[i]["c"] > series[i]: hold_duration += 1 else: break start = max(0, len(bars) - window_bars) rejection_count = sum( 1 for i in range(start, len(bars)) if bars[i]["c"] < series[i] <= bars[i]["h"] ) return hold_duration, rejection_count VWAP_STRONG_SUPPORT = "STRONG_VWAP_SUPPORT" VWAP_RECLAIM = "VWAP_RECLAIM" VWAP_HOLD = "VWAP_HOLD" VWAP_EXTENSION = "VWAP_EXTENSION" VWAP_FAILURE = "VWAP_FAILURE" VWAP_FLAT = "VWAP_FLAT" VWAP_DECLINING = "VWAP_DECLINING" def _classify_vwap(above: bool, slope_class: str, hold_duration: int, rejection_count: int, extended: bool, cfg: dict) -> str: """ Exhaustive, priority-ordered, mutually exclusive classification -- see module docstring / config.json's stream_features._note_vwap_classes for the full reasoning behind each branch. A symbol repeatedly holding above a positively-sloped VWAP (STRONG_VWAP_SUPPORT) should read as meaningfully more confident than one merely above a flat VWAP this instant (VWAP_HOLD) -- that's the whole point of this classification existing on top of the raw price_vs_vwap_pct/slope fields. """ if above and extended: return VWAP_EXTENSION if above and slope_class == "negative": # Price is still above VWAP, but the line itself is now dropping -- # an early warning distinct from FAILURE (price hasn't lost VWAP # yet) and distinct from ordinary HOLD (the reference is weakening). return VWAP_DECLINING if above and hold_duration >= cfg["vwap_min_hold_bars_for_support"]: return VWAP_STRONG_SUPPORT if above and hold_duration > 0: return VWAP_RECLAIM if above: return VWAP_HOLD if slope_class == "negative" or rejection_count >= cfg["vwap_min_rejections_for_failure"]: return VWAP_FAILURE return VWAP_FLAT def compute_features(symbol: str, bars: list, bars_sub: list, quote, avg_vol_baseline: float = None, prior: "StreamFeatures" = None, sub_bucket_seconds: int = 30, as_of: datetime = None, cfg: dict = None) -> StreamFeatures: """ bars: 1-min bars, oldest first (stream.StreamManager.get_bars()). bars_sub: sub-minute bars, oldest first, forming bucket included (stream.StreamManager.get_bars_sub()) -- each carries tick_count/upticks/downticks alongside o/h/l/c/v. quote: (bid, ask, timestamp) or None. avg_vol_baseline: same "expected normal volume" baseline entry_engine/intraday_health already take -- used for relative_volume-style comparisons. None is handled (volume.relative_volume simply omitted from output). prior: this symbol's previous StreamFeatures reading, or None on the first call -- see module docstring's two-kinds-of- acceleration note. Caller-held; this function never stores anything itself. sub_bucket_seconds: must match whatever streaming. sub_minute_bucket_seconds actually is, so time-normalized calculations (velocity, acceleration) use the real elapsed time rather than an assumed 30. as_of: [FEATURE 2026-09-05] the moment this reading is "as of," used for computed_at (and therefore the elapsed-time base for acceleration vs `prior`). Defaults to real wall-clock time (datetime.now()), correct for live use where a bar really does arrive at roughly the same real-world instant it's timestamped. A REPLAY/BACKTEST caller MUST pass the bar's own timestamp here instead -- otherwise computed_at would be the real time the backtest script happens to execute, completely unrelated to the simulated market time being replayed, which would corrupt every acceleration calculation that compares consecutive readings' elapsed time. """ cfg = {**DEFAULT_CONFIG, **(cfg or get_config().get("stream_features", {}))} now_ts = as_of or datetime.now(timezone.utc) missing = [] if len(bars) < cfg["min_bars_1m"]: return _insufficient(symbol, ["price", "velocity", "slope", "volume", "vwap", "ema", "volatility", "rsi", "trade_flow"], computed_at=now_ts) now_bar = bars[-1] last_price = now_bar["c"] # ---------------- price ---------------- price = {"last": last_price} for label, n in (("price_1m", 1), ("price_3m", 3), ("price_5m", 5)): price[label] = bars[-1 - n]["c"] if len(bars) > n else None if price[label] is None: missing.append(label) if len(bars_sub) >= 2: price["price_sub"] = bars_sub[-2]["c"] else: price["price_sub"] = None missing.append("price_sub") # ---------------- velocity (% per second) ---------------- velocity = {} if len(bars) >= 2: prev_bar = bars[-2] elapsed = _bucket_seconds_between(prev_bar, now_bar) velocity["velocity_1m"] = _pct_change(last_price, prev_bar["c"]) / elapsed else: velocity["velocity_1m"] = None missing.append("velocity_1m") if len(bars_sub) >= 2: prev_sub = bars_sub[-2] elapsed_sub = max(sub_bucket_seconds, 1) velocity["velocity_sub"] = _pct_change(last_price, prev_sub["c"]) / elapsed_sub else: velocity["velocity_sub"] = None missing.append("velocity_sub") # ---------------- slope: cross-horizon, stateless ---------------- slope = { "slope_1m": _horizon_slope_pct(bars, 1), "slope_3m": _horizon_slope_pct(bars, 3), "slope_5m": _horizon_slope_pct(bars, 5), } if len(bars_sub) >= cfg["slope_sub_lookback_buckets"]: per_bucket = normalized_slope_pct( _closes(_window(bars_sub, cfg["slope_sub_lookback_buckets"]))) slope["slope_sub"] = per_bucket * (60.0 / max(sub_bucket_seconds, 1)) else: slope["slope_sub"] = None missing.append("slope_sub") # ---------------- acceleration ---------------- # (1) cross-horizon waterfall -- see module docstring. Positive means # shorter horizons are moving further/faster in the SAME direction as # the longest horizon (accelerating); negative means shorter horizons # have already weakened relative to it (decelerating), independent of # whether the longer-horizon trend itself is still positive. horizon_chain = [slope["slope_5m"], slope["slope_3m"], slope["slope_1m"]] if slope["slope_sub"] is not None: horizon_chain.append(slope["slope_sub"]) long_horizon_sign = 1 if horizon_chain[0] > 0 else (-1 if horizon_chain[0] < 0 else 0) momentum_acceleration = (horizon_chain[-1] - horizon_chain[0]) * long_horizon_sign \ if long_horizon_sign != 0 else (horizon_chain[-1] - horizon_chain[0]) acceleration = {"momentum_acceleration": round(momentum_acceleration, 4)} # (2) true time-derivative vs the caller-supplied prior reading. # Uses now_ts (this call's as_of/computed_at), NOT now_bar["t"] # directly -- they coincide when the caller passes as_of=bar["t"] # (the intended backtest usage), but now_ts is the semantically # correct "as of" reference for this reading either way. if prior is not None and not prior.insufficient_data: elapsed = max((now_ts - prior.computed_at).total_seconds(), 1.0) \ if isinstance(prior.computed_at, datetime) else 1.0 v_now = velocity.get("velocity_sub") v_prev = prior.velocity.get("velocity_sub") if prior.velocity else None if v_now is not None and v_prev is not None: acceleration["velocity_acceleration"] = round((v_now - v_prev) / elapsed, 6) else: acceleration["velocity_acceleration"] = None s_now = slope.get("slope_1m") s_prev = prior.slope.get("slope_1m") if prior.slope else None if s_now is not None and s_prev is not None: acceleration["slope_acceleration"] = round((s_now - s_prev) / elapsed, 6) else: acceleration["slope_acceleration"] = None else: acceleration["velocity_acceleration"] = None acceleration["slope_acceleration"] = None # ---------------- volume ---------------- volume = { "volume_1m": now_bar["v"], "volume_3m": sum(b["v"] for b in _window(bars, 3)), "volume_5m": sum(b["v"] for b in _window(bars, 5)), } if len(bars_sub) >= 1: volume["volume_sub"] = bars_sub[-1]["v"] else: volume["volume_sub"] = None vols_1m = [b["v"] for b in bars] if len(vols_1m) >= 4: mid = len(vols_1m) // 2 early = sum(vols_1m[:mid]) / mid or 1e-9 recent = sum(vols_1m[mid:]) / (len(vols_1m) - mid) volume["volume_acceleration"] = round(recent / early, 3) else: volume["volume_acceleration"] = None missing.append("volume_acceleration") if avg_vol_baseline: volume["relative_volume"] = round(sum(vols_1m) / avg_vol_baseline, 3) else: volume["relative_volume"] = None # ---------------- VWAP ---------------- v_wap = vwap(bars) v_slope = vwap_slope(bars, lookback=min(3, len(bars) - 1)) if len(bars) > 1 else 0.0 window_bars = _window(bars, cfg["vwap_time_window_bars"]) time_above = sum(1 for b in window_bars if b["c"] > v_wap) / len(window_bars) * 100.0 vwap_out = { "value": round(v_wap, 4), "slope": round(v_slope, 4), "price_vs_vwap_pct": round(extension_from_vwap_pct(last_price, v_wap), 3), "time_above_pct": round(time_above, 1), "time_below_pct": round(100.0 - time_above, 1), } # ---------------- EMA ---------------- closes_all = _closes(bars) ema9_series = ema_series(closes_all, cfg["ema9_period"]) ema20_series = ema_series(closes_all, cfg["ema20_period"]) ema9_now = ema9_series[-1] if ema9_series else None ema_lookback = min(cfg["ema_slope_lookback_bars"], len(ema9_series)) ema9_slope = normalized_slope_pct(ema9_series[-ema_lookback:]) if ema_lookback >= 2 else 0.0 ema_out = { "ema9": round(ema9_now, 4) if ema9_now is not None else None, "ema20": round(ema20_series[-1], 4) if ema20_series else None, "ema9_slope": round(ema9_slope, 4), "price_vs_ema9_pct": round(_pct_change(last_price, ema9_now), 3) if ema9_now else None, } # ---------------- volatility ---------------- a = atr(bars, period=min(cfg["atr_period"], max(2, len(bars) - 1))) atr_pct = (a / last_price * 100.0) if last_price else 0.0 tightness_recent = consolidation_tightness(bars, lookback=min(3, len(bars))) tightness_lookback = cfg["range_expansion_lookback_bars"] tightness_earlier = consolidation_tightness( bars[:-min(3, len(bars))] or bars, lookback=tightness_lookback) range_expansion = (tightness_earlier > 0 and (tightness_recent - tightness_earlier) / tightness_earlier * 100.0) volatility = { "atr": round(a, 4), "atr_pct": round(atr_pct, 3), "vwap_distance_atr": round((last_price - v_wap) / a, 3) if a else None, "range_expansion_pct": round(range_expansion, 1) if range_expansion is not False else None, } ema_out["price_vs_ema9_atr"] = (round((last_price - ema9_now) / a, 3) if (a and ema9_now is not None) else None) # ---------------- VWAP intelligence (hold/reject/classification) ---------------- # [Phase 3] See _vwap_hold_and_rejections()/_classify_vwap() above -- # both derived purely from `bars` + the running vwap_series(), no new # persistent state. Placed after volatility (not inside the VWAP block # above) because the extension check needs vwap_distance_atr, computed # just above. # [Phase 3] v_slope above is a raw price delta (vwap_slope()'s own # docstring: "series[-1] - series[-1-lookback]"), not a percentage -- # normalize it against the VWAP level itself before classifying, so # this comparison is apples-to-apples across stocks at different price # levels the same way normalized_slope_pct() already does for price/ # momentum slopes elsewhere in this project. hold_duration, rejection_count = _vwap_hold_and_rejections(bars, cfg["vwap_time_window_bars"]) vwap_slope_pct = (v_slope / v_wap * 100.0) if v_wap else 0.0 slope_class = classify_slope(vwap_slope_pct, cfg["vwap_flat_slope_threshold_pct"]) vwap_atr_dist = volatility.get("vwap_distance_atr") extended = vwap_atr_dist is not None and abs(vwap_atr_dist) > cfg["vwap_max_extension_atr"] vwap_out["hold_duration_bars"] = hold_duration vwap_out["rejection_count"] = rejection_count vwap_out["classification"] = _classify_vwap( above=last_price > v_wap, slope_class=slope_class, hold_duration=hold_duration, rejection_count=rejection_count, extended=extended, cfg=cfg) # ---------------- RSI ---------------- rsi_period = cfg["rsi_period"] rsi_value = rsi(bars, period=rsi_period) rsi_has_full_history = len(bars) >= rsi_period + 1 if not rsi_has_full_history: missing.append("rsi (insufficient history -- neutral 50.0 default, not a real reading)") rsi_out = {"value": round(rsi_value, 1), "has_full_history": rsi_has_full_history} # ---------------- quote ---------------- if quote: bid, ask, _ts = quote quote_out = { "bid": bid, "ask": ask, "spread": round(ask - bid, 4), "spread_pct": round(spread_pct(bid, ask), 3), } else: quote_out = {"bid": None, "ask": None, "spread": None, "spread_pct": None} missing.append("quote") # ---------------- trade flow ---------------- if bars_sub: current_sub = bars_sub[-1] tick_count = current_sub.get("tick_count", 0) upticks = current_sub.get("upticks", 0) downticks = current_sub.get("downticks", 0) trade_flow = { "tick_count_sub": tick_count, "trades_per_second": round(tick_count / max(sub_bucket_seconds, 1), 3), "uptick_ratio": round(upticks / tick_count, 3) if tick_count else None, "downtick_ratio": round(downticks / tick_count, 3) if tick_count else None, } else: trade_flow = {"tick_count_sub": None, "trades_per_second": None, "uptick_ratio": None, "downtick_ratio": None} missing.append("trade_flow") # ---------------- pressure (-100 strong selling .. +100 strong buying) ---------------- # [FEATURE 2026-09-05] Not a separate module -- the project's original # design brief describes this as its own "engine" (section 13) but # section 5's actual recommended-module list has no pressure_engine.py, # meaning it's meant to live alongside the other stream-derived # features (velocity, volume) it's built from, not as a standalone # file. Three components, each normalized to roughly [-1, 1] then # blended by pressure_weights: # uptick (uptick_ratio - downtick_ratio) from trade_flow -- direct # measure of buy vs sell print dominance this sub-bucket. # velocity very recent price velocity (velocity_sub) scaled against # pressure_velocity_scale_pct_per_sec (what counts as # "fast" for this instrument) -- a configurable reference, # not a universal constant. # volume volume_acceleration centered on 1.0 (no change). # Missing components are simply excluded from the weighted average # (re-normalized over whatever's available) rather than treated as 0, # so a quiet sub-bucket with no trade_flow data doesn't silently drag # the score toward neutral. pw = cfg["pressure_weights"] components = {} uptick_r, downtick_r = trade_flow.get("uptick_ratio"), trade_flow.get("downtick_ratio") if uptick_r is not None and downtick_r is not None: components["uptick"] = max(-1.0, min(1.0, uptick_r - downtick_r)) v_sub = velocity.get("velocity_sub") if v_sub is not None: scale = cfg["pressure_velocity_scale_pct_per_sec"] or 1e-9 components["velocity"] = max(-1.0, min(1.0, v_sub / scale)) vol_accel = volume.get("volume_acceleration") if vol_accel is not None: components["volume"] = max(-1.0, min(1.0, vol_accel - 1.0)) if components: weight_sum = sum(pw[k] for k in components) pressure_score = round(sum(components[k] * pw[k] for k in components) / weight_sum * 100.0, 1) else: pressure_score = None missing.append("pressure") pressure_slope = None if pressure_score is not None and prior is not None and not prior.insufficient_data and prior.pressure: prior_score = prior.pressure.get("score") if prior_score is not None: pressure_slope = round(pressure_score - prior_score, 2) pressure_out = {"score": pressure_score, "slope": pressure_slope} return StreamFeatures( symbol=symbol, computed_at=now_ts, insufficient_data=False, missing=missing, meta={"sub_bucket_seconds": sub_bucket_seconds}, price=price, velocity=velocity, slope=slope, acceleration=acceleration, volume=volume, vwap=vwap_out, ema=ema_out, volatility=volatility, rsi=rsi_out, quote=quote_out, trade_flow=trade_flow, pressure=pressure_out, )