risk-metrics-calculation

name	risk-metrics-calculation
description	Calculate portfolio risk metrics including VaR, CVaR, Sharpe, Sortino, and drawdown analysis. Use when measuring portfolio risk, implementing risk limits, or building risk monitoring systems.
version	1.0.0
cluster	databases-data

Risk Metrics Calculation

Comprehensive risk measurement toolkit for portfolio management, including Value at Risk, Expected Shortfall, and drawdown analysis.

When to Use This Skill

• Measuring portfolio risk
• Implementing risk limits
• Building risk dashboards
• Calculating risk-adjusted returns
• Setting position sizes
• Regulatory reporting

Core Concepts

1. Risk Metric Categories

Category	Metrics	Use Case
Volatility	Std Dev, Beta	General risk
Tail Risk	VaR, CVaR	Extreme losses
Drawdown	Max DD, Calmar	Capital preservation
Risk-Adjusted	Sharpe, Sortino	Performance

2. Time Horizons

Intraday:   Minute/hourly VaR for day traders
Daily:      Standard risk reporting
Weekly:     Rebalancing decisions
Monthly:    Performance attribution
Annual:     Strategic allocation

Implementation

Pattern 1: Core Risk Metrics

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 = .returns[.returns < threshold]
         (downside) == :
             
        dd = downside.std()
         annualized:
            dd *= np.sqrt(.ann_factor)
         dd

     () -> :
        
        aligned = pd.concat([.returns, market_returns], axis=).dropna()
         (aligned) < :
             np.nan
        cov = np.cov(aligned.iloc[:, ], aligned.iloc[:, ])
         cov[, ] / cov[, ]  cov[, ] !=   

    
     () -> :
        
         -np.percentile(.returns, ( - confidence) * )

     () -> :
        
        z_score = stats.norm.ppf(confidence)
         .returns.mean() - z_score * .returns.std()

     () -> :
        
        z = stats.norm.ppf(confidence)
        s = stats.skew(.returns)  
        k = stats.kurtosis(.returns)  

        
        z_cf = (z + (z** - ) * s /  +
                (z** - *z) * k /  -
                (*z** - *z) * s** / )

         -(.returns.mean() + z_cf * .returns.std())

    
     () -> :
        
        var = .var_historical(confidence)
         -.returns[.returns <= -var].mean()

    
     () -> pd.Series:
        
        cumulative = ( + .returns).cumprod()
        running_max = cumulative.cummax()
         (cumulative - running_max) / running_max

     () -> :
        
         .drawdowns().()

     () -> :
        
        dd = .drawdowns()
         dd[dd < ].mean()  (dd < ).()  

     () -> [, ]:
        
        dd = .drawdowns()
        in_drawdown = dd < 

        
        drawdown_starts = in_drawdown & ~in_drawdown.shift().fillna()
        drawdown_ends = ~in_drawdown & in_drawdown.shift().fillna()

        durations = []
        current_duration = 

         i  ((dd)):
             in_drawdown.iloc[i]:
                current_duration += 
             current_duration > :
                durations.append(current_duration)
                current_duration = 

         current_duration > :
            durations.append(current_duration)

         {
            : (durations)  durations  ,
            : np.mean(durations)  durations  ,
            : current_duration
        }

    
     () -> :
        
        excess_return = .returns.mean() * .ann_factor - .rf_rate
        vol = .volatility(annualized=)
         excess_return / vol  vol >   

     () -> :
        
        excess_return = .returns.mean() * .ann_factor - .rf_rate
        dd = .downside_deviation(threshold=, annualized=)
         excess_return / dd  dd >   

     () -> :
        
        annual_return = ( + .returns).prod() ** (.ann_factor / (.returns)) - 
        max_dd = (.max_drawdown())
         annual_return / max_dd  max_dd >   

     () -> :
        
        returns_above = .returns[.returns > threshold] - threshold
        returns_below = threshold - .returns[.returns <= threshold]

         returns_below.() == :
             np.inf

         returns_above.() / returns_below.()

    
     () -> :
        
        active_returns = .returns - benchmark_returns
        tracking_error = active_returns.std() * np.sqrt(.ann_factor)
        active_return = active_returns.mean() * .ann_factor
         active_return / tracking_error  tracking_error >   

    
     () -> [, ]:
        
        dd_stats = .drawdown_duration()

         {
            
            : ( + .returns).prod() - ,
            : ( + .returns).prod() ** (.ann_factor / (.returns)) - ,

            
            : .volatility(),
            : .downside_deviation(),

            
            : .var_historical(),
            : .var_historical(),
            : .cvar(),

            
            : .max_drawdown(),
            : .avg_drawdown(),
            : dd_stats[],

            
            : .sharpe_ratio(),
            : .sortino_ratio(),
            : .calmar_ratio(),
            : .omega_ratio(),

            
            : stats.skew(.returns),
            : stats.kurtosis(.returns),
        }

Pattern 2: Portfolio Risk

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 = .returns.cov() * .ann_factor
        port_vol = .portfolio_volatility()

        
        mrc = (cov_matrix @ .weights) / port_vol
         mrc

     () -> pd.Series:
        
        mrc = .marginal_risk_contribution()
         .weights * mrc

     () -> pd.Series:
        
         scipy.optimize  minimize

        n = (.returns.columns)
        cov_matrix = .returns.cov() * .ann_factor

         ():
            port_vol = np.sqrt(weights @ cov_matrix @ weights)
            mrc = (cov_matrix @ weights) / port_vol
            rc = weights * mrc
            target_rc = port_vol / n  
             np.((rc - target_rc) ** )

        constraints = [
            {: , :  w: np.(w) - },  
        ]
        bounds = [(, )  _  (n)]  
        x0 = np.array([/n] * n)

        result = minimize(
            risk_budget_objective,
            x0,
            method=,
            bounds=bounds,
            constraints=constraints
        )

         pd.Series(result.x, index=.returns.columns)

     () -> pd.DataFrame:
        
         .returns.corr()

     () -> :
        
        asset_vols = .returns.std() * np.sqrt(.ann_factor)
        weighted_vol = (.weights * asset_vols).()
        port_vol = .portfolio_volatility()
         weighted_vol / port_vol  port_vol >   

     () -> :
        
        port_returns = .returns @ .weights
        active_returns = port_returns - benchmark_returns
         active_returns.std() * np.sqrt(.ann_factor)

     () -> pd.DataFrame:
        
        port_returns = .returns @ .weights
        threshold = np.percentile(port_returns, threshold_percentile)
        stress_mask = port_returns <= threshold
         .returns[stress_mask].corr()

Pattern 3: Rolling Risk Metrics

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, ( - confidence) * ),
            raw=
        )

     () -> pd.Series:
        
         ():
            cumulative = ( + returns).cumprod()
            running_max = cumulative.cummax()
            drawdowns = (cumulative - running_max) / running_max
             drawdowns.()

         .returns.rolling(.window).apply(max_dd, raw=)

     () -> pd.Series:
        
         ():
            port_ret = window_data.iloc[:, ]
            mkt_ret = window_data.iloc[:, ]
            cov = np.cov(port_ret, mkt_ret)
             cov[, ] / cov[, ]  cov[, ] !=   

        combined = pd.concat([.returns, market_returns], axis=)
         combined.rolling(.window).apply(
             x: calc_beta(x.to_frame()),
            raw=
        ).iloc[:, ]

     () -> pd.Series:
        
        vol = .rolling_volatility()

         ():
             v < low_threshold:
                 
             v > high_threshold:
                 
            :
                 

         vol.apply(classify)

Pattern 4: Stress Testing

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 .weights   :
            port_returns = (crisis_returns @ .weights)
        :
            port_returns = crisis_returns

        total_return = ( + port_returns).prod() - 
        max_dd = ._calculate_max_dd(port_returns)
        worst_day = port_returns.()

         {
            : scenario_name,
            : ,
            : total_return,
            : max_dd,
            : worst_day,
            : port_returns.std() * np.sqrt()
        }

     () -> :
        
         .weights  :
             ValueError()

        total_impact = 
         asset, shock  shocks.items():
             asset  .weights.index:
                total_impact += .weights[asset] * shock

         total_impact

     () -> [, ]:
        
        mean = .returns.mean()
        vol = .returns.std() * vol_multiplier

        simulations = np.random.normal(
            mean,
            vol,
            (n_simulations, horizon_days)
        )

        total_returns = ( + simulations).prod(axis=) - 

         {
            : -total_returns.mean(),
            : -np.percentile(total_returns, ),
            : -np.percentile(total_returns, ),
            : -total_returns.(),
            : (total_returns < -).mean()
        }

     () -> :
        cumulative = ( + returns).cumprod()
        running_max = cumulative.cummax()
        drawdowns = (cumulative - running_max) / running_max
         drawdowns.()

Quick Reference

# 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}")

Best Practices

Do's

• Use multiple metrics - No single metric captures all risk
• Consider tail risk - VaR isn't enough, use CVaR
• Rolling analysis - Risk changes over time
• Stress test - Historical and hypothetical
• Document assumptions - Distribution, lookback, etc.

Don'ts

• Don't rely on VaR alone - Underestimates tail risk
• Don't assume normality - Returns are fat-tailed
• Don't ignore correlation - Increases in stress
• Don't use short lookbacks - Miss regime changes
• Don't forget transaction costs - Affects realized risk

Resources

• Risk Management and Financial Institutions (John Hull)
• Quantitative Risk Management (McNeil, Frey, Embrechts)
• pyfolio Documentation