Trademind/src/trademind/features.py

"""Feature-Engineering: aus einer Kerzenserie normierte Merkmalsvektoren bauen.

Alle Features sind bewusst skalenfrei (Verhältnisse, Prozentwerte, z-Scores), damit ein
Modell über verschiedene Symbole und Preisniveaus hinweg lernen kann.

Die Indikatoren werden einmal über die gesamte Serie berechnet (``build_feature_matrix``);
Backtests laufen dadurch linear statt quadratisch.
"""

from __future__ import annotations

from dataclasses import dataclass

import numpy as np

from .config import DerivativesConfig, RuleConfig
from .indicators import atr, donchian_position, ema, macd, roc, rolling_std, rsi, sma
from .models import Candles

BASE_FEATURE_NAMES: tuple[str, ...] = (
    "ema_spread",          # (EMA_fast - EMA_slow) / Preis  [%]
    "ema_fast_dist",       # (Preis - EMA_fast) / Preis     [%]
    "trend_dist",          # (Preis - EMA_trend) / Preis    [%]
    "rsi_norm",            # (RSI - 50) / 50
    "rsi_slope",           # RSI-Änderung über 3 Bars / 50
    "macd_hist",           # MACD-Histogramm / Preis        [%]
    "macd_hist_slope",
    "atr_pct",             # ATR / Preis                    [%]
    "vol_ratio",           # kurzfristige vs. langfristige Kursvolatilität
    "roc_3",
    "roc_12",
    "donchian_pos",        # Lage in der 20-Bar-Range, zentriert auf 0
    "volume_z",            # z-Score des Volumens
    "body_ratio",          # Kerzenkörper / Range
    "upper_wick",
    "lower_wick",
    "time_sin",            # zyklische Tageszeit
    "time_cos",
)

# Zusatzmerkmale aus dem Terminmarkt – nur aktiv, wenn strategy.derivatives.enabled.
FUNDING_FEATURE_NAMES: tuple[str, ...] = (
    "funding_bps",         # Funding Rate der laufenden Periode, in Basispunkten
    "funding_trend",       # Abweichung vom Mittel der letzten Perioden
)
OI_FEATURE_NAMES: tuple[str, ...] = (
    "oi_change",           # Veränderung des Open Interest über oi_change_bars  [%]
    "oi_price_divergence", # OI-Veränderung × Kursrichtung: neue Positionen oder Glattstellung
)

# Rückwärtskompatibler Name für die Basisausstattung.
FEATURE_NAMES = BASE_FEATURE_NAMES
N_FEATURES = len(BASE_FEATURE_NAMES)
MIN_BARS = 140
_CLIP_LIMIT = 8.0


def feature_names(derivatives: DerivativesConfig | None = None) -> tuple[str, ...]:
    """Merkmalsnamen für die gegebene Konfiguration – die Anzahl hängt davon ab."""
    names = BASE_FEATURE_NAMES
    if derivatives is not None and derivatives.enabled:
        if derivatives.funding_rate:
            names += FUNDING_FEATURE_NAMES
        if derivatives.open_interest:
            names += OI_FEATURE_NAMES
    return names


def n_features(derivatives: DerivativesConfig | None = None) -> int:
    return len(feature_names(derivatives))


@dataclass(slots=True)
class FeatureSnapshot:
    """Merkmalsvektor plus Roh-Kennzahlen, die Risiko und Regelwerk zusätzlich brauchen."""

    values: np.ndarray
    names: tuple[str, ...]
    index: int
    price: float
    atr: float
    rsi: float
    rsi_prev: float
    ema_fast: float
    ema_slow: float
    ema_fast_prev: float
    ema_slow_prev: float
    trend_ema: float
    timestamp: int

    def as_dict(self) -> dict[str, float]:
        return {name: float(v) for name, v in zip(self.names, self.values, strict=True)}


def required_bars(rules: RuleConfig) -> int:
    """Minimale Anzahl Kerzen, damit alle Indikatoren belastbare Werte liefern."""
    return max(MIN_BARS, rules.trend_filter_period + 10, rules.slow_ema * 3, rules.rsi_period * 4)


def _clean(arr: np.ndarray, fill: float | np.ndarray = 0.0) -> np.ndarray:
    out = np.asarray(arr, dtype=np.float64).copy()
    bad = ~np.isfinite(out)
    if np.any(bad):
        out[bad] = fill[bad] if isinstance(fill, np.ndarray) else fill
    return out


def _pct(numerator: np.ndarray, price: np.ndarray) -> np.ndarray:
    with np.errstate(divide="ignore", invalid="ignore"):
        return np.where(price > 0, numerator / price * 100.0, 0.0)


@dataclass(slots=True)
class FeatureMatrix:
    """Alle Merkmalsvektoren einer Serie plus die Roh-Indikatoren."""

    values: np.ndarray            # (n, N_FEATURES)
    timestamp: np.ndarray
    price: np.ndarray
    atr: np.ndarray
    rsi: np.ndarray
    ema_fast: np.ndarray
    ema_slow: np.ndarray
    trend_ema: np.ndarray
    first_valid: int              # ab hier sind die Zeilen belastbar
    names: tuple[str, ...] = BASE_FEATURE_NAMES

    def __len__(self) -> int:
        return int(self.values.shape[0])

    def is_valid(self, index: int) -> bool:
        idx = index if index >= 0 else len(self) + index
        return self.first_valid <= idx < len(self)

    def snapshot(self, index: int = -1) -> FeatureSnapshot | None:
        n = len(self)
        idx = index if index >= 0 else n + index
        if not self.is_valid(idx):
            return None
        prev = max(idx - 1, 0)
        return FeatureSnapshot(
            values=self.values[idx].copy(),
            names=self.names,
            index=idx,
            price=float(self.price[idx]),
            atr=float(self.atr[idx]),
            rsi=float(self.rsi[idx]),
            rsi_prev=float(self.rsi[prev]),
            ema_fast=float(self.ema_fast[idx]),
            ema_slow=float(self.ema_slow[idx]),
            ema_fast_prev=float(self.ema_fast[prev]),
            ema_slow_prev=float(self.ema_slow[prev]),
            trend_ema=float(self.trend_ema[idx]),
            timestamp=int(self.timestamp[idx]),
        )


def _derivative_columns(
    close: np.ndarray, series, config: DerivativesConfig
) -> list[np.ndarray]:
    """Spalten aus Funding Rate und Open Interest, in derselben Reihenfolge wie die Namen."""
    columns: list[np.ndarray] = []
    n = close.size

    if config.funding_rate:
        rate = np.asarray(getattr(series, "funding_rate", None), dtype=np.float64) \
            if series is not None else np.full(n, np.nan)
        rate = _clean(rate, 0.0)
        # Anteil je 8h → Basispunkte. Typisch ±1 bp, in Extremphasen ±10 bp.
        funding_bps = rate * 10_000.0
        # Abweichung vom gleitenden Mittel der letzten Perioden: Zuspitzung oder Entspannung.
        baseline = _clean(sma(funding_bps, 24), 0.0)
        columns += [funding_bps, funding_bps - baseline]

    if config.open_interest:
        oi = np.asarray(getattr(series, "open_interest", None), dtype=np.float64) \
            if series is not None else np.full(n, np.nan)
        oi = _clean(oi, 0.0)
        lag = min(config.oi_change_bars, max(n - 1, 1))
        previous = np.concatenate([np.full(min(lag, n), oi[0] if n else 0.0), oi[:-lag]])[:n]
        with np.errstate(divide="ignore", invalid="ignore"):
            oi_change = np.where(previous > 0, (oi - previous) / previous * 100.0, 0.0)
        prev_close = np.concatenate([np.full(min(lag, n), close[0] if n else 0.0), close[:-lag]])[:n]
        with np.errstate(divide="ignore", invalid="ignore"):
            price_change = np.where(prev_close > 0, (close - prev_close) / prev_close * 100.0, 0.0)
        # Gleiches Vorzeichen = frisches Geld in Richtung des Trends, Gegenzeichen =
        # Glattstellungen. Das Produkt fasst beides in einer Zahl zusammen.
        divergence = np.sign(oi_change) * np.abs(price_change)
        columns += [_clean(oi_change), _clean(divergence)]

    return columns


def build_feature_matrix(
    candles: Candles,
    rules: RuleConfig,
    derivatives: DerivativesConfig | None = None,
    series=None,
) -> FeatureMatrix | None:
    """Berechnet Indikatoren und Merkmalsvektoren für die gesamte Serie.

    ``series`` ist eine :class:`~trademind.derivatives.DerivativeSeries` passend zu den
    Kerzen; fehlt sie bei aktivierten Terminmarktmerkmalen, werden die Spalten neutral
    gefüllt, damit die Modelldimension gleich bleibt.

    Gibt ``None`` zurück, wenn die Historie kürzer als ``required_bars`` ist.
    """
    n = len(candles)
    need = required_bars(rules)
    if n < need:
        return None

    close = np.asarray(candles.close, dtype=np.float64)
    high = np.asarray(candles.high, dtype=np.float64)
    low = np.asarray(candles.low, dtype=np.float64)
    open_ = np.asarray(candles.open, dtype=np.float64)
    volume = np.asarray(candles.volume, dtype=np.float64)
    price = np.where(close > 0, close, np.nan)

    ema_fast = ema(close, rules.fast_ema)
    ema_slow = ema(close, rules.slow_ema)
    trend_period = rules.trend_filter_period or rules.slow_ema * 4
    trend_ema = ema(close, trend_period)

    rsi_arr = _clean(rsi(close, rules.rsi_period), 50.0)
    atr_raw = atr(high, low, close, rules.atr_period)
    atr_arr = _clean(atr_raw, close * 0.005)
    atr_arr = np.where(atr_arr > 0, atr_arr, np.maximum(close * 0.005, 1e-9))

    _, _, macd_hist = macd(close, rules.fast_ema, rules.slow_ema, 9)
    macd_hist = _clean(macd_hist)
    roc3 = _clean(roc(close, 3))
    roc12 = _clean(roc(close, 12))
    dpos = _clean(donchian_position(high, low, close, 20), 0.5)

    vol_short = _clean(rolling_std(close, 10))
    vol_long = _clean(rolling_std(close, 50))
    with np.errstate(divide="ignore", invalid="ignore"):
        vol_ratio = np.where(vol_long > 1e-12, vol_short / vol_long, 1.0)

    volume_mean = _clean(sma(volume, 50), float(np.mean(volume)) if volume.size else 0.0)
    volume_std = _clean(rolling_std(volume, 50))
    with np.errstate(divide="ignore", invalid="ignore"):
        volume_z = np.where(volume_std > 1e-12, (volume - volume_mean) / volume_std, 0.0)

    bar_range = np.maximum(high - low, 1e-12)
    body = np.abs(close - open_) / bar_range
    upper_wick = (high - np.maximum(open_, close)) / bar_range
    lower_wick = (np.minimum(open_, close) - low) / bar_range

    seconds_of_day = (np.asarray(candles.timestamp, dtype=np.int64) // 1000) % 86_400
    angle = 2.0 * np.pi * seconds_of_day.astype(np.float64) / 86_400.0

    rsi_prev = np.concatenate([np.full(min(3, n), rsi_arr[0]), rsi_arr[:-3]])[:n] if n > 3 else rsi_arr
    hist_prev = np.concatenate([macd_hist[:1], macd_hist[:-1]])

    columns = [
        _pct(ema_fast - ema_slow, price),
        _pct(close - ema_fast, price),
        _pct(close - trend_ema, price),
        (rsi_arr - 50.0) / 50.0,
        (rsi_arr - rsi_prev) / 50.0,
        _pct(macd_hist, price),
        _pct(macd_hist - hist_prev, price),
        _pct(atr_arr, price),
        vol_ratio - 1.0,
        roc3 * 100.0,
        roc12 * 100.0,
        dpos - 0.5,
        volume_z,
        body,
        upper_wick,
        lower_wick,
        np.sin(angle),
        np.cos(angle),
    ]
    if derivatives is not None and derivatives.enabled:
        columns += _derivative_columns(close, series, derivatives)
    values = np.column_stack([_clean(col) for col in columns])
    np.clip(values, -_CLIP_LIMIT, _CLIP_LIMIT, out=values)

    return FeatureMatrix(
        values=values,
        timestamp=np.asarray(candles.timestamp, dtype=np.int64),
        price=_clean(close),
        atr=atr_arr,
        rsi=rsi_arr,
        ema_fast=_clean(ema_fast, close),
        ema_slow=_clean(ema_slow, close),
        trend_ema=_clean(trend_ema, close),
        first_valid=need - 1,
        names=feature_names(derivatives),
    )


def compute_features(candles: Candles, rules: RuleConfig, index: int = -1) -> FeatureSnapshot | None:
    """Bequemlichkeits-Wrapper für einen einzelnen Zeitpunkt (Live-Loop, Tests)."""
    matrix = build_feature_matrix(candles, rules)
    if matrix is None:
        return None
    return matrix.snapshot(index)