backend/features/volatility.py

"""
Volatility Features — Historical volatility, compression detection, regime.
"""
import logging

import numpy as np
import pandas as pd

logger = logging.getLogger(__name__)


def compute_volatility_features(df: pd.DataFrame, ticker: str = "") -> pd.DataFrame:
    """
    Compute volatility-related features.

    Adds:
        - Historical volatility (10d, 20d, 60d)
        - Volatility ratio (short-term vs long-term)
        - Bollinger bandwidth compression detection
        - Average True Range % of price
        - Volatility regime classification
    """
    if df.empty or len(df) < 60:
        logger.warning(f"{ticker}: Insufficient data for volatility features")
        return df

    features = df.copy()
    log_returns = np.log(features["Close"] / features["Close"].shift(1))

    # ── Historical Volatility (annualized) ──
    features["hv_10d"] = log_returns.rolling(10).std() * np.sqrt(252) * 100
    features["hv_20d"] = log_returns.rolling(20).std() * np.sqrt(252) * 100
    features["hv_60d"] = log_returns.rolling(60).std() * np.sqrt(252) * 100

    # ── Volatility Ratio (short vs long — compression signal) ──
    features["vol_ratio_10_60"] = features["hv_10d"] / features["hv_60d"].replace(0, np.nan)

    # ── Compression Detection ──
    # When short-term vol < 0.7x long-term vol, a breakout may be incoming
    features["vol_compressed"] = (features["vol_ratio_10_60"] < 0.7).astype(int)
    features["vol_expanding"] = (features["vol_ratio_10_60"] > 1.3).astype(int)

    # ── True Range metrics ──
    high_low = features["High"] - features["Low"]
    high_pc = abs(features["High"] - features["Close"].shift(1))
    low_pc = abs(features["Low"] - features["Close"].shift(1))
    features["true_range"] = pd.concat([high_low, high_pc, low_pc], axis=1).max(axis=1)
    features["true_range_pct"] = (features["true_range"] / features["Close"]) * 100

    # ── Daily Range as % (intraday volatility proxy) ──
    features["daily_range_pct"] = ((features["High"] - features["Low"]) / features["Close"]) * 100
    features["avg_range_20d"] = features["daily_range_pct"].rolling(20).mean()

    # ── Volatility Regime ──
    def _vol_regime(hv20):
        if pd.isna(hv20):
            return "unknown"
        if hv20 < 15:
            return "low"
        elif hv20 < 30:
            return "medium"
        elif hv20 < 50:
            return "high"
        else:
            return "extreme"

    features["vol_regime"] = features["hv_20d"].apply(_vol_regime)

    logger.info(f"{ticker}: Computed volatility features")
    return features