Add risk management engine: VaR/CVaR, stress testing, compliance monitor

risk_management.py

@@ -0,0 +1,703 @@
+"""Risk Management Engine: VaR, CVaR, Stress Testing, Compliance Layer
+
+THE MOST IMPORTANT MODULE IN ALPHA FORGE.
+
+A trader without risk management is a trader who will blow up.
+This is not optional. This is the difference between compounding wealth
+and losing everything in one bad week.
+"""
+import numpy as np
+import pandas as pd
+from scipy import stats
+from typing import Dict, List, Optional, Tuple, Callable
+from dataclasses import dataclass
+import warnings
+warnings.filterwarnings('ignore')
+
+
+@dataclass
+class RiskLimits:
+    """Risk limit configuration — THE CONTRACT WITH YOURSELF"""
+    
+    # Position-level limits
+    max_position_pct: float = 0.20        # No single position > 20%
+    max_sector_pct: float = 0.40          # No sector > 40%
+    max_leverage: float = 2.0              # Gross exposure < 200%
+    
+    # Portfolio-level limits
+    max_portfolio_volatility: float = 0.15  # Annual vol < 15%
+    max_drawdown_limit: float = 0.15        # Circuit breaker at 15%
+    daily_loss_limit: float = 0.03          # Stop if down 3% in a day
+    monthly_loss_limit: float = 0.06        # Stop if down 6% in a month
+    
+    # VaR/CVaR limits
+    daily_var_limit: float = 0.02           # 1-day 95% VaR < 2%
+    daily_cvar_limit: float = 0.03          # 1-day 95% CVaR < 3%
+    
+    # Compliance / regulatory
+    max_turnover_annual: float = 3.0         # Max 300% annual turnover
+    min_holding_days: int = 1                # No day-trading (if applicable)
+    restricted_symbols: List[str] = None     # Blacklist
+    
+    def __post_init__(self):
+        if self.restricted_symbols is None:
+            self.restricted_symbols = []
+
+
+class ValueAtRisk:
+    """
+    Value at Risk (VaR) and Conditional VaR (CVaR) Engine.
+    
+    VaR answers: "What is the maximum loss I can expect with X% confidence?"
+    CVaR answers: "If I lose more than VaR, how bad does it get on average?"
+    
+    Three methods:
+    1. Historical Simulation (non-parametric, uses actual returns)
+    2. Parametric (assumes normal/t distribution)
+    3. Monte Carlo (simulates many scenarios)
+    """
+    
+    def __init__(self, confidence: float = 0.95, 
+                 method: str = 'historical',
+                 distribution: str = 't'):
+        """
+        Args:
+            confidence: Confidence level (e.g., 0.95 = 95%)
+            method: 'historical', 'parametric', 'monte_carlo'
+            distribution: For parametric: 'normal' or 't'
+        """
+        self.confidence = confidence
+        self.method = method
+        self.distribution = distribution
+        self.alpha = 1 - confidence
+    
+    def historical_var(self, returns: np.ndarray, 
+                       weights: Optional[np.ndarray] = None) -> Dict:
+        """
+        Historical Simulation VaR.
+        
+        Advantages: No distribution assumption, captures fat tails
+        Disadvantages: Requires lots of history, assumes past repeats
+        """
+        if weights is not None:
+            port_returns = returns @ weights
+        else:
+            port_returns = returns
+        
+        # VaR is the (1-confidence) percentile
+        var = np.percentile(port_returns, self.alpha * 100)
+        
+        # CVaR (Expected Shortfall): average of returns beyond VaR
+        cvar = port_returns[port_returns <= var].mean() if (port_returns <= var).any() else var
+        
+        # Additional metrics
+        var_pct = var * 100
+        cvar_pct = cvar * 100
+        
+        # Breaches: How often did actual losses exceed VaR?
+        breaches = np.sum(port_returns < var)
+        breach_rate = breaches / len(port_returns)
+        
+        return {
+            'method': 'historical',
+            'confidence': self.confidence,
+            'var': var,
+            'var_pct': var_pct,
+            'cvar': cvar,
+            'cvar_pct': cvar_pct,
+            'breaches': breaches,
+            'breach_rate': breach_rate,
+            'expected_breach_rate': self.alpha,
+            'n_observations': len(port_returns)
+        }
+    
+    def parametric_var(self, returns: np.ndarray,
+                       weights: Optional[np.ndarray] = None,
+                       cov_matrix: Optional[np.ndarray] = None) -> Dict:
+        """
+        Parametric (variance-covariance) VaR.
+        
+        Assumes returns follow a known distribution (normal or Student's t).
+        
+        Advantages: Computationally fast, works with small samples
+        Disadvantages: Assumes distribution, misses fat tails
+        """
+        if weights is not None and cov_matrix is not None:
+            # Multi-asset portfolio
+            port_mean = weights @ returns.mean(axis=0)
+            port_var = weights @ cov_matrix @ weights
+            port_std = np.sqrt(port_var)
+        elif weights is not None:
+            port_returns = returns @ weights
+            port_mean = port_returns.mean()
+            port_std = port_returns.std()
+        else:
+            port_mean = returns.mean()
+            port_std = returns.std()
+        
+        if self.distribution == 'normal':
+            z = stats.norm.ppf(self.alpha)
+            var = port_mean + z * port_std
+            # CVaR for normal: mu - sigma * phi(z) / alpha
+            cvar = port_mean - port_std * stats.norm.pdf(z) / self.alpha
+            
+        elif self.distribution == 't':
+            # Fit Student's t distribution
+            df, loc, scale = stats.t.fit(returns.flatten() if returns.ndim == 1 else returns.mean(axis=1))
+            t_alpha = stats.t.ppf(self.alpha, df, loc=loc, scale=scale)
+            var = t_alpha
+            
+            # CVaR for t distribution (approximation)
+            # ES = - (scale * (df + t_alpha^2) / (df - 1)) * pdf(t_alpha) / alpha + loc
+            pdf_at_alpha = stats.t.pdf(t_alpha, df, loc=loc, scale=scale)
+            cvar = loc - (scale * (df + t_alpha**2) / (df - 1)) * pdf_at_alpha / self.alpha
+        
+        else:
+            raise ValueError(f"Unknown distribution: {self.distribution}")
+        
+        return {
+            'method': 'parametric',
+            'distribution': self.distribution,
+            'confidence': self.confidence,
+            'var': var,
+            'var_pct': var * 100,
+            'cvar': cvar,
+            'cvar_pct': cvar * 100,
+            'port_mean': port_mean,
+            'port_std': port_std
+        }
+    
+    def monte_carlo_var(self, returns: np.ndarray,
+                      weights: Optional[np.ndarray] = None,
+                      cov_matrix: Optional[np.ndarray] = None,
+                      n_simulations: int = 100000) -> Dict:
+        """
+        Monte Carlo Simulation VaR.
+        
+        Simulates many future scenarios using fitted distributions.
+        
+        Advantages: Captures complex distributions, can include correlations
+        Disadvantages: Computationally expensive, model-dependent
+        """
+        if weights is not None and cov_matrix is not None:
+            # Multi-asset simulation
+            means = returns.mean(axis=0)
+            
+            # Cholesky decomposition for correlated random variables
+            try:
+                L = np.linalg.cholesky(cov_matrix)
+            except:
+                # Add small diagonal to ensure positive definite
+                L = np.linalg.cholesky(cov_matrix + np.eye(len(cov_matrix)) * 1e-8)
+            
+            # Simulate
+            n_assets = len(means)
+            random_normals = np.random.randn(n_simulations, n_assets)
+            simulated_returns = random_normals @ L.T
+            
+            # Portfolio returns
+            port_sim = simulated_returns @ weights
+        else:
+            # Single asset
+            if returns.ndim == 2:
+                returns = returns.mean(axis=1)
+            
+            mean = returns.mean()
+            std = returns.std()
+            
+            simulated_returns = np.random.normal(mean, std, n_simulations)
+            port_sim = simulated_returns
+        
+        # Calculate VaR/CVaR from simulations
+        var = np.percentile(port_sim, self.alpha * 100)
+        cvar = port_sim[port_sim <= var].mean() if (port_sim <= var).any() else var
+        
+        return {
+            'method': 'monte_carlo',
+            'confidence': self.confidence,
+            'var': var,
+            'var_pct': var * 100,
+            'cvar': cvar,
+            'cvar_pct': cvar * 100,
+            'n_simulations': n_simulations
+        }
+    
+    def calculate(self, returns: np.ndarray,
+                  weights: Optional[np.ndarray] = None,
+                  cov_matrix: Optional[np.ndarray] = None,
+                  n_simulations: int = 100000) -> Dict:
+        """Calculate VaR using configured method"""
+        if self.method == 'historical':
+            return self.historical_var(returns, weights)
+        elif self.method == 'parametric':
+            return self.parametric_var(returns, weights, cov_matrix)
+        elif self.method == 'monte_carlo':
+            return self.monte_carlo_var(returns, weights, cov_matrix, n_simulations)
+        else:
+            raise ValueError(f"Unknown method: {self.method}")
+    
+    def backtest_var(self, returns: np.ndarray,
+                     window: int = 252,
+                     weights: Optional[np.ndarray] = None) -> pd.DataFrame:
+        """
+        Backtest VaR: rolling window VaR calculation and breach testing.
+        
+        If VaR at 95% confidence is breached more than 5% of the time,
+        your model is underestimating risk.
+        
+        Returns DataFrame with rolling VaR and breach indicators.
+        """
+        results = []
+        n = len(returns)
+        
+        for i in range(window, n):
+            hist = returns[i-window:i]
+            actual = returns[i] if returns.ndim == 1 else returns[i]
+            
+            var_result = self.historical_var(hist, weights)
+            var_threshold = var_result['var']
+            
+            port_return = actual if weights is None else actual @ weights
+            
+            results.append({
+                'date': i,
+                'actual_return': port_return,
+                'var': var_threshold,
+                'breach': port_return < var_threshold,
+                'excess_loss': max(0, -(port_return - var_threshold))
+            })
+        
+        df = pd.DataFrame(results)
+        df['breach_rate'] = df['breach'].rolling(window).mean()
+        
+        return df
+
+
+class StressTesting:
+    """
+    Stress Testing Engine.
+    
+    Tests portfolio under extreme but plausible scenarios.
+    The goal: "Can I survive 2008? March 2020? Flash Crash?"
+    
+    Scenarios:
+    1. Historical events (replay actual market crises)
+    2. Factor shocks (move one risk factor dramatically)
+    3. Custom scenarios (user-defined)
+    """
+    
+    HISTORICAL_SCENARIOS = {
+        'black_monday_1987': {
+            'sp500_shock': -0.20,
+            'volatility_multiplier': 5.0,
+            'correlation_spike': 0.9,
+            'description': 'Black Monday — S&P 500 drops 20% in one day, vol explodes'
+        },
+        'financial_crisis_2008': {
+            'sp500_shock': -0.50,
+            'duration_months': 18,
+            'credit_spread_widening': 0.05,
+            'description': '2008 Financial Crisis — 50% drawdown over 18 months'
+        },
+        'covid_crash_2020': {
+            'sp500_shock': -0.34,
+            'duration_days': 23,
+            'vix_spike': 80,
+            'recovery_days': 150,
+            'description': 'March 2020 COVID crash — 34% in 23 days, VIX to 82'
+        },
+        'flash_crash_2010': {
+            'sp500_shock': -0.09,
+            'duration_minutes': 36,
+            'liquidity_evaporation': True,
+            'description': 'May 6, 2010 — 9% drop in 36 minutes, liquidity vanished'
+        },
+        'russian_default_1998': {
+            'sp500_shock': -0.15,
+            'flight_to_quality': True,
+            'emergency_rate_cut': 0.005,
+            'description': '1998 Russian default/LTCM — 15% drop, flight to Treasuries'
+        }
+    }
+    
+    def __init__(self, portfolio_returns: pd.Series,
+                 benchmark_returns: Optional[pd.Series] = None,
+                 current_weights: Optional[np.ndarray] = None):
+        self.portfolio_returns = portfolio_returns
+        self.benchmark_returns = benchmark_returns
+        self.current_weights = current_weights
+        
+    def run_scenario(self, scenario_name: str) -> Dict:
+        """Run a predefined historical scenario"""
+        if scenario_name not in self.HISTORICAL_SCENARIOS:
+            raise ValueError(f"Unknown scenario: {scenario_name}")
+        
+        scenario = self.HISTORICAL_SCENARIOS[scenario_name]
+        
+        # Estimate portfolio beta to S&P 500
+        if self.benchmark_returns is not None:
+            aligned = pd.concat([self.portfolio_returns, self.benchmark_returns], axis=1).dropna()
+            cov = np.cov(aligned.iloc[:, 0], aligned.iloc[:, 1])
+            beta = cov[0, 1] / cov[1, 1] if cov[1, 1] > 0 else 1.0
+        else:
+            beta = 1.0
+        
+        # Projected portfolio shock
+        shock = scenario['sp500_shock'] * beta
+        
+        # Drawdown from shock
+        drawdown = shock
+        
+        # Time to recover (based on historical averages)
+        recovery_days = scenario.get('recovery_days', 500)
+        recovery_months = recovery_days / 21  # Trading days per month
+        
+        return {
+            'scenario': scenario_name,
+            'description': scenario['description'],
+            'projected_shock': shock,
+            'projected_drawdown': drawdown,
+            'estimated_recovery_months': recovery_months,
+            'survival_check': abs(drawdown) < 0.50,  # Can you survive?
+            'would_blow_up': abs(drawdown) > 0.90  # Would you be ruined?
+        }
+    
+    def run_all_scenarios(self) -> pd.DataFrame:
+        """Run all predefined historical scenarios"""
+        results = []
+        for name in self.HISTORICAL_SCENARIOS:
+            result = self.run_scenario(name)
+            results.append(result)
+        
+        return pd.DataFrame(results)
+    
+    def factor_shock(self, factor_returns: pd.DataFrame,
+                     shock_factors: Dict[str, float]) -> Dict:
+        """
+        Apply factor shocks and project portfolio impact.
+        
+        Args:
+            factor_returns: Factor returns DataFrame (columns = factors)
+            shock_factors: Dict of {factor_name: shock_magnitude}
+        
+        Returns:
+            Projected portfolio impact
+        """
+        # Regress portfolio returns on factors
+        from sklearn.linear_model import LinearRegression
+        
+        aligned = pd.concat([self.portfolio_returns, factor_returns], axis=1).dropna()
+        if len(aligned) < 100:
+            return {'error': 'Insufficient data for factor regression'}
+        
+        X = aligned.iloc[:, 1:].values
+        y = aligned.iloc[:, 0].values
+        
+        reg = LinearRegression().fit(X, y)
+        
+        # Apply shocks
+        shock_vector = np.zeros(len(factor_returns.columns))
+        for factor, shock in shock_factors.items():
+            if factor in factor_returns.columns:
+                idx = list(factor_returns.columns).index(factor)
+                shock_vector[idx] = shock
+        
+        projected_impact = reg.coef_ @ shock_vector
+        
+        return {
+            'projected_return': projected_impact,
+            'projected_drawdown': projected_impact,
+            'factor_exposures': dict(zip(factor_returns.columns, reg.coef_)),
+            'r_squared': reg.score(X, y)
+        }
+    
+    def custom_scenario(self, 
+                        shock_returns: Dict[str, float],
+                        weights: Optional[np.ndarray] = None) -> Dict:
+        """
+        Run a custom scenario with specific asset shocks.
+        
+        Args:
+            shock_returns: Dict of {asset_name: return_shock}
+            weights: Portfolio weights (if None, use current_weights)
+        """
+        if weights is None:
+            weights = self.current_weights
+        
+        if weights is None:
+            return {'error': 'No weights provided'}
+        
+        assets = list(shock_returns.keys())
+        shocks = np.array([shock_returns.get(a, 0) for a in assets])
+        
+        portfolio_impact = weights[:len(shocks)] @ shocks
+        
+        return {
+            'scenario': 'custom',
+            'shocked_assets': shock_returns,
+            'portfolio_impact': portfolio_impact,
+            'projected_drawdown': portfolio_impact
+        }
+
+
+class ComplianceMonitor:
+    """
+    Real-time compliance monitoring.
+    
+    Checks every decision against risk limits.
+    
+    Outputs:
+    - GREEN: Within all limits
+    - YELLOW: Approaching limits (warning)
+    - RED: Limit breached (HARD STOP)
+    """
+    
+    def __init__(self, limits: RiskLimits):
+        self.limits = limits
+        self.violations = []
+        self.warnings = []
+    
+    def check_position_limits(self, weights: np.ndarray,
+                              sector_map: Optional[Dict[int, str]] = None) -> Dict:
+        """Check if positions comply with limits"""
+        status = {'status': 'GREEN', 'violations': [], 'warnings': []}
+        
+        # Individual position limits
+        for i, w in enumerate(weights):
+            if abs(w) > self.limits.max_position_pct:
+                status['violations'].append(
+                    f"Position {i}: {w*100:.1f}% > max {self.limits.max_position_pct*100:.0f}%"
+                )
+                status['status'] = 'RED'
+        
+        # Sector limits
+        if sector_map is not None:
+            sectors = {}
+            for i, sector in sector_map.items():
+                if i < len(weights):
+                    sectors[sector] = sectors.get(sector, 0) + abs(weights[i])
+            
+            for sector, total in sectors.items():
+                if total > self.limits.max_sector_pct:
+                    status['violations'].append(
+                        f"Sector {sector}: {total*100:.1f}% > max {self.limits.max_sector_pct*100:.0f}%"
+                    )
+                    status['status'] = 'RED'
+        
+        # Leverage
+        gross_leverage = np.sum(np.abs(weights))
+        if gross_leverage > self.limits.max_leverage:
+            status['violations'].append(
+                f"Gross leverage: {gross_leverage:.2f}x > max {self.limits.max_leverage:.1f}x"
+            )
+            status['status'] = 'RED'
+        
+        return status
+    
+    def check_var_limits(self, var_result: Dict) -> Dict:
+        """Check if VaR is within limits"""
+        status = {'status': 'GREEN', 'violations': [], 'warnings': []}
+        
+        daily_var = abs(var_result.get('var_pct', 0)) / 100
+        daily_cvar = abs(var_result.get('cvar_pct', 0)) / 100
+        
+        if daily_var > self.limits.daily_var_limit:
+            status['violations'].append(
+                f"Daily VaR: {daily_var*100:.2f}% > limit {self.limits.daily_var_limit*100:.0f}%"
+            )
+            status['status'] = 'RED'
+        elif daily_var > self.limits.daily_var_limit * 0.8:
+            status['warnings'].append(
+                f"Daily VaR: {daily_var*100:.2f}% approaching limit"
+            )
+            if status['status'] == 'GREEN':
+                status['status'] = 'YELLOW'
+        
+        if daily_cvar > self.limits.daily_cvar_limit:
+            status['violations'].append(
+                f"Daily CVaR: {daily_cvar*100:.2f}% > limit {self.limits.daily_cvar_limit*100:.0f}%"
+            )
+            status['status'] = 'RED'
+        
+        return status
+    
+    def check_drawdown(self, current_drawdown: float) -> Dict:
+        """Check if drawdown is within limits"""
+        status = {'status': 'GREEN', 'violations': [], 'warnings': []}
+        
+        if current_drawdown < -self.limits.max_drawdown_limit:
+            status['violations'].append(
+                f"Drawdown: {current_drawdown*100:.1f}% > limit {self.limits.max_drawdown_limit*100:.0f}%"
+            )
+            status['status'] = 'RED'
+        elif current_drawdown < -self.limits.max_drawdown_limit * 0.7:
+            status['warnings'].append(
+                f"Drawdown: {current_drawdown*100:.1f}% approaching limit"
+            )
+            if status['status'] == 'GREEN':
+                status['status'] = 'YELLOW'
+        
+        return status
+    
+    def full_compliance_check(self,
+                               weights: np.ndarray,
+                               var_result: Dict,
+                               current_drawdown: float,
+                               sector_map: Optional[Dict[int, str]] = None) -> Dict:
+        """Run all compliance checks and return overall status"""
+        pos_check = self.check_position_limits(weights, sector_map)
+        var_check = self.check_var_limits(var_result)
+        dd_check = self.check_drawdown(current_drawdown)
+        
+        # Aggregate
+        all_violations = (pos_check.get('violations', []) + 
+                         var_check.get('violations', []) + 
+                         dd_check.get('violations', []))
+        all_warnings = (pos_check.get('warnings', []) + 
+                       var_check.get('warnings', []) + 
+                       dd_check.get('warnings', []))
+        
+        # Determine overall status
+        if any(s.get('status') == 'RED' for s in [pos_check, var_check, dd_check]):
+            overall_status = 'RED'
+        elif any(s.get('status') == 'YELLOW' for s in [pos_check, var_check, dd_check]):
+            overall_status = 'YELLOW'
+        else:
+            overall_status = 'GREEN'
+        
+        result = {
+            'overall_status': overall_status,
+            'position_check': pos_check,
+            'var_check': var_check,
+            'drawdown_check': dd_check,
+            'violations': all_violations,
+            'warnings': all_warnings,
+            'can_trade': overall_status != 'RED',
+            'should_reduce': overall_status == 'YELLOW'
+        }
+        
+        # Log
+        if overall_status == 'RED':
+            self.violations.append(result)
+        elif overall_status == 'YELLOW':
+            self.warnings.append(result)
+        
+        return result
+
+
+def run_full_risk_assessment(returns: pd.DataFrame,
+                                weights: np.ndarray,
+                                current_drawdown: float = 0.0,
+                                sector_map: Optional[Dict[int, str]] = None,
+                                confidence: float = 0.95) -> Dict:
+    """
+    Complete risk assessment — run this EVERY DAY before trading.
+    
+    Returns:
+        Comprehensive risk report with VaR, stress tests, and compliance status
+    """
+    print("=" * 70)
+    print("  DAILY RISK ASSESSMENT")
+    print("=" * 70)
+    
+    # 1. VaR Calculation
+    print("\n[1/4] Value at Risk Analysis...")
+    var_engine = ValueAtRisk(confidence=confidence, method='historical')
+    var_result = var_engine.calculate(returns.values, weights=weights)
+    
+    var_engine_param = ValueAtRisk(confidence=confidence, method='parametric', distribution='t')
+    var_param = var_engine_param.calculate(returns.values, weights=weights)
+    
+    print(f"  Historical VaR ({confidence*100:.0f}%): {var_result['var_pct']:.2f}%")
+    print(f"  Historical CVaR ({confidence*100:.0f}%): {var_result['cvar_pct']:.2f}%")
+    print(f"  Parametric VaR (Student-t): {var_param['var_pct']:.2f}%")
+    print(f"  Parametric CVaR (Student-t): {var_param['cvar_pct']:.2f}%")
+    
+    # 2. Stress Testing
+    print("\n[2/4] Stress Testing...")
+    port_returns = (returns.values @ weights) if returns.values.ndim == 2 else returns.values
+    stress = StressTesting(
+        pd.Series(port_returns),
+        benchmark_returns=returns.iloc[:, 0] if len(returns.columns) > 0 else None
+    )
+    stress_results = stress.run_all_scenarios()
+    
+    for _, row in stress_results.iterrows():
+        icon = '✓' if not row['would_blow_up'] else '✗'
+        print(f"  {icon} {row['scenario']}: {row['projected_drawdown']*100:.1f}% drawdown, "
+              f"recovery {row['estimated_recovery_months']:.0f} months")
+    
+    # 3. Compliance
+    print("\n[3/4] Compliance Check...")
+    limits = RiskLimits()
+    compliance = ComplianceMonitor(limits)
+    compliance_result = compliance.full_compliance_check(
+        weights, var_result, current_drawdown, sector_map
+    )
+    
+    status_icon = {'GREEN': '✓', 'YELLOW': '⚠', 'RED': '✗'}
+    icon = status_icon.get(compliance_result['overall_status'], '?')
+    print(f"  {icon} OVERALL STATUS: {compliance_result['overall_status']}")
+    
+    if compliance_result['violations']:
+        print(f"  VIOLATIONS:")
+        for v in compliance_result['violations']:
+            print(f"    - {v}")
+    
+    if compliance_result['warnings']:
+        print(f"  WARNINGS:")
+        for w in compliance_result['warnings']:
+            print(f"    - {w}")
+    
+    # 4. Risk Summary
+    print("\n[4/4] Risk Summary...")
+    
+    # Portfolio statistics
+    port_mean = np.mean(port_returns) * 252
+    port_vol = np.std(port_returns) * np.sqrt(252)
+    sharpe = (port_mean - 0.04) / port_vol if port_vol > 0 else 0
+    max_dd = (np.cumprod(1 + port_returns) - np.maximum.accumulate(np.cumprod(1 + port_returns))).min()
+    
+    summary = {
+        'daily_var_95': var_result['var_pct'],
+        'daily_cvar_95': var_result['cvar_pct'],
+        'annual_return': port_mean,
+        'annual_volatility': port_vol,
+        'sharpe_ratio': sharpe,
+        'max_drawdown': max_dd,
+        'compliance_status': compliance_result['overall_status'],
+        'can_trade': compliance_result['can_trade'],
+        'should_reduce': compliance_result['should_reduce'],
+        'stress_tests': stress_results.to_dict('records'),
+        'violations': compliance_result['violations'],
+        'warnings': compliance_result['warnings']
+    }
+    
+    print(f"\n  Annual Return:      {port_mean*100:.1f}%")
+    print(f"  Annual Volatility:  {port_vol*100:.1f}%")
+    print(f"  Sharpe Ratio:       {sharpe:.2f}")
+    print(f"  Max Drawdown:       {max_dd*100:.1f}%")
+    print(f"  Daily VaR (95%):    {var_result['var_pct']:.2f}%")
+    print(f"  Daily CVaR (95%):   {var_result['cvar_pct']:.2f}%")
+    print(f"\n  {'='*70}")
+    print(f"  CAN TRADE TODAY: {compliance_result['can_trade']}")
+    print(f"  {'='*70}")
+    
+    return summary
+
+
+if __name__ == '__main__':
+    # Test risk management
+    np.random.seed(42)
+    
+    # Generate synthetic portfolio
+    n_days = 1000
+    n_assets = 5
+    returns = np.random.randn(n_days, n_assets) * 0.02
+    weights = np.array([0.20, 0.20, 0.20, 0.20, 0.20])
+    
+    returns_df = pd.DataFrame(returns, columns=[f'Asset_{i}' for i in range(n_assets)])
+    
+    summary = run_full_risk_assessment(
+        returns_df, weights, current_drawdown=-0.05
+    )