Volatility Arbitrage Frameworks for AI Agents

What Is Volatility Arbitrage?

Volatility arbitrage is a market-neutral strategy that profits from the difference between an option's implied volatility (IV) — the market's expectation of future price movement encoded in the option price — and the realized volatility (RV) that actually occurs over the option's life.

Unlike directional trading, vol arb is agnostic to whether the underlying asset goes up or down. The bet is purely on the size of price swings. If you believe an asset will move less than the market implies, you sell volatility. If you expect larger swings, you buy volatility.

The Vol Risk Premium (VRP)

In most liquid markets — equities, FX, crypto — implied volatility systematically exceeds realized volatility over time. This volatility risk premium exists because option sellers demand compensation for bearing uncertainty. Buyers pay a structural premium for insurance. AI agents can systematically harvest this premium by selling options and delta-hedging the directional risk.

Vol Arb Variants

• Calendar spread arb: Front-month IV vs back-month IV mispricing
• Skew arb: Put skew vs realized downside vol asymmetry
• Cross-asset vol arb: Correlated assets with diverging IVs
• Term structure arb: Vol curve shape vs historical term structure
• Realized vs implied: The pure VRP harvest strategy

Understanding the Volatility Surface

The volatility surface is a 3-D structure mapping implied volatility across all strikes and expiries for a given underlying. It encodes the market's entire distribution of future price outcomes. Vol arb opportunities often manifest as surface distortions — local inconsistencies that violate theoretical arbitrage bounds.

Surface Dimensions

• Strike axis (moneyness): Deep OTM puts carry higher IV than ATM (vol skew / smile)
• Time axis (term structure): Near-term vol often differs from long-dated expectations
• Bid-ask spread: The tradeable width defines real arb bounds

Arbitrage-Free Constraints

A valid vol surface must satisfy several no-arbitrage conditions. Any violation represents a theoretical (though not always practical) arb opportunity:

import numpy as np
from scipy.interpolate import RectBivariateSpline
from scipy.stats import norm

class VolSurface:
    """Implied volatility surface with arbitrage detection."""

    def __init__(self, strikes: list, expiries: list, ivs: np.ndarray):
        # ivs[i, j] = IV for strike strikes[i], expiry expiries[j]
        self.strikes = np.array(strikes)
        self.expiries = np.array(expiries)  # in years
        self.ivs = ivs
        self._spline = RectBivariateSpline(strikes, expiries, ivs, kx=3, ky=3)

    def get_iv(self, strike: float, expiry: float) -> float:
        """Interpolate IV for arbitrary strike/expiry."""
        return float(self._spline(strike, expiry))

    def calendar_violations(self) -> list:
        """Find calendar spread arbitrage (total variance must be non-decreasing)."""
        violations = []
        for i, K in enumerate(self.strikes):
            for j in range(len(self.expiries) - 1):
                T1, T2 = self.expiries[j], self.expiries[j+1]
                var1 = self.ivs[i, j]**2 * T1
                var2 = self.ivs[i, j+1]**2 * T2
                if var2 < var1 - 0.0001:  # tolerance
                    violations.append({
                        'strike': K,
                        'front_expiry': T1,
                        'back_expiry': T2,
                        'magnitude': var1 - var2
                    })
        return violations

    def butterfly_violations(self) -> list:
        """Find butterfly arb (density must be non-negative)."""
        violations = []
        dK = np.diff(self.strikes)[0]  # assume uniform grid
        for j, T in enumerate(self.expiries):
            for i in range(1, len(self.strikes) - 1):
                iv_left = self.ivs[i-1, j]
                iv_mid  = self.ivs[i, j]
                iv_right = self.ivs[i+1, j]
                # Breeden-Litzenberger density approximation
                call_price = lambda iv, K: self._bs_call(
                    self.strikes[i], K, T, iv)
                c_left  = call_price(iv_left,  self.strikes[i-1])
                c_mid   = call_price(iv_mid,   self.strikes[i])
                c_right = call_price(iv_right, self.strikes[i+1])
                density = (c_left - 2*c_mid + c_right) / dK**2
                if density < 0:
                    violations.append({
                        'strike': self.strikes[i],
                        'expiry': T,
                        'density': density
                    })
        return violations

    def _bs_call(self, S, K, T, sigma, r=0.0) -> float:
        if T <= 0: return max(0, S - K)
        d1 = (np.log(S/K) + (r + 0.5*sigma**2)*T) / (sigma*np.sqrt(T))
        d2 = d1 - sigma*np.sqrt(T)
        return S*norm.cdf(d1) - K*np.exp(-r*T)*norm.cdf(d2)

Greeks and Delta-Neutral Hedging

A vol arb position earns from vol mispricing, but it also carries directional exposure (delta), time decay (theta), and second-order vol sensitivity (vanna, volga). An agent must manage these exposures continuously.

Key Greeks for Vol Arb

P&L of a Delta-Hedged Option

The daily P&L of a delta-hedged option position is driven by the gamma P&L vs theta decay trade-off:

When realized vol exceeds implied, a long gamma (long option) position earns. When realized falls short, a short gamma (short option) position earns. The vol arb thesis is simply a bet on which regime will persist.

import numpy as np
from scipy.stats import norm

class BlackScholesGreeks:
    """Fast analytical Greeks calculator."""

    def __init__(self, S, K, T, sigma, r=0.0, option_type='call'):
        self.S = S; self.K = K; self.T = T
        self.sigma = sigma; self.r = r
        self.option_type = option_type
        if T > 0:
            self.d1 = (np.log(S/K) + (r + 0.5*sigma**2)*T) / (sigma*np.sqrt(T))
            self.d2 = self.d1 - sigma*np.sqrt(T)
        else:
            self.d1 = self.d2 = np.inf if S >= K else -np.inf

    @property
    def delta(self) -> float:
        if self.option_type == 'call':
            return norm.cdf(self.d1)
        return norm.cdf(self.d1) - 1

    @property
    def gamma(self) -> float:
        return norm.pdf(self.d1) / (self.S * self.sigma * np.sqrt(self.T))

    @property
    def theta(self) -> float:
        # Per calendar day
        term1 = -(self.S * norm.pdf(self.d1) * self.sigma) / (2 * np.sqrt(self.T))
        if self.option_type == 'call':
            term2 = -self.r * self.K * np.exp(-self.r*self.T) * norm.cdf(self.d2)
        else:
            term2 = self.r * self.K * np.exp(-self.r*self.T) * norm.cdf(-self.d2)
        return (term1 + term2) / 365

    @property
    def vega(self) -> float:
        # Per 1% change in vol (expressed as decimal)
        return self.S * norm.pdf(self.d1) * np.sqrt(self.T) * 0.01

    @property
    def vanna(self) -> float:
        return -norm.pdf(self.d1) * self.d2 / self.sigma

    @property
    def volga(self) -> float:
        return self.S * norm.pdf(self.d1) * np.sqrt(self.T) * self.d1 * self.d2 / self.sigma

    def gamma_pnl(self, realized_vol: float, dt: float = 1/252) -> float:
        """Expected daily gamma P&L given realized vol."""
        return 0.5 * self.gamma * self.S**2 * (realized_vol**2 - self.sigma**2) * dt


class DeltaHedger:
    """Real-time delta hedging engine for vol arb positions."""

    def __init__(self, api_key: str, hedge_threshold: float = 0.05):
        self.api_key = api_key
        self.hedge_threshold = hedge_threshold  # rehedge if |delta_drift| > 5%
        self.positions = {}
        self.hedge_position = 0.0  # units of underlying held

    def add_position(self, option_id: str, greeks: BlackScholesGreeks, quantity: float):
        self.positions[option_id] = {'greeks': greeks, 'qty': quantity}

    def net_delta(self) -> float:
        total = 0.0
        for pos in self.positions.values():
            total += pos['greeks'].delta * pos['qty']
        total += self.hedge_position
        return total

    def needs_rehedge(self) -> bool:
        return abs(self.net_delta()) > self.hedge_threshold

    def compute_hedge_trade(self) -> float:
        """Returns quantity of underlying to buy/sell to flatten delta."""
        target_hedge = -sum(
            pos['greeks'].delta * pos['qty']
            for pos in self.positions.values()
        )
        return target_hedge - self.hedge_position

Realized Volatility Estimation

The quality of a vol arb strategy depends critically on accurate realized vol estimates. Several estimators exist, each with different bias and efficiency trade-offs.

Estimator Comparison

import numpy as np
import pandas as pd

class RealizedVolEstimator:
    """Multiple realized volatility estimators."""

    def __init__(self, df: pd.DataFrame, trading_days: int = 252):
        """df must have columns: open, high, low, close (all log-prices OK)."""
        self.o = np.log(df['open'])
        self.h = np.log(df['high'])
        self.l = np.log(df['low'])
        self.c = np.log(df['close'])
        self.c_prev = self.c.shift(1)
        self.ann = trading_days

    def close_to_close(self, window: int = 20) -> pd.Series:
        returns = self.c.diff()
        return returns.rolling(window).std() * np.sqrt(self.ann)

    def parkinson(self, window: int = 20) -> pd.Series:
        hl_sq = (self.h - self.l)**2
        factor = 1 / (4 * np.log(2))
        daily_var = factor * hl_sq
        return np.sqrt(daily_var.rolling(window).mean() * self.ann)

    def garman_klass(self, window: int = 20) -> pd.Series:
        hl_term = 0.5 * (self.h - self.l)**2
        co_term = -(2*np.log(2)-1) * (self.c - self.o)**2
        daily_var = (hl_term + co_term)
        return np.sqrt(daily_var.rolling(window).mean() * self.ann)

    def yang_zhang(self, window: int = 20) -> pd.Series:
        oc = self.o - self.c_prev
        co = self.c - self.o
        sigma_oc = oc.rolling(window).var()
        sigma_co = co.rolling(window).var()
        k = 0.34 / (1.34 + (window+1)/(window-1))
        rs_term = (self.h-self.o)*(self.h-self.c) + (self.l-self.o)*(self.l-self.c)
        sigma_rs = rs_term.rolling(window).mean()
        daily_var = sigma_oc + k*sigma_co + (1-k)*sigma_rs
        return np.sqrt(daily_var * self.ann)

    def best_estimate(self, window: int = 20) -> pd.Series:
        """Yang-Zhang with bias correction."""
        raw = self.yang_zhang(window)
        # Small-sample bias correction
        correction = 1 + 1 / (4 * (window - 1))
        return raw / correction

Vol Arb Framework Class

A complete VolArbFramework class ties together surface analysis, realized vol estimation, signal generation, position sizing, and execution. This is the core engine your agent deploys.

import asyncio
import httpx
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime

PURPLE_FLEA_BASE = "https://api.purpleflea.com/v1"

@dataclass
class VolArbSignal:
    underlying: str
    strike: float
    expiry: float      # years to expiry
    option_type: str   # 'call' or 'put'
    implied_vol: float
    realized_vol: float
    vrp: float         # IV - RV (positive = sell opportunity)
    z_score: float     # standardized signal strength
    recommended_action: str  # 'sell_vol', 'buy_vol', 'hold'

@dataclass
class VolPosition:
    signal: VolArbSignal
    option_units: float   # positive = long, negative = short
    delta_hedge: float    # units of underlying for delta hedge
    vega_exposure: float  # target vega in dollars
    entry_price: float
    entry_time: datetime = field(default_factory=datetime.utcnow)

class VolArbFramework:
    """
    Complete volatility arbitrage framework for AI agents.
    Operates on Purple Flea trading infrastructure.
    """

    def __init__(self,
                 api_key: str,
                 max_vega_usd: float = 10000,
                 vrp_threshold: float = 0.04,
                 z_entry: float = 1.5,
                 z_exit: float = 0.3):
        self.api_key = api_key   # pf_live_... style key
        self.max_vega_usd = max_vega_usd
        self.vrp_threshold = vrp_threshold  # minimum IV-RV gap to trade
        self.z_entry = z_entry
        self.z_exit = z_exit
        self.positions: list[VolPosition] = []
        self.signal_history: list[VolArbSignal] = []
        self.hedger = DeltaHedger(api_key)
        self.vol_estimator = None  # set after loading OHLC data
        self._session = None

    async def initialize(self, underlying: str, lookback_days: int = 60):
        """Fetch OHLC data and warm up estimators."""
        self._session = httpx.AsyncClient(
            headers={"Authorization": f"Bearer {self.api_key}"},
            timeout=30
        )
        ohlc = await self._fetch_ohlc(underlying, lookback_days)
        self.vol_estimator = RealizedVolEstimator(ohlc)
        self.current_rv = float(self.vol_estimator.best_estimate().iloc[-1])

    async def _fetch_ohlc(self, underlying: str, days: int) -> pd.DataFrame:
        resp = await self._session.get(
            f"{PURPLE_FLEA_BASE}/market/ohlc",
            params={"asset": underlying, "days": days, "interval": "1d"}
        )
        data = resp.json()["candles"]
        return pd.DataFrame(data, columns=['open','high','low','close'])

    async def scan_surface(self, underlying: str) -> list[VolArbSignal]:
        """Fetch live options chain and identify vol arb signals."""
        resp = await self._session.get(
            f"{PURPLE_FLEA_BASE}/options/chain",
            params={"asset": underlying}
        )
        chain = resp.json()["options"]
        signals = []
        for opt in chain:
            iv = opt["implied_vol"]
            vrp = iv - self.current_rv
            # Compute z-score vs rolling signal history
            z = self._zscore(vrp)
            action = "hold"
            if z > self.z_entry and vrp > self.vrp_threshold:
                action = "sell_vol"
            elif z < -self.z_entry and vrp < -self.vrp_threshold:
                action = "buy_vol"
            sig = VolArbSignal(
                underlying=underlying,
                strike=opt["strike"],
                expiry=opt["days_to_expiry"] / 365,
                option_type=opt["type"],
                implied_vol=iv,
                realized_vol=self.current_rv,
                vrp=vrp,
                z_score=z,
                recommended_action=action
            )
            signals.append(sig)
            self.signal_history.append(sig)
        return [s for s in signals if s.recommended_action != "hold"]

    def _zscore(self, current_vrp: float) -> float:
        if len(self.signal_history) < 30:
            return 0.0
        historical_vrps = [s.vrp for s in self.signal_history[-252:]]
        mu = np.mean(historical_vrps)
        sigma = np.std(historical_vrps)
        if sigma < 0.0001: return 0.0
        return (current_vrp - mu) / sigma

    def size_position(self, signal: VolArbSignal, spot: float) -> float:
        """Kelly-capped vega sizing."""
        greeks = BlackScholesGreeks(
            S=spot, K=signal.strike,
            T=signal.expiry, sigma=signal.implied_vol,
            option_type=signal.option_type
        )
        vega_per_contract = greeks.vega * spot * 100  # assume 100x multiplier
        if vega_per_contract <= 0: return 0
        max_contracts = self.max_vega_usd / vega_per_contract
        # Scale by signal strength (z-score normalized to [0, 1])
        strength = min(1.0, abs(signal.z_score) / (3 * self.z_entry))
        return max(1, int(max_contracts * strength))

    async def execute_vol_arb(self, signal: VolArbSignal, spot: float):
        """Execute option + delta hedge via Purple Flea API."""
        qty = self.size_position(signal, spot)
        if qty == 0:
            return
        direction = "sell" if signal.recommended_action == "sell_vol" else "buy"
        # Leg 1: Option
        opt_resp = await self._session.post(
            f"{PURPLE_FLEA_BASE}/orders/option",
            json={
                "asset": signal.underlying,
                "strike": signal.strike,
                "expiry_days": int(signal.expiry * 365),
                "option_type": signal.option_type,
                "direction": direction,
                "quantity": qty,
                "order_type": "limit",
                "limit_iv": signal.implied_vol
            }
        )
        opt_order = opt_resp.json()
        # Leg 2: Delta hedge in underlying
        greeks = BlackScholesGreeks(
            S=spot, K=signal.strike,
            T=signal.expiry, sigma=signal.implied_vol,
            option_type=signal.option_type
        )
        hedge_qty = -greeks.delta * qty * (1 if direction == "buy" else -1)
        hedge_direction = "buy" if hedge_qty > 0 else "sell"
        hedge_resp = await self._session.post(
            f"{PURPLE_FLEA_BASE}/orders/spot",
            json={
                "asset": signal.underlying,
                "direction": hedge_direction,
                "quantity": abs(hedge_qty),
                "order_type": "market"
            }
        )
        return {"option_order": opt_order, "hedge_order": hedge_resp.json()}

    async def run_hedging_loop(self, interval_seconds: int = 60):
        """Continuous delta rehedging loop."""
        while True:
            if self.hedger.needs_rehedge():
                trade_qty = self.hedger.compute_hedge_trade()
                direction = "buy" if trade_qty > 0 else "sell"
                await self._session.post(
                    f"{PURPLE_FLEA_BASE}/orders/spot",
                    json={
                        "asset": "BTC",
                        "direction": direction,
                        "quantity": abs(trade_qty),
                        "order_type": "market"
                    }
                )
                self.hedger.hedge_position += trade_qty
            await asyncio.sleep(interval_seconds)

Term Structure Arbitrage

The volatility term structure — how IV varies with time to expiry — often creates arb opportunities when the curve moves to extremes. Calendar spreads and vol roll-down trades exploit these dynamics.

Vol Roll-Down

When the vol term structure is upward sloping (longer expiries have higher IV), a short option position benefits not just from theta decay but from the roll-down as the option moves to shorter expiry and lower IV — even if implied vol stays constant.

Calendar Spread Construction

class CalendarArbAnalyzer:
    """Identify calendar spread arb in vol term structure."""

    def __init__(self, surface: VolSurface):
        self.surface = surface

    def term_structure_slope(self, strike: float) -> pd.Series:
        """dIV/dT at given strike across all expiries."""
        expiries = self.surface.expiries
        ivs = [self.surface.get_iv(strike, T) for T in expiries]
        slopes = np.gradient(ivs, expiries)
        return pd.Series(slopes, index=expiries, name='iv_slope')

    def find_roll_down_candidates(self,
                                     min_slope: float = 0.02,
                                     min_expiry: float = 14/365) -> list:
        """ATM options with best roll-down characteristics."""
        candidates = []
        atm_strike = self.surface.strikes[len(self.surface.strikes) // 2]
        slopes = self.term_structure_slope(atm_strike)
        for T, slope in slopes.items():
            if T >= min_expiry and slope > min_slope:
                iv = self.surface.get_iv(atm_strike, T)
                candidates.append({
                    'expiry': T,
                    'iv': iv,
                    'slope': slope,
                    'daily_roll_bps': slope * (1/365) * 10000
                })
        return sorted(candidates, key=lambda x: -x['daily_roll_bps'])

    def optimal_calendar_spread(self) -> dict:
        """Find front/back month pair maximizing vega-weighted mispricing."""
        violations = self.surface.calendar_violations()
        if not violations:
            return {}
        # Sort by magnitude of violation
        best = max(violations, key=lambda x: x['magnitude'])
        return {
            'strike': best['strike'],
            'buy_expiry': best['back_expiry'],   # buy the back month
            'sell_expiry': best['front_expiry'],  # sell the front month
            'magnitude': best['magnitude'],
            'trade': 'calendar_spread'
        }

Skew Arbitrage

Volatility skew describes how IV varies across strikes at a fixed expiry. Put skew — the premium of OTM put IV over ATM IV — reflects demand for downside protection. Skew arb bets on mean reversion of this premium versus the actual realized skew of returns.

Measuring Skew

class SkewArbAnalyzer:
    """Risk reversal and butterfly skew arbitrage signals."""

    def __init__(self, surface: VolSurface, spot: float):
        self.surface = surface
        self.spot = spot
        self.history_rr: list[float] = []
        self.history_bf: list[float] = []

    def _delta_to_strike(self, delta: float, T: float, atm_iv: float) -> float:
        """Approximate strike from delta (put convention, delta > 0)."""
        d1_target = norm.ppf(delta)
        log_moneyness = d1_target * atm_iv * np.sqrt(T) - 0.5 * atm_iv**2 * T
        return self.spot * np.exp(-log_moneyness)

    def risk_reversal(self, T: float) -> float:
        atm_iv = self.surface.get_iv(self.spot, T)
        put_25_strike = self._delta_to_strike(0.25, T, atm_iv)
        call_25_strike = self._delta_to_strike(0.75, T, atm_iv)  # call 25d
        iv_put = self.surface.get_iv(put_25_strike, T)
        iv_call = self.surface.get_iv(call_25_strike, T)
        rr = iv_put - iv_call
        self.history_rr.append(rr)
        return rr

    def butterfly(self, T: float) -> float:
        atm_iv = self.surface.get_iv(self.spot, T)
        put_25_strike = self._delta_to_strike(0.25, T, atm_iv)
        call_25_strike = self._delta_to_strike(0.75, T, atm_iv)
        iv_put = self.surface.get_iv(put_25_strike, T)
        iv_call = self.surface.get_iv(call_25_strike, T)
        bf = (iv_put + iv_call) / 2 - atm_iv
        self.history_bf.append(bf)
        return bf

    def skew_signal(self, T: float, z_threshold: float = 1.5) -> dict:
        rr = self.risk_reversal(T)
        bf = self.butterfly(T)
        rr_z = self._zscore(rr, self.history_rr)
        bf_z = self._zscore(bf, self.history_bf)
        trades = []
        if rr_z > z_threshold:
            trades.append({'type': 'sell_rr', 'z': rr_z,
                           'desc': 'Sell put, buy call (fade skew premium)'})
        elif rr_z < -z_threshold:
            trades.append({'type': 'buy_rr', 'z': rr_z,
                           'desc': 'Buy put, sell call (long skew)'})
        if bf_z > z_threshold:
            trades.append({'type': 'sell_fly', 'z': bf_z,
                           'desc': 'Sell wings, buy body (fade vol of vol)'})
        return {'rr': rr, 'bf': bf, 'rr_z': rr_z, 'bf_z': bf_z, 'trades': trades}

    def _zscore(self, current: float, history: list) -> float:
        if len(history) < 20: return 0.0
        mu, sigma = np.mean(history), np.std(history)
        return (current - mu) / sigma if sigma > 1e-8 else 0.0

Transaction Cost Modeling

Options markets have wide bid-ask spreads. A vol arb with positive theoretical edge can easily lose money once transaction costs are modeled in. Your agent must compute the break-even IV edge before executing any trade.

class TransactionCostModel:
    """Model total cost of entering a vol arb position."""

    def __init__(self,
                 option_fee_per_contract: float = 0.002,  # 0.2% of premium
                 spot_fee_bps: float = 3,                  # 3 bps on hedge
                 slippage_bps: float = 2):                  # 2 bps assumed slippage
        self.option_fee_pct = option_fee_per_contract
        self.spot_fee_bps = spot_fee_bps / 10000
        self.slippage_bps = slippage_bps / 10000

    def option_cost(self, premium: float, quantity: int) -> float:
        return premium * quantity * self.option_fee_pct

    def hedge_cost(self, spot: float, delta_units: float) -> float:
        notional = spot * abs(delta_units)
        return notional * (self.spot_fee_bps + self.slippage_bps)

    def total_round_trip_cost(self,
                              premium: float,
                              quantity: int,
                              spot: float,
                              delta: float) -> float:
        # Entry + exit (2x) for option; entry + continuous rehedge estimate
        opt_cost = self.option_cost(premium, quantity) * 2
        hedge_cost_entry = self.hedge_cost(spot, delta * quantity)
        # Estimate rehedge cost: assume 5 rehedges over option life
        rehedge_cost = hedge_cost_entry * 5
        return opt_cost + hedge_cost_entry + rehedge_cost

    def break_even_iv_edge(self, premium: float, quantity: int,
                           spot: float, delta: float, vega: float) -> float:
        """Minimum IV edge needed to cover round-trip costs."""
        total_cost = self.total_round_trip_cost(premium, quantity, spot, delta)
        if vega * quantity == 0: return float('inf')
        return total_cost / (vega * quantity)

    def is_trade_viable(self,
                        iv_edge: float,  # positive = selling expensive vol
                        premium: float,
                        quantity: int,
                        spot: float,
                        delta: float,
                        vega: float) -> dict:
        be = self.break_even_iv_edge(premium, quantity, spot, delta, vega)
        net_edge = abs(iv_edge) - be
        return {
            'viable': net_edge > 0,
            'break_even_edge': be,
            'gross_edge': abs(iv_edge),
            'net_edge': net_edge,
            'edge_to_cost_ratio': abs(iv_edge) / be if be > 0 else 0
        }

Risk Management for Vol Arb

Vol arb carries unique risks distinct from directional trading. Understanding and monitoring these is essential for agent longevity.

Primary Risk Factors

• Vol-of-vol (vvol): Sudden IV spikes can cause large mark-to-market losses on short vega positions before realized vol confirms the move
• Gap risk: Overnight or weekend gaps can cause realized vol to spike suddenly, devastating short gamma positions
• Correlation breakdown: Multi-asset vol arb can suffer when inter-asset correlations shift sharply
• Liquidity risk: Options can become illiquid precisely when you need to exit — volatility spikes drain market depth
• Model risk: Black-Scholes Greeks are only approximations; higher-order terms matter near expiry or at extremes

class VolArbRiskManager:
    """Portfolio-level risk controls for vol arb positions."""

    def __init__(self,
                 max_net_vega_usd: float = 50000,
                 max_net_gamma_usd: float = 5000,
                 max_drawdown_pct: float = 0.15,
                 daily_stop_loss_usd: float = 2000):
        self.max_net_vega_usd = max_net_vega_usd
        self.max_net_gamma_usd = max_net_gamma_usd
        self.max_drawdown_pct = max_drawdown_pct
        self.daily_stop_loss_usd = daily_stop_loss_usd
        self.daily_pnl = 0.0
        self.peak_pnl = 0.0

    def portfolio_greeks(self, positions: list[VolPosition], spot: float) -> dict:
        net_vega = net_gamma = net_delta = 0.0
        for pos in positions:
            sig = pos.signal
            g = BlackScholesGreeks(
                S=spot, K=sig.strike, T=sig.expiry,
                sigma=sig.implied_vol, option_type=sig.option_type
            )
            qty = pos.option_units
            net_vega  += g.vega * qty * spot * 100
            net_gamma += g.gamma * qty * spot**2 * 100 * 0.01
            net_delta += g.delta * qty
        net_delta += sum(pos.delta_hedge for pos in positions)
        return {'net_vega': net_vega, 'net_gamma': net_gamma, 'net_delta': net_delta}

    def risk_check(self, positions: list, spot: float, pnl: float) -> dict:
        greeks = self.portfolio_greeks(positions, spot)
        violations = []
        if abs(greeks['net_vega']) > self.max_net_vega_usd:
            violations.append(f"Vega limit: {greeks['net_vega']:.0f} USD")
        if abs(greeks['net_gamma']) > self.max_net_gamma_usd:
            violations.append(f"Gamma limit: {greeks['net_gamma']:.0f} USD")
        self.daily_pnl += pnl
        if self.daily_pnl < -self.daily_stop_loss_usd:
            violations.append(f"Daily stop loss: {self.daily_pnl:.0f} USD")
        self.peak_pnl = max(self.peak_pnl, self.daily_pnl)
        drawdown = (self.peak_pnl - self.daily_pnl) / (abs(self.peak_pnl) + 1)
        if drawdown > self.max_drawdown_pct:
            violations.append(f"Drawdown limit: {drawdown*100:.1f}%")
        return {
            'pass': len(violations) == 0,
            'violations': violations,
            'greeks': greeks,
            'drawdown_pct': drawdown * 100
        }

    def iv_spike_scenario(self, positions: list, spot: float,
                          iv_shock: float = 0.15) -> float:
        """Estimate P&L if IV jumps by iv_shock (e.g., 0.15 = +15 vol points)."""
        total_vega_pnl = 0.0
        for pos in positions:
            sig = pos.signal
            g = BlackScholesGreeks(
                S=spot, K=sig.strike, T=sig.expiry,
                sigma=sig.implied_vol, option_type=sig.option_type
            )
            # First order vega P&L + second order volga
            vega_pnl = g.vega * pos.option_units * spot * 100 * iv_shock * 100
            volga_pnl = 0.5 * g.volga * pos.option_units * (iv_shock * 100)**2
            total_vega_pnl += vega_pnl + volga_pnl
        return total_vega_pnl

Purple Flea API Integration

Purple Flea's trading API supports multi-leg options execution, real-time Greeks streaming, and automated delta hedging — all the primitives needed for a production vol arb agent.

import asyncio
from vol_arb_framework import VolArbFramework, VolArbRiskManager, TransactionCostModel

async def main():
    # Initialize with your pf_live_ API key
    API_KEY = "pf_live_your_agent_key_here"

    framework = VolArbFramework(
        api_key=API_KEY,
        max_vega_usd=15000,
        vrp_threshold=0.03,   # minimum 3 vol point edge
        z_entry=1.5,
        z_exit=0.5
    )
    risk_mgr = VolArbRiskManager(
        max_net_vega_usd=50000,
        max_net_gamma_usd=5000,
        daily_stop_loss_usd=3000
    )
    cost_model = TransactionCostModel(
        option_fee_per_contract=0.002,
        spot_fee_bps=3
    )

    await framework.initialize("BTC", lookback_days=90)
    print(f"Current BTC RV (Yang-Zhang): {framework.current_rv:.1%}")

    # Scan options chain for vol arb signals
    signals = await framework.scan_surface("BTC")
    print(f"Found {len(signals)} actionable vol arb signals")

    for sig in signals[:5]:  # process top 5
        # Check transaction costs
        spot = 45000  # example BTC price
        greeks = BlackScholesGreeks(
            S=spot, K=sig.strike, T=sig.expiry,
            sigma=sig.implied_vol, option_type=sig.option_type
        )
        viability = cost_model.is_trade_viable(
            iv_edge=sig.vrp,
            premium=greeks.delta * spot * 0.01,  # approximate
            quantity=1,
            spot=spot,
            delta=greeks.delta,
            vega=greeks.vega
        )
        if not viability['viable']:
            print(f"Skip {sig.strike} {sig.option_type}: insufficient edge after costs")
            continue

        # Risk check
        risk = risk_mgr.risk_check(framework.positions, spot, 0)
        if not risk['pass']:
            print(f"Risk violations: {risk['violations']}")
            break

        # Execute
        result = await framework.execute_vol_arb(sig, spot)
        print(f"Executed: {sig.recommended_action} {sig.strike} {sig.option_type}, Z={sig.z_score:.2f}")

    # Start continuous delta hedging in background
    hedge_task = asyncio.create_task(
        framework.run_hedging_loop(interval_seconds=30)
    )
    await asyncio.gather(hedge_task)

if __name__ == "__main__":
    asyncio.run(main())

Available API Endpoints

Backtesting and Performance Metrics

Before deploying a vol arb strategy live, agents must backtest on realistic simulated surfaces with proper transaction cost assumptions. Key metrics to evaluate:

• Sharpe Ratio: Target > 1.5 for a well-constructed vol arb strategy
• Sortino Ratio: Better measure given vol arb's asymmetric drawdown profile
• Maximum Drawdown: Critical — short vol strategies can have severe tail drawdowns
• Win Rate: Short vol typically wins 60-75% of days; long vol wins less but with convexity
• Vol Capture Ratio: What fraction of the theoretical VRP edge is captured after costs
• Correlation to market: Vol arb should be nearly zero-beta to underlying direction

class VolArbBacktester:
    """Simple mark-to-model backtester for vol arb strategies."""

    def __init__(self, price_series: pd.Series, iv_series: pd.Series,
                 hedge_freq: int = 1):
        self.S = price_series
        self.IV = iv_series
        self.hedge_freq = hedge_freq  # rehedge every N days
        self.rv_estimator = RealizedVolEstimator(
            pd.DataFrame({'open': self.S, 'high': self.S*1.001,
                          'low': self.S*0.999, 'close': self.S})
        )

    def run(self, window: int = 30) -> pd.DataFrame:
        rv = self.rv_estimator.yang_zhang(window)
        vrp = self.IV - rv
        # Daily gamma P&L of short straddle, unit vega
        daily_returns = self.S.pct_change()
        annualized_daily_rv = daily_returns.rolling(1).std() * np.sqrt(252)
        gamma_pnl = 0.5 * (annualized_daily_rv**2 - self.IV**2) * (-1)  # short gamma
        # Theta earned per day (assume short ATM straddle ≈ IV/sqrt(252)*S/40)
        theta_per_day = self.IV / np.sqrt(252) * self.S * 0.4 / 100
        daily_pnl = gamma_pnl + theta_per_day
        cum_pnl = daily_pnl.cumsum()
        peak = cum_pnl.cummax()
        drawdown = (peak - cum_pnl)
        return pd.DataFrame({
            'vrp': vrp, 'daily_pnl': daily_pnl,
            'cum_pnl': cum_pnl, 'drawdown': drawdown
        })

    def metrics(self, results: pd.DataFrame) -> dict:
        rets = results['daily_pnl'].dropna()
        sharpe = np.sqrt(252) * rets.mean() / rets.std()
        sortino_denom = rets[rets < 0].std()
        sortino = np.sqrt(252) * rets.mean() / sortino_denom if sortino_denom > 0 else 0
        max_dd = results['drawdown'].max()
        win_rate = (rets > 0).mean()
        return {
            'total_pnl': results['cum_pnl'].iloc[-1],
            'sharpe': sharpe,
            'sortino': sortino,
            'max_drawdown': max_dd,
            'win_rate': win_rate,
            'avg_vrp': results['vrp'].mean()
        }

Deploying Your Vol Arb Agent

A production vol arb agent on Purple Flea requires careful deployment configuration to handle reconnection, position recovery, and graceful shutdown during high-volatility events.

import signal
import logging

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("vol_arb_agent")

class VolArbAgent:
    """Production-grade vol arb agent with lifecycle management."""

    def __init__(self, api_key: str, underlying: str = "BTC"):
        self.api_key = api_key
        self.underlying = underlying
        self.framework = VolArbFramework(api_key)
        self.risk_mgr = VolArbRiskManager()
        self.cost_model = TransactionCostModel()
        self._shutdown = False

    async def start(self):
        log.info("Vol arb agent starting...")
        signal.signal(signal.SIGTERM, self._handle_shutdown)
        signal.signal(signal.SIGINT, self._handle_shutdown)

        await self.framework.initialize(self.underlying)
        log.info(f"RV estimate: {self.framework.current_rv:.1%}")

        hedge_task = asyncio.create_task(
            self.framework.run_hedging_loop(interval_seconds=60)
        )
        scan_task = asyncio.create_task(self._scan_loop())
        await asyncio.gather(hedge_task, scan_task, return_exceptions=True)

    async def _scan_loop(self):
        while not self._shutdown:
            try:
                signals = await self.framework.scan_surface(self.underlying)
                log.info(f"Scan complete: {len(signals)} signals")
                for sig in signals:
                    if self._shutdown: break
                    risk = self.risk_mgr.risk_check(
                        self.framework.positions, 45000, 0)
                    if not risk['pass']:
                        log.warning(f"Risk limit: {risk['violations']}")
                        break
                    await self.framework.execute_vol_arb(sig, 45000)
            except Exception as e:
                log.error(f"Scan error: {e}")
            await asyncio.sleep(300)  # scan every 5 minutes

    def _handle_shutdown(self, *args):
        log.info("Shutdown signal received, closing positions...")
        self._shutdown = True

if __name__ == "__main__":
    agent = VolArbAgent(api_key="pf_live_your_key_here")
    asyncio.run(agent.start())

Further Reading

• Statistical Arbitrage Frameworks for AI Agents — pairs and basket arb
• On-Chain Options Trading for Agents — DeFi options venues
• Options Greeks Complete Guide — deeper Greek analysis
• Purple Flea Trading API Docs — full endpoint reference
• Perpetual Funding Rate Arbitrage — related vol strategies
• Purple Flea Research Paper — agent financial infrastructure