Connecting to Deribit's options data for implied volatility surface construction and historical backtesting requires reliable, low-latency market data feeds. In this hands-on guide, I walk through the complete integration architecture using HolySheep's Tardis.dev relay service, which provides real-time and historical data from Deribit, Binance, Bybit, OKX, and Deribit at a fraction of traditional market data costs.
As of May 2026, HolySheep offers AI inference at GPT-4.1: $8/MTok, Claude Sonnet 4.5: $15/MTok, Gemini 2.5 Flash: $2.50/MTok, and DeepSeek V3.2: $0.42/MTok. For a typical 10M token/month workload processing options Greeks and volatility surfaces, this translates to:
- GPT-4.1: $80/month
- Claude Sonnet 4.5: $150/month
- Gemini 2.5 Flash: $25/month
- DeepSeek V3.2: $4.20/month
Using DeepSeek V3.2 through HolySheep saves $75.80/month compared to GPT-4.1, and the relay supports WeChat/Alipay with <50ms latency. Sign up here to receive free credits on registration.
Why HolySheep for Crypto Market Data
HolySheep's Tardis.dev relay provides institutional-grade market data feeds for crypto derivatives exchanges. The service includes:
- Trades, Order Book, Liquidations, and Funding Rates for Binance, Bybit, OKX, and Deribit
- Historical data replay with microsecond precision timestamps
- WebSocket streaming with automatic reconnection and backpressure handling
- Rate ¥1=$1 — an 85%+ savings compared to ¥7.3 pricing on competing platforms
- Multi-currency support via WeChat Pay, Alipay, and international cards
Prerequisites
- HolySheep account with Tardis.dev API key
- Python 3.9+ with asyncio support
- pandas, numpy, scipy for numerical analysis
- Access to Deribit testnet or production endpoints
Architecture Overview
Our risk model validation pipeline consists of three layers:
- Data Ingestion Layer: HolySheep Tardis.dev WebSocket feeds for real-time options data
- Volatility Surface Engine: Interpolate implied volatilities across strikes and expirations
- Model Validation Layer: Compare model-implied Greeks against observed market behavior
Connecting to Deribit via HolySheep
The HolySheep relay exposes Deribit data through a unified WebSocket interface. Here is the complete connection handler for options order books and trades:
#!/usr/bin/env python3
"""
Deribit Options Data Ingestion via HolySheep Tardis.dev Relay
Connects to Deribit perpetuals and options markets for IV surface construction
"""
import asyncio
import json
import logging
from datetime import datetime, timezone
from typing import Dict, List, Optional
import pandas as pd
import numpy as np
import websockets
from websockets.exceptions import ConnectionClosed
HolySheep Relay Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_WS_URL = "wss://stream.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Replace with your key
Deribit-specific channel configurations
EXCHANGE = "deribit"
INSTRUMENTS = ["BTC-25JUN26-95000-C", "BTC-25JUN26-100000-C", "BTC-25JUN26-105000-C"]
SUBSCRIPTIONS = {
"book": "data", # Order book snapshots
"trades": "trades", # Trade tape
"ticker": "ticker" # Last price, mark price, IV
}
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)
class DeribitDataRelay:
"""
HolySheep Tardis.dev relay client for Deribit options data.
Handles authentication, subscription management, and message parsing.
"""
def __init__(self, api_key: str, symbols: List[str]):
self.api_key = api_key
self.symbols = symbols
self.connected = False
self.order_books: Dict[str, Dict] = {}
self.trades: List[Dict] = []
self.tickers: Dict[str, Dict] = {}
async def authenticate(self, ws: websockets.WebSocketClientProtocol):
"""Send authentication message to HolySheep relay"""
auth_msg = {
"type": "auth",
"apiKey": self.api_key,
"service": "tardis"
}
await ws.send(json.dumps(auth_msg))
response = await ws.recv()
result = json.loads(response)
if result.get("status") == "authenticated":
self.connected = True
logger.info("HolySheep relay authentication successful")
return True
else:
logger.error(f"Authentication failed: {result}")
return False
async def subscribe_channels(self, ws: websockets.WebSocketClientProtocol):
"""Subscribe to Deribit options channels via HolySheep relay"""
for symbol in self.symbols:
# Subscribe to order book
book_sub = {
"type": "subscribe",
"channel": "book",
"exchange": EXCHANGE,
"symbol": symbol,
"depth": 10 # 10-level order book
}
await ws.send(json.dumps(book_sub))
# Subscribe to trades
trade_sub = {
"type": "subscribe",
"channel": "trades",
"exchange": EXCHANGE,
"symbol": symbol
}
await ws.send(json.dumps(trade_sub))
# Subscribe to ticker (includes IV)
ticker_sub = {
"type": "subscribe",
"channel": "ticker",
"exchange": EXCHANGE,
"symbol": symbol
}
await ws.send(json.dumps(ticker_sub))
logger.info(f"Subscribed to {symbol} on Deribit via HolySheep")
async def handle_message(self, msg: Dict):
"""Process incoming market data messages"""
channel = msg.get("channel", "")
data = msg.get("data", {})
timestamp = pd.Timestamp(data.get("timestamp", datetime.now(timezone.utc).timestamp() * 1000))
if channel.startswith("book"):
symbol = msg.get("symbol", "")
self.order_books[symbol] = {
"timestamp": timestamp,
"bids": data.get("bids", []),
"asks": data.get("asks", []),
"mid_price": self._calculate_mid(data)
}
elif channel.startswith("trades"):
for trade in data if isinstance(data, list) else [data]:
self.trades.append({
"timestamp": pd.Timestamp(trade.get("timestamp")),
"symbol": trade.get("symbol"),
"side": trade.get("side"),
"price": float(trade.get("price", 0)),
"size": float(trade.get("size", 0)),
"trade_id": trade.get("id")
})
elif channel.startswith("ticker"):
symbol = msg.get("symbol", "")
self.tickers[symbol] = {
"timestamp": timestamp,
"last_price": float(data.get("last", 0)),
"mark_price": float(data.get("markPrice", 0)),
"best_bid": float(data.get("bestBidPrice", 0)),
"best_ask": float(data.get("bestAskPrice", 0)),
"mark_iv": float(data.get("markIv", 0)) * 100 if data.get("markIv") else None
}
def _calculate_mid(self, book_data: Dict) -> Optional[float]:
"""Calculate mid price from order book"""
bids = book_data.get("bids", [])
asks = book_data.get("asks", [])
if bids and asks:
return (float(bids[0][0]) + float(asks[0][0])) / 2
return None
async def run(self, duration_seconds: int = 60):
"""
Main data collection loop.
In production, this would run continuously.
"""
uri = f"{HOLYSHEEP_WS_URL}/market_data"
try:
async with websockets.connect(uri, ping_interval=20, ping_timeout=10) as ws:
await self.authenticate(ws)
await self.subscribe_channels(ws)
end_time = asyncio.get_event_loop().time() + duration_seconds
while asyncio.get_event_loop().time() < end_time:
try:
message = await asyncio.wait_for(ws.recv(), timeout=5.0)
data = json.loads(message)
await self.handle_message(data)
except asyncio.TimeoutError:
continue
except ConnectionClosed as e:
logger.warning(f"Connection closed: {e}, reconnecting...")
break
except Exception as e:
logger.error(f"WebSocket error: {e}")
raise
finally:
self.connected = False
return self._compile_results()
def _compile_results(self) -> Dict:
"""Compile collected data into analysis-ready format"""
df_trades = pd.DataFrame(self.trades) if self.trades else pd.DataFrame()
df_tickers = pd.DataFrame.from_dict(self.tickers, orient='index')
return {
"trades": df_trades,
"tickers": df_tickers,
"order_books": self.order_books,
"collection_duration": len(df_trades) / max(1, len(df_trades.drop_duplicates('timestamp'))) if not df_trades.empty else 0
}
async def main():
"""Demo: Collect 60 seconds of BTC option data"""
client = DeribitDataRelay(
api_key=API_KEY,
symbols=INSTRUMENTS
)
logger.info("Starting Deribit options data collection via HolySheep...")
results = await client.run(duration_seconds=60)
print(f"\n=== Data Collection Summary ===")
print(f"Trades collected: {len(results['trades'])}")
print(f"Ticker updates: {len(results['tickers'])}")
print(f"Order books: {len(results['order_books'])}")
if not results['tickers'].empty:
print("\n=== Mark IV Snapshot ===")
print(results['tickers'][['mark_iv', 'mark_price']].to_string())
if __name__ == "__main__":
asyncio.run(main())
Implied Volatility Surface Construction
Once we have tick data from the HolySheep relay, we construct the implied volatility surface using SABR model interpolation. This enables us to backtest strike-wise volatility smiles and identify regime changes:
#!/usr/bin/env python3
"""
Implied Volatility Surface Construction from Deribit Options Data
Uses SABR model for vol smile interpolation and surface validation
"""
import pandas as pd
import numpy as np
from scipy.optimize import brentq, minimize
from scipy.interpolate import CubicSpline, RectBivariateSpline
from typing import Tuple, Dict, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import warnings
Market constants
RISK_FREE_RATE = 0.05 # BTC funding rate proxy
VOL_OF_VOL = 0.4 # SABR parameter: vol of vol
CORRELATION = -0.3 # SABR parameter: correlation (typical for crypto)
@dataclass
class OptionContract:
"""Single option contract with market data"""
symbol: str
expiry: datetime
strike: float
option_type: str # 'call' or 'put'
market_price: float
spot_price: float
mark_iv: float
bid_price: float
ask_price: float
@property
def time_to_expiry(self) -> float:
return (self.expiry - datetime.now()).days / 365.25
@property
def moneyness(self) -> float:
return np.log(self.strike / self.spot_price)
class BlackScholes:
"""Black-Scholes option pricing with Greeks"""
@staticmethod
def norm_cdf(x: float) -> float:
"""Cumulative distribution function of standard normal"""
return 0.5 * (1 + np.math.erf(x / np.sqrt(2)))
@staticmethod
def norm_pdf(x: float) -> float:
"""Probability density function of standard normal"""
return np.exp(-0.5 * x**2) / np.sqrt(2 * np.pi)
@classmethod
def price(cls, S: float, K: float, T: float, r: float, sigma: float,
option_type: str) -> float:
"""Calculate option price using Black-Scholes"""
if T <= 0 or sigma <= 0:
return max(0, S - K) if option_type == 'call' else max(0, K - S)
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if option_type.lower() == 'call':
return S * cls.norm_cdf(d1) - K * np.exp(-r * T) * cls.norm_cdf(d2)
else:
return K * np.exp(-r * T) * cls.norm_cdf(-d2) - S * cls.norm_cdf(-d1)
@classmethod
def implied_vol(cls, market_price: float, S: float, K: float, T: float,
r: float, option_type: str) -> Optional[float]:
"""Calculate implied volatility using Newton-Raphson"""
if market_price <= 0 or T <= 0:
return None
# Intrinsic value check
intrinsic = max(S - K, 0) if option_type == 'call' else max(K - S, 0)
if market_price <= intrinsic:
return None
def objective(sigma):
return cls.price(S, K, T, r, sigma, option_type) - market_price
try:
iv = brentq(objective, 1e-6, 10.0, xtol=1e-8)
return iv
except (ValueError, RuntimeError):
return None
@classmethod
def greeks(cls, S: float, K: float, T: float, r: float, sigma: float,
option_type: str) -> Dict[str, float]:
"""Calculate option Greeks"""
if T <= 0 or sigma <= 0:
return {'delta': 0, 'gamma': 0, 'theta': 0, 'vega': 0}
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if option_type.lower() == 'call':
delta = cls.norm_cdf(d1)
theta = (-S * cls.norm_pdf(d1) * sigma / (2 * np.sqrt(T))
- r * K * np.exp(-r * T) * cls.norm_cdf(d2)) / 365
else:
delta = cls.norm_cdf(d1) - 1
theta = (-S * cls.norm_pdf(d1) * sigma / (2 * np.sqrt(T))
+ r * K * np.exp(-r * T) * cls.norm_cdf(-d2)) / 365
gamma = cls.norm_pdf(d1) / (S * sigma * np.sqrt(T))
vega = S * cls.norm_pdf(d1) * np.sqrt(T) / 100 # Per 1% vol move
return {'delta': delta, 'gamma': gamma, 'theta': theta, 'vega': vega}
class SABRSurface:
"""
SABR Stochastic Volatility Model for IV Surface Construction.
Calibrates to market smiles and enables interpolation across strikes/expirations.
"""
def __init__(self, options: list, spot: float, r: float = RISK_FREE_RATE):
self.options = options
self.spot = spot
self.r = r
self.calibrated_params = {}
def calibrate_strike_slice(self, expiry: datetime,
initial_guess: Tuple = (0.5, 0.5, -0.3, 0.1)) -> Dict:
"""
Calibrate SABR parameters for a single expiry.
[alpha, rho, nu, rho] = [vol level, skew, vol of vol, correlation]
"""
T = (expiry - datetime.now()).days / 365.25
expiry_opts = [o for o in self.options if abs((o.expiry - expiry).days) < 1]
if len(expiry_opts) < 3:
warnings.warn(f"Insufficient options for {expiry}, need at least 3")
return {}
strikes = np.array([o.strike for o in expiry_opts])
market_ivs = np.array([o.mark_iv for o in expiry_opts if o.mark_iv])
if len(market_ivs) < 3:
return {}
def sabr_implied_vol(F, K, T, alpha, rho, nu):
"""SABR implied volatility formula (Hagan 2002)"""
if F == K:
# ATM formula
term1 = alpha / (F ** (1 - rho))
term2 = 1 + ((1 - rho)**2 / 24 * alpha**2 / (F ** (2 - 2*rho))
+ 0.25 * rho * nu * alpha / (F ** (1 - rho))
+ (2 - 3*rho**2) / 24 * nu**2) * T
return term1 * term2
else:
FK = F * K
log FK = np.log(F / K)
z = nu / alpha * (FK) ** ((1 - rho) / 2) * log FK
# Expansion terms
sqrt_term = np.sqrt(1 - 2*rho*z + z**2)
z_over_x = np.log((sqrt_term + z - rho) / (1 - rho))
numerator = alpha * (FK) ** ((1 - rho) / 2)
denominator = (1 - rho)**2 / 24 * log FK**2 + sqrt_term
return numerator / denominator * (1 + ((1 - rho)**2/24 * alpha**2/(FK**(1-rho))
+ 0.25*rho*nu*alpha/(FK**((1-rho)/2))
+ (2-3*rho**2)/24*nu**2) * T)
def objective(params):
alpha, vol_of_vol, corr, beta = params
model_ivs = []
F = self.spot * np.exp(self.r * T)
for strike, mkt_iv in zip(strikes, market_ivs):
try:
mdl_iv = sabr_implied_vol(F, strike, T, alpha, corr, vol_of_vol)
model_ivs.append(mdl_iv)
except:
model_ivs.append(mkt_iv)
return np.sum((np.array(model_ivs) - market_ivs)**2)
result = minimize(objective, initial_guess, method='L-BFGS-B',
bounds=[(0.01, 3.0), (0.01, 2.0), (-0.99, 0.99), (0, 1)])
self.calibrated_params[expiry] = result.x
return dict(zip(['alpha', 'nu', 'rho', 'beta'], result.x))
def build_surface(self) -> pd.DataFrame:
"""Build complete IV surface across strikes and expiries"""
expiry_groups = {}
for opt in self.options:
exp_str = opt.expiry.strftime('%Y-%m-%d')
if exp_str not in expiry_groups:
expiry_groups[exp_str] = []
expiry_groups[exp_str].append(opt)
surface_data = []
for exp_str, opts in expiry_groups.items():
expiry = opts[0].expiry
self.calibrate_strike_slice(expiry)
strikes = [o.strike for o in opts]
ivs = [o.mark_iv for o in opts if o.mark_iv]
# Cubic spline interpolation
if len(strikes) >= 4 and len(ivs) >= 4:
sorted_idx = np.argsort(strikes)
interp = CubicSpline(np.array(strikes)[sorted_idx],
np.array(ivs)[sorted_idx])
for strike in np.linspace(min(strikes), max(strikes), 50):
surface_data.append({
'expiry': exp_str,
'strike': strike,
'implied_vol': interp(strike),
'moneyness': np.log(strike / self.spot)
})
return pd.DataFrame(surface_data)
def validate_model(self, holdout_opts: list) -> Dict[str, float]:
"""
Out-of-sample validation: compare model-predicted Greeks
against observed P&L from trades.
"""
predictions = []
actuals = []
for opt in holdout_opts:
# Calculate model Greeks
greeks = BlackScholes.greeks(
opt.spot_price, opt.strike, opt.time_to_expiry,
self.r, opt.mark_iv, opt.option_type
)
# Predict P&L from small vol move
vol_shock = 0.01 # 1% vol shock
new_price = BlackScholes.price(
opt.spot_price, opt.strike, opt.time_to_expiry,
self.r, opt.mark_iv + vol_shock, opt.option_type
)
predicted_pnl = (new_price - opt.market_price) * opt.market_price * 0.1
predictions.append(predicted_pnl)
actuals.append(opt.market_price * opt.bid_price) # Simplified actual
predictions = np.array(predictions)
actuals = np.array(actuals)
mae = np.mean(np.abs(predictions - actuals))
rmse = np.sqrt(np.mean((predictions - actuals)**2))
r_squared = 1 - np.sum((actuals - predictions)**2) / np.sum((actuals - np.mean(actuals))**2)
return {
'MAE': mae,
'RMSE': rmse,
'R_squared': r_squared,
'sample_size': len(holdout_opts)
}
Demo usage with synthetic data
if __name__ == "__main__":
from datetime import datetime, timedelta
# Synthetic BTC options data
spot = 95000
expiries = [datetime.now() + timedelta(days=d) for d in [7, 30, 60]]
strikes = np.linspace(85000, 105000, 13)
options = []
for exp in expiries:
for K in strikes:
iv = 0.5 + 0.1 * (1 - np.abs(np.log(K/spot)) * 5) + np.random.normal(0, 0.02)
price = BlackScholes.price(spot, K, (exp - datetime.now()).days/365.25,
RISK_FREE_RATE, max(iv, 0.1), 'call')
options.append(OptionContract(
symbol=f"BTC-{exp.strftime('%d%b%y').upper()}-{int(K)}",
expiry=exp, strike=K, option_type='call',
market_price=price, spot_price=spot, mark_iv=max(iv, 0.1),
bid_price=price * 0.99, ask_price=price * 1.01
))
# Build and validate surface
surface = SABRSurface(options, spot)
iv_surface = surface.build_surface()
print("=== IV Surface Summary ===")
print(iv_surface.groupby('expiry')['implied_vol'].describe().to_string())
# Model validation
validation_result = surface.validate_model(options[:10])
print(f"\n=== Model Validation ===")
print(f"MAE: ${validation_result['MAE']:.4f}")
print(f"RMSE: ${validation_result['RMSE']:.4f}")
print(f"R²: {validation_result['R_squared']:.4f}")
Real-Time Risk Monitoring Pipeline
Combine the data relay with the volatility surface engine for continuous risk monitoring. This pipeline calculates portfolio-level Greeks and alerts on volatility regime changes:
#!/usr/bin/env python3
"""
Real-Time Options Risk Monitor
Combines HolySheep relay data with IV surface for live risk management
"""
import asyncio
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from collections import deque
from typing import Dict, List, Optional
import logging
import json
Import from previous modules
from deribit_relay import DeribitDataRelay
from iv_surface import BlackScholes, SABRSurface, OptionContract
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class OptionsRiskMonitor:
"""
Real-time risk monitoring for Deribit options portfolio.
Tracks Greeks, VaR, and volatility regime changes.
"""
def __init__(self, portfolio: Dict[str, float], spot_price: float):
"""
Args:
portfolio: Dict of {symbol: position_size} (positive=long, negative=short)
spot_price: Current underlying price
"""
self.portfolio = portfolio
self.spot = spot_price
self.position_greeks = {}
self.pnl_history = deque(maxlen=1000)
self.iv_history = deque(maxlen=500)
self.alerts = []
# Risk limits
self.delta_limit = 0.5
self.gamma_limit = 0.1
self.vega_limit = 0.25
def update_position_greeks(self, option_data: Dict):
"""Update Greeks for a single position"""
symbol = option_data['symbol']
if symbol not in self.portfolio:
return
position_size = self.portfolio[symbol]
greeks = BlackScholes.greeks(
self.spot,
option_data['strike'],
option_data['time_to_expiry'],
RISK_FREE_RATE,
option_data['implied_vol'],
option_data['option_type']
)
# Scale by position size
scaled_greeks = {
'delta': greeks['delta'] * position_size,
'gamma': greeks['gamma'] * position_size,
'theta': greeks['theta'] * position_size,
'vega': greeks['vega'] * position_size
}
self.position_greeks[symbol] = {
**scaled_greeks,
'market_value': option_data['market_price'] * position_size,
'iv': option_data['implied_vol']
}
# Record IV for regime detection
self.iv_history.append({
'timestamp': datetime.now(),
'symbol': symbol,
'iv': option_data['implied_vol'],
'moneyness': np.log(option_data['strike'] / self.spot)
})
def calculate_portfolio_greeks(self) -> Dict[str, float]:
"""Aggregate Greeks across all positions"""
total = {'delta': 0, 'gamma': 0, 'theta': 0, 'vega': 0, 'market_value': 0}
for pos in self.position_greeks.values():
for greek in ['delta', 'gamma', 'theta', 'vega']:
total[greek] += pos[greek]
total['market_value'] += pos.get('market_value', 0)
# Per-unit (for easier interpretation)
total['delta_per_unit'] = total['delta'] / max(1, len(self.position_greeks))
total['gamma_per_unit'] = total['gamma'] / max(1, len(self.position_greeks))
total['vega_per_unit'] = total['vega'] / max(1, len(self.position_greeks))
return total
def check_risk_limits(self, portfolio_greeks: Dict) -> List[Dict]:
"""Check if any risk limits are breached"""
breaches = []
if abs(portfolio_greeks['delta']) > self.delta_limit:
breaches.append({
'type': 'DELTA_BREACH',
'limit': self.delta_limit,
'actual': portfolio_greeks['delta'],
'severity': 'HIGH' if abs(portfolio_greeks['delta']) > self.delta_limit * 2 else 'MEDIUM'
})
if abs(portfolio_greeks['gamma']) > self.gamma_limit:
breaches.append({
'type': 'GAMMA_BREACH',
'limit': self.gamma_limit,
'actual': portfolio_greeks['gamma'],
'severity': 'HIGH'
})
if abs(portfolio_greeks['vega']) > self.vega_limit:
breaches.append({
'type': 'VEGA_BREACH',
'limit': self.vega_limit,
'actual': portfolio_greeks['vega'],
'severity': 'MEDIUM'
})
return breaches
def detect_volatility_regime(self) -> Dict:
"""
Detect volatility regime changes using IV surface dynamics.
Returns regime classification and trend indicators.
"""
if len(self.iv_history) < 50:
return {'regime': 'INSUFFICIENT_DATA', 'confidence': 0}
df_iv = pd.DataFrame(list(self.iv_history))
df_iv.set_index('timestamp', inplace=True)
# Calculate IV percentile over rolling window
df_iv['iv_percentile'] = df_iv.groupby('symbol')['iv'].transform(
lambda x: x.rolling(50, min_periods=20).apply(lambda y: pd.Series(y).rank(pct=True).iloc[-1])
)
current_ivs = df_iv.groupby('symbol')['iv'].last()
mean_iv = current_ivs.mean()
# Regime classification
if mean_iv > 0.8:
regime = 'HIGH_VOL'
elif mean_iv < 0.3:
regime = 'LOW_VOL'
else:
regime = 'NORMAL_VOL'
# IV trend
recent_mean = df_iv['iv'].tail(20).mean()
older_mean = df_iv['iv'].head(20).mean()
trend = 'INCREASING' if recent_mean > older_mean * 1.05 else 'DECREASING' if recent_mean < older_mean * 0.95 else 'STABLE'
return {
'regime': regime,
'trend': trend,
'mean_iv': mean_iv,
'confidence': min(1.0, len(self.iv_history) / 200)
}
def calculate_var(self, confidence: float = 0.95, horizon_hours: int = 1) -> Dict:
"""
Historical simulation VaR calculation.
Uses past P&L history to estimate quantile losses.
"""
if len(self.pnl_history) < 30:
return {'var': None, 'confidence': confidence, 'message': 'Insufficient history'}
pnl_array = np.array(list(self.pnl_history))
# Scale by horizon
scale_factor = np.sqrt(horizon_hours / 24)
scaled_pnl = pnl_array * scale_factor
var = np.percentile(scaled_pnl, (1 - confidence) * 100)
cvar = np.mean(scaled_pnl[scaled_pnl <= var]) if len(scaled_pnl[scaled_pnl <= var]) > 0 else var
return {
'var': abs(var),
'cvar': abs(cvar),
'confidence': confidence,
'horizon_hours': horizon_hours
}
def generate_risk_report(self) -> Dict:
"""Generate comprehensive risk report"""
portfolio_greeks = self.calculate_portfolio_greeks()
breaches = self.check_risk_limits(portfolio_greeks)
regime = self.detect_volatility_regime()
var = self.calculate_var()
report = {
'timestamp': datetime.now().isoformat(),
'spot_price': self.spot,
'portfolio_size': len(self.position_greeks),
'total_market_value': portfolio_greeks['market_value'],
'portfolio_greeks': portfolio_greeks,
'risk_limits': {
'delta': {'limit': self.delta_limit, 'breached': abs(portfolio_greeks['delta']) > self.delta_limit},
'gamma': {'limit': self.gamma_limit, 'breached': abs(portfolio_greeks['gamma']) > self.gamma_limit},
'vega': {'limit': self.vega_limit, 'breached': abs(portfolio_greeks['vega']) > self.vega_limit}
},
'breaches': breaches,
'volatility_regime': regime,
'var_95_1h': var,
'alerts': self.alerts[-10:] # Last 10 alerts
}
return report
async def monitoring_demo():
"""Demo: Real-time monitoring with HolySheep relay"""
# Sample portfolio (long 5 ATM calls, short 3 OTM puts)
portfolio = {
'BTC-25JUN26-95000-C': 5,
'BTC-25JUN26-100000-C': -3
}
monitor = OptionsRiskMonitor(portfolio, spot_price=95000)
# Simulate position updates
for i in range(100):
# Simulate market data update
iv = 0.55 + np.random.normal(0, 0.02)
monitor.update_position_greeks({
'symbol': 'BTC-25JUN26-95000-C',
'strike': 95000,
'time_to_expiry': 54/365.25,
'implied_vol': iv,
'option_type': 'call',
'market_price': 5000 + np.random.normal(0, 100)
})
# Simulate P&L
monitor.pnl_history.append(np.random.normal(0, 500))
if i % 20 == 0:
report = monitor.generate_risk_report()
print(f"\n=== Risk Report {i} ===")
print(f"Spot: ${report['spot_price']}")
print(f"Portfolio Delta: {report['portfolio_greeks']['delta']:.4f}")
print(f"Portfolio Vega: {report['portfolio_greeks']['vega']:.4f}")
print(f"Vol Regime: {report['volatility_regime']['regime']} ({report['volatility_regime']['trend']})")
print(f"VaR (95%, 1h): ${report['var_95_1h'].get('var', 'N/A')}")
print(f"