Volatility Trading for AI Agents: VIX, Realized Vol, and Dispersion

import numpy as np
import pandas as pd
from scipy.stats import norm
import requests

# ---- Vol computation utilities ----

def historical_vol(prices: pd.Series, window: int = 30) -> pd.Series:
    """Annualized historical volatility over rolling window."""
    log_returns = np.log(prices / prices.shift(1))
    return log_returns.rolling(window).std() * np.sqrt(252) * 100

def realized_vol_intraday(intraday_prices: pd.Series, freq: '5min') -> float:
    """Realized variance from high-frequency 5-min returns, annualized."""
    resampled = intraday_prices.resample(freq).last().dropna()
    log_returns = np.log(resampled / resampled.shift(1)).dropna()
    daily_rv = np.sqrt((log_returns ** 2).sum())
    return daily_rv * np.sqrt(252) * 100

def iv_rv_zscore(iv_series: pd.Series, rv_series: pd.Series, lookback: int = 126) -> pd.Series:
    """Z-score of IV-RV spread against 6-month rolling history."""
    spread = iv_series - rv_series
    mu = spread.rolling(lookback).mean()
    sigma = spread.rolling(lookback).std()
    return (spread - mu) / sigma

# ---- ATM IV from option chain ----

def fetch_atm_iv(ticker: str, expiry_days: int = 30) -> float:
    """Fetch nearest ATM IV for a given ticker and tenor."""
    # In production: connect to Purple Flea Trading API or broker feed
    url = f"https://purpleflea.com/trading-api/options/{ticker}/atm-iv"
    params = {"expiry_days": expiry_days, "moneyness": "atm"}
    resp = requests.get(url, params=params, headers={"X-API-Key": API_KEY})
    resp.raise_for_status()
    return resp.json()["iv"]

# ---- Main signal generation ----

class IVRVSpreadAgent:
    def __init__(self, ticker: str, api_key: str):
        self.ticker = ticker
        self.api_key = api_key
        self.iv_history = []
        self.rv_history = []

    def update(self, current_iv: float, current_rv: float):
        self.iv_history.append(current_iv)
        self.rv_history.append(current_rv)

    def signal(self, entry_z: float = 1.5) -> str:
        """Return 'short_vol', 'long_vol', or 'neutral'."""
        if len(self.iv_history) < 126:
            return "neutral"
        iv_s = pd.Series(self.iv_history)
        rv_s = pd.Series(self.rv_history)
        z = iv_rv_zscore(iv_s, rv_s).iloc[-1]
        if z > entry_z:
            return "short_vol"  # IV expensive, sell options
        elif z < -entry_z:
            return "long_vol"   # IV cheap, buy options
        return "neutral"

    def execute_short_vol(self, notional: float = 10_000):
        """Sell ATM straddle: sell call + put at same strike, same expiry."""
        print(f"[SHORT VOL] Selling ATM straddle on {self.ticker}, notional=${notional:,}")
        # Purple Flea Trading API call here
        payload = {
            "action": "sell_straddle",
            "ticker": self.ticker,
            "notional": notional,
            "expiry_days": 30,
        }
        return requests.post(
            "https://purpleflea.com/trading-api/execute",
            json=payload,
            headers={"X-API-Key": self.api_key}
        ).json()

import numpy as np
from scipy.interpolate import RectBivariateSpline
import matplotlib.pyplot as plt

# ---- Build volatility surface from option quotes ----

class VolSurface:
    """Bivariate spline interpolation over (strike, tenor) grid."""

    def __init__(self, strikes: np.ndarray, tenors: np.ndarray, iv_grid: np.ndarray):
        """
        strikes: array of moneyness ratios (e.g., 0.8, 0.9, 1.0, 1.1, 1.2)
        tenors: array of days to expiry (e.g., 7, 14, 30, 60, 90, 180)
        iv_grid: 2D array (len(strikes) x len(tenors)) of implied vols
        """
        self.spline = RectBivariateSpline(strikes, tenors, iv_grid)

    def get_iv(self, strike: float, tenor: float) -> float:
        """Interpolate IV for arbitrary (strike, tenor) pair."""
        return float(self.spline(strike, tenor))

    def term_structure_slope(self, strike: float = 1.0) -> float:
        """Slope of IV from 30d to 90d at ATM — positive = contango."""
        iv_30 = self.get_iv(strike, 30)
        iv_90 = self.get_iv(strike, 90)
        return (iv_90 - iv_30) / 60  # per day slope

    def skew(self, tenor: float = 30) -> float:
        """Risk-reversal: IV(0.9 put) - IV(1.1 call). Negative = put skew."""
        return self.get_iv(0.9, tenor) - self.get_iv(1.1, tenor)

    def calendar_spread_signal(self, short_tenor: 30, long_tenor: 90,
                                 min_slope: 0.05) -> bool:
        """Enter calendar if term structure steep enough to be profitable."""
        slope = self.term_structure_slope()
        return slope > min_slope

# ---- Example: build surface from Purple Flea data ----

def build_surface_from_api(ticker: str, api_key: str) -> VolSurface:
    strikes = np.array([0.8, 0.9, 0.95, 1.0, 1.05, 1.1, 1.2])
    tenors  = np.array([7, 14, 30, 60, 90, 180])
    iv_grid = np.zeros((len(strikes), len(tenors)))

    for i, k in enumerate(strikes):
        for j, t in enumerate(tenors):
            iv_grid[i, j] = fetch_iv_from_api(ticker, k, t, api_key)

    return VolSurface(strikes, tenors, iv_grid)

# Term structure agent decision loop
def term_structure_agent(surface: VolSurface) -> dict:
    slope = surface.term_structure_slope()
    skew_val = surface.skew()
    calendar_ok = surface.calendar_spread_signal(30, 90)

    decision = {
        "slope_per_day": round(slope, 5),
        "skew_30d": round(skew_val, 2),
        "calendar_signal": calendar_ok,
        "action": "sell_30d_buy_90d" if calendar_ok else "hold"
    }
    return decision

import numpy as np
import pandas as pd
import requests

# ---- Implied correlation calculator ----

def implied_correlation(
    index_iv: float,
    constituent_ivs: list,
    weights: list,
    assume_equal_corr: bool = True
) -> float:
    """
    Back out implied average pairwise correlation from index IV.
    Simplification: assume equal pairwise correlation rho.
    index_var = sum(w_i^2 * sigma_i^2) + rho * sum_{i!=j}(w_i*w_j*sigma_i*sigma_j)
    """
    w = np.array(weights)
    s = np.array(constituent_ivs) / 100  # to decimal
    idx_var = (index_iv / 100) ** 2

    # Sum of squared weighted variances
    sum_ww_ss = np.sum((w ** 2) * (s ** 2))

    # Cross terms without correlation
    cross = 0.0
    n = len(w)
    for i in range(n):
        for j in range(i+1, n):
            cross += 2 * w[i] * w[j] * s[i] * s[j]

    if cross == 0:
        return 0.0

    rho = (idx_var - sum_ww_ss) / cross
    return np.clip(rho, -1.0, 1.0)


class DispersionAgent:
    def __init__(self, index: str, constituents: list, weights: list, api_key: str):
        self.index = index
        self.constituents = constituents
        self.weights = weights
        self.api_key = api_key
        self.positions = {}

    def fetch_ivs(self) -> tuple:
        """Fetch ATM IV for index and all constituents."""
        index_iv = fetch_atm_iv(self.index, 30)
        const_ivs = [fetch_atm_iv(t, 30) for t in self.constituents]
        return index_iv, const_ivs

    def dispersion_signal(self, entry_corr_threshold: float = 0.75) -> dict:
        """
        Enter dispersion if implied correlation > threshold.
        High implied corr = index vol expensive vs constituents.
        """
        idx_iv, const_ivs = self.fetch_ivs()
        rho = implied_correlation(idx_iv, const_ivs, self.weights)
        weighted_avg_iv = np.dot(self.weights, const_ivs)

        signal = {
            "index_iv": idx_iv,
            "weighted_stock_iv": weighted_avg_iv,
            "implied_corr": round(rho, 3),
            "iv_spread": round(idx_iv - weighted_avg_iv, 2),
            "trade": "dispersion" if rho > entry_corr_threshold else "hold"
        }
        return signal

    def execute_dispersion(self, index_notional: float = 50_000):
        """Sell index straddle, buy constituent straddles weighted by index weights."""
        signal = self.dispersion_signal()
        if signal["trade"] != "dispersion":
            return {"status": "no_trade", "reason": "implied corr below threshold"}

        orders = []

        # Short index straddle
        orders.append({
            "action": "sell_straddle",
            "ticker": self.index,
            "notional": index_notional,
            "expiry_days": 30
        })

        # Long constituent straddles
        for ticker, w in zip(self.constituents, self.weights):
            orders.append({
                "action": "buy_straddle",
                "ticker": ticker,
                "notional": index_notional * w,
                "expiry_days": 30
            })

        # Execute via Purple Flea Trading API
        resp = requests.post(
            "https://purpleflea.com/trading-api/batch-execute",
            json={"orders": orders},
            headers={"X-API-Key": self.api_key}
        )
        return resp.json()

# ---- Run the agent ----
if __name__ == "__main__":
    agent = DispersionAgent(
        index="SPY",
        constituents=["AAPL", "MSFT", "NVDA", "AMZN", "GOOGL"],
        weights=[0.07, 0.065, 0.062, 0.056, 0.044],
        api_key="YOUR_PURPLEFLEA_API_KEY"
    )
    result = agent.execute_dispersion(index_notional=50_000)
    print(result)

import asyncio
import logging
from dataclasses import dataclass
from typing import Optional
import requests

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
log = logging.getLogger("vol_arb_agent")

@dataclass
class VolArbitrageConfig:
    api_key: str
    max_notional_per_trade: float = 25_000
    max_total_notional: float = 100_000
    iv_rv_entry_z: float = 1.5
    dispersion_corr_threshold: float = 0.75
    stop_loss_multiple: float = 3.0
    gamma_hedge_threshold: float = 2.0
    scan_interval_seconds: int = 300


class VolArbitrageAgent:
    def __init__(self, config: VolArbitrageConfig):
        self.config = config
        self.active_positions = []
        self.total_notional = 0.0

    def get_account_balance(self) -> float:
        resp = requests.get(
            "https://purpleflea.com/wallet-api/balance",
            headers={"X-API-Key": self.config.api_key}
        )
        return resp.json()["usdc_balance"]

    def check_risk_limits(self, proposed_notional: float) -> bool:
        remaining = self.config.max_total_notional - self.total_notional
        return proposed_notional <= min(self.config.max_notional_per_trade, remaining)

    def classify_regime(self, vix: float) -> str:
        if vix < 15:    return "calm"
        elif vix < 22: return "normal"
        elif vix < 35: return "elevated"
        else:          return "crisis"

    async def run(self):
        log.info("Vol Arb Agent starting — Purple Flea Trading API")
        while True:
            try:
                vix = self.fetch_vix()
                regime = self.classify_regime(vix)
                log.info(f"VIX={vix:.1f} | Regime={regime}")

                if regime in ("calm", "normal"):
                    # Primary: short vol via IV-RV spread
                    await self.scan_iv_rv_opportunities()
                    # Secondary: dispersion if implied corr elevated
                    await self.scan_dispersion_opportunities()
                elif regime == "elevated":
                    # Mixed: small positions, focus on calendar spreads
                    await self.scan_term_structure_opportunities()
                else:
                    # Crisis: de-risk, exit shorts, consider long vol
                    log.warning("Crisis regime — liquidating short vol positions")
                    await self.exit_short_vol_positions()

                await asyncio.sleep(self.config.scan_interval_seconds)

            except Exception as e:
                log.error(f"Agent error: {e}")
                await asyncio.sleep(60)

    def fetch_vix(self) -> float:
        resp = requests.get(
            "https://purpleflea.com/trading-api/market-data/vix",
            headers={"X-API-Key": self.config.api_key}
        )
        return resp.json()["vix"]


if __name__ == "__main__":
    config = VolArbitrageConfig(
        api_key="YOUR_PURPLEFLEA_API_KEY",
        max_notional_per_trade=20_000,
        max_total_notional=80_000
    )
    agent = VolArbitrageAgent(config)
    asyncio.run(agent.run())