indicators.py

"""Deterministic indicator math — no AI, no guessing.
All formulas match TradingView built-ins exactly.
Indicators run on CLOSED candles only (caller must drop the forming bar).
"""
from __future__ import annotations
import pandas as pd


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


def rma(series: pd.Series, length: int) -> pd.Series:
    return series.ewm(alpha=1.0 / length, adjust=False).mean()


def rsi(close: pd.Series, length: int = 14) -> pd.Series:
    delta = close.diff()
    gain = rma(delta.clip(lower=0.0), length)
    loss = rma(-delta.clip(upper=0.0), length)
    rs = gain / loss.replace(0.0, 1e-12)
    return 100.0 - (100.0 / (1.0 + rs))


def true_range(df: pd.DataFrame) -> pd.Series:
    prev = df["close"].shift(1)
    return pd.concat([
        df["high"] - df["low"],
        (df["high"] - prev).abs(),
        (df["low"] - prev).abs(),
    ], axis=1).max(axis=1)


def atr(df: pd.DataFrame, length: int = 14) -> pd.Series:
    return rma(true_range(df), length)


def swing_pivots(df: pd.DataFrame, k: int = 3):
    highs, lows = [], []
    h, l = df["high"].values, df["low"].values
    for i in range(k, len(df) - k):
        wh = h[i - k: i + k + 1]
        wl = l[i - k: i + k + 1]
        if h[i] == wh.max() and (wh == h[i]).sum() == 1:
            highs.append(float(h[i]))
        if l[i] == wl.min() and (wl == l[i]).sum() == 1:
            lows.append(float(l[i]))
    return highs, lows


def nearest_levels(price, pivot_highs, pivot_lows, lookback=20):
    pts = pivot_highs[-lookback:] + pivot_lows[-lookback:]
    supports = [p for p in pts if p < price]
    resistances = [p for p in pts if p > price]
    return (max(supports) if supports else None,
            min(resistances) if resistances else None)


def structure_tag(df: pd.DataFrame) -> str:
    e20 = ema(df["close"], 20)
    e50 = ema(df["close"], 50)
    close = df["close"].iloc[-1]
    e20_now, e20_prev = e20.iloc[-1], e20.iloc[-6]
    e50_now = e50.iloc[-1]
    if e20_now > e50_now and close > e20_now and e20_now > e20_prev:
        return "uptrend"
    if e20_now < e50_now and close < e20_now and e20_now < e20_prev:
        return "downtrend"
    return "range"


def compute_taker_ratio(df: pd.DataFrame, lookback: int = 10) -> float | None:
    """Return taker-buy ratio over the last `lookback` CLOSED candles.

    Formula: sum(taker_buy_vol[-lookback:]) / sum(volume[-lookback:])

    Closed candles = df[:-1]  (the last row is the forming/live bar).
    Returns None if data is missing, insufficient, or volume is zero.
    Clamps result to [0.0, 1.0] to guard against exchange data errors.

    Window choice — lookback=10 on 15m = 150 min (2.5 h):
      • Too short (≤3): single-candle spikes dominate; ratio is noisy.
      • Too long (≥20): captures prior sessions; signal becomes stale.
      • 10 candles smooths intra-hour noise while staying within the same
        trading session, making it actionable for the scorer's 15m signals.
    """
    if "taker_buy_vol" not in df.columns:
        return None

    closed = df.iloc[:-1]          # drop the live/forming candle
    if len(closed) < lookback:     # insufficient history
        return None

    window = closed.iloc[-lookback:]
    tbv = pd.to_numeric(window["taker_buy_vol"], errors="coerce")
    vol = pd.to_numeric(window["volume"],        errors="coerce")

    # Drop rows where either column is NaN so they cancel symmetrically
    mask = tbv.notna() & vol.notna()
    tbv, vol = tbv[mask], vol[mask]

    if len(tbv) == 0:              # all NaN after cleaning
        return None

    total_vol = vol.sum()
    if total_vol == 0.0:           # all-zero volume (halted / bad data)
        return None

    ratio = float(tbv.sum() / total_vol)
    return max(0.0, min(1.0, ratio))   # clamp: handles taker_buy_vol > volume


def analyze_timeframe(df: pd.DataFrame) -> dict:
    if len(df) < 60:
        return {"error": f"insufficient history ({len(df)} candles)"}
    close = df["close"]
    e20 = ema(close, 20)
    e50 = ema(close, 50)
    r = rsi(close, 14)
    a = atr(df, 14)
    vol = df["volume"]
    vol_avg20 = vol.rolling(20).mean()
    ph, pl = swing_pivots(df, k=3)
    last_close = float(close.iloc[-1])
    sup, res = nearest_levels(last_close, ph, pl)
    return {
        "close": round(last_close, 8),
        "ema20": round(float(e20.iloc[-1]), 6),
        "ema50": round(float(e50.iloc[-1]), 6),
        "rsi14": round(float(r.iloc[-1]), 2),
        "atr14": round(float(a.iloc[-1]), 6),
        "atr_pct": round(float(a.iloc[-1]) / last_close * 100, 3),
        "vol_ratio": round(float(vol.iloc[-1] / vol_avg20.iloc[-1]), 2)
                     if vol_avg20.iloc[-1] > 0 else None,
        "support": round(sup, 6) if sup else None,
        "resistance": round(res, 6) if res else None,
        "structure": structure_tag(df),
    }