Volatility Surface Analyzer

Volatility Surface Analyzer

Overview

Analyzes realized vs implied volatility, constructs vol surfaces, detects mispricing, and generates vol-based trading signals. Essential for options-aware FX trading.

1. Realized Volatility Engine

import pandas as pd
import numpy as np
from scipy import stats

class VolatilityAnalyzer:
    """Complete volatility analysis toolkit."""

    @staticmethod
    def realized_vol(returns: pd.Series, windows: list[int] = [5, 10, 20, 60, 252]) -> pd.DataFrame:
        """Multi-window realized volatility (annualized)."""
        result = pd.DataFrame(index=returns.index)
        for w in windows:
            result[f"rv_{w}d"] = returns.rolling(w).std() * np.sqrt(252) * 100
        return result

    @staticmethod
    def volatility_cone(returns: pd.Series, windows: list[int] = [5, 10, 20, 60]) -> dict:
        """Vol cone: distribution of RV at each window — shows where current vol sits historically."""
        cone = {}
        for w in windows:
            rv = returns.rolling(w).std() * np.sqrt(252) * 100
            rv = rv.dropna()
            current = rv.iloc[-1]
            percentile = (rv < current).mean() * 100
            cone[f"{w}d"] = {
                "current": round(current, 2),
                "percentile": round(percentile, 1),
                "min": round(rv.min(), 2), "p25": round(rv.quantile(0.25), 2),
                "median": round(rv.median(), 2), "p75": round(rv.quantile(0.75), 2),
                "max": round(rv.max(), 2),
                "regime": "HIGH" if percentile > 75 else "LOW" if percentile < 25 else "NORMAL",
            }
        return cone

    @staticmethod
    def iv_rank(current_iv: float, iv_high_52w: float, iv_low_52w: float) -> dict:
        """IV Rank: where current IV sits in 52-week range (0-100)."""
        iv_range = iv_high_52w - iv_low_52w
        rank = ((current_iv - iv_low_52w) / iv_range * 100) if iv_range > 0 else 50
        return {
            "iv_rank": round(rank, 1),
            "current_iv": current_iv,
            "52w_high": iv_high_52w, "52w_low": iv_low_52w,
            "signal": "SELL VOL (premium rich)" if rank > 80 else "BUY VOL (premium cheap)" if rank < 20 else "NEUTRAL",
        }

    @staticmethod
    def variance_risk_premium(realized_vol: float, implied_vol: float) -> dict:
        """VRP = IV - RV. Positive = vol sellers earn premium."""
        vrp = implied_vol - realized_vol
        return {
            "vrp": round(vrp, 2),
            "implied_vol": round(implied_vol, 2),
            "realized_vol": round(realized_vol, 2),
            "signal": "SELL VOL — IV overpricing risk" if vrp > 3 else "BUY VOL — IV underpricing risk" if vrp < -2 else "FAIR",
            "note": "Positive VRP is normal (insurance premium). Extreme readings are actionable.",
        }

    @staticmethod
    def vol_term_structure(ivs_by_expiry: dict) -> dict:
        """Analyze IV term structure shape: contango (normal) vs backwardation (fear)."""
        expiries = sorted(ivs_by_expiry.keys())
        ivs = [ivs_by_expiry[e] for e in expiries]
        if len(ivs) < 2:
            return {"shape": "insufficient data"}
        slope = (ivs[-1] - ivs[0]) / len(ivs)
        shape = "CONTANGO (normal)" if slope > 0.5 else "BACKWARDATION (fear/event)" if slope < -0.5 else "FLAT"
        return {
            "shape": shape, "slope": round(slope, 3),
            "front_iv": round(ivs[0], 2), "back_iv": round(ivs[-1], 2),
            "signal": "Risk-off positioning" if "BACKWARDATION" in shape else "Normal conditions",
            "term_structure": {e: round(v, 2) for e, v in zip(expiries, ivs)},
        }

    @staticmethod
    def vol_regime_signal(returns: pd.Series) -> dict:
        """Generate trading signals from vol regime analysis."""
        cone = VolatilityAnalyzer.volatility_cone(returns)
        rv_20 = cone.get("20d", {})
        regime = rv_20.get("regime", "NORMAL")
        return {
            "regime": regime,
            "rv_20d": rv_20.get("current", 0),
            "percentile": rv_20.get("percentile", 50),
            "strategy_advice": {
                "HIGH": "Widen stops, reduce size. Vol expansion phase — breakout strategies work.",
                "LOW": "Tighten stops, normal size. Vol compression — expect breakout soon. Set range orders.",
                "NORMAL": "Standard parameters. Follow directional signals.",
            }.get(regime, ""),
        }