Upload stress_test.py

stress_test.py

@@ -0,0 +1,160 @@
+"""Stress Testing Engine - Simulate extreme market scenarios."""
+import numpy as np
+import pandas as pd
+from typing import Dict, List, Optional
+import warnings
+warnings.filterwarnings('ignore')
+
+
+class StressTestEngine:
+    """Simulate portfolio performance under extreme scenarios."""
+    
+    def __init__(self):
+        self.scenarios = {
+            'crisis_2008': {
+                'name': '2008 Financial Crisis',
+                'returns_shock': -0.05,
+                'vol_shock': 3.0,
+                'correlation_shock': 2.0,
+                'duration': 60,
+                'description': 'Lehman collapse + credit freeze'
+            },
+            'covid_crash': {
+                'name': 'COVID-19 Crash',
+                'returns_shock': -0.04,
+                'vol_shock': 4.0,
+                'correlation_shock': 1.5,
+                'duration': 30,
+                'description': 'Pandemic-driven market panic'
+            },
+            'rate_hike': {
+                'name': 'Aggressive Rate Hike',
+                'returns_shock': -0.02,
+                'vol_shock': 1.5,
+                'correlation_shock': 1.2,
+                'duration': 90,
+                'description': 'Central bank tightening cycle'
+            },
+            'vol_spike': {
+                'name': 'Volatility Spike',
+                'returns_shock': -0.01,
+                'vol_shock': 6.0,
+                'correlation_shock': 1.8,
+                'duration': 15,
+                'description': 'VIX explosion event'
+            },
+            'flash_crash': {
+                'name': 'Flash Crash',
+                'returns_shock': -0.03,
+                'vol_shock': 5.0,
+                'correlation_shock': 1.0,
+                'duration': 5,
+                'description': 'Algorithmic trading cascade'
+            }
+        }
+        self.results = []
+        
+    def run_scenario(self, 
+                     portfolio: Dict[str, float],
+                     base_returns: pd.DataFrame,
+                     scenario_name: str,
+                     initial_value: float = 1_000_000) -> Dict:
+        """Run a stress test scenario."""
+        scenario = self.scenarios.get(scenario_name)
+        if scenario is None:
+            raise ValueError(f"Unknown scenario: {scenario_name}")
+        
+        # Extract base parameters
+        mu = base_returns.mean().values
+        sigma = base_returns.std().values
+        corr = base_returns.corr().values
+        
+        # Apply shocks
+        mu_shocked = mu + scenario['returns_shock']
+        sigma_shocked = sigma * scenario['vol_shock']
+        
+        # Shock correlations
+        n = len(mu)
+        corr_shocked = corr * scenario['correlation_shock']
+        np.fill_diagonal(corr_shocked, 1.0)
+        corr_shocked = np.clip(corr_shocked, -1, 1)
+        
+        # Build shocked covariance
+        D = np.diag(sigma_shocked)
+        cov_shocked = D @ corr_shocked @ D
+        
+        # Simulate returns
+        np.random.seed(42)
+        n_days = scenario['duration']
+        shocked_returns = np.random.multivariate_normal(mu_shocked / 252, cov_shocked / 252, n_days)
+        
+        # Portfolio simulation
+        weights = np.array([portfolio.get(c, 0) for c in base_returns.columns])
+        weights /= weights.sum()
+        
+        port_returns = shocked_returns @ weights
+        equity = initial_value * np.cumprod(1 + port_returns)
+        
+        # Metrics
+        total_return = (equity[-1] / initial_value) - 1
+        max_idx = np.argmax(np.maximum.accumulate(equity) - equity)
+        max_drawdown = (np.min(equity) / np.maximum.accumulate(equity)[np.argmin(equity)]) - 1
+        
+        peak_equity = np.max(equity)
+        recovery_days = (np.argmax(equity >= peak_equity * 0.95) if np.any(equity >= peak_equity * 0.95) else n_days)
+        
+        result = {
+            'scenario': scenario['name'],
+            'total_return': total_return,
+            'max_drawdown': max_drawdown,
+            'min_equity': np.min(equity),
+            'recovery_days': recovery_days,
+            'sharpe': np.mean(port_returns) / np.std(port_returns) * np.sqrt(252) if np.std(port_returns) > 0 else 0,
+            'var_99': -np.percentile(port_returns, 1),
+            'cvar_99': -port_returns[port_returns <= np.percentile(port_returns, 1)].mean(),
+            'daily_vol': np.std(port_returns) * np.sqrt(252)
+        }
+        
+        self.results.append(result)
+        return result
+    
+    def run_all_scenarios(self, portfolio: Dict[str, float], base_returns: pd.DataFrame) -> pd.DataFrame:
+        """Run all stress test scenarios."""
+        results = []
+        for scenario_name in self.scenarios:
+            result = self.run_scenario(portfolio, base_returns, scenario_name)
+            results.append(result)
+        return pd.DataFrame(results)
+    
+    def survival_analysis(self, portfolio: Dict[str, float], base_returns: pd.DataFrame, n_simulations: int = 1000) -> Dict:
+        """Monte Carlo survival analysis."""
+        mu = base_returns.mean().values
+        cov = base_returns.cov().values * 252
+        weights = np.array([portfolio.get(c, 0) for c in base_returns.columns])
+        weights /= weights.sum()
+        
+        port_mu = weights @ mu
+        port_var = weights @ cov @ weights
+        port_sigma = np.sqrt(port_var)
+        
+        np.random.seed(42)
+        years = 5
+        daily_returns = np.random.normal(port_mu / 252, port_sigma / np.sqrt(252), (n_simulations, years * 252))
+        
+        equity_curves = np.cumprod(1 + daily_returns, axis=1)
+        
+        final_values = equity_curves[:, -1]
+        var_95_final = -np.percentile(final_values, 5)
+        prob_loss = np.mean(final_values < 1.0)
+        prob_ruin = np.mean(np.min(equity_curves, axis=1) < 0.5)
+        
+        return {
+            'expected_final_value': np.mean(final_values),
+            'median_final_value': np.median(final_values),
+            'worst_case_5th': np.percentile(final_values, 5),
+            'best_case_95th': np.percentile(final_values, 95),
+            'probability_of_loss': prob_loss,
+            'probability_of_ruin': prob_ruin,
+            'n_simulations': n_simulations,
+            'horizon_years': years
+        }