data/indicators.py

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")),
    }