import pandas as pd import numpy as np def compute_indicators(df: pd.DataFrame) -> pd.DataFrame: """Add RSI, MA, MACD, Bollinger Bands to a OHLCV DataFrame.""" df = df.copy() close = df["close"].astype(float) # Moving averages df["ma20"] = close.rolling(20).mean().round(2) df["ma50"] = close.rolling(50).mean().round(2) # RSI(14) df["rsi"] = _rsi(close, 14).round(2) # MACD (12,26,9) ema12 = close.ewm(span=12, adjust=False).mean() ema26 = close.ewm(span=26, adjust=False).mean() df["macd"] = (ema12 - ema26).round(2) df["macd_signal"] = df["macd"].ewm(span=9, adjust=False).mean().round(2) df["macd_hist"] = (df["macd"] - df["macd_signal"]).round(2) # Bollinger Bands (20, 2) std20 = close.rolling(20).std() df["bb_upper"] = (df["ma20"] + 2 * std20).round(2) df["bb_lower"] = (df["ma20"] - 2 * std20).round(2) return df def _rsi(series: pd.Series, period: int = 14) -> pd.Series: delta = series.diff() gain = delta.clip(lower=0) loss = -delta.clip(upper=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) return 100 - (100 / (1 + rs)) def get_latest_indicators(df: pd.DataFrame) -> dict: """Return the most recent indicator values as a dict.""" if df.empty: return {} row = df.iloc[-1] def safe(val): if pd.isna(val): return None return float(val) return { "rsi": safe(row.get("rsi")), "ma20": safe(row.get("ma20")), "ma50": safe(row.get("ma50")), "macd": safe(row.get("macd")), "macd_signal": safe(row.get("macd_signal")), "macd_hist": safe(row.get("macd_hist")), "bb_upper": safe(row.get("bb_upper")), "bb_lower": safe(row.get("bb_lower")), }