"""Dynamic Hedging System using Options Model.""" import numpy as np import pandas as pd from typing import Dict, Optional import warnings warnings.filterwarnings('ignore') class DynamicHedgingEngine: """Dynamic hedging using options model integration.""" def __init__(self, max_hedge_ratio: float = 0.5, hedge_cost: float = 0.002, delta_threshold: float = 0.1): self.max_hedge_ratio = max_hedge_ratio self.hedge_cost = hedge_cost self.delta_threshold = delta_threshold self.hedge_history = [] def compute_hedge_ratio(self, portfolio_delta: float, portfolio_gamma: float, volatility: float, risk_aversion: float = 2.0) -> float: """ Compute optimal hedge ratio. Args: portfolio_delta: Current delta exposure portfolio_gamma: Current gamma exposure volatility: Current volatility risk_aversion: Risk aversion parameter Returns: Hedge ratio (0 to max_hedge_ratio) """ # Base hedge from delta delta_hedge = abs(portfolio_delta) if portfolio_delta > self.delta_threshold else 0 # Gamma adjustment (more hedging when gamma is negative/high) gamma_hedge = abs(portfolio_gamma) * volatility * 0.5 if portfolio_gamma > 0 else 0 # Combine with risk aversion raw_hedge = (delta_hedge + gamma_hedge) * risk_aversion / 2.0 # Cap at max hedge_ratio = np.clip(raw_hedge, 0, self.max_hedge_ratio) self.hedge_history.append({ 'hedge_ratio': hedge_ratio, 'delta_hedge': delta_hedge, 'gamma_hedge': gamma_hedge }) return hedge_ratio def compute_portfolio_greeks(self, positions: Dict[str, float], option_greeks: pd.DataFrame) -> Dict[str, float]: """Compute portfolio-level Greeks.""" total_delta = 0 total_gamma = 0 total_theta = 0 total_vega = 0 for ticker, position in positions.items(): if ticker in option_greeks.index: row = option_greeks.loc[ticker] total_delta += row.get('delta', 0) * position total_gamma += row.get('gamma', 0) * position total_theta += row.get('theta', 0) * position total_vega += row.get('vega', 0) * position else: # Equity position: delta = 1 per share total_delta += position return { 'delta': total_delta, 'gamma': total_gamma, 'theta': total_theta, 'vega': total_vega } def compute_delta_neutral_weights(self, portfolio_greeks: Dict[str, float], hedge_instruments: Dict[str, float]) -> Dict[str, float]: """Compute delta-neutral hedge weights.""" total_delta = portfolio_greeks['delta'] if abs(total_delta) < 0.01: return {k: 0 for k in hedge_instruments} # Distribute hedge proportionally hedge_weights = {} total_hedge_delta = sum(abs(v) for v in hedge_instruments.values()) for instrument, delta in hedge_instruments.items(): if total_hedge_delta > 0: hedge_weights[instrument] = -total_delta * abs(delta) / total_hedge_delta else: hedge_weights[instrument] = 0 return hedge_weights def get_hedge_performance(self) -> Dict: """Get hedging performance statistics.""" if not self.hedge_history: return {} hedge_ratios = [h['hedge_ratio'] for h in self.hedge_history] return { 'avg_hedge_ratio': np.mean(hedge_ratios), 'max_hedge_ratio': np.max(hedge_ratios), 'hedge_utilization': np.mean(hedge_ratios) / self.max_hedge_ratio, 'n_rebalances': len(self.hedge_history) }