risk-stress-testing

---

name: risk-stress-testing compatibility: opencode completeness: 95 content-types:

• code
• guidance
• config
• do-dont description: '"Implements stress test scenarios and portfolio resilience analysis for risk management and algorithmic trading execution."' license: MIT maturity: stable metadata: domain: trading output-format: code related-skills: backtest-drawdown-analysis, exchange-order-execution-api role: implementation scope: implementation triggers: portfolio, resilience, risk stress testing, risk-stress-testing, scenarios, unit tests, testing, test automation archetypes:
  • tactical anti_triggers:
  • brainstorming
  • vague ideation
  • no risk management response_profile: verbosity: low directive_strength: high abstraction_level: operational version: "1.0.0"

---

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

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

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
• Monte Carlo Stress Testing Methods
• Regulatory Stress Testing Frameworks
• Portfolio Stress Testing Techniques
• Historical Scenario Analysis