""" 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