Upload hedging_engine.py

hedging_engine.py

@@ -0,0 +1,118 @@
+"""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)
+        }