""" Full rule-based technical analysis engine. Analyses 4H, 1H, 15min timeframes for each asset. Computes: EMA stack, RSI, MACD, Bollinger Bands, Stochastic, ADX, VWAP, OBV, ATR, Fibonacci retracements, key S/R levels, and swing structure. Chart snapshots (candlestick PNGs) are rendered for vision-capable models — structural analysis is delegated to the vision pipeline. """ from __future__ import annotations import logging from typing import Any import numpy as np import pandas as pd logger = logging.getLogger("gap_system.analysis.technical") # ═══════════════════════════════════════════════════════════════════════════════ # CORE INDICATORS # ═══════════════════════════════════════════════════════════════════════════════ def ema(series: pd.Series, period: int) -> pd.Series: return series.ewm(span=period, adjust=False).mean() def sma(series: pd.Series, period: int) -> pd.Series: return series.rolling(window=period).mean() def rsi(close: pd.Series, period: int = 14) -> pd.Series: delta = close.diff() gain = delta.where(delta > 0, 0.0) loss = -delta.where(delta < 0, 0.0) avg_gain = gain.ewm(span=period, adjust=False).mean() avg_loss = loss.ewm(span=period, adjust=False).mean() rs = avg_gain / avg_loss.replace(0, np.nan) return 100 - (100 / (1 + rs)) def atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series: prev_close = close.shift(1) tr = pd.concat([ high - low, (high - prev_close).abs(), (low - prev_close).abs(), ], axis=1).max(axis=1) return tr.ewm(span=period, adjust=False).mean() def bollinger_bands(close: pd.Series, period: int = 20, num_std: float = 2.0) -> tuple[pd.Series, pd.Series, pd.Series]: """Bollinger Bands. Returns (upper, middle, lower).""" mid = close.rolling(window=period).mean() std = close.rolling(window=period).std() upper = mid + num_std * std lower = mid - num_std * std return upper, mid, lower # ═══════════════════════════════════════════════════════════════════════════════ # MOMENTUM INDICATORS # ═══════════════════════════════════════════════════════════════════════════════ def macd(close: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> tuple[pd.Series, pd.Series, pd.Series]: fast_ema = close.ewm(span=fast, adjust=False).mean() slow_ema = close.ewm(span=slow, adjust=False).mean() macd_line = fast_ema - slow_ema signal_line = macd_line.ewm(span=signal, adjust=False).mean() histogram = macd_line - signal_line return macd_line, signal_line, histogram def stochastic(high: pd.Series, low: pd.Series, close: pd.Series, k_period: int = 14, d_period: int = 3) -> tuple[float, float]: """Stochastic %K and %D. Returns current (K, D) values.""" lowest_low = low.rolling(window=k_period).min() highest_high = high.rolling(window=k_period).max() denom = highest_high - lowest_low k_raw = 100 * (close - lowest_low) / denom.replace(0, np.nan) k_smooth = k_raw.rolling(window=d_period).mean() d_smooth = k_smooth.rolling(window=d_period).mean() k_val = float(k_smooth.iloc[-1]) if not pd.isna(k_smooth.iloc[-1]) else 50.0 d_val = float(d_smooth.iloc[-1]) if not pd.isna(d_smooth.iloc[-1]) else 50.0 return k_val, d_val # ═══════════════════════════════════════════════════════════════════════════════ # TREND STRENGTH # ═══════════════════════════════════════════════════════════════════════════════ def adx(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> tuple[float, float, float]: """ Average Directional Index. Returns (ADX, +DI, -DI). ADX > 25 = trending, > 50 = strong trend, < 20 = ranging. """ prev_high = high.shift(1) prev_low = low.shift(1) prev_close = close.shift(1) plus_dm = (high - prev_high).where((high - prev_high) > (prev_low - low), 0.0) plus_dm = plus_dm.where(plus_dm > 0, 0.0) minus_dm = (prev_low - low).where((prev_low - low) > (high - prev_high), 0.0) minus_dm = minus_dm.where(minus_dm > 0, 0.0) tr = pd.concat([ high - low, (high - prev_close).abs(), (low - prev_close).abs(), ], axis=1).max(axis=1) atr_val = tr.ewm(span=period, adjust=False).mean() plus_di = 100 * (plus_dm.ewm(span=period, adjust=False).mean() / atr_val.replace(0, np.nan)) minus_di = 100 * (minus_dm.ewm(span=period, adjust=False).mean() / atr_val.replace(0, np.nan)) dx = (abs(plus_di - minus_di) / (plus_di + minus_di).replace(0, np.nan)) * 100 adx_val = dx.ewm(span=period, adjust=False).mean() a = float(adx_val.iloc[-1]) if not pd.isna(adx_val.iloc[-1]) else 0.0 p = float(plus_di.iloc[-1]) if not pd.isna(plus_di.iloc[-1]) else 0.0 m = float(minus_di.iloc[-1]) if not pd.isna(minus_di.iloc[-1]) else 0.0 return a, p, m # ═══════════════════════════════════════════════════════════════════════════════ # VOLATILITY INDICATORS # ═══════════════════════════════════════════════════════════════════════════════ def bollinger_bands(close: pd.Series, window: int = 20, num_std: float = 2.0) -> tuple[pd.Series, pd.Series, pd.Series]: rolling_mean = close.rolling(window=window).mean() rolling_std = close.rolling(window=window).std() upper_band = rolling_mean + (rolling_std * num_std) lower_band = rolling_mean - (rolling_std * num_std) return upper_band, rolling_mean, lower_band # ═══════════════════════════════════════════════════════════════════════════════ # VOLUME INDICATORS # ═══════════════════════════════════════════════════════════════════════════════ def obv(close: pd.Series, volume: pd.Series) -> pd.Series: """On Balance Volume — accumulation/distribution pressure gauge.""" direction = close.diff().apply(lambda x: 1 if x > 0 else (-1 if x < 0 else 0)) return (volume * direction).cumsum() def vwap_approx(high: pd.Series, low: pd.Series, close: pd.Series, volume: pd.Series) -> float: """Session VWAP approximation using typical price × volume.""" typical_price = (high + low + close) / 3 cumulative_tpv = (typical_price * volume).cumsum() cumulative_vol = volume.cumsum() vwap_series = cumulative_tpv / cumulative_vol.replace(0, np.nan) val = vwap_series.iloc[-1] return float(val) if not pd.isna(val) else float(close.iloc[-1]) # ═══════════════════════════════════════════════════════════════════════════════ # SUPPORT / RESISTANCE (Swing-Based) # ═══════════════════════════════════════════════════════════════════════════════ def find_swing_highs(high: pd.Series, lookback: int = 5) -> list[dict]: swings: list[dict] = [] vals = high.values for i in range(lookback, len(vals) - lookback): window = vals[i - lookback:i + lookback + 1] if vals[i] == max(window): swings.append({"index": i, "price": float(vals[i])}) return swings[-10:] def find_swing_lows(low: pd.Series, lookback: int = 5) -> list[dict]: swings: list[dict] = [] vals = low.values for i in range(lookback, len(vals) - lookback): window = vals[i - lookback:i + lookback + 1] if vals[i] == min(window): swings.append({"index": i, "price": float(vals[i])}) return swings[-10:] def find_support_resistance(swing_highs: list[dict], swing_lows: list[dict], current_price: float) -> dict: """ Nearest horizontal support (below price) and resistance (above price) derived from swing pivot clustering. """ resistance = None support = None for s in reversed(swing_highs): if s["price"] > current_price: resistance = s["price"] break for s in reversed(swing_lows): if s["price"] < current_price: support = s["price"] break return {"resistance": resistance, "support": support} # ═══════════════════════════════════════════════════════════════════════════════ # FIBONACCI RETRACEMENTS # ═══════════════════════════════════════════════════════════════════════════════ def fibonacci(high: pd.Series, low: pd.Series) -> dict: highest = float(high.max()) lowest = float(low.min()) diff = highest - lowest return { "1.000": round(highest, 5), "0.786": round(highest - diff * 0.786, 5), "0.618": round(highest - diff * 0.618, 5), "0.500": round(highest - diff * 0.500, 5), "0.382": round(highest - diff * 0.382, 5), "0.236": round(highest - diff * 0.236, 5), "0.000": round(lowest, 5), } # ═══════════════════════════════════════════════════════════════════════════════ # KEY LEVELS (Previous week/day H/L + round numbers) # ═══════════════════════════════════════════════════════════════════════════════ def find_key_levels(df: pd.DataFrame, asset: str) -> dict: """Previous week H/L, previous day H/L, round numbers.""" result: dict[str, float | None] = { "prev_week_high": None, "prev_week_low": None, "prev_day_high": None, "prev_day_low": None, "round_above": None, "round_below": None, } if df.empty: return result current_price = float(df["close"].iloc[-1]) if len(df) >= 2: result["prev_day_high"] = float(df["high"].iloc[-2]) result["prev_day_low"] = float(df["low"].iloc[-2]) if len(df) >= 10: week_data = df.iloc[-10:-5] result["prev_week_high"] = float(week_data["high"].max()) result["prev_week_low"] = float(week_data["low"].min()) if "XAU" in asset or "GOLD" in asset.upper(): step = 50.0 else: step = 0.0050 # 50 pips result["round_above"] = float(np.ceil(current_price / step) * step) result["round_below"] = float(np.floor(current_price / step) * step) return result # ═══════════════════════════════════════════════════════════════════════════════ # DIVERGENCE DETECTION # ═══════════════════════════════════════════════════════════════════════════════ def find_divergence(close: pd.Series, indicator: pd.Series, lookback: int = 30) -> str: """ Very simple peak/trough divergence check between Price and an Indicator. Looks at the last local minimum and maximum over the lookback period compared to current price. """ if len(close) < lookback + 5: return "none" recent_close = close.iloc[-lookback:-5] recent_ind = indicator.iloc[-lookback:-5] try: min_idx = recent_close.idxmin() max_idx = recent_close.idxmax() prev_low = close[min_idx] prev_ind_low = indicator[min_idx] prev_high = close[max_idx] prev_ind_high = indicator[max_idx] curr_close = close.iloc[-1] curr_ind = indicator.iloc[-1] # Regular Bearish: Higher High in price, Lower High in indicator if curr_close > prev_high and curr_ind < prev_ind_high: return "bearish (regular)" # Regular Bullish: Lower Low in price, Higher Low in indicator if curr_close < prev_low and curr_ind > prev_ind_low: return "bullish (regular)" except Exception: pass return "none" # ═══════════════════════════════════════════════════════════════════════════════ # SINGLE TIMEFRAME ANALYSIS # ═══════════════════════════════════════════════════════════════════════════════ def analyse_timeframe(df: pd.DataFrame, asset: str) -> dict: """Full indicator suite for one timeframe. Returns structured dict.""" if df.empty or len(df) < 30: return {"bias": "NEUTRAL", "error": "insufficient data"} close = df["close"] high = df["high"] low = df["low"] current_price = float(close.iloc[-1]) # ── EMAs ── ema20 = ema(close, 20) ema50 = ema(close, 50) ema200 = ema(close, 200) if len(close) >= 200 else ema(close, len(close)) ema20_val = float(ema20.iloc[-1]) ema50_val = float(ema50.iloc[-1]) ema200_val = float(ema200.iloc[-1]) if ema20_val > ema50_val > ema200_val: ema_stack = "bullish" elif ema20_val < ema50_val < ema200_val: ema_stack = "bearish" else: ema_stack = "mixed" above_20 = current_price > ema20_val above_50 = current_price > ema50_val above_200 = current_price > ema200_val # ── RSI ── rsi_series = rsi(close) rsi_val = float(rsi_series.iloc[-1]) if len(close) >= 14 else 50.0 # ── Divergence ── div = find_divergence(close, rsi_series) # ── ATR ── atr_val = float(atr(high, low, close).iloc[-1]) if len(close) >= 14 else 0.0 # ── MACD ── macd_l, signal_l, hist_l = macd(close) macdh = float(hist_l.iloc[-1]) macd_signal = "bullish" if macdh > 0 else "bearish" # Detect MACD crossover (signal within last 3 bars) macd_cross = "none" if len(hist_l) >= 4: prev_hist = [float(hist_l.iloc[i]) for i in range(-4, -1)] if macdh > 0 and any(h <= 0 for h in prev_hist): macd_cross = "bullish_cross" elif macdh < 0 and any(h >= 0 for h in prev_hist): macd_cross = "bearish_cross" # ── Bollinger Bands ── up_b, mid_b, low_b = bollinger_bands(close) bb_up = float(up_b.iloc[-1]) bb_low = float(low_b.iloc[-1]) bb_mid = float(mid_b.iloc[-1]) bandwidth = (bb_up - bb_low) / bb_mid if bb_mid > 0 else 0 bb_state = "expanding" if bandwidth > 0.02 else "squeezing" if current_price > bb_up: bb_pos = "above_upper" elif current_price < bb_low: bb_pos = "below_lower" else: bb_pos = "inside" # ── Stochastic ── stoch_k, stoch_d = stochastic(high, low, close) if stoch_k > 80: stoch_zone = "overbought" elif stoch_k < 20: stoch_zone = "oversold" else: stoch_zone = "neutral" # ── ADX ── adx_val, plus_di, minus_di = adx(high, low, close) if adx_val > 50: trend_strength = "strong" elif adx_val > 25: trend_strength = "moderate" else: trend_strength = "weak/ranging" di_bias = "bullish" if plus_di > minus_di else "bearish" # ── VWAP & OBV ── has_volume = "volume" in df.columns and df["volume"].sum() > 0 vwap_val = None obv_trend = "N/A" vol_above_avg = False if has_volume: vol = df["volume"] vwap_val = round(vwap_approx(high, low, close, vol), 5) obv_series = obv(close, vol) obv_sma = obv_series.rolling(20).mean() if not pd.isna(obv_sma.iloc[-1]): obv_trend = "accumulating" if obv_series.iloc[-1] > obv_sma.iloc[-1] else "distributing" vol_avg = vol.rolling(20).mean().iloc[-1] if not pd.isna(vol_avg) and vol_avg > 0: vol_above_avg = float(vol.iloc[-1]) > vol_avg # ── Swings + S/R ── swing_highs = find_swing_highs(high) swing_lows = find_swing_lows(low) sr = find_support_resistance(swing_highs, swing_lows, current_price) # ── Fibonacci ── fib_levels = fibonacci(high, low) # ── Key Levels ── key_levels = find_key_levels(df, asset) # ═══ BIAS SCORING ═══ bull_score = 0 bear_score = 0 # EMA stack (weight: 2) if ema_stack == "bullish": bull_score += 2 elif ema_stack == "bearish": bear_score += 2 # Price vs EMAs (weight: 1) if above_20 and above_50: bull_score += 1 elif not above_20 and not above_50: bear_score += 1 # RSI (weight: 1) if rsi_val > 60: bull_score += 1 elif rsi_val < 40: bear_score += 1 # MACD (weight: 1) if macd_signal == "bullish": bull_score += 1 elif macd_signal == "bearish": bear_score += 1 # MACD crossover (weight: 1 — fresh signal) if macd_cross == "bullish_cross": bull_score += 1 elif macd_cross == "bearish_cross": bear_score += 1 # Stochastic (weight: 1 — mean reversion) if stoch_zone == "oversold": bull_score += 1 elif stoch_zone == "overbought": bear_score += 1 # ADX direction (weight: 1 — only if trending) if adx_val > 25: if di_bias == "bullish": bull_score += 1 else: bear_score += 1 # Bollinger Band position (weight: 1 — mean reversion) if bb_pos == "below_lower": bull_score += 1 elif bb_pos == "above_upper": bear_score += 1 # OBV — accumulation / distribution (weight: 1) if obv_trend == "accumulating": bull_score += 1 elif obv_trend == "distributing": bear_score += 1 # ── Final bias ── total = bull_score + bear_score if total == 0: bias = "NEUTRAL" elif bull_score > bear_score + 1: bias = "BULLISH" elif bear_score > bull_score + 1: bias = "BEARISH" else: bias = "RANGING" return { "bias": bias, "bull_score": bull_score, "bear_score": bear_score, # EMAs "ema_stack": ema_stack, "ema_20": round(ema20_val, 5), "ema_50": round(ema50_val, 5), "ema_200": round(ema200_val, 5), "price_above_20": bool(above_20), "price_above_50": bool(above_50), "price_above_200": bool(above_200), # Momentum "rsi": round(rsi_val, 1), "macd_signal": macd_signal, "macd_hist": round(macdh, 5), "macd_cross": macd_cross, "stoch_k": round(stoch_k, 1), "stoch_d": round(stoch_d, 1), "stoch_zone": stoch_zone, # Trend strength "adx": round(adx_val, 1), "plus_di": round(plus_di, 1), "minus_di": round(minus_di, 1), "trend_strength": trend_strength, "di_bias": di_bias, # Volatility "atr": round(atr_val, 5), "bb_state": bb_state, "bb_pos": bb_pos, "bb_upper": round(bb_up, 5), "bb_lower": round(bb_low, 5), # Volume "volume_above_avg": bool(vol_above_avg), "obv_trend": obv_trend, "vwap": vwap_val, # Structure "resistance": sr["resistance"], "support": sr["support"], "swing_highs": [s["price"] for s in swing_highs[-3:]], "swing_lows": [s["price"] for s in swing_lows[-3:]], "fibonacci": fib_levels, "key_levels": key_levels, "rsi_divergence": div, } # ═══════════════════════════════════════════════════════════════════════════════ # FULL MULTI-TIMEFRAME ANALYSIS # ═══════════════════════════════════════════════════════════════════════════════ async def analyse_asset(asset: str) -> dict: """ Run full technical analysis on 4H, 1H, 15min for one asset. Returns weighted bias, confluence score, and chart paths. """ from data.price_data import fetch_candles from analysis.chart_renderer import render_candlestick # Optimized candle counts — enough for all indicators, 90% faster than 5000 # 4H × 500 = ~83 trading days (RSI/EMA/ATR need ~150 max) # 1H × 300 = ~12.5 trading days (sufficient for structure detection) # 15m × 200 = ~3.5 trading days (recent momentum only) df_4h = await fetch_candles(asset, "4h", 500) df_1h = await fetch_candles(asset, "1h", 300) df_15m = await fetch_candles(asset, "15m", 200) ta_4h = analyse_timeframe(df_4h, asset) ta_1h = analyse_timeframe(df_1h, asset) ta_15m = analyse_timeframe(df_15m, asset) # Render candlestick charts (incremental — skips if no new candles) chart_paths: dict[str, str] = {} for tf_label, tf_df in [("4h", df_4h), ("1h", df_1h), ("15m", df_15m)]: path = render_candlestick(tf_df, asset, tf_label, num_candles=80) if path: chart_paths[tf_label] = path # Weighted bias (4H=50%, 1H=30%, 15m=20%) bias_scores: dict[str, float] = {"BULLISH": 0.0, "BEARISH": 0.0, "RANGING": 0.0, "NEUTRAL": 0.0} weights = [(ta_4h, 0.50), (ta_1h, 0.30), (ta_15m, 0.20)] for ta, w in weights: bias = ta.get("bias", "NEUTRAL") bias_scores[bias] = bias_scores.get(bias, 0.0) + w max_bias = max(bias_scores, key=lambda k: bias_scores[k]) confluence_score = bias_scores[max_bias] if max_bias in ("RANGING", "NEUTRAL") or confluence_score < 0.4: weighted_bias = "NEUTRAL" else: weighted_bias = max_bias # Key invalidation level: nearest support/resistance from 4H or 1H if weighted_bias == "BULLISH": invalidation = ta_4h.get("support") or ta_1h.get("support") elif weighted_bias == "BEARISH": invalidation = ta_4h.get("resistance") or ta_1h.get("resistance") else: invalidation = None # Fallback engine AI-override metrics (enhanced) try: current_close = float(df_15m['close'].iloc[-1]) # 3 candles = 45 mins ~ 40 mins close_40m_ago = float(df_15m['close'].iloc[-4]) # 96 candles = 24 hours (1 Day = 96 15m candles) close_1d_ago = float(df_15m['close'].iloc[-97]) # 288 candles = 72 hours (3 Days) close_3d_ago = float(df_15m['close'].iloc[-289]) # Previous Week Low (using 4H: 1 trading week = 30 candles. So [-60:-30] is previous week) prev_week_low = float(df_4h['low'].iloc[-60:-30].min()) prev_week_high = float(df_4h['high'].iloc[-60:-30].max()) # Multi-timeframe momentum scoring for better gap direction trend_40m = 1 if current_close > close_40m_ago else -1 trend_1d = 1 if current_close > close_1d_ago else -1 trend_3d = 1 if current_close > close_3d_ago else -1 # Weighted momentum score: recent momentum matters most momentum_score = trend_40m * 0.20 + trend_1d * 0.40 + trend_3d * 0.40 # ATR-based volatility regime (last 14 4H candles) atr_14 = ta_4h.get("atr", 0) atr_slow = float(df_4h['high'].iloc[-50:-14].max() - df_4h['low'].iloc[-50:-14].min()) / 36 if len(df_4h) > 50 else atr_14 volatility_regime = "HIGH" if atr_14 > atr_slow * 1.5 else "LOW" if atr_14 < atr_slow * 0.7 else "NORMAL" # RSI extremes from 4H for overbought/oversold context rsi_4h = ta_4h.get("rsi", 50) # Infer gap direction from momentum if momentum_score > 0.3: fallback_direction = "BULLISH" elif momentum_score < -0.3: fallback_direction = "BEARISH" else: fallback_direction = "NEUTRAL" fallback_metrics = { "current_close": current_close, "trend_40m": "BEARISH" if current_close < close_40m_ago else "BULLISH", "trend_1d": "BEARISH" if current_close < close_1d_ago else "BULLISH", "trend_3d": "BEARISH" if current_close < close_3d_ago else "BULLISH", "momentum_score": round(momentum_score, 3), "fallback_direction": fallback_direction, "prev_week_low": prev_week_low, "prev_week_high": prev_week_high, "below_prev_week_low": current_close < prev_week_low, "above_prev_week_high": current_close > prev_week_high, "volatility_regime": volatility_regime, "rsi_4h": rsi_4h, "rsi_extreme": "OVERBOUGHT" if rsi_4h > 70 else "OVERSOLD" if rsi_4h < 30 else "NEUTRAL", } except Exception as e: logger.error("Fallback metrics error for %s: %s", asset, e) fallback_metrics = {} result = { "asset": asset, "4H": ta_4h, "1H": ta_1h, "15min": ta_15m, "weighted_bias": weighted_bias, "confluence_score": round(confluence_score, 2), "key_invalidation": invalidation, "chart_paths": chart_paths, "fallback_metrics": fallback_metrics, } logger.info( "TA %s: %s (confluence %.0f%%) | 4H=%s 1H=%s 15m=%s | Charts: %d", asset, weighted_bias, confluence_score * 100, ta_4h.get("bias"), ta_1h.get("bias"), ta_15m.get("bias"), len(chart_paths), ) return result # ═══════════════════════════════════════════════════════════════════════════════ # BLOOMBERG-STYLE TEXT DIGEST # ═══════════════════════════════════════════════════════════════════════════════ def format_technical_digest(ta_results: dict) -> str: """Format TA result as institutional Bloomberg-style text for agents.""" if not ta_results: return "No technical analysis available." lines = [f"═══ INSTITUTIONAL TECHNICAL ANALYSIS — {ta_results['asset']} ═══"] lines.append(f"Weighted Bias: {ta_results['weighted_bias']} (confluence {ta_results['confluence_score']:.0%})") if ta_results.get("key_invalidation"): lines.append(f"Key Invalidation: {ta_results['key_invalidation']}") lines.append("") for tf in ["4H", "1H", "15min"]: data = ta_results.get(tf, {}) if not data or "error" in data: lines.append(f"[{tf}] Insufficient data\n") continue rsi_v = data.get("rsi", 0.0) macd_s = data.get("macd_signal", "N/A") macd_h = data.get("macd_hist", 0.0) macd_x = data.get("macd_cross", "none") atr_v = data.get("atr", 0.0) bb_state = data.get("bb_state", "N/A") bb_pos = data.get("bb_pos", "N/A") stk = data.get("stoch_k", 0.0) std = data.get("stoch_d", 0.0) stoch_z = data.get("stoch_zone", "neutral") adx_v = data.get("adx", 0.0) pdi = data.get("plus_di", 0.0) mdi = data.get("minus_di", 0.0) ts = data.get("trend_strength", "N/A") obv_t = data.get("obv_trend", "N/A") vwap_v = data.get("vwap") div_str = data.get("rsi_divergence", "none") vol_str = "↑ above avg" if data.get("volume_above_avg") else "↓ below avg" bull_s = data.get("bull_score", 0) bear_s = data.get("bear_score", 0) lines.append(f"[{tf}] Bias: {data.get('bias', 'NEUTRAL')}") lines.append(f" • Trend: {data.get('ema_stack', '?').upper()} Stack | ADX {adx_v:.1f} (+DI {pdi:.1f}, -DI {mdi:.1f})") lines.append(f" • Momentum: RSI {rsi_v:.1f} (Divergence: {div_str.upper()}) | MACD {macd_h:.4f} ({macd_x}) | Stoch {stk:.1f}/{std:.1f} ({stoch_z})") lines.append(f" • Volume/Volty: OBV Trend {obv_t} | ATR {atr_v:.4f} | BB {bb_state} ({bb_pos})") resistance = data.get("resistance") support = data.get("support") if resistance or support: lines.append(f" S/R: Support={support or 'N/A'} Resistance={resistance or 'N/A'}") fibs = data.get("fibonacci", {}) if fibs: lines.append(f" Fib: 0.618={fibs.get('0.618', 'N/A')} 0.500={fibs.get('0.500', 'N/A')} 0.382={fibs.get('0.382', 'N/A')}") kl = data.get("key_levels", {}) if kl: lines.append(f" Levels: Prev Day H={kl.get('prev_day_high', 'N/A')} L={kl.get('prev_day_low', 'N/A')} | Round ↑={kl.get('round_above', 'N/A')} ↓={kl.get('round_below', 'N/A')}") lines.append("") return "\n".join(lines).strip()