Stress Testing and Scenario Analysis for Portfolios

1import numpy as np
2import pandas as pd
3from typing import Dict, List, Tuple
4from dataclasses import dataclass
5from datetime import datetime
6
7@dataclass
8class StressScenario:
9    """Container for stress test scenario"""
10    name: str
11    description: str
12    start_date: str
13    end_date: str
14    factor_shocks: Dict[str, float]  # Factor -> return
15    portfolio_pnl: float = 0.0
16    pnl_pct: float = 0.0
17
18class HistoricalStressTester:
19    """
20    Historical stress testing framework
21    
22    Applies historical crisis scenarios to current portfolio
23    """
24    
25    # Pre-defined historical scenarios
26    SCENARIOS = {
27        'lehman_crisis': {
28            'name': 'Lehman Brothers Collapse',
29            'description': 'Sep 15-30, 2008: Lehman bankruptcy and credit freeze',
30            'start_date': '2008-09-15',
31            'end_date': '2008-09-30'
32        },
33        'covid_crash': {
34            'name': 'COVID-19 Market Crash',
35            'description': 'Feb 19 - Mar 23, 2020: Pandemic-driven selloff',
36            'start_date': '2020-02-19',
37            'end_date': '2020-03-23'
38        },
39        'dotcom_bubble': {
40            'name': 'Dot-com Bubble Burst',
41            'description': 'Mar-Apr 2000: Tech stock collapse',
42            'start_date': '2000-03-10',
43            'end_date': '2000-04-14'
44        },
45        'flash_crash': {
46            'name': 'Flash Crash',
47            'description': 'May 6, 2010: Intraday market crash',
48            'start_date': '2010-05-06',
49            'end_date': '2010-05-06'
50        },
51        'brexit': {
52            'name': 'Brexit Vote',
53            'description': 'Jun 23-24, 2016: UK votes to leave EU',
54            'start_date': '2016-06-23',
55            'end_date': '2016-06-24'
56        },
57        'volmageddon': {
58            'name': 'Volmageddon',
59            'description': 'Feb 5, 2018: VIX spike and vol ETF collapse',
60            'start_date': '2018-02-05',
61            'end_date': '2018-02-06'
62        },
63        'russian_default': {
64            'name': 'Russian Default / LTCM',
65            'description': 'Aug-Sep 1998: Russian default and LTCM crisis',
66            'start_date': '1998-08-17',
67            'end_date': '1998-09-30'
68        }
69    }
70    
71    def __init__(self, historical_data: pd.DataFrame):
72        """
73        Args:
74            historical_data: DataFrame with asset prices (columns) and dates (index)
75        """
76        self.historical_data = historical_data
77    
78    def calculate_scenario_shocks(self, 
79                                  scenario_name: str) -> Dict[str, float]:
80        """
81        Calculate factor returns during a historical scenario
82        
83        Args:
84            scenario_name: Name of scenario (from SCENARIOS dict)
85            
86        Returns:
87            Dict mapping asset -> total return during scenario
88        """
89        if scenario_name not in self.SCENARIOS:
90            raise ValueError(f"Unknown scenario: {scenario_name}")
91        
92        scenario = self.SCENARIOS[scenario_name]
93        start_date = scenario['start_date']
94        end_date = scenario['end_date']
95        
96        # Get prices at start and end of scenario
97        try:
98            start_prices = self.historical_data.loc[start_date]
99            end_prices = self.historical_data.loc[end_date]
100        except KeyError:
101            # If exact dates not available, use nearest
102            start_prices = self.historical_data.asof(pd.Timestamp(start_date))
103            end_prices = self.historical_data.asof(pd.Timestamp(end_date))
104        
105        # Calculate returns
106        returns = (end_prices - start_prices) / start_prices
107        
108        return returns.to_dict()
109    
110    def stress_test_portfolio(self,
111                             positions: pd.DataFrame,
112                             scenarios: List[str] = None) -> pd.DataFrame:
113        """
114        Stress test portfolio against historical scenarios
115        
116        Args:
117            positions: DataFrame with columns ['asset', 'quantity', 'current_price']
118            scenarios: List of scenario names (default: all scenarios)
119            
120        Returns:
121            DataFrame with scenario results
122        """
123        if scenarios is None:
124            scenarios = list(self.SCENARIOS.keys())
125        
126        # Calculate current portfolio value
127        portfolio_value = (positions['quantity'] * positions['current_price']).sum()
128        
129        results = []
130        
131        for scenario_name in scenarios:
132            # Get factor shocks
133            factor_shocks = self.calculate_scenario_shocks(scenario_name)
134            
135            # Calculate portfolio P&L
136            pnl = 0.0
137            
138            for _, row in positions.iterrows():
139                asset = row['asset']
140                quantity = row['quantity']
141                current_price = row['current_price']
142                
143                if asset in factor_shocks:
144                    shock = factor_shocks[asset]
145                    asset_pnl = quantity * current_price * shock
146                    pnl += asset_pnl
147            
148            scenario_info = self.SCENARIOS[scenario_name]
149            
150            results.append({
151                'scenario': scenario_info['name'],
152                'description': scenario_info['description'],
153                'start_date': scenario_info['start_date'],
154                'end_date': scenario_info['end_date'],
155                'portfolio_pnl': pnl,
156                'pnl_pct': (pnl / portfolio_value * 100) if portfolio_value > 0 else 0,
157                'portfolio_value': portfolio_value,
158                'stressed_value': portfolio_value + pnl
159            })
160        
161        df_results = pd.DataFrame(results)
162        
163        # Sort by worst P&L
164        df_results = df_results.sort_values('portfolio_pnl')
165        
166        return df_results
167    
168    def decompose_scenario_pnl(self,
169                              positions: pd.DataFrame,
170                              scenario_name: str) -> pd.DataFrame:
171        """
172        Decompose scenario P&L by asset
173        
174        Args:
175            positions: Portfolio positions
176            scenario_name: Scenario to analyze
177            
178        Returns:
179            DataFrame with P&L contribution by asset
180        """
181        factor_shocks = self.calculate_scenario_shocks(scenario_name)
182        
183        contributions = []
184        
185        for _, row in positions.iterrows():
186            asset = row['asset']
187            quantity = row['quantity']
188            current_price = row['current_price']
189            position_value = quantity * current_price
190            
191            if asset in factor_shocks:
192                shock = factor_shocks[asset]
193                pnl = position_value * shock
194                
195                contributions.append({
196                    'asset': asset,
197                    'position_value': position_value,
198                    'shock_pct': shock * 100,
199                    'pnl': pnl,
200                    'pnl_pct': (pnl / position_value * 100) if position_value != 0 else 0
201                })
202        
203        df_contrib = pd.DataFrame(contributions)
204        df_contrib = df_contrib.sort_values('pnl')
205        
206        return df_contrib
207
208
209# Example usage
210if __name__ == "__main__":
211    # Generate sample historical data
212    np.random.seed(42)
213    dates = pd.date_range(start='1998-01-01', end='2025-11-25', freq='D')
214    
215    # Simulate asset prices with crisis periods
216    n_days = len(dates)
217    
218    # Create realistic price paths with embedded crises
219    spy_prices = 100 * np.exp(np.cumsum(np.random.randn(n_days) * 0.01))
220    
221    # Inject 2008 crisis
222    crisis_2008_start = dates.get_loc('2008-09-15')
223    crisis_2008_end = dates.get_loc('2008-09-30')
224    spy_prices[crisis_2008_start:crisis_2008_end] *= np.linspace(1.0, 0.85, crisis_2008_end - crisis_2008_start)
225    
226    # Inject COVID crash
227    covid_start = dates.get_loc('2020-02-19')
228    covid_end = dates.get_loc('2020-03-23')
229    spy_prices[covid_start:covid_end] *= np.linspace(1.0, 0.70, covid_end - covid_start)
230    
231    historical_data = pd.DataFrame({
232        'SPY': spy_prices,
233        'TLT': 100 * np.exp(np.cumsum(np.random.randn(n_days) * 0.005)),
234        'GLD': 100 * np.exp(np.cumsum(np.random.randn(n_days) * 0.008)),
235    }, index=dates)
236    
237    # Current portfolio
238    positions = pd.DataFrame({
239        'asset': ['SPY', 'TLT', 'GLD'],
240        'quantity': [1000, 500, 200],
241        'current_price': [450.0, 95.0, 180.0]
242    })
243    
244    # Stress test
245    stress_tester = HistoricalStressTester(historical_data)
246    results = stress_tester.stress_test_portfolio(positions)
247    
248    print("Historical Stress Test Results:")
249    print("=" * 100)
250    print(results.to_string(index=False))
251    
252    print("\n\nWorst Scenario Decomposition:")
253    print("=" * 80)
254    worst_scenario = results.iloc[0]['scenario']
255    
256    # Find scenario key
257    scenario_key = [k for k, v in stress_tester.SCENARIOS.items() 
258                   if v['name'] == worst_scenario][0]
259    
260    decomp = stress_tester.decompose_scenario_pnl(positions, scenario_key)
261    print(decomp.to_string(index=False))
262

1class HypotheticalStressTester:
2    """
3    Hypothetical stress testing with custom factor shocks
4    
5    Allows defining arbitrary scenarios with factor shocks
6    """
7    
8    def __init__(self):
9        self.scenarios = {}
10    
11    def add_scenario(self,
12                    name: str,
13                    description: str,
14                    factor_shocks: Dict[str, float]):
15        """
16        Add a hypothetical scenario
17        
18        Args:
19            name: Scenario name
20            description: Scenario description
21            factor_shocks: Dict mapping factor -> shock (e.g., {'SPY': -0.30})
22        """
23        self.scenarios[name] = {
24            'description': description,
25            'factor_shocks': factor_shocks
26        }
27    
28    def add_predefined_scenarios(self):
29        """Add common hypothetical scenarios"""
30        
31        # Equity crash
32        self.add_scenario(
33            name='Equity Crash -30%',
34            description='Severe equity market selloff',
35            factor_shocks={
36                'SPY': -0.30,
37                'QQQ': -0.35,
38                'IWM': -0.40,
39                'EFA': -0.28,
40                'EEM': -0.45
41            }
42        )
43        
44        # Interest rate shock
45        self.add_scenario(
46            name='Rate Shock +200bps',
47            description='Sudden interest rate spike',
48            factor_shocks={
49                'TLT': -0.15,  # Long bonds hurt most
50                'IEF': -0.08,  # Intermediate bonds
51                'SHY': -0.02,  # Short bonds least affected
52                'SPY': -0.10,  # Equities down (higher discount rate)
53            }
54        )
55        
56        # Credit crisis
57        self.add_scenario(
58            name='Credit Crisis',
59            description='Credit spread widening +300bps',
60            factor_shocks={
61                'HYG': -0.20,  # High yield bonds
62                'LQD': -0.10,  # Investment grade bonds
63                'SPY': -0.25,  # Equities down
64                'GLD': 0.15,   # Flight to safety
65            }
66        )
67        
68        # Inflation shock
69        self.add_scenario(
70            name='Inflation Shock',
71            description='Unexpected inflation spike',
72            factor_shocks={
73                'TLT': -0.12,  # Bonds down
74                'TIP': 0.05,   # TIPS up
75                'GLD': 0.20,   # Gold up
76                'DBC': 0.25,   # Commodities up
77                'SPY': -0.08,  # Equities mixed
78            }
79        )
80        
81        # Deflation shock
82        self.add_scenario(
83            name='Deflation Shock',
84            description='Deflationary spiral',
85            factor_shocks={
86                'TLT': 0.15,   # Bonds up
87                'SPY': -0.20,  # Equities down
88                'GLD': -0.05,  # Gold down
89                'DBC': -0.30,  # Commodities crash
90            }
91        )
92        
93        # Geopolitical crisis
94        self.add_scenario(
95            name='Geopolitical Crisis',
96            description='Major geopolitical event (war, terrorism)',
97            factor_shocks={
98                'SPY': -0.15,
99                'EEM': -0.25,  # Emerging markets hit hard
100                'GLD': 0.10,   # Flight to safety
101                'USO': 0.30,   # Oil spike
102                'VIX': 1.50,   # Volatility explosion
103            }
104        )
105    
106    def stress_test_portfolio(self,
107                             positions: pd.DataFrame,
108                             scenarios: List[str] = None) -> pd.DataFrame:
109        """
110        Stress test portfolio against hypothetical scenarios
111        
112        Args:
113            positions: Portfolio positions
114            scenarios: List of scenario names (default: all)
115            
116        Returns:
117            DataFrame with results
118        """
119        if scenarios is None:
120            scenarios = list(self.scenarios.keys())
121        
122        portfolio_value = (positions['quantity'] * positions['current_price']).sum()
123        
124        results = []
125        
126        for scenario_name in scenarios:
127            if scenario_name not in self.scenarios:
128                continue
129            
130            scenario = self.scenarios[scenario_name]
131            factor_shocks = scenario['factor_shocks']
132            
133            # Calculate P&L
134            pnl = 0.0
135            
136            for _, row in positions.iterrows():
137                asset = row['asset']
138                quantity = row['quantity']
139                current_price = row['current_price']
140                
141                if asset in factor_shocks:
142                    shock = factor_shocks[asset]
143                    asset_pnl = quantity * current_price * shock
144                    pnl += asset_pnl
145            
146            results.append({
147                'scenario': scenario_name,
148                'description': scenario['description'],
149                'portfolio_pnl': pnl,
150                'pnl_pct': (pnl / portfolio_value * 100) if portfolio_value > 0 else 0,
151                'portfolio_value': portfolio_value,
152                'stressed_value': portfolio_value + pnl
153            })
154        
155        df_results = pd.DataFrame(results)
156        df_results = df_results.sort_values('portfolio_pnl')
157        
158        return df_results
159    
160    def create_combined_scenario(self,
161                                name: str,
162                                description: str,
163                                component_scenarios: List[str],
164                                weights: List[float] = None):
165        """
166        Create a combined scenario from multiple scenarios
167        
168        Args:
169            name: New scenario name
170            description: Description
171            component_scenarios: List of existing scenario names
172            weights: Weights for each component (default: equal weight)
173        """
174        if weights is None:
175            weights = [1.0 / len(component_scenarios)] * len(component_scenarios)
176        
177        # Combine factor shocks
178        combined_shocks = {}
179        
180        for scenario_name, weight in zip(component_scenarios, weights):
181            if scenario_name not in self.scenarios:
182                continue
183            
184            for factor, shock in self.scenarios[scenario_name]['factor_shocks'].items():
185                if factor not in combined_shocks:
186                    combined_shocks[factor] = 0.0
187                combined_shocks[factor] += shock * weight
188        
189        self.add_scenario(name, description, combined_shocks)
190
191
192# Example usage
193if __name__ == "__main__":
194    # Portfolio
195    positions = pd.DataFrame({
196        'asset': ['SPY', 'TLT', 'GLD', 'HYG', 'EEM'],
197        'quantity': [1000, 500, 200, 300, 400],
198        'current_price': [450.0, 95.0, 180.0, 75.0, 40.0]
199    })
200    
201    # Hypothetical stress test
202    hypo_tester = HypotheticalStressTester()
203    hypo_tester.add_predefined_scenarios()
204    
205    results = hypo_tester.stress_test_portfolio(positions)
206    
207    print("Hypothetical Stress Test Results:")
208    print("=" * 100)
209    print(results.to_string(index=False))
210    
211    # Create combined scenario
212    hypo_tester.create_combined_scenario(
213        name='Perfect Storm',
214        description='Equity crash + credit crisis + rate shock',
215        component_scenarios=['Equity Crash -30%', 'Credit Crisis', 'Rate Shock +200bps'],
216        weights=[0.5, 0.3, 0.2]
217    )
218    
219    perfect_storm = hypo_tester.stress_test_portfolio(
220        positions, 
221        scenarios=['Perfect Storm']
222    )
223    
224    print("\n\nPerfect Storm Scenario:")
225    print("=" * 100)
226    print(perfect_storm.to_string(index=False))
227

1from scipy.optimize import minimize, differential_evolution
2
3class ReverseStressTester:
4    """
5    Reverse stress testing: find scenarios that cause specific losses
6    
7    Uses optimization to find factor shocks that produce target P&L
8    """
9    
10    def __init__(self, 
11                 factors: List[str],
12                 max_shock: float = 0.50):
13        """
14        Args:
15            factors: List of risk factors
16            max_shock: Maximum allowed shock per factor (e.g., 0.50 = 50%)
17        """
18        self.factors = factors
19        self.max_shock = max_shock
20    
21    def find_breaking_scenario(self,
22                              positions: pd.DataFrame,
23                              target_loss: float,
24                              method: str = 'minimize_shock') -> Dict:
25        """
26        Find scenario that produces target loss
27        
28        Args:
29            positions: Portfolio positions
30            target_loss: Target P&L (negative for loss)
31            method: 'minimize_shock' (smallest shocks) or 'realistic' (plausible shocks)
32            
33        Returns:
34            Dict with scenario shocks and statistics
35        """
36        # Build position mapping
37        position_map = {}
38        for _, row in positions.iterrows():
39            asset = row['asset']
40            value = row['quantity'] * row['current_price']
41            position_map[asset] = value
42        
43        # Objective function: minimize sum of squared shocks
44        def objective(shocks):
45            if method == 'minimize_shock':
46                # Minimize magnitude of shocks
47                return np.sum(shocks ** 2)
48            else:
49                # Minimize "unrealisticness" (prefer correlated shocks)
50                # This is a simplified version
51                return np.sum(np.abs(shocks))
52        
53        # Constraint: P&L must equal target
54        def pnl_constraint(shocks):
55            pnl = 0.0
56            for i, factor in enumerate(self.factors):
57                if factor in position_map:
58                    pnl += position_map[factor] * shocks[i]
59            return pnl - target_loss
60        
61        # Bounds: shocks between -max_shock and +max_shock
62        bounds = [(-self.max_shock, self.max_shock)] * len(self.factors)
63        
64        # Initial guess: uniform shock
65        x0 = np.zeros(len(self.factors))
66        
67        # Solve optimization
68        from scipy.optimize import minimize
69        
70        result = minimize(
71            objective,
72            x0,
73            method='SLSQP',
74            bounds=bounds,
75            constraints={'type': 'eq', 'fun': pnl_constraint}
76        )
77        
78        if not result.success:
79            # Try differential evolution (global optimizer)
80            def combined_objective(shocks):
81                pnl = 0.0
82                for i, factor in enumerate(self.factors):
83                    if factor in position_map:
84                        pnl += position_map[factor] * shocks[i]
85                
86                # Penalize deviation from target
87                pnl_penalty = (pnl - target_loss) ** 2 * 1e6
88                
89                # Minimize shock magnitude
90                shock_penalty = np.sum(shocks ** 2)
91                
92                return pnl_penalty + shock_penalty
93            
94            result = differential_evolution(
95                combined_objective,
96                bounds=bounds,
97                maxiter=1000
98            )
99        
100        # Extract shocks
101        shocks = result.x
102        
103        # Calculate actual P&L
104        actual_pnl = 0.0
105        for i, factor in enumerate(self.factors):
106            if factor in position_map:
107                actual_pnl += position_map[factor] * shocks[i]
108        
109        # Build result
110        scenario_shocks = {factor: shocks[i] for i, factor in enumerate(self.factors)}
111        
112        return {
113            'target_loss': target_loss,
114            'actual_pnl': actual_pnl,
115            'factor_shocks': scenario_shocks,
116            'max_shock': max(abs(shocks)),
117            'avg_shock': np.mean(np.abs(shocks)),
118            'num_factors_shocked': np.sum(np.abs(shocks) > 0.01)
119        }
120    
121    def find_multiple_scenarios(self,
122                               positions: pd.DataFrame,
123                               target_loss: float,
124                               n_scenarios: int = 5) -> List[Dict]:
125        """
126        Find multiple different scenarios that produce same loss
127        
128        Shows that many paths can lead to same outcome
129        """
130        scenarios = []
131        
132        for i in range(n_scenarios):
133            # Add randomness to find different solutions
134            np.random.seed(i * 42)
135            
136            result = self.find_breaking_scenario(
137                positions,
138                target_loss,
139                method='minimize_shock' if i % 2 == 0 else 'realistic'
140            )
141            
142            scenarios.append(result)
143        
144        return scenarios
145
146
147# Example usage
148if __name__ == "__main__":
149    # Portfolio
150    positions = pd.DataFrame({
151        'asset': ['SPY', 'TLT', 'GLD', 'HYG'],
152        'quantity': [1000, 500, 200, 300],
153        'current_price': [450.0, 95.0, 180.0, 75.0]
154    })
155    
156    portfolio_value = (positions['quantity'] * positions['current_price']).sum()
157    
158    # Reverse stress test: find scenario that loses 20% of portfolio
159    target_loss = -0.20 * portfolio_value
160    
161    reverse_tester = ReverseStressTester(
162        factors=['SPY', 'TLT', 'GLD', 'HYG'],
163        max_shock=0.50
164    )
165    
166    result = reverse_tester.find_breaking_scenario(positions, target_loss)
167    
168    print(f"Portfolio Value: ${portfolio_value:,.2f}")
169    print(f"Target Loss: ${target_loss:,.2f} ({target_loss/portfolio_value*100:.1f}%)")
170    print(f"\nBreaking Scenario:")
171    print("=" * 60)
172    
173    for factor, shock in result['factor_shocks'].items():
174        if abs(shock) > 0.01:
175            print(f"{factor}: {shock*100:+.2f}%")
176    
177    print(f"\nActual P&L: ${result['actual_pnl']:,.2f}")
178    print(f"Max single factor shock: {result['max_shock']*100:.2f}%")
179

1class CCARStressTester:
2    """
3    CCAR/DFAST stress testing framework
4    
5    Implements regulatory scenarios for bank stress testing
6    """
7    
8    def __init__(self):
9        self.scenarios = self._load_ccar_scenarios()
10    
11    def _load_ccar_scenarios(self) -> Dict:
12        """
13        Load CCAR scenarios (simplified version)
14        
15        In production, load from Federal Reserve published scenarios
16        """
17        return {
18            'baseline': {
19                'name': 'Baseline',
20                'description': 'Expected economic conditions',
21                'gdp_growth': 0.02,  # 2% annual GDP growth
22                'unemployment': 0.04,  # 4% unemployment
23                'equity_return': 0.08,  # 8% equity return
24                'house_price_change': 0.03,  # 3% house price growth
25                'rate_10y_change': 0.005,  # 50bps increase in 10Y rate
26                'credit_spread_change': 0.0,  # No change
27            },
28            'adverse': {
29                'name': 'Adverse',
30                'description': 'Moderate recession',
31                'gdp_growth': -0.02,  # -2% GDP
32                'unemployment': 0.08,  # 8% unemployment
33                'equity_return': -0.20,  # -20% equity
34                'house_price_change': -0.10,  # -10% house prices
35                'rate_10y_change': -0.01,  # -100bps (flight to quality)
36                'credit_spread_change': 0.02,  # +200bps credit spreads
37            },
38            'severely_adverse': {
39                'name': 'Severely Adverse',
40                'description': 'Severe recession (2008-like)',
41                'gdp_growth': -0.05,  # -5% GDP
42                'unemployment': 0.12,  # 12% unemployment
43                'equity_return': -0.40,  # -40% equity
44                'house_price_change': -0.25,  # -25% house prices
45                'rate_10y_change': -0.02,  # -200bps
46                'credit_spread_change': 0.04,  # +400bps credit spreads
47            }
48        }
49    
50    def map_macro_to_factors(self, 
51                            scenario_name: str) -> Dict[str, float]:
52        """
53        Map macroeconomic scenario to asset factor shocks
54        
55        Args:
56            scenario_name: 'baseline', 'adverse', or 'severely_adverse'
57            
58        Returns:
59            Dict mapping assets to expected returns
60        """
61        scenario = self.scenarios[scenario_name]
62        
63        # Map macro variables to asset returns
64        # This is a simplified model - in production, use econometric models
65        
66        factor_shocks = {}
67        
68        # Equities
69        factor_shocks['SPY'] = scenario['equity_return']
70        factor_shocks['QQQ'] = scenario['equity_return'] * 1.2  # Tech more volatile
71        factor_shocks['IWM'] = scenario['equity_return'] * 1.3  # Small cap more volatile
72        
73        # Bonds
74        # Duration ~17 for TLT, so -200bps = +34% price increase
75        factor_shocks['TLT'] = -scenario['rate_10y_change'] * 17
76        factor_shocks['IEF'] = -scenario['rate_10y_change'] * 7  # Duration ~7
77        
78        # Credit
79        # HYG has duration ~4 and spread duration ~4
80        rate_effect = -scenario['rate_10y_change'] * 4
81        spread_effect = -scenario['credit_spread_change'] * 4
82        factor_shocks['HYG'] = rate_effect + spread_effect
83        
84        # REITs (sensitive to house prices and rates)
85        factor_shocks['VNQ'] = scenario['house_price_change'] - scenario['rate_10y_change'] * 10
86        
87        # Gold (safe haven)
88        if scenario['equity_return'] < -0.10:
89            factor_shocks['GLD'] = 0.15  # Flight to safety
90        else:
91            factor_shocks['GLD'] = 0.02
92        
93        return factor_shocks
94    
95    def stress_test_portfolio(self,
96                             positions: pd.DataFrame) -> pd.DataFrame:
97        """
98        Run CCAR stress test on portfolio
99        
100        Args:
101            positions: Portfolio positions
102            
103        Returns:
104            DataFrame with results for all scenarios
105        """
106        portfolio_value = (positions['quantity'] * positions['current_price']).sum()
107        
108        results = []
109        
110        for scenario_name in ['baseline', 'adverse', 'severely_adverse']:
111            factor_shocks = self.map_macro_to_factors(scenario_name)
112            scenario = self.scenarios[scenario_name]
113            
114            # Calculate P&L
115            pnl = 0.0
116            
117            for _, row in positions.iterrows():
118                asset = row['asset']
119                quantity = row['quantity']
120                current_price = row['current_price']
121                
122                if asset in factor_shocks:
123                    shock = factor_shocks[asset]
124                    asset_pnl = quantity * current_price * shock
125                    pnl += asset_pnl
126            
127            results.append({
128                'scenario': scenario['name'],
129                'description': scenario['description'],
130                'gdp_growth': scenario['gdp_growth'] * 100,
131                'unemployment': scenario['unemployment'] * 100,
132                'equity_return': scenario['equity_return'] * 100,
133                'portfolio_pnl': pnl,
134                'pnl_pct': (pnl / portfolio_value * 100) if portfolio_value > 0 else 0,
135                'stressed_value': portfolio_value + pnl,
136                'capital_ratio_impact': (pnl / portfolio_value) * 100  # Simplified
137            })
138        
139        return pd.DataFrame(results)
140
141
142# Example usage
143if __name__ == "__main__":
144    # Bank portfolio
145    positions = pd.DataFrame({
146        'asset': ['SPY', 'TLT', 'HYG', 'VNQ', 'GLD'],
147        'quantity': [10000, 5000, 8000, 3000, 1000],
148        'current_price': [450.0, 95.0, 75.0, 85.0, 180.0]
149    })
150    
151    ccar_tester = CCARStressTester()
152    results = ccar_tester.stress_test_portfolio(positions)
153    
154    print("CCAR Stress Test Results:")
155    print("=" * 120)
156    print(results.to_string(index=False))
157