backend/features/technical.py

"""
Technical Features — Compute all standard TA indicators using pandas-ta.
Returns a feature DataFrame that can be fed into the signal engine.
"""
import logging

import pandas as pd
import pandas_ta as ta

from config import SIGNAL_PARAMS as SP

logger = logging.getLogger(__name__)


def compute_technical_features(df: pd.DataFrame, ticker: str = "") -> pd.DataFrame:
    """
    Compute comprehensive technical indicators for a single ticker's OHLCV DataFrame.

    Args:
        df: DataFrame with columns Open, High, Low, Close, Volume (DatetimeIndex)
        ticker: For logging

    Returns:
        DataFrame with original OHLCV + all computed features
    """
    if df.empty or len(df) < 50:
        logger.warning(f"{ticker}: Insufficient data ({len(df)} rows) for technical features")
        return df

    features = df.copy()

    # ── Trend Indicators ──
    features[f"sma_{SP['sma_short']}"] = ta.sma(features["Close"], length=SP["sma_short"])
    features[f"sma_{SP['sma_medium']}"] = ta.sma(features["Close"], length=SP["sma_medium"])
    features[f"sma_{SP['sma_long']}"] = ta.sma(features["Close"], length=SP["sma_long"])
    features[f"ema_{SP['ema_fast']}"] = ta.ema(features["Close"], length=SP["ema_fast"])
    features[f"ema_{SP['ema_slow']}"] = ta.ema(features["Close"], length=SP["ema_slow"])

    # MACD
    macd = ta.macd(features["Close"], fast=SP["macd_fast"], slow=SP["macd_slow"], signal=SP["macd_signal"])
    if macd is not None:
        features = pd.concat([features, macd], axis=1)

    # ADX (trend strength)
    adx = ta.adx(features["High"], features["Low"], features["Close"], length=14)
    if adx is not None:
        features = pd.concat([features, adx], axis=1)

    # ── Momentum Indicators ──
    features["rsi_14"] = ta.rsi(features["Close"], length=14)
    features["rsi_7"] = ta.rsi(features["Close"], length=7)

    stoch = ta.stoch(features["High"], features["Low"], features["Close"])
    if stoch is not None:
        features = pd.concat([features, stoch], axis=1)

    features["willr"] = ta.willr(features["High"], features["Low"], features["Close"])
    features["roc_10"] = ta.roc(features["Close"], length=10)
    features["roc_20"] = ta.roc(features["Close"], length=20)
    features["cci_20"] = ta.cci(features["High"], features["Low"], features["Close"], length=20)

    # ── Volatility Indicators ──
    bb = ta.bbands(features["Close"], length=SP["bb_period"], std=SP["bb_std"])
    if bb is not None:
        features = pd.concat([features, bb], axis=1)

    features["atr_14"] = ta.atr(features["High"], features["Low"], features["Close"], length=SP["atr_period"])
    features["atr_pct"] = (features["atr_14"] / features["Close"]) * 100  # ATR as % of price

    kc = ta.kc(features["High"], features["Low"], features["Close"])
    if kc is not None:
        features = pd.concat([features, kc], axis=1)

    # ── Volume Indicators ──
    features["obv"] = ta.obv(features["Close"], features["Volume"])
    features[f"vol_sma_{SP['volume_sma_period']}"] = ta.sma(features["Volume"], length=SP["volume_sma_period"])
    features["vol_ratio"] = features["Volume"] / features[f"vol_sma_{SP['volume_sma_period']}"]

    # ── Price Position Relative to MAs ──
    features["price_vs_sma20"] = (features["Close"] / features[f"sma_{SP['sma_short']}"]) - 1
    features["price_vs_sma50"] = (features["Close"] / features[f"sma_{SP['sma_medium']}"]) - 1
    features["price_vs_sma200"] = (features["Close"] / features[f"sma_{SP['sma_long']}"]) - 1

    # ── Crossover Signals (binary) ──
    sma_s = features[f"sma_{SP['sma_short']}"]
    sma_m = features[f"sma_{SP['sma_medium']}"]
    features["golden_cross"] = ((sma_s > sma_m) & (sma_s.shift(1) <= sma_m.shift(1))).astype(int)
    features["death_cross"] = ((sma_s < sma_m) & (sma_s.shift(1) >= sma_m.shift(1))).astype(int)

    ema_f = features[f"ema_{SP['ema_fast']}"]
    ema_sl = features[f"ema_{SP['ema_slow']}"]
    features["ema_bullish_cross"] = ((ema_f > ema_sl) & (ema_f.shift(1) <= ema_sl.shift(1))).astype(int)

    # ── Support / Resistance (rolling highs/lows) ──
    lookback = SP["breakout_lookback"]
    features["resistance"] = features["High"].rolling(lookback).max()
    features["support"] = features["Low"].rolling(lookback).min()
    features["at_resistance"] = (features["Close"] >= features["resistance"] * 0.98).astype(int)
    features["at_support"] = (features["Close"] <= features["support"] * 1.02).astype(int)

    # ── Bollinger Band Position ──
    bb_upper = features.get(f"BBU_{SP['bb_period']}_{SP['bb_std']}")
    bb_lower = features.get(f"BBL_{SP['bb_period']}_{SP['bb_std']}")
    if bb_upper is not None and bb_lower is not None:
        bb_width = bb_upper - bb_lower
        features["bb_pct"] = (features["Close"] - bb_lower) / bb_width.replace(0, float("nan"))
        features["bb_squeeze"] = (bb_width / features["Close"] * 100)  # Width as % of price

    logger.info(f"{ticker}: Computed {len(features.columns) - len(df.columns)} technical features")
    return features


def get_latest_features(features_df: pd.DataFrame) -> dict:
    """Extract the latest row of features as a flat dict (for signal engine)."""
    if features_df.empty:
        return {}
    last = features_df.iloc[-1]
    return {k: (round(float(v), 4) if pd.notna(v) else None) for k, v in last.items()}