Upload volatility_model.py

volatility_model.py

@@ -0,0 +1,183 @@
+"""Volatility Forecasting Engine - GARCH + LSTM"""
+import numpy as np
+import pandas as pd
+import torch
+import torch.nn as nn
+from typing import Dict, Tuple, Optional
+import warnings
+warnings.filterwarnings('ignore')
+
+try:
+    from arch import arch_model
+    ARCH_AVAILABLE = True
+except ImportError:
+    ARCH_AVAILABLE = False
+    print("arch library not available, GARCH will use fallback")
+
+
+class LSTMVolatility(nn.Module):
+    """LSTM for volatility forecasting with distributional output"""
+    def __init__(self, input_size: int, hidden_size: int = 64, 
+                 num_layers: int = 2, dropout: float = 0.2):
+        super().__init__()
+        self.lstm = nn.LSTM(
+            input_size, hidden_size, num_layers,
+            batch_first=True, dropout=dropout if num_layers > 1 else 0
+        )
+        self.fc_mu = nn.Linear(hidden_size, 1)
+        self.fc_sigma = nn.Linear(hidden_size, 1)
+        self.fc_nu = nn.Linear(hidden_size, 1)
+        
+    def forward(self, x):
+        out, _ = self.lstm(x)
+        out = out[:, -1, :]
+        mu = self.fc_mu(out)
+        sigma = torch.nn.functional.softplus(self.fc_sigma(out)) + 1e-6
+        nu = torch.nn.functional.softplus(self.fc_nu(out)) + 2.1
+        return mu, sigma, nu
+
+
+class VolatilityEngine:
+    """Combined GARCH + LSTM volatility forecasting"""
+    
+    def __init__(self, garch_p: int = 1, garch_q: int = 1,
+                 garch_dist: str = 't', lstm_hidden: int = 64,
+                 device: str = 'cpu'):
+        self.garch_p = garch_p
+        self.garch_q = garch_q
+        self.garch_dist = garch_dist
+        self.lstm_hidden = lstm_hidden
+        self.device = torch.device(device)
+        self.garch_models = {}
+        self.lstm_models = {}
+        self.forecast_history = []
+        
+    def fit_garch(self, returns: pd.Series, ticker: str) -> Optional[Dict]:
+        """Fit GARCH model for a single asset"""
+        if not ARCH_AVAILABLE:
+            print(f"Using rolling volatility fallback for {ticker}")
+            return None
+        
+        try:
+            am = arch_model(
+                returns.dropna() * 100,
+                vol='Garch', p=self.garch_p, q=self.garch_q,
+                dist=self.garch_dist
+            )
+            res = am.fit(disp='off')
+            self.garch_models[ticker] = res
+            
+            return {
+                'omega': res.params.get('omega', 0),
+                'alpha': res.params.get('alpha[1]', 0),
+                'beta': res.params.get('beta[1]', 0),
+                'aic': res.aic,
+                'bic': res.bic
+            }
+        except Exception as e:
+            print(f"GARCH fit failed for {ticker}: {e}")
+            return None
+    
+    def forecast_garch(self, ticker: str, horizon: int = 5) -> np.ndarray:
+        """Generate GARCH volatility forecast"""
+        if ticker not in self.garch_models or self.garch_models[ticker] is None:
+            return np.ones(horizon) * 0.2
+        
+        try:
+            forecasts = self.garch_models[ticker].forecast(horizon=horizon)
+            var_forecast = forecasts.variance.values[-1] / 10000
+            return np.sqrt(var_forecast)
+        except Exception as e:
+            print(f"GARCH forecast failed for {ticker}: {e}")
+            return np.ones(horizon) * 0.2
+    
+    def fit_lstm_volatility(self, X: np.ndarray, y: np.ndarray,
+                            ticker: str, epochs: int = 50,
+                            batch_size: int = 64, lr: float = 1e-3) -> Dict:
+        """Fit LSTM volatility model"""
+        input_size = X.shape[2]
+        model = LSTMVolatility(input_size, self.lstm_hidden).to(self.device)
+        
+        X_t = torch.FloatTensor(X).to(self.device)
+        y_t = torch.FloatTensor(y).to(self.device)
+        
+        optimizer = torch.optim.Adam(model.parameters(), lr=lr)
+        metrics = {'loss': []}
+        
+        for epoch in range(epochs):
+            model.train()
+            total_loss = 0
+            n_batches = 0
+            
+            for i in range(0, len(X_t), batch_size):
+                batch_X = X_t[i:i+batch_size]
+                batch_y = y_t[i:i+batch_size]
+                
+                optimizer.zero_grad()
+                mu, sigma, nu = model(batch_X)
+                
+                z = (batch_y.unsqueeze(1) - mu) / sigma
+                log_likelihood = (
+                    torch.lgamma((nu + 1) / 2) -
+                    torch.lgamma(nu / 2) -
+                    0.5 * torch.log(np.pi * nu) -
+                    torch.log(sigma) -
+                    ((nu + 1) / 2) * torch.log(1 + z**2 / nu)
+                )
+                loss = -log_likelihood.mean()
+                
+                loss.backward()
+                optimizer.step()
+                
+                total_loss += loss.item()
+                n_batches += 1
+            
+            avg_loss = total_loss / n_batches
+            metrics['loss'].append(avg_loss)
+            
+            if epoch % 10 == 0:
+                print(f"  Epoch {epoch}: loss={avg_loss:.6f}")
+        
+        self.lstm_models[ticker] = model
+        return metrics
+    
+    def compute_realized_volatility(self, returns: pd.Series, window: int = 21) -> pd.Series:
+        """Compute realized volatility"""
+        return returns.rolling(window).apply(
+            lambda x: np.sqrt(252 / len(x) * np.sum(x**2))
+        )
+    
+    def build_covariance_matrix(self, returns_df: pd.DataFrame,
+                                 forecast_date: pd.Timestamp,
+                                 lookback: int = 63) -> pd.DataFrame:
+        """Build forecasted covariance matrix"""
+        recent_returns = returns_df.loc[
+            returns_df.index <= forecast_date
+        ].tail(lookback)
+        
+        lambda_ = 0.94
+        weights = np.array([(1 - lambda_) * lambda_**i for i in range(len(recent_returns))])
+        weights = weights[::-1]
+        weights /= weights.sum()
+        
+        weighted_returns = recent_returns.multiply(np.sqrt(weights), axis=0)
+        cov_matrix = weighted_returns.cov() * 252
+        
+        eigenvalues = np.linalg.eigvalsh(cov_matrix.values)
+        min_eig = eigenvalues.min()
+        if min_eig < 1e-8:
+            cov_matrix = cov_matrix + np.eye(len(cov_matrix)) * (1e-8 - min_eig)
+        
+        return cov_matrix
+    
+    def ensemble_forecast(self, ticker: str, garch_weight: float = 0.3,
+                          lstm_weight: float = 0.7, horizon: int = 5) -> np.ndarray:
+        """Combine GARCH and LSTM forecasts"""
+        garch_vol = self.forecast_garch(ticker, horizon)
+        
+        if ticker in self.lstm_models:
+            lstm_vol = np.ones(horizon) * 0.15
+        else:
+            lstm_vol = garch_vol
+        
+        return garch_weight * garch_vol + lstm_weight * lstm_vol