# risk-metrics-calculation — detailed patterns and worked examples ## Implementation ### Pattern 1: Core Risk Metrics ```python import numpy as np import pandas as pd from scipy import stats from typing import Dict, Optional, Tuple class RiskMetrics: """Core risk metric calculations.""" def __init__(self, returns: pd.Series, rf_rate: float = 0.02): """ Args: returns: Series of periodic returns rf_rate: Annual risk-free rate """ self.returns = returns self.rf_rate = rf_rate self.ann_factor = 252 # Trading days per year # Volatility Metrics def volatility(self, annualized: bool = True) -> float: """Standard deviation of returns.""" vol = self.returns.std() if annualized: vol *= np.sqrt(self.ann_factor) return vol def downside_deviation(self, threshold: float = 0, annualized: bool = True) -> float: """Standard deviation of returns below threshold.""" downside = self.returns[self.returns < threshold] if len(downside) == 0: return 0.0 dd = downside.std() if annualized: dd *= np.sqrt(self.ann_factor) return dd def beta(self, market_returns: pd.Series) -> float: """Beta relative to market.""" aligned = pd.concat([self.returns, market_returns], axis=1).dropna() if len(aligned) < 2: return np.nan cov = np.cov(aligned.iloc[:, 0], aligned.iloc[:, 1]) return cov[0, 1] / cov[1, 1] if cov[1, 1] != 0 else 0 # Value at Risk def var_historical(self, confidence: float = 0.95) -> float: """Historical VaR at confidence level.""" return -np.percentile(self.returns, (1 - confidence) * 100) def var_parametric(self, confidence: float = 0.95) -> float: """Parametric VaR assuming normal distribution.""" z_score = stats.norm.ppf(confidence) return self.returns.mean() - z_score * self.returns.std() def var_cornish_fisher(self, confidence: float = 0.95) -> float: """VaR with Cornish-Fisher expansion for non-normality.""" z = stats.norm.ppf(confidence) s = stats.skew(self.returns) # Skewness k = stats.kurtosis(self.returns) # Excess kurtosis # Cornish-Fisher expansion z_cf = (z + (z**2 - 1) * s / 6 + (z**3 - 3*z) * k / 24 - (2*z**3 - 5*z) * s**2 / 36) return -(self.returns.mean() + z_cf * self.returns.std()) # Conditional VaR (Expected Shortfall) def cvar(self, confidence: float = 0.95) -> float: """Expected Shortfall / CVaR / Average VaR.""" var = self.var_historical(confidence) return -self.returns[self.returns <= -var].mean() # Drawdown Analysis def drawdowns(self) -> pd.Series: """Calculate drawdown series.""" cumulative = (1 + self.returns).cumprod() running_max = cumulative.cummax() return (cumulative - running_max) / running_max def max_drawdown(self) -> float: """Maximum drawdown.""" return self.drawdowns().min() def avg_drawdown(self) -> float: """Average drawdown.""" dd = self.drawdowns() return dd[dd < 0].mean() if (dd < 0).any() else 0 def drawdown_duration(self) -> Dict[str, int]: """Drawdown duration statistics.""" dd = self.drawdowns() in_drawdown = dd < 0 # Find drawdown periods drawdown_starts = in_drawdown & ~in_drawdown.shift(1).fillna(False) drawdown_ends = ~in_drawdown & in_drawdown.shift(1).fillna(False) durations = [] current_duration = 0 for i in range(len(dd)): if in_drawdown.iloc[i]: current_duration += 1 elif current_duration > 0: durations.append(current_duration) current_duration = 0 if current_duration > 0: durations.append(current_duration) return { "max_duration": max(durations) if durations else 0, "avg_duration": np.mean(durations) if durations else 0, "current_duration": current_duration } # Risk-Adjusted Returns def sharpe_ratio(self) -> float: """Annualized Sharpe ratio.""" excess_return = self.returns.mean() * self.ann_factor - self.rf_rate vol = self.volatility(annualized=True) return excess_return / vol if vol > 0 else 0 def sortino_ratio(self) -> float: """Sortino ratio using downside deviation.""" excess_return = self.returns.mean() * self.ann_factor - self.rf_rate dd = self.downside_deviation(threshold=0, annualized=True) return excess_return / dd if dd > 0 else 0 def calmar_ratio(self) -> float: """Calmar ratio (return / max drawdown).""" annual_return = (1 + self.returns).prod() ** (self.ann_factor / len(self.returns)) - 1 max_dd = abs(self.max_drawdown()) return annual_return / max_dd if max_dd > 0 else 0 def omega_ratio(self, threshold: float = 0) -> float: """Omega ratio.""" returns_above = self.returns[self.returns > threshold] - threshold returns_below = threshold - self.returns[self.returns <= threshold] if returns_below.sum() == 0: return np.inf return returns_above.sum() / returns_below.sum() # Information Ratio def information_ratio(self, benchmark_returns: pd.Series) -> float: """Information ratio vs benchmark.""" active_returns = self.returns - benchmark_returns tracking_error = active_returns.std() * np.sqrt(self.ann_factor) active_return = active_returns.mean() * self.ann_factor return active_return / tracking_error if tracking_error > 0 else 0 # Summary def summary(self) -> Dict[str, float]: """Generate comprehensive risk summary.""" dd_stats = self.drawdown_duration() return { # Returns "total_return": (1 + self.returns).prod() - 1, "annual_return": (1 + self.returns).prod() ** (self.ann_factor / len(self.returns)) - 1, # Volatility "annual_volatility": self.volatility(), "downside_deviation": self.downside_deviation(), # VaR & CVaR "var_95_historical": self.var_historical(0.95), "var_99_historical": self.var_historical(0.99), "cvar_95": self.cvar(0.95), # Drawdowns "max_drawdown": self.max_drawdown(), "avg_drawdown": self.avg_drawdown(), "max_drawdown_duration": dd_stats["max_duration"], # Risk-Adjusted "sharpe_ratio": self.sharpe_ratio(), "sortino_ratio": self.sortino_ratio(), "calmar_ratio": self.calmar_ratio(), "omega_ratio": self.omega_ratio(), # Distribution "skewness": stats.skew(self.returns), "kurtosis": stats.kurtosis(self.returns), } ``` ### Pattern 2: Portfolio Risk ```python class PortfolioRisk: """Portfolio-level risk calculations.""" def __init__( self, returns: pd.DataFrame, weights: Optional[pd.Series] = None ): """ Args: returns: DataFrame with asset returns (columns = assets) weights: Portfolio weights (default: equal weight) """ self.returns = returns self.weights = weights if weights is not None else \ pd.Series(1/len(returns.columns), index=returns.columns) self.ann_factor = 252 def portfolio_return(self) -> float: """Weighted portfolio return.""" return (self.returns @ self.weights).mean() * self.ann_factor def portfolio_volatility(self) -> float: """Portfolio volatility.""" cov_matrix = self.returns.cov() * self.ann_factor port_var = self.weights @ cov_matrix @ self.weights return np.sqrt(port_var) def marginal_risk_contribution(self) -> pd.Series: """Marginal contribution to risk by asset.""" cov_matrix = self.returns.cov() * self.ann_factor port_vol = self.portfolio_volatility() # Marginal contribution mrc = (cov_matrix @ self.weights) / port_vol return mrc def component_risk(self) -> pd.Series: """Component contribution to total risk.""" mrc = self.marginal_risk_contribution() return self.weights * mrc def risk_parity_weights(self, target_vol: float = None) -> pd.Series: """Calculate risk parity weights.""" from scipy.optimize import minimize n = len(self.returns.columns) cov_matrix = self.returns.cov() * self.ann_factor def risk_budget_objective(weights): port_vol = np.sqrt(weights @ cov_matrix @ weights) mrc = (cov_matrix @ weights) / port_vol rc = weights * mrc target_rc = port_vol / n # Equal risk contribution return np.sum((rc - target_rc) ** 2) constraints = [ {"type": "eq", "fun": lambda w: np.sum(w) - 1}, # Weights sum to 1 ] bounds = [(0.01, 1.0) for _ in range(n)] # Min 1%, max 100% x0 = np.array([1/n] * n) result = minimize( risk_budget_objective, x0, method="SLSQP", bounds=bounds, constraints=constraints ) return pd.Series(result.x, index=self.returns.columns) def correlation_matrix(self) -> pd.DataFrame: """Asset correlation matrix.""" return self.returns.corr() def diversification_ratio(self) -> float: """Diversification ratio (higher = more diversified).""" asset_vols = self.returns.std() * np.sqrt(self.ann_factor) weighted_vol = (self.weights * asset_vols).sum() port_vol = self.portfolio_volatility() return weighted_vol / port_vol if port_vol > 0 else 1 def tracking_error(self, benchmark_returns: pd.Series) -> float: """Tracking error vs benchmark.""" port_returns = self.returns @ self.weights active_returns = port_returns - benchmark_returns return active_returns.std() * np.sqrt(self.ann_factor) def conditional_correlation( self, threshold_percentile: float = 10 ) -> pd.DataFrame: """Correlation during stress periods.""" port_returns = self.returns @ self.weights threshold = np.percentile(port_returns, threshold_percentile) stress_mask = port_returns <= threshold return self.returns[stress_mask].corr() ``` ### Pattern 3: Rolling Risk Metrics ```python class RollingRiskMetrics: """Rolling window risk calculations.""" def __init__(self, returns: pd.Series, window: int = 63): """ Args: returns: Return series window: Rolling window size (default: 63 = ~3 months) """ self.returns = returns self.window = window def rolling_volatility(self, annualized: bool = True) -> pd.Series: """Rolling volatility.""" vol = self.returns.rolling(self.window).std() if annualized: vol *= np.sqrt(252) return vol def rolling_sharpe(self, rf_rate: float = 0.02) -> pd.Series: """Rolling Sharpe ratio.""" rolling_return = self.returns.rolling(self.window).mean() * 252 rolling_vol = self.rolling_volatility() return (rolling_return - rf_rate) / rolling_vol def rolling_var(self, confidence: float = 0.95) -> pd.Series: """Rolling historical VaR.""" return self.returns.rolling(self.window).apply( lambda x: -np.percentile(x, (1 - confidence) * 100), raw=True ) def rolling_max_drawdown(self) -> pd.Series: """Rolling maximum drawdown.""" def max_dd(returns): cumulative = (1 + returns).cumprod() running_max = cumulative.cummax() drawdowns = (cumulative - running_max) / running_max return drawdowns.min() return self.returns.rolling(self.window).apply(max_dd, raw=False) def rolling_beta(self, market_returns: pd.Series) -> pd.Series: """Rolling beta vs market.""" def calc_beta(window_data): port_ret = window_data.iloc[:, 0] mkt_ret = window_data.iloc[:, 1] cov = np.cov(port_ret, mkt_ret) return cov[0, 1] / cov[1, 1] if cov[1, 1] != 0 else 0 combined = pd.concat([self.returns, market_returns], axis=1) return combined.rolling(self.window).apply( lambda x: calc_beta(x.to_frame()), raw=False ).iloc[:, 0] def volatility_regime( self, low_threshold: float = 0.10, high_threshold: float = 0.20 ) -> pd.Series: """Classify volatility regime.""" vol = self.rolling_volatility() def classify(v): if v < low_threshold: return "low" elif v > high_threshold: return "high" else: return "normal" return vol.apply(classify) ``` ### Pattern 4: Stress Testing ```python class StressTester: """Historical and hypothetical stress testing.""" # Historical crisis periods HISTORICAL_SCENARIOS = { "2008_financial_crisis": ("2008-09-01", "2009-03-31"), "2020_covid_crash": ("2020-02-19", "2020-03-23"), "2022_rate_hikes": ("2022-01-01", "2022-10-31"), "dot_com_bust": ("2000-03-01", "2002-10-01"), "flash_crash_2010": ("2010-05-06", "2010-05-06"), } def __init__(self, returns: pd.Series, weights: pd.Series = None): self.returns = returns self.weights = weights def historical_stress_test( self, scenario_name: str, historical_data: pd.DataFrame ) -> Dict[str, float]: """Test portfolio against historical crisis period.""" if scenario_name not in self.HISTORICAL_SCENARIOS: raise ValueError(f"Unknown scenario: {scenario_name}") start, end = self.HISTORICAL_SCENARIOS[scenario_name] # Get returns during crisis crisis_returns = historical_data.loc[start:end] if self.weights is not None: port_returns = (crisis_returns @ self.weights) else: port_returns = crisis_returns total_return = (1 + port_returns).prod() - 1 max_dd = self._calculate_max_dd(port_returns) worst_day = port_returns.min() return { "scenario": scenario_name, "period": f"{start} to {end}", "total_return": total_return, "max_drawdown": max_dd, "worst_day": worst_day, "volatility": port_returns.std() * np.sqrt(252) } def hypothetical_stress_test( self, shocks: Dict[str, float] ) -> float: """ Test portfolio against hypothetical shocks. Args: shocks: Dict of {asset: shock_return} """ if self.weights is None: raise ValueError("Weights required for hypothetical stress test") total_impact = 0 for asset, shock in shocks.items(): if asset in self.weights.index: total_impact += self.weights[asset] * shock return total_impact def monte_carlo_stress( self, n_simulations: int = 10000, horizon_days: int = 21, vol_multiplier: float = 2.0 ) -> Dict[str, float]: """Monte Carlo stress test with elevated volatility.""" mean = self.returns.mean() vol = self.returns.std() * vol_multiplier simulations = np.random.normal( mean, vol, (n_simulations, horizon_days) ) total_returns = (1 + simulations).prod(axis=1) - 1 return { "expected_loss": -total_returns.mean(), "var_95": -np.percentile(total_returns, 5), "var_99": -np.percentile(total_returns, 1), "worst_case": -total_returns.min(), "prob_10pct_loss": (total_returns < -0.10).mean() } def _calculate_max_dd(self, returns: pd.Series) -> float: cumulative = (1 + returns).cumprod() running_max = cumulative.cummax() drawdowns = (cumulative - running_max) / running_max return drawdowns.min() ``` ## Quick Reference ```python # Daily usage metrics = RiskMetrics(returns) print(f"Sharpe: {metrics.sharpe_ratio():.2f}") print(f"Max DD: {metrics.max_drawdown():.2%}") print(f"VaR 95%: {metrics.var_historical(0.95):.2%}") # Full summary summary = metrics.summary() for metric, value in summary.items(): print(f"{metric}: {value:.4f}") ```

quantitative-trading

Install

Contentsdiscovered from source

agents ×6