""" APEX V16 — TechSnapshot. Single source of truth for the per-symbol tech state consumed by Brains. Replaces V15's `tech: dict` (cf V15-BUG-3 "init_data with hidden fallbacks"): typed fields with sane defaults eliminate the "key dimenticata che fa partire un fallback nascosto" bug class. Field inventory (mapped to V15 build_tech_dict + the keys both Brains read): Identity / market data: symbol, price, open Bars / timing: candle_time — UTC unix timestamp of the latest M5 bar. Stable for 5 min: Brain dedup key. RSI multi-TF: rsi (M5), rsi_prev (M5 t-1), rsi_h1, rsi_h4 ATR / volatility: atr_m5_points — ATR M5 in price points (NOT ticks). atr_ratio — atr / 20-bar avg. vol_regime — "HIGH" if ratio > 1.5 else "NORMAL". vol_spike — last bar volume > 2x 20-bar avg. Trend / structure: market_structure — Enum-string: BULLISH_EXPANSION / BEARISH_EXPANSION / RANGING h1_struct_bull / h1_struct_bear — HH+HL / LH+LL on last 6 H1. trend_maturity — max consecutive up/down H1 closes (last 5). regime — TRENDING / TRENDING_SOFT / BREAKOUT / RANGING. regime_reason — human reason from determine_regime. regime_near_trending — diagnostics list (calibration log fodder). EMA / divergence / MACD: deviation_pct — (close - EMA50_H1) / EMA50_H1 * 100. divergence — "BULLISH" / "BEARISH" / "NONE". macd_decelerating — bool. Candle / pattern: candle_strength — body / 10-bar avg body. hammer, shooting_star, bull_engulfing, bear_engulfing, doji, doji_type, piercing, dark_cloud, morning_star, evening_star, volume_weak, buy_absorption, sell_absorption. VWAP: vwap, vwap_deviation_pct. Bias H4: bias, allowed_direction, h1_compatibility, h1_reason. Swing structural (lazy — populated only when caller passes a direction): swing_data — dict {swing_found, swing_type, swing_price, ...} or {} when not requested. Anti-revenge / housekeeping: consecutive_sl_count — how many SL fires this trading day. Tick mechanics (mirrored from config_futures for downstream price math): tick_size, tick_value Experimental / non-canonical: extra: dict[str, Any] — orderflow features, ad-hoc experiments. The builder `build_tech_snapshot(...)` is async (it awaits the MarketDataProvider) but does no other I/O. Returns None when bars are insufficient (V15 parity). """ from __future__ import annotations import asyncio import dataclasses from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Optional import pandas as pd from analysis.bias import BiasData, compute_algo_bias from analysis.indicators import ( calc_absorption, calc_atr, calc_candle_strength, calc_divergence, calc_doji, calc_engulfing, calc_hammer, calc_macd_decel_and_hist, calc_market_structure, calc_piercing_dark, calc_rsi, calc_star_patterns, calc_volume_weak, calc_vwap_intraday, identify_swing_levels, ) from analysis.market_data import MarketDataProvider from analysis.regime import determine_regime # ============================================================ # CACHE — candle-keyed snapshot memoization # ============================================================ # # Key: (symbol, candle_time_M5). The bar OPEN time is stable for the # entire 5-minute candle window, so all derived indicators are too. # Only `candle_age_seconds` (and the derived `is_candle_closed`) varies # with wall-clock and is refreshed at lookup time via _refresh_candle_age. # # Bounded LRU (insertion-ordered dict + cap). 64 entries is plenty: # 10 assets × 6 historical M5 = 60 entries worst case under normal flow. # Tests use clear_tech_cache() to reset between runs. _TECH_CACHE: "dict[tuple[str, int], TechSnapshot]" = {} _TECH_CACHE_MAX: int = 64 def clear_tech_cache() -> None: """Test-only: drop all cached snapshots.""" _TECH_CACHE.clear() def invalidate_tech_cache(symbol: Optional[str] = None) -> None: """V17 bug #2 hook: drop cached snapshots ad-hoc. Called by orchestrator from: - _handle_exit / _check_external_close: ensure next scan rebuilds fresh after a trade closes (consecutive_sl_count may have changed). - reconcile_startup post-reconnect: broker state may have shifted. Args: symbol: drop only entries for this symbol; None drops everything. """ if symbol is None: _TECH_CACHE.clear() return keys_to_drop = [k for k in _TECH_CACHE if k[0] == symbol] for k in keys_to_drop: _TECH_CACHE.pop(k, None) class StaleDataError(Exception): """V17 bug #2: raised when broker hasn't published the expected M5 candle after retries. Caller skips the scan WITHOUT marking the candle as already-evaluated, so the next tick retries naturally.""" def __init__( self, symbol: str, candle_time_seen: int, candle_time_expected: int, ) -> None: self.symbol = symbol self.candle_time_seen = candle_time_seen self.candle_time_expected = candle_time_expected super().__init__( f"{symbol}: broker lag — last bar at {candle_time_seen}, " f"expected ≥ {candle_time_expected}" ) def _store_with_lru(key: "tuple[str, int]", snap: "TechSnapshot") -> None: if key in _TECH_CACHE: _TECH_CACHE.pop(key) _TECH_CACHE[key] = snap while len(_TECH_CACHE) > _TECH_CACHE_MAX: _TECH_CACHE.pop(next(iter(_TECH_CACHE))) def _refresh_candle_age( snap: "TechSnapshot", now_utc: Optional[datetime], ) -> "TechSnapshot": """ Return a copy of `snap` with `candle_age_seconds` and `is_candle_closed` recomputed against `now_utc`. All other fields are stable for the lifetime of the M5 candle. """ if snap.candle_time <= 0: return snap _now = now_utc if now_utc is not None else datetime.now(timezone.utc) age = (_now.timestamp() - snap.candle_time) - V16_M5_BAR_SECONDS is_closed = age >= 0 return dataclasses.replace( snap, candle_age_seconds=age, is_candle_closed=is_closed, ) # ============================================================ # DATACLASS # ============================================================ @dataclass(frozen=True) class TechSnapshot: """ Frozen typed snapshot of indicator state for one symbol at one tick. Frozen = no accidental mutation by Brain code (every field reflects the "moment of computation"). When the orchestrator advances time it builds a fresh TechSnapshot — never mutates the previous one. """ # --- identity --- symbol: str price: float open: float # --- timing --- candle_time: int # UTC unix seconds (bar OPEN); 0 if unavailable is_candle_closed: bool # True iff candle_time>0 AND now>=open+bar_duration candle_age_seconds: float # seconds since close; <0 in-flight; 0.0 if unavailable # --- RSI multi-TF --- rsi: float rsi_prev: float rsi_h1: float rsi_h4: float # --- ATR / volatility --- atr_m5_points: float # ATR M5 in price points (V16 standard) atr_ratio: float vol_regime: str # "HIGH" | "NORMAL" vol_spike: bool # --- structure / regime --- market_structure: str # MarketStructure enum value h1_struct_bull: bool h1_struct_bear: bool trend_maturity: int regime: str # Regime enum value regime_reason: str regime_near_trending: list[str] # --- trend deviation / momentum --- deviation_pct: float # (close - EMA50_H1) / EMA50_H1 * 100 divergence: str # "BULLISH" | "BEARISH" | "NONE" macd_decelerating: bool macd_hist_last: float # ultimo bar dell'istogramma MACD # (>0 momentum bullish, <0 bearish). # Letto dal gate MACD_ACCELERATING_AGAINST. # --- candle / pattern flags --- candle_strength: float hammer: bool shooting_star: bool bull_engulfing: bool bear_engulfing: bool doji: bool doji_type: Optional[str] # "dragonfly"/"gravestone"/"standard"/None piercing: bool dark_cloud: bool morning_star: bool evening_star: bool volume_weak: bool buy_absorption: bool sell_absorption: bool # --- VWAP --- vwap: float vwap_deviation_pct: float # --- bias H4 --- bias: str # "RIALZISTA"/"RIBASSISTA"/"NEUTRO" allowed_direction: str # "BUY"/"SELL"/"BOTH"/"NONE" h1_compatibility: float h1_reason: str # --- swing structural (lazy) --- swing_data: dict[str, Any] = field(default_factory=dict) # --- anti-revenge / housekeeping --- consecutive_sl_count: int = 0 # --- tick mechanics --- tick_size: float = 0.0 tick_value: float = 0.0 # --- experimental / non-canonical --- extra: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: """ Flat dict view for consumers that pre-date TechSnapshot. Used at the trade_opener boundary (V16-day-1 trade_opener still accepts tech_now: dict). New consumers should read fields via attribute access; this helper is a thin compatibility bridge. """ from dataclasses import asdict return asdict(self) # ============================================================ # BUILDER # ============================================================ V16_TF_M5 = "5min" V16_TF_H1 = "1hour" V16_TF_H4 = "4hour" # Bar duration in seconds, used to derive candle_age_seconds from # candle_time (which is bar OPEN time per broker convention). Brain # entry-evaluation gating reads is_candle_closed / candle_age_seconds # to call AI only in the post-close window (5-60s typical). V16_M5_BAR_SECONDS: int = 300 # Minimum bar counts (V15 parity) _MIN_M5 = 30 _MIN_H1 = 10 _MIN_H4 = 5 # ============================================================ # Dashboard log enrichment (V18 12-mag) # # La dashboard web mostra per ogni entry_approved/entry_rejected i # segnali tecnici "RADAR style" (RSI multi-TF, pattern, ATR, h1_compat, # MACD, bias, struttura). Per evitare di duplicare la stessa estrazione # nei due brain (_reject) e nell'orchestrator (_log_entry_approved), # centralizziamo qui la conversione TechSnapshot → dict di campi log. # # Defensive con getattr: alcuni campi (`candle_pattern`, `bouncing_rsi`) # non esistono sul TechSnapshot V16 — vengono derivati o lasciati vuoti # per backward-compat con dashboard che li attende. # ============================================================ def tech_log_fields(tech: Optional["TechSnapshot"]) -> dict: """Return the standard set of tech fields to embed in brain_log entries. Empty dict when `tech` is None — call sites can safely splat the result with `**` even on reject paths that never built a snapshot. """ if tech is None: return {} # `candle_pattern` non è un campo nativo: deriviamo una stringa breve # dai flag pattern individuali (primo match vince, V15 RADAR parity). pattern_flags = ( ("bull_engulfing", "BULL_ENGULF"), ("bear_engulfing", "BEAR_ENGULF"), ("hammer", "HAMMER"), ("shooting_star", "SHOOTING_STAR"), ("morning_star", "MORNING_STAR"), ("evening_star", "EVENING_STAR"), ("piercing", "PIERCING"), ("dark_cloud", "DARK_CLOUD"), ("doji", "DOJI"), ) pattern_label = "" for attr, label in pattern_flags: if bool(getattr(tech, attr, False)): pattern_label = label break # `bouncing_rsi`: RSI in risalita dalla precedente lettura. Derivato # da rsi vs rsi_prev — non un attr nativo, ma utile sul radar UI. rsi_now = float(getattr(tech, "rsi", 0.0) or 0.0) rsi_prev = float(getattr(tech, "rsi_prev", 0.0) or 0.0) bouncing_rsi = bool(getattr(tech, "bouncing_rsi", rsi_now > rsi_prev)) return { "rsi_m5": round(rsi_now, 1), "rsi_h1": round(float(getattr(tech, "rsi_h1", 0.0) or 0.0), 1), "rsi_h4": round(float(getattr(tech, "rsi_h4", 0.0) or 0.0), 1), "h1_compat": round(float(getattr(tech, "h1_compatibility", 0.0) or 0.0), 2), "macd_hist": round(float(getattr(tech, "macd_hist_last", 0.0) or 0.0), 6), "macd_decel": bool(getattr(tech, "macd_decelerating", False)), "pattern": pattern_label, "atr_ratio": round(float(getattr(tech, "atr_ratio", 0.0) or 0.0), 2), "bias": str(getattr(tech, "bias", "") or ""), "h1_struct_bull": bool(getattr(tech, "h1_struct_bull", False)), "h1_struct_bear": bool(getattr(tech, "h1_struct_bear", False)), "volume_weak": bool(getattr(tech, "volume_weak", False)), "divergence": str(getattr(tech, "divergence", "") or ""), "bouncing_rsi": bouncing_rsi, } async def build_tech_snapshot( *, symbol: str, provider: MarketDataProvider, tick_size: float, tick_value: float, consecutive_sl_count: int = 0, bars_m5: int = 200, bars_h1: int = 100, bars_h4: int = 50, direction_for_swing: Optional[str] = None, now_utc: Optional[datetime] = None, min_candle_time: int = 0, lag_max_retries: int = 2, lag_retry_sleep_seconds: float = 1.5, ) -> Optional[TechSnapshot]: """ Async-fetch bars from `provider` and compute the full TechSnapshot. Args: symbol: instrument root. provider: MarketDataProvider supplying OHLCV bars. tick_size, tick_value: from config_futures.ASSETS_MAP[symbol]. consecutive_sl_count: anti-revenge counter (caller-managed). bars_m5/h1/h4: how many bars to request per timeframe. direction_for_swing: if "BUY"/"SELL", populate swing_data; None leaves it {}. Returns: TechSnapshot, or None if any timeframe has insufficient bars (V15 build_tech_dict parity). Caching: candle-keyed by (symbol, candle_time_M5). On a cache hit only the M5 probe fetch runs; H1+H4 fetches and indicator math are skipped. `candle_age_seconds` / `is_candle_closed` are recomputed at lookup so callers always see fresh wall-clock-derived fields. Cache bypass: when `direction_for_swing` is set OR `consecutive_sl_count != 0`, the cache is neither read nor written (these inputs change the snapshot but aren't in the key). The orchestrator's hot path passes neither, so the cache applies there. V17 bug #2 — broker lag sentinel: If `min_candle_time` > 0, verify df5.iloc[-1].time >= min_candle_time. If lower (broker hasn't published the expected M5 candle yet), retry the M5 fetch up to `lag_max_retries` times with `lag_retry_sleep_seconds` backoff. If still lagged after retries, raise StaleDataError so the caller can skip without consuming the candle. """ cache_bypass = ( direction_for_swing is not None or consecutive_sl_count != 0 ) # ── PHASE 1: M5 probe (1 REST) ───────────────────────────── df5 = await provider.get_bars(symbol, V16_TF_M5, bars_m5) if df5 is None or len(df5) < _MIN_M5: return None candle_time_probe = _extract_candle_time(df5) # V17 bug #2: broker lag sentinel + retry if min_candle_time > 0 and candle_time_probe < min_candle_time: for _ in range(lag_max_retries): await asyncio.sleep(lag_retry_sleep_seconds) df5 = await provider.get_bars(symbol, V16_TF_M5, bars_m5) if df5 is None or len(df5) < _MIN_M5: continue candle_time_probe = _extract_candle_time(df5) if candle_time_probe >= min_candle_time: break if candle_time_probe < min_candle_time: raise StaleDataError( symbol=symbol, candle_time_seen=candle_time_probe, candle_time_expected=min_candle_time, ) # ── PHASE 2: cache lookup ────────────────────────────────── cache_key: Optional[tuple[str, int]] = ( (symbol, candle_time_probe) if (candle_time_probe > 0 and not cache_bypass) else None ) if cache_key is not None and cache_key in _TECH_CACHE: return _refresh_candle_age(_TECH_CACHE[cache_key], now_utc) # ── PHASE 3: full build (H1+H4 + indicators) ─────────────── df1 = await provider.get_bars(symbol, V16_TF_H1, bars_h1) df4 = await provider.get_bars(symbol, V16_TF_H4, bars_h4) if df1 is None or df4 is None: return None if len(df1) < _MIN_H1 or len(df4) < _MIN_H4: return None # ── ATR M5 ──────────────────────────────────────────────── atr_series = calc_atr(df5) atr_cur = float(atr_series.iloc[-1]) atr_avg = float(atr_series.rolling(20).mean().iloc[-1]) if pd.isna(atr_cur) or pd.isna(atr_avg) or atr_avg <= 0: return None atr_ratio = atr_cur / (atr_avg + 1e-9) vol_regime = "HIGH" if atr_cur > atr_avg * 1.5 else "NORMAL" # ── H1 structure ────────────────────────────────────────── struct = calc_market_structure(df1, symbol) # ── RSI multi-TF ────────────────────────────────────────── rsi_m5_series = calc_rsi(df5) rsi_h1_series = calc_rsi(df1) rsi_h4_series = calc_rsi(df4) rsi_cur = float(rsi_m5_series.iloc[-1]) rsi_prev = float(rsi_m5_series.iloc[-2]) rsi_h1 = float(rsi_h1_series.iloc[-1]) rsi_h4 = float(rsi_h4_series.iloc[-1]) # ── EMA50 H1 (deviation_pct) ────────────────────────────── ema50_h1 = float(df1["close"].ewm(span=50).mean().iloc[-1]) last_h1_close = float(df1["close"].iloc[-1]) deviation_pct = ((last_h1_close - ema50_h1) / ema50_h1) * 100.0 # ── Vol spike (M5) ──────────────────────────────────────── vol_col = "volume" if "volume" in df5.columns else "tick_volume" if vol_col in df5.columns: last_vol = float(df5[vol_col].iloc[-1]) avg_vol = float(df5[vol_col].rolling(20).mean().iloc[-1]) vol_spike = bool(last_vol > avg_vol * 2) else: vol_spike = False # ── Candle patterns ─────────────────────────────────────── hammer, shooting_star = calc_hammer(df5) bull_eng, bear_eng = calc_engulfing(df5) doji_type, has_doji = calc_doji(df5) piercing, dark_cloud = calc_piercing_dark(df5) morning_star, evening_star = calc_star_patterns(df5) volume_weak = calc_volume_weak(df5) buy_absorption, sell_absorption = calc_absorption(df5, atr_cur) # ── VWAP ────────────────────────────────────────────────── vwap, vwap_dev = calc_vwap_intraday(df5) # ── Candle time (M5 last bar) ───────────────────────────── candle_time = _extract_candle_time(df5) # candle_time is bar OPEN time per broker convention; close = open # + V16_M5_BAR_SECONDS. age >= 0 means closed; age < 0 means in-flight. # candle_time == 0 (unavailable) -> fail closed: is_closed=False. _now = now_utc if now_utc is not None else datetime.now(timezone.utc) if candle_time > 0: candle_age_seconds = (_now.timestamp() - candle_time) - V16_M5_BAR_SECONDS is_candle_closed = candle_age_seconds >= 0 else: candle_age_seconds = 0.0 is_candle_closed = False # ── Regime classifier ───────────────────────────────────── regime, regime_reason, near_trending = determine_regime( rsi_h1=rsi_h1, market_structure=struct["market_structure"], trend_maturity=int(struct["trend_maturity"]), atr_ratio=atr_ratio, vol_spike=vol_spike, h1_struct_bull=bool(struct["h1_struct_bull"]), h1_struct_bear=bool(struct["h1_struct_bear"]), ) # ── Bias H4 ─────────────────────────────────────────────── bias: BiasData = compute_algo_bias(df4) # ── Swing data (lazy, only if direction requested) ──────── swing_data: dict[str, Any] = {} if direction_for_swing in ("BUY", "SELL"): # price_decimals from tick_size if tick_size > 0 and tick_size < 1: import math price_decimals = max(0, -int(math.floor(math.log10(tick_size)))) else: price_decimals = 2 swing_data = identify_swing_levels( df5, direction=direction_for_swing, lookback=30, atr_ratio=atr_ratio, price_decimals=price_decimals, entry_price=float(df5["close"].iloc[-1]), ) macd_decel, macd_hist_last = calc_macd_decel_and_hist(df5) snap = TechSnapshot( symbol=symbol, price=float(df5["close"].iloc[-1]), open=float(df5["open"].iloc[-1]), candle_time=candle_time, is_candle_closed=is_candle_closed, candle_age_seconds=candle_age_seconds, rsi=rsi_cur, rsi_prev=rsi_prev, rsi_h1=rsi_h1, rsi_h4=rsi_h4, atr_m5_points=atr_cur, atr_ratio=atr_ratio, vol_regime=vol_regime, vol_spike=vol_spike, market_structure=struct["market_structure"], h1_struct_bull=bool(struct["h1_struct_bull"]), h1_struct_bear=bool(struct["h1_struct_bear"]), trend_maturity=int(struct["trend_maturity"]), regime=regime, regime_reason=regime_reason, regime_near_trending=near_trending, deviation_pct=deviation_pct, divergence=calc_divergence(df5, rsi_m5_series), macd_decelerating=macd_decel, macd_hist_last=macd_hist_last, candle_strength=calc_candle_strength(df5), hammer=hammer, shooting_star=shooting_star, bull_engulfing=bull_eng, bear_engulfing=bear_eng, doji=has_doji, doji_type=doji_type, piercing=piercing, dark_cloud=dark_cloud, morning_star=morning_star, evening_star=evening_star, volume_weak=volume_weak, buy_absorption=buy_absorption, sell_absorption=sell_absorption, vwap=vwap, vwap_deviation_pct=vwap_dev, bias=bias.bias, allowed_direction=bias.allowed_direction, h1_compatibility=bias.h1_compatibility, h1_reason=bias.h1_reason, swing_data=swing_data, consecutive_sl_count=consecutive_sl_count, tick_size=tick_size, tick_value=tick_value, ) if cache_key is not None: _store_with_lru(cache_key, snap) return snap def _extract_candle_time(df5: pd.DataFrame) -> int: """ Extract M5 last-bar timestamp as UTC unix seconds. Used by Brain same-candle dedup. Returns 0 when unavailable. """ col_ts = None for c in ("time", "timestamp", "datetime"): if c in df5.columns: col_ts = c break if col_ts: try: ts_val = df5[col_ts].iloc[-1] if isinstance(ts_val, (pd.Timestamp, datetime)): return int(pd.Timestamp(ts_val).timestamp()) return int(ts_val) except Exception: return 0 if isinstance(df5.index, pd.DatetimeIndex): try: return int(df5.index[-1].timestamp()) except Exception: return 0 return 0