Risk analytics and Decomposition
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
import statsmodels.api as sm
@dataclass
class FactorExposures:
market_beta: float
size: float
value: float
momentum: float
quality: float
volatility: float
class RiskAnalytics:
def __init__(self, returns: pd.DataFrame, factor_returns: pd.DataFrame = None, benchmark_returns: pd.Series = None):
"""
Initialize RiskAnalytics with historical returns data
Args:
returns: DataFrame of asset returns (assets in columns)
factor_returns: DataFrame of factor returns for risk decomposition
benchmark_returns: Series of benchmark returns for tracking error analysis
"""
self.returns = returns
self.factor_returns = factor_returns
self.benchmark_returns = benchmark_returns
self.cov_matrix = returns.cov()
def calculate_var(self, weights: Dict[str, float], confidence_level: float = 0.95, timeframe: int = 1) -> float:
"""
Calculate Value at Risk using historical simulation
Args:
weights: Dictionary of asset weights
confidence_level: VaR confidence level (default 95%)
timeframe: Time horizon in days
Returns:
Value at Risk estimate
"""
portfolio_returns = self._get_portfolio_returns(weights)
var = -np.percentile(portfolio_returns, (1 - confidence_level) * 100)
return var * np.sqrt(timeframe)
def calculate_cvar(self, weights: Dict[str, float], confidence_level: float = 0.95) -> float:
"""
Calculate Conditional Value at Risk (Expected Shortfall)
Args:
weights: Dictionary of asset weights
confidence_level: CVaR confidence level (default 95%)
Returns:
Conditional Value at Risk estimate
"""
portfolio_returns = self._get_portfolio_returns(weights)
var = -np.percentile(portfolio_returns, (1 - confidence_level) * 100)
return -portfolio_returns[portfolio_returns <= -var].mean()
def decompose_factor_risk(self, weights: Dict[str, float]) -> Dict[str, float]:
"""
Decompose portfolio risk into factor contributions
Args:
weights: Dictionary of asset weights
Returns:
Dictionary of factor risk contributions
"""
if self.factor_returns is None:
raise ValueError("Factor returns not provided")
# Calculate factor exposures through regression
portfolio_returns = self._get_portfolio_returns(weights)
factor_exposures = self._calculate_factor_exposures(portfolio_returns)
# Calculate factor contributions
factor_cov = self.factor_returns.cov()
factor_risk = np.sqrt(factor_exposures @ factor_cov @ factor_exposures)
factor_contrib = {}
for factor in self.factor_returns.columns:
factor_idx = self.factor_returns.columns.get_loc(factor)
contrib = factor_exposures[factor_idx] * (factor_cov.iloc[factor_idx] @ factor_exposures) / factor_risk
factor_contrib[factor] = contrib
return factor_contrib
def calculate_tracking_error(self, weights: Dict[str, float]) -> float:
"""
Calculate ex-post tracking error versus benchmark
Args:
weights: Dictionary of asset weights
Returns:
Annualized tracking error
"""
if self.benchmark_returns is None:
raise ValueError("Benchmark returns not provided")
portfolio_returns = self._get_portfolio_returns(weights)
active_returns = portfolio_returns - self.benchmark_returns
tracking_error = np.std(active_returns) * np.sqrt(252) # Annualized
return tracking_error
def stress_test(self, weights: Dict[str, float], scenarios: Dict[str, Dict[str, float]]) -> Dict[str, float]:
"""
Perform stress testing under different scenarios
Args:
weights: Dictionary of asset weights
scenarios: Dictionary of scenarios with asset return assumptions
Returns:
Dictionary of scenario returns
"""
results = {}
for scenario_name, scenario_returns in scenarios.items():
portfolio_return = sum(weights[asset] * scenario_returns[asset]
for asset in weights.keys())
results[scenario_name] = portfolio_return
return results
def risk_attribution(self, weights: Dict[str, float]) -> Dict[str, float]:
"""
Calculate risk attribution by asset
Args:
weights: Dictionary of asset weights
Returns:
Dictionary of risk contributions by asset
"""
weight_array = np.array([weights[col] for col in self.returns.columns])
portfolio_vol = np.sqrt(weight_array @ self.cov_matrix @ weight_array)
# Marginal contribution to risk
mctr = self.cov_matrix @ weight_array / portfolio_vol
# Component contribution to risk
cctr = {asset: weights[asset] * mctr[self.returns.columns.get_loc(asset)]
for asset in self.returns.columns}
return cctr
def _get_portfolio_returns(self, weights: Dict[str, float]) -> pd.Series:
"""Calculate historical portfolio returns given weights"""
return self.returns @ pd.Series(weights)
def _calculate_factor_exposures(self, portfolio_returns: pd.Series) -> np.ndarray:
"""Calculate factor exposures through regression"""
X = sm.add_constant(self.factor_returns)
factor_reg = sm.OLS(portfolio_returns, X).fit()
return factor_reg.params[1:] # Exclude constant
def demo_risk_analytics():
"""Demonstrate usage of RiskAnalytics class"""
# Sample data
np.random.seed(42)
dates = pd.date_range(start='2020-01-01', end='2023-12-31', freq='D')
assets = ['AAPL', 'MSFT', 'JPM', 'JNJ', 'PG']
# Generate random returns
returns = pd.DataFrame(
np.random.normal(0.0005, 0.02, (len(dates), len(assets))),
index=dates,
columns=assets
)
# Generate factor returns
factors = ['Market', 'Size', 'Value', 'Momentum']
factor_returns = pd.DataFrame(
np.random.normal(0.0003, 0.015, (len(dates), len(factors))),
index=dates,
columns=factors
)
# Sample portfolio
weights = {
'AAPL': 0.3,
'MSFT': 0.25,
'JPM': 0.2,
'JNJ': 0.15,
'PG': 0.1
}
# Initialize risk analytics
risk_analytics = RiskAnalytics(
returns=returns,
factor_returns=factor_returns,
benchmark_returns=returns['AAPL'] # Using AAPL as sample benchmark
)
# Calculate various risk metrics
print("\nRisk Analytics Results:")
print(f"VaR (95%): {risk_analytics.calculate_var(weights):.4f}")
print(f"CVaR (95%): {risk_analytics.calculate_cvar(weights):.4f}")
print(f"Tracking Error: {risk_analytics.calculate_tracking_error(weights):.4f}")
# Risk attribution
risk_contrib = risk_analytics.risk_attribution(weights)
print("\nRisk Attribution by Asset:")
for asset, contrib in risk_contrib.items():
print(f"{asset}: {contrib:.4f}")
# Factor risk decomposition
factor_risk = risk_analytics.decompose_factor_risk(weights)
print("\nFactor Risk Decomposition:")
for factor, risk in factor_risk.items():
print(f"{factor}: {risk:.4f}")
# Stress testing
scenarios = {
'Market_Crash': {'AAPL': -0.15, 'MSFT': -0.12, 'JPM': -0.18, 'JNJ': -0.08, 'PG': -0.05},
'Tech_Boom': {'AAPL': 0.10, 'MSFT': 0.12, 'JPM': 0.03, 'JNJ': 0.02, 'PG': 0.01},
'Recession': {'AAPL': -0.08, 'MSFT': -0.07, 'JPM': -0.12, 'JNJ': -0.04, 'PG': -0.03}
}
stress_results = risk_analytics.stress_test(weights, scenarios)
print("\nStress Test Results:")
for scenario, result in stress_results.items():
print(f"{scenario}: {result:.4f}")
if __name__ == "__main__":
demo_risk_analytics()
- Risk Metrics:
- VaR (95%): 1.50% daily loss threshold
- CVaR (95%): 1.90% expected loss beyond VaR
- Tracking Error: 25.42% annualized vs benchmark
- Risk Attribution:
- Shows how each asset contributes to total portfolio risk
- AAPL has highest contribution (0.38%), followed by MSFT (0.26%)
- Defensive stocks (JNJ, PG) have lower risk contributions
- Factor Risk Decomposition:
- Momentum has highest factor risk contribution (0.04%)
- Size factor shows small contribution (0.01%)
- Market and Value factors show minimal impact
- Stress Test Results:
- Market Crash scenario: -12.80% loss
- Tech Boom scenario: +7.00% gain
- Recession scenario: -7.45% loss
This module can be extended to include:
- Dynamic correlation analysis
- Liquidity risk metrics
- More sophisticated stress testing scenarios
- Custom factor models
- Real-time risk monitoring capabilities
Naixian Zhang 