""" intraday_health.py [FEATURE 2026-08-17] Continuously evaluates every symbol in premarket_20 during the trading day to answer one question: "is this stock still behaving like a healthy intraday candidate?" -- independent of whether the bot currently holds a position in it. CRITICAL SEPARATION OF RESPONSIBILITIES (do not blur this): Health state's PRIMARY job is still to gate which symbols are ELIGIBLE to be picked as a NEW entry or slot-replacement candidate (see top_stocks.find_replacement() and monitor.py's _scan_for_entries()) -- is_eligible_for_entry() and is_confirmed_fading() (consulted by position_manager.py's trailing- stop fade-confirmation check, but only AT the moment price actually touches the stop) are unaffected by the exception below. [FEATURE 2026-09-11] One deliberate, narrow exception: an OPEN position whose health has been confirmed deteriorating (raw state STALE or worse) for intraday_health.exit_confirm_reads consecutive evaluations in a row is force-closed by monitor.py's _update_intraday_health(), independent of whether price has ever touched the trailing stop. Added live, same day, after TJGC sat open 3+ hours with a degrading, never-recovering STALE health score while price stayed inside a tight range and never threatened the stop -- the fade-confirmation path only ever runs AT a stop touch, so a position that just goes quiet and stale without ever approaching the stop had no exit path at all. See should_force_exit_on_deterioration() below; gated by intraday_health.exit_on_deterioration_enabled (default on). Health states, best to worst: HEALTHY - positive/supported slope, volume participation, healthy structure (higher highs / higher lows) WATCH - has weakened but not yet confirmed stale/unhealthy STALE - flat/low-participation; NOT necessarily unhealthy, just quiet (a stock can stay healthy while consolidating). [2026-08-20] Also reached when price_slope is still positive but momentum is decelerating AND at least one other weak signal corroborates (volume declining, lower-highs/lower-lows, or price already lost VWAP) -- a fading bounce that hasn't fully broken down yet, but isn't confirming fresh strength either. UNHEALTHY - negative slope + VWAP lost + deteriorating structure (lower highs / lower lows) REMOVED - confirmed UNHEALTHY for enough consecutive evaluations in a row; dropped from the active candidate pool until the next full rescan brings a fresh premarket_20 State transitions require hysteresis (config: intraday_health.*) so a single noisy bar can't flip a symbol's state back and forth. A "severe" reading (see _is_severe()) is the one deliberately-documented exception: it can demote a symbol to UNHEALTHY immediately, skipping the normal confirm-count requirement, but it still cannot jump straight to REMOVED without going through the same consecutive-unhealthy count everyone else does. """ from dataclasses import dataclass, field from datetime import datetime, timezone from config_loader import get_config from logger_setup import get_logger from indicators import ( vwap, vwap_series, relative_volume, volume_acceleration, is_higher_highs_higher_lows, is_lower_highs_lower_lows, consolidation_tightness, distance_from_high_pct, atr, normalized_slope_pct, classify_slope, ) import data_store import market_time log = get_logger("intraday_health") STATES_BEST_TO_WORST = ["HEALTHY", "WATCH", "STALE", "UNHEALTHY", "REMOVED"] @dataclass class HealthReading: symbol: str health_score: float raw_state: str # this evaluation's un-confirmed read confirmed_state: str # the persisted, hysteresis-confirmed state price_slope_pct: float vwap_slope_pct: float momentum_slope_pct: float price_slope_class: str vwap_slope_class: str momentum_slope_class: str breakdown: dict = field(default_factory=dict) flags: dict = field(default_factory=dict) def _clamp(x, lo=0.0, hi=100.0): return max(lo, min(hi, x)) def _momentum_series(closes: list) -> list: """Bar-over-bar % returns -- the input to the momentum-slope regression. Whether *momentum itself* is accelerating or decaying is a different question from whether price is rising, which is exactly why this gets its own slope rather than reusing price.""" out = [] for i in range(1, len(closes)): prev = closes[i - 1] if prev: out.append((closes[i] - prev) / prev * 100.0) return out def _is_severe(price_slope_pct: float, price_above_vwap: bool, lower_hl: bool, cfg: dict) -> bool: """ A deliberately narrow "skip the confirm count" exception, per the project's explicit instruction: single-bad-tick churn should be avoided EXCEPT for a genuinely severe risk condition. Severe here means: sharply negative regression slope, price already lost VWAP, AND the bar structure itself is confirming lower-highs/lower-lows -- three independent signals agreeing, not just one noisy print. """ return (price_slope_pct <= -cfg["severe_negative_slope_pct"] and not price_above_vwap and lower_hl) def compute_health(symbol: str, bars: list, avg_vol_baseline: float) -> HealthReading: """ Pure evaluation function -- no state, no hysteresis, no file I/O. Given a symbol's recent intraday bars, returns this instant's raw health reading. update_health_state() is what applies hysteresis on top of this to produce the persisted, confirmed state. """ cfg = get_config()["intraday_health"] lookback = min(cfg["slope_lookback_bars"], len(bars)) if len(bars) < 3: return HealthReading( symbol=symbol, health_score=0.0, raw_state="STALE", confirmed_state="STALE", price_slope_pct=0.0, vwap_slope_pct=0.0, momentum_slope_pct=0.0, price_slope_class="flat", vwap_slope_class="flat", momentum_slope_class="flat", flags={"insufficient_data": True}, ) window = bars[-lookback:] closes = [b["c"] for b in window] volumes = [b["v"] for b in window] current_price = closes[-1] v_wap = vwap(bars) vwap_ser = vwap_series(bars)[-lookback:] price_slope_pct = normalized_slope_pct(closes) vwap_slope_pct = normalized_slope_pct(vwap_ser) momentum_slope_pct = normalized_slope_pct(_momentum_series(closes)) if len(closes) >= 3 else 0.0 flat_thresh = cfg["flat_slope_threshold_pct"] price_slope_class = classify_slope(price_slope_pct, flat_thresh) vwap_slope_class = classify_slope(vwap_slope_pct, flat_thresh) momentum_slope_class = classify_slope(momentum_slope_pct, flat_thresh) price_above_vwap = current_price > v_wap struct_lookback = min(cfg["structure_lookback_bars"], len(bars)) hh_hl = is_higher_highs_higher_lows(bars, lookback=struct_lookback) lower_hl = is_lower_highs_lower_lows(bars, lookback=struct_lookback) total_volume = sum(volumes) rvol = relative_volume(total_volume, avg_vol_baseline) vol_accel = volume_acceleration(volumes) recent_high = max(b["h"] for b in bars) dist_from_high = distance_from_high_pct(current_price, recent_high) tight_recent = consolidation_tightness(bars, lookback=max(3, lookback // 2)) tight_earlier_bars = bars[:-max(3, lookback // 2)] or bars tight_earlier = consolidation_tightness(tight_earlier_bars, lookback=max(3, lookback // 2)) range_contracting = (tight_earlier > 0 and (tight_earlier - tight_recent) / tight_earlier * 100.0 >= cfg["stale_range_contraction_pct"]) a = atr(bars, period=min(cfg["atr_period"], max(2, len(bars) - 1))) # ---- breakdown (transparent, like scorer.py's pattern) ---- w = cfg["weights"] breakdown = {} breakdown["price_slope"] = w["price_slope"] * ( 1.0 if price_slope_class == "positive" else cfg["price_slope_flat_credit"] if price_slope_class == "flat" else cfg["price_slope_negative_credit"]) vwap_support = price_above_vwap and vwap_slope_class in ("positive", "flat") breakdown["vwap_support"] = w["vwap_support"] * ( 1.0 if (price_above_vwap and vwap_slope_class == "positive") else cfg["vwap_support_partial_credit"] if vwap_support else cfg["vwap_support_weak_credit"]) volume_declining = vol_accel < cfg["stale_volume_decline_ratio"] rvol_target = cfg["volume_participation_rvol_target"] breakdown["volume_participation"] = w["volume_participation"] * _clamp( (rvol / rvol_target) if not volume_declining else (rvol / rvol_target) * cfg["volume_declining_score_multiplier"], 0, 1) if hh_hl: structure_credit = 1.0 elif lower_hl: structure_credit = 0.0 else: structure_credit = cfg["structure_mixed_credit"] breakdown["structure"] = w["structure"] * structure_credit breakdown["range_expansion"] = w["range_expansion"] * ( cfg["range_contracting_score_multiplier"] if range_contracting else 1.0) max_dist = cfg["max_distance_from_recent_high_pct"] breakdown["distance_from_recent_high"] = w["distance_from_recent_high"] * _clamp( (max_dist - dist_from_high) / max_dist, 0, 1) total = sum(breakdown.values()) max_possible = sum(w.values()) health_score = round(_clamp(total / max_possible * 100.0, 0, 100), 2) momentum_fading = momentum_slope_class == "negative" flags = { "price_above_vwap": price_above_vwap, "higher_highs_higher_lows": hh_hl, "lower_highs_lower_lows": lower_hl, "volume_declining": volume_declining, "range_contracting": range_contracting, "rvol": round(rvol, 2), "atr": round(a, 4), "no_new_highs": dist_from_high > cfg["no_new_highs_threshold_pct"], "momentum_fading": momentum_fading, } # ---- raw state classification ---- # # [FEATURE 2026-08-20] momentum_slope_pct/momentum_slope_class were # computed above (and stored on every HealthReading) since this # module's introduction, but never actually consulted here -- price # direction and the RATE of that price movement are different # questions, and a symbol can keep a positive price_slope for # several bars while its underlying momentum is already # decelerating. That's exactly the pattern behind QUBT's two real # losing entries on 2026-08-20 (09:33 and 09:45): both confirmed # with price_slope=positive (one HEALTHY, one WATCH) and both # stopped out within minutes. # # Two changes: # 1. HEALTHY now additionally requires momentum not be fading -- # a positive price slope propped up by decelerating momentum # doesn't earn the top tier. # 2. A new, deliberately corroborated-only branch: fading # momentum ALONE does not downgrade a reading (single-signal # noise is exactly what this module's hysteresis and # "_is_severe" pattern already goes out of its way to avoid -- # see this module's docstring). It only reclassifies to STALE # when at least one other independent weak signal agrees: # volume declining, lower-highs/lower-lows structure, or price # having already lost VWAP. This mirrors the existing UNHEALTHY # branch's require-multiple-signals pattern rather than # inventing a new, looser standard. # # STALE (unlike WATCH) is NOT entry-eligible -- see # is_eligible_for_entry() below -- and entry_engine.py's fresh # health-reading gate hard-disqualifies any non-HEALTHY/WATCH raw # state at the moment of entry. So this isn't just a relabeling: # a fading-momentum, corroborated reading now actually blocks a # new entry, the same real protection any other STALE read gets. severe = _is_severe(price_slope_pct, price_above_vwap, lower_hl, cfg) if severe: raw_state = "UNHEALTHY" elif (price_slope_class == "positive" and vwap_support and not volume_declining and not lower_hl and not momentum_fading): raw_state = "HEALTHY" elif (price_slope_class == "negative" and not price_above_vwap and lower_hl): raw_state = "UNHEALTHY" elif (price_slope_class == "flat" and (volume_declining or range_contracting or flags["no_new_highs"])): raw_state = "STALE" elif momentum_fading and (volume_declining or lower_hl or not price_above_vwap): raw_state = "STALE" else: raw_state = "WATCH" return HealthReading( symbol=symbol, health_score=health_score, raw_state=raw_state, confirmed_state=raw_state, # overwritten by update_health_state() price_slope_pct=round(price_slope_pct, 3), vwap_slope_pct=round(vwap_slope_pct, 3), momentum_slope_pct=round(momentum_slope_pct, 3), price_slope_class=price_slope_class, vwap_slope_class=vwap_slope_class, momentum_slope_class=momentum_slope_class, breakdown=breakdown, flags=flags, ) def update_health_state(reading: HealthReading, persisted: dict, severe: bool = False) -> dict: """ Applies hysteresis on top of a single raw reading. `persisted` is this symbol's existing entry from state/intraday_health.json (a fresh {} if never seen before). Returns the new persisted entry; caller is responsible for saving it back via data_store. """ cfg = get_config()["intraday_health"] prev_state = persisted.get("state", "WATCH") consecutive_bad = persisted.get("consecutive_bad", 0) consecutive_good = persisted.get("consecutive_good", 0) consecutive_deteriorating = persisted.get("consecutive_deteriorating", 0) is_bad_read = reading.raw_state in ("UNHEALTHY",) is_good_read = reading.raw_state == "HEALTHY" # [FEATURE 2026-09-11] Deliberately looser than is_bad_read: STALE # already means "not entry-eligible" (see is_eligible_for_entry()), # and an OPEN position sitting in STALE for many reads in a row -- # never bad enough to hit UNHEALTHY, never recovering to WATCH/ # HEALTHY either -- is exactly the TJGC pattern this counter exists # to catch. Any raw_state that isn't HEALTHY/WATCH counts. is_deteriorating_read = reading.raw_state not in ("HEALTHY", "WATCH") consecutive_bad = consecutive_bad + 1 if is_bad_read else 0 consecutive_good = consecutive_good + 1 if is_good_read else 0 consecutive_deteriorating = consecutive_deteriorating + 1 if is_deteriorating_read else 0 new_state = prev_state if prev_state == "REMOVED": # Sticky: once REMOVED, stays REMOVED regardless of further bad # reads (avoids the exact UNHEALTHY<->REMOVED oscillation this # comment is warning about). It can still recover organically # via consecutive_good below, or get a clean slate at the next # full rescan (monitor.py rebuilds premarket_20 and this # symbol's history is dropped along with it). new_state = "REMOVED" elif severe: # [confirm-bars skip] documented exception -- a genuinely severe # reading demotes to UNHEALTHY immediately rather than waiting # for unhealthy_confirm_reads, but STILL cannot jump straight to # REMOVED; that still needs its own consecutive-count below. new_state = "UNHEALTHY" elif is_bad_read and consecutive_bad >= cfg["unhealthy_confirm_reads"]: new_state = "UNHEALTHY" elif is_good_read and consecutive_good >= cfg["healthy_confirm_reads"]: new_state = "HEALTHY" elif reading.raw_state == "STALE" and prev_state in ("HEALTHY", "WATCH"): new_state = "STALE" elif reading.raw_state == "WATCH" and prev_state == "STALE": new_state = "WATCH" elif reading.raw_state == "WATCH" and prev_state == "HEALTHY" and not is_good_read: new_state = "WATCH" # else: hold prev_state -- a single off reading isn't enough to move it if is_good_read and consecutive_good >= cfg["healthy_confirm_reads"]: new_state = "HEALTHY" # recovery overrides stickiness once confirmed if new_state == "UNHEALTHY" and (prev_state == "UNHEALTHY" or prev_state == "REMOVED"): if consecutive_bad >= cfg["remove_after_consecutive_unhealthy"]: new_state = "REMOVED" if new_state != prev_state: log.info(f"[HEALTH] {reading.symbol} {prev_state} -> {new_state} " f"(score={reading.health_score}, price_slope={reading.price_slope_class}, " f"vwap_slope={reading.vwap_slope_class})") return { "state": new_state, "consecutive_bad": consecutive_bad, "consecutive_good": consecutive_good, "consecutive_deteriorating": consecutive_deteriorating, "health_score": reading.health_score, "price_slope_class": reading.price_slope_class, "vwap_slope_class": reading.vwap_slope_class, "momentum_slope_class": reading.momentum_slope_class, "updated_at": datetime.now(timezone.utc).isoformat(), } def evaluate_symbol(symbol: str, bars: list, avg_vol_baseline: float, persisted_state: dict) -> tuple: """ Convenience wrapper: compute this instant's reading, apply hysteresis against the persisted state for this symbol, and return (HealthReading with confirmed_state filled in, new_persisted_entry). Does not touch disk -- caller batches the save. """ cfg = get_config()["intraday_health"] reading = compute_health(symbol, bars, avg_vol_baseline) severe = _is_severe(reading.price_slope_pct, reading.flags.get("price_above_vwap", False), reading.flags.get("lower_highs_lower_lows", False), cfg) entry = persisted_state.get(symbol, {}) new_entry = update_health_state(reading, entry, severe=severe) reading.confirmed_state = new_entry["state"] return reading, new_entry def prune_stale_health_state(health_state: dict) -> dict: """ [BUGFIX 2026-08-31] state/intraday_health.json is loaded wholesale at every call site (monitor.py's _scan_for_entries(), _update_intraday_ health(), _run_intraday_full_rescan()) with no date filtering, so entries from prior sessions pile up indefinitely (confirmed live: entries going back to 2026-08-18 still present). update_health_state() above assumed a symbol reappearing in premarket_20 would only ever see its OWN history via that same-session round-trip and would "get a clean slate at the next full rescan" (see the REMOVED-stickiness comment) -- but a full rescan only rebuilds premarket_20, it never touches this dict, so a symbol that returns to premarket_20 days later inherits its old consecutive_bad/consecutive_good counters and, worse, a sticky REMOVED/UNHEALTHY state from a session that has nothing to do with today, instead of starting fresh at WATCH. Called once at session startup (see monitor.py's SessionOrchestrator. __init__) so every symbol's hysteresis state actually starts clean for the day, same fix pattern as position_manager.py's _prune_stale_closed_positions(). """ today = market_time.now_et().date() stale = [] for symbol, entry in health_state.items(): updated_at = entry.get("updated_at") if not updated_at or market_time.et_date(datetime.fromisoformat(updated_at)) != today: stale.append(symbol) for symbol in stale: del health_state[symbol] if stale: log.info(f"[STARTUP] Pruned {len(stale)} stale health-state entr(y/ies) from a prior day") return health_state def is_eligible_for_entry(confirmed_state: str) -> bool: """HEALTHY and WATCH are still tradeable; STALE/UNHEALTHY/REMOVED are not NEW-entry eligible. An existing open position in a STALE/UNHEALTHY symbol is otherwise left alone by this module -- that's position_manager.py's call, not this module's -- except for the one narrow force-exit case in should_force_exit_on_deterioration().""" return confirmed_state in ("HEALTHY", "WATCH") def should_force_exit_on_deterioration(new_entry: dict, cfg: dict) -> bool: """ [FEATURE 2026-09-11] Consulted by monitor.py's _update_intraday_ health() for symbols with an OPEN position, right after that symbol's new_entry (this evaluation's post-hysteresis persisted state, from update_health_state()/evaluate_symbol()) is computed. True once consecutive_deteriorating has reached cfg["exit_confirm_reads"] -- the same "N confirmations in a row" hysteresis pattern this module already uses everywhere else (healthy_confirm_reads / unhealthy_confirm_reads), applied here to a standing open position instead of pool eligibility. See this module's docstring for why this one exception exists. """ if not cfg.get("exit_on_deterioration_enabled", True): return False return new_entry.get("consecutive_deteriorating", 0) >= cfg.get("exit_confirm_reads", 3) def is_confirmed_fading(reading: HealthReading) -> bool: """ [FEATURE 2026-08-18] Used by position_manager.py's trailing-stop fade-confirmation check (config: stop.fade_confirmation). Reuses the exact same UNHEALTHY classification compute_health() already applies everywhere else in the system (negative price slope AND VWAP lost AND confirmed lower-highs/lower-lows structure, OR the severe- override condition) -- deliberately not a separate, looser definition of "fading" invented just for the exit path. """ return reading.raw_state == "UNHEALTHY"