PaperclipOrg
Claude Agent Skill · by Wshobson

Risk Metrics Calculation

Calculates the full suite of portfolio risk metrics you actually need for trading and risk management. Handles VaR (historical, parametric, and Cornish-Fisher),

Install
Terminal · npx
$npx skills add https://github.com/wshobson/agents --skill risk-metrics-calculation
Works with Paperclip

How Risk Metrics Calculation fits into a Paperclip company.

Risk Metrics Calculation drops into any Paperclip agent that handles this kind of work. Assign it to a specialist inside a pre-configured PaperclipOrg company and the skill becomes available on every heartbeat — no prompt engineering, no tool wiring.

S
SaaS FactoryPaired

Pre-configured AI company — 18 agents, 18 skills, one-time purchase.

$27$59
Explore pack
Source file
SKILL.md551 lines
Expand
---name: risk-metrics-calculationdescription: 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.--- # 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 tradersDaily:      Standard risk reportingWeekly:     Rebalancing decisionsMonthly:    Performance attributionAnnual:     Strategic allocation``` ## Implementation ### Pattern 1: Core Risk Metrics ```pythonimport numpy as npimport pandas as pdfrom scipy import statsfrom 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 ```pythonclass 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 ```pythonclass 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 ```pythonclass 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 usagemetrics = 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 summarysummary = 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