AH-Oil-Trader/core/analysis/indicators.py

"""
core/analysis/indicators.py — Mathematische Indikatoren & Helper
"""

from __future__ import annotations
import math

from core.config import ANGLE_LR_BARS


def _ema(values: list, period: int) -> list:
    """Exponentieller gleitender Durchschnitt."""
    k = 2.0 / (period + 1)
    out = [values[0]]
    for v in values[1:]:
        out.append(v * k + out[-1] * (1 - k))
    return out


def calc_trend_angle(closes: list, n: int = ANGLE_LR_BARS) -> float:
    """
    Lineare Regression über die letzten n Schlusskurse.
    Liefert 0°–180°:   0° = starker Aufwärtstrend
                      90° = seitwärts
                     180° = starker Abwärtstrend
    """
    data = closes[-n:] if len(closes) >= n else closes
    k = len(data)
    if k < 2:
        return 90.0
    xm = (k - 1) / 2.0
    ym = sum(data) / k
    num = sum((i - xm) * (v - ym) for i, v in enumerate(data))
    den = sum((i - xm) ** 2 for i in range(k))
    if den == 0:
        return 90.0
    slope     = num / den
    slope_pct = slope / (ym or 1) * 100
    angle_rad = math.atan(slope_pct * 18)
    angle     = 90.0 - math.degrees(angle_rad)
    return max(0.0, min(180.0, angle))


FIB_RATIOS = (0.236, 0.382, 0.500, 0.618, 0.786)


def calc_rsi(closes: list, period: int = 14) -> float:
    """Wilder-RSI über die letzten `period` Schlusskurse. 0–100."""
    if len(closes) < period + 1:
        return 50.0
    gains, losses = 0.0, 0.0
    for i in range(1, period + 1):
        diff = closes[i] - closes[i - 1]
        if diff >= 0:
            gains += diff
        else:
            losses -= diff
    avg_g = gains  / period
    avg_l = losses / period
    for i in range(period + 1, len(closes)):
        diff = closes[i] - closes[i - 1]
        g = max(diff, 0.0)
        l = max(-diff, 0.0)
        avg_g = (avg_g * (period - 1) + g) / period
        avg_l = (avg_l * (period - 1) + l) / period
    if avg_l == 0:
        return 100.0
    rs = avg_g / avg_l
    return 100.0 - 100.0 / (1.0 + rs)


def calc_atr(highs: list, lows: list, closes: list, period: int = 14) -> float | None:
    """Average True Range, Wilder-Smoothing. Liefert None bei zu wenig Daten."""
    n = len(highs)
    if n < period + 1 or n != len(lows) or n != len(closes):
        return None
    trs = []
    for i in range(1, n):
        tr = max(highs[i] - lows[i],
                 abs(highs[i] - closes[i - 1]),
                 abs(lows[i]  - closes[i - 1]))
        trs.append(tr)
    if len(trs) < period:
        return None
    atr = sum(trs[:period]) / period
    for tr in trs[period:]:
        atr = (atr * (period - 1) + tr) / period
    return atr


def calc_volume_ratio(bars: list, lookback: int = 20) -> float:
    """
    Verhältnis des letzten Bar-Tick-Volumens zum Ø der vorherigen lookback Bars.
    > 1.5 = überdurchschnittlich, < 0.7 = unterdurchschnittlich
    """
    vols = []
    for b in bars:
        try:
            vols.append(float(b["tick_volume"] or 0))
        except Exception:
            vols.append(0.0)
    if len(vols) < 3:
        return 1.0
    window = vols[-(lookback + 1):-1]
    avg = sum(window) / len(window) if window else 0.0
    return round(vols[-1] / avg, 2) if avg > 0 else 1.0


def calc_fib_levels(highs: list, lows: list, lookback: int = 50) -> dict | None:
    """
    Fibonacci-Retracement-Level aus dem größten Swing-High/Low im lookback-Fenster.
    """
    n = len(highs)
    if n < 10 or n != len(lows):
        return None
    start     = max(0, n - lookback)
    win_h     = highs[start:]
    win_l     = lows[start:]
    hi_rel    = max(range(len(win_h)), key=lambda i: win_h[i])
    lo_rel    = min(range(len(win_l)), key=lambda i: win_l[i])
    swing_high = win_h[hi_rel]
    swing_low  = win_l[lo_rel]
    rng        = swing_high - swing_low
    if rng <= 0:
        return None
    up_levels   = {round(r * 100, 1): round(swing_high - r * rng, 5) for r in FIB_RATIOS}
    down_levels = {round(r * 100, 1): round(swing_low  + r * rng, 5) for r in FIB_RATIOS}
    hi_idx = start + hi_rel
    lo_idx = start + lo_rel
    return {
        "swing_high": swing_high,
        "swing_low":  swing_low,
        "high_idx":   hi_idx,
        "low_idx":    lo_idx,
        "range":      rng,
        "direction":  "up" if hi_idx > lo_idx else "down",
        "levels":     {"up": up_levels, "down": down_levels},
    }


def calc_fib_distance(price: float, fib: dict | None, direction: str = "up") -> dict:
    """Abstand des Preises zum nächsten Schlüssel-Fib-Level (38.2, 50.0, 61.8)."""
    empty = {"nearest_ratio": None, "nearest_price": None, "dist_pct": None}
    if not fib or not price:
        return empty
    levels = fib["levels"].get(direction, {})
    key    = {k: v for k, v in levels.items() if k in (38.2, 50.0, 61.8)}
    if not key:
        return empty
    nearest_ratio, nearest_price = min(key.items(), key=lambda kv: abs(kv[1] - price))
    rng      = fib.get("range", 1)
    dist_pct = abs(nearest_price - price) / rng * 100 if rng > 0 else None
    return {
        "nearest_ratio": nearest_ratio,
        "nearest_price": nearest_price,
        "dist_pct":      round(dist_pct, 1) if dist_pct is not None else None,
    }


def detect_regime(angles: dict) -> str:
    """
    Klassifiziert den Markt-Zustand anhand der 4 TF-Winkel.
    Rückgabewerte: 'trend_up', 'trend_down', 'range', 'transition'
    """
    if not angles:
        return "transition"
    vals = [angles.get(tf, 90.0) for tf in ("M5", "M15", "M30", "H1")]
    spread = max(vals) - min(vals)
    if all(v < 80 for v in vals):
        return "trend_up"
    if all(v > 100 for v in vals):
        return "trend_down"
    if spread < 25 and all(75 <= v <= 105 for v in vals):
        return "range"
    return "transition"


def calc_sma(values: list, period: int = 50) -> float | None:
    """Simple Moving Average über die letzten `period` Werte."""
    if len(values) < period:
        return None
    return sum(values[-period:]) / period


def calc_vwap(bars: list) -> dict | None:
    """
    Daily VWAP (Volume Weighted Average Price) aus Intraday-Bars.
    Reset täglich um 00:00 UTC.

    reclaim=True: Preis war in den letzten 3 Bars unter VWAP,
                  aktuelle Bar schloss darüber → VWAP-Reclaim-Signal.

    Rückgabe: {'vwap', 'price_vs_vwap': 'above'|'below'|'at',
               'reclaim': bool, 'n_bars': int, 'diff_pct': float}
    """
    import time as _t
    from datetime import datetime, timezone as _tz
    if not bars:
        return None
    now_ts    = _t.time()
    today_utc = datetime.fromtimestamp(now_ts, tz=_tz.utc).replace(
        hour=0, minute=0, second=0, microsecond=0)
    today_ts  = today_utc.timestamp()

    today_bars = [b for b in bars if int(b["time"]) >= today_ts]
    if len(today_bars) < 2:
        return None

    cum_tpv = 0.0
    cum_vol = 0.0
    vwap_series = []
    for b in today_bars:
        tp  = (float(b["high"]) + float(b["low"]) + float(b["close"])) / 3
        vol = float(b["tick_volume"] or 1)
        cum_tpv += tp * vol
        cum_vol += vol
        vwap_series.append(cum_tpv / cum_vol if cum_vol > 0 else tp)

    vwap = vwap_series[-1]
    cur  = float(today_bars[-1]["close"])
    n    = len(today_bars)

    # Reclaim: letzte ≥2 Bars unter VWAP, jetzt drüber
    prev_below = (n >= 3 and all(
        float(today_bars[i]["close"]) < vwap_series[i]
        for i in range(max(0, n - 3), n - 1)
    ))
    reclaim = prev_below and (cur > vwap)

    diff_pct = (cur - vwap) / vwap * 100 if vwap > 0 else 0.0
    return {
        "vwap":          round(vwap, 3),
        "price_vs_vwap": ("above" if cur > vwap * 1.0001 else
                          "below" if cur < vwap * 0.9999 else "at"),
        "reclaim":       reclaim,
        "n_bars":        n,
        "diff_pct":      round(diff_pct, 2),
    }


def calc_sr_distance(price: float, sr: dict | None, atr: float | None) -> dict:
    """Distanz zum nächsten Support/Resistance in ATR-Einheiten."""
    if not sr or not atr or atr <= 0 or not price:
        return {"support_atr": None, "resistance_atr": None}
    sups = sr.get("supports") or []
    ress = sr.get("resistances") or []
    s_dists = [price - s["price"] for s in sups if s.get("price") and s["price"] < price]
    r_dists = [r["price"] - price for r in ress if r.get("price") and r["price"] > price]
    return {
        "support_atr":    (min(s_dists) / atr) if s_dists else None,
        "resistance_atr": (min(r_dists) / atr) if r_dists else None,
    }