crypto_trading/backend/app/engine/indicators.py

from __future__ import annotations

from collections.abc import Sequence

from app.schemas import Candle

Number = float | None


def sma(values: Sequence[float], period: int) -> list[Number]:
    if period <= 0:
        raise ValueError("period must be positive")
    out: list[Number] = [None] * len(values)
    window_sum = 0.0
    for i, value in enumerate(values):
        window_sum += value
        if i >= period:
            window_sum -= values[i - period]
        if i >= period - 1:
            out[i] = window_sum / period
    return out


def ema(values: Sequence[float], period: int) -> list[Number]:
    if period <= 0:
        raise ValueError("period must be positive")
    out: list[Number] = [None] * len(values)
    if len(values) < period:
        return out
    multiplier = 2 / (period + 1)
    seed = sum(values[:period]) / period
    out[period - 1] = seed
    prev = seed
    for i in range(period, len(values)):
        prev = (values[i] - prev) * multiplier + prev
        out[i] = prev
    return out


def rsi(values: Sequence[float], period: int = 14) -> list[Number]:
    if period <= 0:
        raise ValueError("period must be positive")
    out: list[Number] = [None] * len(values)
    if len(values) <= period:
        return out
    gains = 0.0
    losses = 0.0
    for i in range(1, period + 1):
        change = values[i] - values[i - 1]
        gains += max(change, 0.0)
        losses += max(-change, 0.0)
    avg_gain = gains / period
    avg_loss = losses / period
    out[period] = _rsi_value(avg_gain, avg_loss)
    for i in range(period + 1, len(values)):
        change = values[i] - values[i - 1]
        gain = max(change, 0.0)
        loss = max(-change, 0.0)
        avg_gain = (avg_gain * (period - 1) + gain) / period
        avg_loss = (avg_loss * (period - 1) + loss) / period
        out[i] = _rsi_value(avg_gain, avg_loss)
    return out


def _rsi_value(avg_gain: float, avg_loss: float) -> float:
    if avg_loss == 0:
        return 100.0
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))


def true_range(candles: Sequence[Candle]) -> list[float]:
    out: list[float] = []
    for i, candle in enumerate(candles):
        if i == 0:
            out.append(candle.high - candle.low)
            continue
        prev_close = candles[i - 1].close
        out.append(
            max(
                candle.high - candle.low,
                abs(candle.high - prev_close),
                abs(candle.low - prev_close),
            )
        )
    return out


def atr(candles: Sequence[Candle], period: int = 14) -> list[Number]:
    if period <= 0:
        raise ValueError("period must be positive")
    tr = true_range(candles)
    out: list[Number] = [None] * len(tr)
    if len(tr) < period:
        return out
    prev = sum(tr[:period]) / period
    out[period - 1] = prev
    for i in range(period, len(tr)):
        prev = (prev * (period - 1) + tr[i]) / period
        out[i] = prev
    return out


def rolling_std(values: Sequence[float], period: int) -> list[Number]:
    if period <= 0:
        raise ValueError("period must be positive")
    out: list[Number] = [None] * len(values)
    for i in range(period - 1, len(values)):
        window = values[i - period + 1 : i + 1]
        mean = sum(window) / period
        variance = sum((x - mean) ** 2 for x in window) / period
        out[i] = variance**0.5
    return out


def macd(
    values: Sequence[float],
    fast: int = 12,
    slow: int = 26,
    signal: int = 9,
) -> tuple[list[Number], list[Number], list[Number]]:
    """Return (macd line, signal line, histogram), each aligned with ``values``."""
    ema_fast = ema(values, fast)
    ema_slow = ema(values, slow)
    macd_line: list[Number] = [
        (f - s) if f is not None and s is not None else None
        for f, s in zip(ema_fast, ema_slow, strict=True)
    ]
    signal_line: list[Number] = [None] * len(values)
    start = next((i for i, value in enumerate(macd_line) if value is not None), None)
    if start is not None:
        valid = [value for value in macd_line[start:] if value is not None]
        for offset, value in enumerate(ema(valid, signal)):
            signal_line[start + offset] = value
    histogram: list[Number] = [
        (m - s) if m is not None and s is not None else None
        for m, s in zip(macd_line, signal_line, strict=True)
    ]
    return macd_line, signal_line, histogram


def roc(values: Sequence[float], period: int = 10) -> list[Number]:
    """Rate of change over ``period`` bars, as a fraction (0.02 = +2%)."""
    if period <= 0:
        raise ValueError("period must be positive")
    out: list[Number] = [None] * len(values)
    for i in range(period, len(values)):
        previous = values[i - period]
        if previous:
            out[i] = (values[i] - previous) / previous
    return out


def last(series: Sequence[Number]) -> Number:
    for value in reversed(series):
        if value is not None:
            return value
    return None