Forex_bot/indicators.py

"""Technical indicator functions.

Pure pandas implementations — no third-party TA library. Each public function
takes a DataFrame with columns [open, high, low, close] indexed by datetime
ascending, and returns either a float or a dict of latest values.
"""
from __future__ import annotations

import pandas as pd


def _require(df: pd.DataFrame, n: int) -> None:
    if len(df) < n:
        raise ValueError(f"Not enough data: need at least {n} rows, got {len(df)}")


def _ema(series: pd.Series, period: int) -> pd.Series:
    return series.ewm(span=period, adjust=False).mean()


def _wilder(series: pd.Series, period: int) -> pd.Series:
    """Wilder's smoothing (RMA) — same as EMA with alpha = 1/period."""
    return series.ewm(alpha=1.0 / period, adjust=False).mean()


def rsi(df: pd.DataFrame, period: int = 14) -> float:
    """Latest RSI using Wilder's smoothing."""
    _require(df, period + 1)
    delta = df["close"].diff()
    gain = delta.clip(lower=0.0)
    loss = -delta.clip(upper=0.0)
    avg_gain = _wilder(gain, period)
    avg_loss = _wilder(loss, period)
    rs = avg_gain / avg_loss.replace(0.0, pd.NA)
    rsi_series = (100.0 - (100.0 / (1.0 + rs))).fillna(100.0)
    return float(rsi_series.iloc[-1])


def macd(df: pd.DataFrame) -> dict[str, float]:
    """Latest MACD line, signal, histogram (12/26/9), plus prior histogram for direction."""
    _require(df, 35)
    fast = _ema(df["close"], 12)
    slow = _ema(df["close"], 26)
    macd_line = fast - slow
    signal_line = _ema(macd_line, 9)
    hist = macd_line - signal_line
    return {
        "macd": float(macd_line.iloc[-1]),
        "signal": float(signal_line.iloc[-1]),
        "hist": float(hist.iloc[-1]),
        "hist_prev": float(hist.iloc[-2]),
    }


def bollinger(df: pd.DataFrame, period: int = 20) -> dict[str, float]:
    """Latest Bollinger Bands (upper/mid/lower, 2 stdev) and current price."""
    _require(df, period + 1)
    close = df["close"]
    mid = close.rolling(period).mean()
    std = close.rolling(period).std(ddof=0)
    return {
        "upper": float((mid + 2.0 * std).iloc[-1]),
        "mid": float(mid.iloc[-1]),
        "lower": float((mid - 2.0 * std).iloc[-1]),
        "price": float(close.iloc[-1]),
    }


def ema_pair(df: pd.DataFrame, fast: int = 20, slow: int = 50) -> dict[str, float]:
    """Latest fast/slow EMA values + current close."""
    _require(df, slow + 1)
    return {
        "fast": float(_ema(df["close"], fast).iloc[-1]),
        "slow": float(_ema(df["close"], slow).iloc[-1]),
        "price": float(df["close"].iloc[-1]),
    }


def atr(df: pd.DataFrame, period: int = 14) -> float:
    """Latest Average True Range (Wilder)."""
    _require(df, period + 1)
    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 float(_wilder(tr, period).iloc[-1])


def adx(df: pd.DataFrame, period: int = 14) -> float:
    """Latest ADX(14) — trend strength, range 0-100."""
    _require(df, period * 2 + 1)
    high, low, close = df["high"], df["low"], df["close"]
    up = high.diff()
    down = -low.diff()
    plus_dm = pd.Series(((up > down) & (up > 0)) * up, index=df.index).clip(lower=0.0)
    minus_dm = pd.Series(((down > up) & (down > 0)) * down, index=df.index).clip(lower=0.0)

    prev_close = close.shift(1)
    tr = pd.concat(
        [(high - low), (high - prev_close).abs(), (low - prev_close).abs()],
        axis=1,
    ).max(axis=1)

    atr_s = _wilder(tr, period)
    plus_di = 100.0 * _wilder(plus_dm, period) / atr_s.replace(0.0, pd.NA)
    minus_di = 100.0 * _wilder(minus_dm, period) / atr_s.replace(0.0, pd.NA)
    dx = 100.0 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0.0, pd.NA)
    adx_series = _wilder(dx.fillna(0.0), period)
    return float(adx_series.iloc[-1])


def stochastic(df: pd.DataFrame, k_period: int = 14, d_period: int = 3) -> dict[str, float]:
    """Latest Stochastic %K (smoothed) and %D, plus prior values for cross detection."""
    _require(df, k_period + d_period + 1)
    high, low, close = df["high"], df["low"], df["close"]
    lowest = low.rolling(k_period).min()
    highest = high.rolling(k_period).max()
    raw_k = 100.0 * (close - lowest) / (highest - lowest).replace(0.0, pd.NA)
    k = raw_k.rolling(d_period).mean()
    d = k.rolling(d_period).mean()
    return {
        "k": float(k.iloc[-1]),
        "d": float(d.iloc[-1]),
        "k_prev": float(k.iloc[-2]),
        "d_prev": float(d.iloc[-2]),
    }