""" utils/indicators.py — Technical Analysis Calculations Computes SMA, EMA, RSI, MACD, Bollinger Bands, ATR, Stochastic, Fibonacci retracements, support/resistance, and pivot points. """ import numpy as np import pandas as pd from typing import Dict, List, Tuple, Optional from utils.config import ( SMA_PERIODS, RSI_PERIOD, MACD_FAST, MACD_SLOW, MACD_SIGNAL, BB_PERIOD, BB_STD, ATR_PERIOD, FIB_LOOKBACK_DAYS, RSI_OVERSOLD, RSI_OVERBOUGHT ) # ─── Moving Averages ────────────────────────────────────────────────────────── def compute_sma(series: pd.Series, period: int) -> pd.Series: return series.rolling(window=period, min_periods=1).mean() def compute_ema(series: pd.Series, period: int) -> pd.Series: return series.ewm(span=period, adjust=False).mean() def compute_all_smas(df: pd.DataFrame) -> pd.DataFrame: for p in SMA_PERIODS: df[f"SMA_{p}"] = compute_sma(df["Close"], p) return df # ─── RSI ────────────────────────────────────────────────────────────────────── def compute_rsi(series: pd.Series, period: int = RSI_PERIOD) -> pd.Series: delta = series.diff() gain = delta.clip(lower=0) loss = (-delta).clip(lower=0) avg_gain = gain.ewm(com=period - 1, min_periods=period).mean() avg_loss = loss.ewm(com=period - 1, min_periods=period).mean() rs = avg_gain / avg_loss.replace(0, np.nan) rsi = 100 - (100 / (1 + rs)) return rsi.fillna(50) # ─── MACD ───────────────────────────────────────────────────────────────────── def compute_macd(series: pd.Series) -> Tuple[pd.Series, pd.Series, pd.Series]: ema_fast = compute_ema(series, MACD_FAST) ema_slow = compute_ema(series, MACD_SLOW) macd_line = ema_fast - ema_slow signal = compute_ema(macd_line, MACD_SIGNAL) histogram = macd_line - signal return macd_line, signal, histogram # ─── Bollinger Bands ────────────────────────────────────────────────────────── def compute_bollinger_bands(series: pd.Series) -> Tuple[pd.Series, pd.Series, pd.Series]: mid = compute_sma(series, BB_PERIOD) std = series.rolling(window=BB_PERIOD, min_periods=1).std() upper = mid + BB_STD * std lower = mid - BB_STD * std return upper, mid, lower # ─── ATR ────────────────────────────────────────────────────────────────────── def compute_atr(df: pd.DataFrame, period: int = ATR_PERIOD) -> pd.Series: high, low, close = df["High"], df["Low"], df["Close"] 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(com=period - 1, adjust=False).mean() # ─── Stochastic Oscillator ──────────────────────────────────────────────────── def compute_stochastic(df: pd.DataFrame, k: int = 14, d: int = 3) -> Tuple[pd.Series, pd.Series]: low_min = df["Low"].rolling(k).min() high_max = df["High"].rolling(k).max() stoch_k = 100 * (df["Close"] - low_min) / (high_max - low_min + 1e-9) stoch_d = stoch_k.rolling(d).mean() return stoch_k.fillna(50), stoch_d.fillna(50) # ─── Fibonacci Retracement ──────────────────────────────────────────────────── def compute_fibonacci(df: pd.DataFrame, lookback: int = FIB_LOOKBACK_DAYS) -> Dict[str, float]: """ Identify swing high and swing low over lookback period, then calculate Fibonacci retracement levels. """ recent = df.tail(lookback) swing_high = recent["High"].max() swing_low = recent["Low"].min() diff = swing_high - swing_low levels = { "0.0% (Low)": swing_low, "23.6%": swing_low + 0.236 * diff, "38.2%": swing_low + 0.382 * diff, "50.0%": swing_low + 0.500 * diff, "61.8%": swing_low + 0.618 * diff, "78.6%": swing_low + 0.786 * diff, "100% (High)": swing_high, } return levels # ─── Support / Resistance ───────────────────────────────────────────────────── def compute_support_resistance(df: pd.DataFrame, n_levels: int = 5) -> Tuple[List[float], List[float]]: """ Detect support and resistance zones using local minima/maxima and volume-weighted price clustering. """ closes = df["Close"].values highs = df["High"].values lows = df["Low"].values volumes= df["Volume"].values if "Volume" in df.columns else np.ones(len(closes)) # Local extrema detection with window=5 window = 5 resistance_raw, support_raw = [], [] for i in range(window, len(closes) - window): if highs[i] == max(highs[i-window:i+window+1]): resistance_raw.append(highs[i]) if lows[i] == min(lows[i-window:i+window+1]): support_raw.append(lows[i]) def cluster_levels(raw: List[float], n: int) -> List[float]: if not raw: return [] arr = np.array(sorted(raw)) tolerance = arr.mean() * 0.015 # 1.5% clustering tolerance clusters = [] current = [arr[0]] for v in arr[1:]: if v - current[-1] <= tolerance: current.append(v) else: clusters.append(np.mean(current)) current = [v] clusters.append(np.mean(current)) return sorted(clusters)[-n:] supports = cluster_levels(support_raw, n_levels) resistances = cluster_levels(resistance_raw, n_levels) return supports, resistances # ─── Pivot Points ───────────────────────────────────────────────────────────── def compute_pivots(df: pd.DataFrame) -> Dict[str, float]: """Classic floor trader pivot points from most recent completed candle.""" last = df.iloc[-2] if len(df) >= 2 else df.iloc[-1] H, L, C = last["High"], last["Low"], last["Close"] P = (H + L + C) / 3 R1 = 2 * P - L S1 = 2 * P - H R2 = P + (H - L) S2 = P - (H - L) R3 = H + 2 * (P - L) S3 = L - 2 * (H - P) return {"P": P, "R1": R1, "R2": R2, "R3": R3, "S1": S1, "S2": S2, "S3": S3} # ─── Full Indicator Bundle ──────────────────────────────────────────────────── def compute_all_indicators(df: pd.DataFrame) -> pd.DataFrame: """Run all indicators and attach as columns to df.""" df = df.copy() df = compute_all_smas(df) df["RSI"] = compute_rsi(df["Close"]) df["MACD"], df["MACD_Signal"], df["MACD_Hist"] = compute_macd(df["Close"]) df["BB_Upper"], df["BB_Mid"], df["BB_Lower"] = compute_bollinger_bands(df["Close"]) df["ATR"] = compute_atr(df) df["Stoch_K"], df["Stoch_D"] = compute_stochastic(df) return df # ─── Summary Snapshot ───────────────────────────────────────────────────────── def get_indicator_snapshot(df: pd.DataFrame) -> Dict: """Return last-row indicator values as a clean dict.""" df = compute_all_indicators(df) last = df.iloc[-1] prev = df.iloc[-2] if len(df) >= 2 else last rsi_val = float(last["RSI"]) rsi_state = ( "Oversold" if rsi_val < RSI_OVERSOLD else "Overbought" if rsi_val > RSI_OVERBOUGHT else "Neutral" ) macd_cross = "Bullish Cross" if (last["MACD"] > last["MACD_Signal"] and prev["MACD"] <= prev["MACD_Signal"]) else \ "Bearish Cross" if (last["MACD"] < last["MACD_Signal"] and prev["MACD"] >= prev["MACD_Signal"]) else "No Cross" supports, resistances = compute_support_resistance(df) fibs = compute_fibonacci(df) pivots = compute_pivots(df) return { "price": float(last["Close"]), "open": float(last["Open"]), "high": float(last["High"]), "low": float(last["Low"]), "volume": float(last.get("Volume", 0)), "rsi": rsi_val, "rsi_state": rsi_state, "macd": float(last["MACD"]), "macd_signal": float(last["MACD_Signal"]), "macd_hist": float(last["MACD_Hist"]), "macd_cross": macd_cross, "sma_20": float(last.get("SMA_20", 0)), "sma_50": float(last.get("SMA_50", 0)), "sma_200": float(last.get("SMA_200", 0)), "bb_upper": float(last["BB_Upper"]), "bb_lower": float(last["BB_Lower"]), "atr": float(last["ATR"]), "stoch_k": float(last["Stoch_K"]), "stoch_d": float(last["Stoch_D"]), "supports": supports, "resistances": resistances, "fibonacci": fibs, "pivots": pivots, }