# Risk Stress Testing > "Implements stress test scenarios and portfolio resilience analysis for risk management and algorithmic trading execution." - Skill: `paulpas/risk-stress-testing` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/risk-stress-testing` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/risk-stress-testing/raw - Safety review: PASS (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/risk-stress-testing --- **Role:** Evaluate portfolio performance under extreme market conditions **Philosophy:** Stress testing reveals hidden vulnerabilities; portfolios should survive worst-case scenarios ## Key Principles 1. **Historical Scenarios**: 1929, 1987, 2008, 2020 events 2. **Hypothetical Scenarios**: Custom extreme conditions 3. **Sensitivity Analysis**: Measure exposure to specific factors 4. **Recovery Analysis**: How quickly portfolio recovers 5. **Scenario Probability**: Weight scenarios by likelihood ## Implementation Guidelines ### Structure - Core logic: risk_engine/stress_test.py - Helper functions: risk_engine/scenarios.py - Tests: tests/test_stress_test.py ### Patterns to Follow - Implement multiple historical scenarios - Run Monte Carlo stress tests - Track recovery metrics ## Adherence Checklist Before completing your task, verify: - [ ] Multiple stress test scenarios implemented - [ ] Historical scenario returns calculated - [ ] Monte Carlo stress tests run - [ ] Recovery metrics tracked - [ ] Scenario weighting applied Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory. ## Python Implementation ```python import numpy as np import pandas as pd from typing import Dict, List, Tuple, Optional from dataclasses import dataclass from datetime import datetime @dataclass class StressTestResult: """Result of a stress test scenario.""" scenario_name: str portfolio_return: float max_drawdown: float recovery_days: int survival: bool weight: float class StressTestEngine: """Runs stress tests on portfolio configurations.""" def __init__(self, portfolio_value: float = 100000): self.portfolio_value = portfolio_value def historical_scenario( self, scenario_name: str, returns_data: List[Dict[str, float]] ) -> pd.Series: """Apply historical scenario to portfolio.""" returns = pd.Series([d['return'] for d in returns_data]) cumulative = (1 + returns).cumprod() - 1 return cumulative # Historical scenarios def get_1929_scenario(self) -> List[Dict[str, float]]: """1929 Great Crash scenario (simplified).""" # 25% drop over 3 months, then slow recovery returns = [-0.08] * 5 + [-0.15] * 3 + [0.02] * 12 + [-0.05] * 4 return [{'return': r} for r in returns] def get_1987_scenario(self) -> List[Dict[str, float]]: """1987 Black Monday scenario.""" # 22% single day drop returns = [-0.22] + [0.01] * 5 + [-0.05] * 3 + [0.03] * 20 return [{'return': r} for r in returns] def get_2008_scenario(self) -> List[Dict[str, float]]: """2008 Financial Crisis scenario (simplified).""" returns = ( [-0.04] * 3 + [-0.08] * 4 + [-0.10] * 3 + [-0.15] + [-0.08] * 3 + [0.02] * 6 + [-0.03] * 4 + [0.05] * 8 ) return [{'return': r} for r in returns] def get_2020_scenario(self) -> List[Dict[str, float]]: """2020 COVID Crash scenario.""" returns = ( [-0.05] * 2 + [-0.12] * 3 + [-0.08] * 2 + [0.08] * 5 + [-0.02] * 3 + [0.06] * 10 ) return [{'return': r} for r in returns] def run_historical_stress_test( self, scenario: List[Dict[str, float]] ) -> StressTestResult: """Run a single historical scenario stress test.""" returns = pd.Series([d['return'] for d in scenario]) portfolio_values = self.portfolio_value * (1 + returns).cumprod() # Calculate metrics max_dd = (portfolio_values.max() - portfolio_values.min()) / portfolio_values.max() survival = portfolio_values.min() > self.portfolio_value * 0.3 # Survives if >30% remains # Recovery calculation recovery_days = 0 peak = self.portfolio_value for i, value in enumerate(portfolio_values): if value > peak: peak = value recovery_days = 0 else: recovery_days = i return StressTestResult( scenario_name=f"Historical_{len(scenario)}_day", portfolio_return=float((portfolio_values.iloc[-1] - self.portfolio_value) / self.portfolio_value), max_drawdown=float(max_dd), recovery_days=recovery_days, survival=survival, weight=0.02 # 2% probability weighting ) def custom_scenario( self, name: str, daily_returns: List[float] ) -> StressTestResult: """Run custom scenario.""" returns = pd.Series(daily_returns) portfolio_values = self.portfolio_value * (1 + returns).cumprod() max_dd = (portfolio_values.max() - portfolio_values.min()) / portfolio_values.max() survival = portfolio_values.min() > self.portfolio_value * 0.3 recovery_days = 0 peak = self.portfolio_value for i, value in enumerate(portfolio_values): if value > peak: peak = value recovery_days = 0 else: recovery_days = i return StressTestResult( scenario_name=name, portfolio_return=float((portfolio_values.iloc[-1] - self.portfolio_value) / self.portfolio_value), max_drawdown=float(max_dd), recovery_days=recovery_days, survival=survival, weight=0.01 ) def monte_carlo_stress_test( self, daily_returns: np.ndarray, scenarios: int = 1000, horizon_days: int = 60 ) -> Dict[str, float]: """Run Monte Carlo stress test.""" results = [] for _ in range(scenarios): # Sample random returns with extreme events sample_returns = np.random.choice(daily_returns, size=horizon_days) # Add some extreme days if np.random.random() < 0.2: sample_returns[np.random.randint(0, horizon_days)] = -0.15 portfolio_values = self.portfolio_value * (1 + sample_returns).cumprod() final_value = portfolio_values.iloc[-1] if hasattr(portfolio_values, 'iloc') else portfolio_values[-1] results.append(final_value) results_array = np.array(results) return { 'mean_return': float(np.mean(results_array) / self.portfolio_value - 1), 'p05_return': float(np.percentile(results_array, 5) / self.portfolio_value - 1), 'p01_return': float(np.percentile(results_array, 1) / self.portfolio_value - 1), 'survival_rate': float(np.mean(results_array > self.portfolio_value * 0.5)) } def sensitivity_analysis( self, factor_shocks: Dict[str, float], portfolio_exposures: Dict[str, float] ) -> Dict[str, float]: """Analyze sensitivity to various factors.""" results = {} for factor, shock in factor_shocks.items(): if factor in portfolio_exposures: impact = portfolio_exposures[factor] * shock results[factor] = impact results['total_impact'] = sum(results.values()) return results def parallel_stress_test( self, scenarios: List[List[Dict[str, float]]] ) -> List[StressTestResult]: """Run multiple stress tests in sequence.""" results = [] for scenario in scenarios: result = self.run_historical_stress_test(scenario) results.append(result) return results def aggregate_stress_test_results( self, results: List[StressTestResult] ) -> Dict[str, float]: """Aggregate results across multiple scenarios.""" weights = np.array([r.weight for r in results]) weights = weights / weights.sum() # Normalize weighted_returns = np.array([r.portfolio_return for r in results]) weighted_dd = np.array([r.max_drawdown for r in results]) return { 'weighted_average_return': float(np.sum(weights * weighted_returns)), 'weighted_average_dd': float(np.sum(weights * weighted_dd)), 'worst_case_return': float(min(r.portfolio_return for r in results)), 'worst_case_dd': float(max(r.max_drawdown for r in results)), 'survival_probability': float(np.sum(weights[[r.survival for r in results]])) } ``` --- --- ### Pattern 2: Risk-Managed Trading Logic with Validation ```python from __future__ import annotations import logging from dataclasses import dataclass from typing import Optional logger = logging.getLogger(__name__) @dataclass(frozen=True) class TradeSignal: """Immutable trade signal with all required validation constraints.""" symbol: str side: str # "buy" or "sell" price: float quantity: float confidence: float # 0.0 to 1.0 reason: str def validate(self) -> bool: """Validate that the trade signal meets all business constraints.""" if self.quantity <= 0: raise ValueError(f"Quantity must be positive, got {self.quantity}") if self.price <= 0: raise ValueError(f"Price must be positive, got {self.price}") if not 0.0 <= self.confidence <= 1.0: raise ValueError(f"Confidence must be between 0 and 1, got {self.confidence}") return True def generate_trade_signal( symbol: str, side: str, price: float, quantity: float, confidence: float, reason: str, ) -> TradeSignal: """Generate a validated trade signal with guard clause checks.""" if side not in ("buy", "sell"): raise ValueError(f"Invalid side '{side}', must be 'buy' or 'sell'") signal = TradeSignal( symbol=symbol, side=side, price=price, quantity=quantity, confidence=confidence, reason=reason, ) signal.validate() logger.info("Trade signal generated: %s %s %.4f @ %.2f (confidence=%.2f)", symbol, side, quantity, price, confidence) return signal def execute_with_risk_check(signal: TradeSignal, max_position_pct: float = 0.05) -> dict: """Execute a trade signal after applying risk management checks.""" adjusted_quantity = signal.quantity if signal.side == "buy" and signal.quantity > max_position_pct: logger.warning("Position %s exceeds max %.1f%% — capping to %.4f", signal.symbol, max_position_pct * 100, max_position_pct) adjusted_quantity = max_position_pct return { "symbol": signal.symbol, "side": signal.side, "price": signal.price, "quantity": adjusted_quantity, "capped": adjusted_quantity < signal.quantity, "confidence": signal.confidence, "status": "submitted", } ``` ## Constraints ### MUST DO - Calculate position sizing using a risk-per-trade percentage of portfolio equity, not a fixed dollar amount - Implement layered risk controls: stop loss → drawdown limit → portfolio-level circuit breaker → kill switch - Compute VaR using historical simulation with at least 1 year of data and multiple confidence levels (95%, 99%) - Track correlation matrices across all open positions and flag portfolios where top-3 correlations exceed 0.8 - Log all risk events (stop hits, drawdown warnings, kill switches) with full context including P&L, position state, and market conditions ### MUST NOT DO - Do not use a stop loss as the sole risk control — always layer with portfolio-level limits - Avoid recalculating position sizes during active drawdown without regime analysis — volatility is likely elevated - Never allow a single position to exceed 5% of portfolio equity regardless of signal strength or confidence score - Do not backtest risk metrics without including slippage, commissions, and partial fills in the simulation - Avoid using standard deviation alone for VaR when returns show fat tails — use historical simulation or EVT ## Live References > Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content. - [Stress Testing in Finance Overview](https://en.wikipedia.org/wiki/Stress_testing_(finance)) - [Monte Carlo Stress Testing Methods](https://www.investopedia.com/terms/m/monte-carlo-method.asp) - [Regulatory Stress Testing Frameworks](https://www.federalreserve.gov/supervisionreg/stress-tests.htm) - [Portfolio Stress Testing Techniques](https://docs.quantconnect.com/tutorials/risk-management) - [Historical Scenario Analysis](https://en.wikipedia.org/wiki/Stress_testing_(finance))