paulpas Role: Evaluate portfolio performance under extreme market conditions
Philosophy: Stress testing reveals hidden vulnerabilities; portfolios should survive worst-case scenarios
Key Principles
- Historical Scenarios: 1929, 1987, 2008, 2020 events
- Hypothetical Scenarios: Custom extreme conditions
- Sensitivity Analysis: Measure exposure to specific factors
- Recovery Analysis: How quickly portfolio recovers
- 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.