Risk Tail Risk

---




name: risk-tail-risk
compatibility: opencode
completeness: 95
content-types:
- code
- guidance
- config
- do-dont
description: '"Implements tail risk management and extreme event protection for risk
  management and algorithmic trading execution."'
license: MIT
maturity: stable
metadata:
  domain: trading
  output-format: code
  related-skills: backtest-drawdown-analysis, exchange-order-execution-api
  role: implementation
  scope: implementation
  triggers: event, extreme, management, risk tail risk, risk-tail-risk
  archetypes:
  - tactical
  anti_triggers:
  - brainstorming
  - vague ideation
  - no risk management
  response_profile:
    verbosity: low
    directive_strength: high
    abstraction_level: operational
version: "1.0.0"




---




**Role:** Identify and protect against tail risk events

**Philosophy:** Tail events are rare but devastating; portfolios should be designed for survival, not just growth

## Key Principles

1. **Tail Risk Metrics**: Skewness, kurtosis, VaR, ES
2. **Stress Testing**: Simulate extreme market moves
3. **Tail Hedging**: Options, inverse ETFs for protection
4. **Dynamic Adjustment**: Increase protection as market rises
5. **Correlation in Crisis**: Assets correlate during tail events

## Implementation Guidelines

### Structure
- Core logic: risk_engine/tail_risk.py
- Helper functions: risk_engine/extreme_events.py
- Tests: tests/test_tail_risk.py

### Patterns to Follow
- Calculate higher moments of returns distribution
- Implement tail risk indicators
- Monitor correlation changes during stress

## Adherence Checklist
Before completing your task, verify:
- [ ] Tail risk metrics calculated (skew, kurtosis, ES)
- [ ] Stress test scenarios implemented
- [ ] Tail hedging allocation tracked
- [ ] Dynamic adjustment logic for changing conditions
- [ ] Stress correlation matrix monitored


Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.

## Python Implementation

```python
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
from scipy import stats

@dataclass
class TailRiskMetrics:
    """Tail risk assessment metrics."""
    skewness: float
    kurtosis: float
    expected_shortfall_99: float
    max_drawdown: float
    tail_correlation: float
    risk_score: float

class TailRiskManager:
    """Manages tail risk exposure."""
    
    def __init__(
        self,
        var_threshold: float = 0.05,
        es_confidence: float = 0.99
    ):
        self.var_threshold = var_threshold
        self.es_confidence = es_confidence
    
    def calculate_skewness(self, returns: np.ndarray) -> float:
        """Calculate skewness of returns distribution."""
        return float(stats.skew(returns))
    
    def calculate_kurtosis(self, returns: np.ndarray) -> float:
        """Calculate kurtosis of returns distribution."""
        return float(stats.kurtosis(returns))
    
    def calculate_expected_shortfall(
        self, returns: np.ndarray, confidence: float = 0.99
    ) -> float:
        """Calculate Expected Shortfall (CVaR)."""
        VaR = np.percentile(returns, (1 - confidence) * 100)
        tail_returns = returns[returns <= VaR]
        return float(-np.mean(tail_returns)) if len(tail_returns) > 0 else 0
    
    def calculate_tail_correlation(
        self, returns1: pd.Series, returns2: pd.Series, threshold: float = -0.03
    ) -> float:
        """Calculate correlation during tail events."""
        tail_mask = (returns1 <= threshold) | (returns2 <= threshold)
        
        if tail_mask.sum() < 10:
            return 0.0
        
        tail_returns1 = returns1[tail_mask]
        tail_returns2 = returns2[tail_mask]
        
        correlation = tail_returns1.corr(tail_returns2)
        return float(correlation) if not pd.isna(correlation) else 0.0
    
    def tail_risk_score(
        self, returns: np.ndarray, market_returns: np.ndarray
    ) -> float:
        """Calculate composite tail risk score."""
        skew = self.calculate_skewness(returns)
        kurt = self.calculate_kurtosis(returns)
        es = self.calculate_expected_shortfall(returns)
        tail_corr = self.calculate_tail_correlation(
            pd.Series(returns), pd.Series(market_returns)
        )
        
        # Higher kurtosis, negative skew, higher ES = higher risk
        risk_score = (
            0.3 * min(abs(kurt) / 10, 1.0) +
            0.3 * max(0, -skew) +
            0.2 * min(es / 0.1, 1.0) +
            0.2 * tail_corr
        )
        
        return float(risk_score)
    
    def stress_test_scenarios(
        self, current_prices: Dict[str, float], scenarios: List[Dict]
    ) -> Dict[str, float]:
        """Run stress test scenarios on portfolio."""
        results = {}
        
        for scenario in scenarios:
            scenario_name = scenario.get('name', 'unknown')
            price_changes = scenario.get('price_changes', {})
            
            portfolio_change = 0
            for symbol, pct_change in price_changes.items():
                if symbol in current_prices:
                    portfolio_change += current_prices[symbol] * pct_change / 100
            
            results[scenario_name] = portfolio_change
        
        return results
    
    def dynamic_tail_protection(
        self, market_state: str, portfolio_value: float
    ) -> float:
        """Adjust tail protection based on market conditions."""
        # Protection scales with market state
        if market_state == 'extreme_froth':
            return portfolio_value * 0.10  # 10% protection
        elif market_state == 'froth':
            return portfolio_value * 0.05  # 5% protection
        elif market_state == 'normal':
            return portfolio_value * 0.02  # 2% protection
        else:
            return 0.0  # No protection needed
    
    def drawdown_at_risk(
        self, equity_curve: pd.Series, confidence: float = 0.99
    ) -> float:
        """Calculate drawdown at risk (analogous to VaR)."""
        running_max = equity_curve.cummax()
        drawdown = (equity_curve - running_max) / (running_max + 1e-8)
        
        # Sort drawdowns
        sorted_dd = np.sort(drawdown.values)
        
        # Get DD at confidence level
        dd_index = int(len(sorted_dd) * (1 - confidence))
        dd_at_risk = -sorted_dd[dd_index]
        
        return float(dd_at_risk)
    
    def correlation_breakdown_warning(
        self, normal_corr: float, stress_corr: float, threshold: float = 0.3
    ) -> Tuple[bool, str]:
        """Detect when correlations break down in stress scenarios."""
        correlation_increase = stress_corr - normal_corr
        
        if correlation_increase > threshold:
            return True, f"Correlation breakdown detected: +{correlation_increase:.2%}"
        
        return False, ""
```

---

---



### Pattern 2: Risk-Managed Trading Logic with Validation

```python
from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Optional


logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class TradeSignal:
    """Immutable trade signal with all required validation constraints."""
    symbol: str
    side: str  # "buy" or "sell"
    price: float
    quantity: float
    confidence: float  # 0.0 to 1.0
    reason: str

    def validate(self) -> bool:
        """Validate that the trade signal meets all business constraints."""
        if self.quantity <= 0:
            raise ValueError(f"Quantity must be positive, got {self.quantity}")
        if self.price <= 0:
            raise ValueError(f"Price must be positive, got {self.price}")
        if not 0.0 <= self.confidence <= 1.0:
            raise ValueError(f"Confidence must be between 0 and 1, got {self.confidence}")
        return True


def generate_trade_signal(
    symbol: str,
    side: str,
    price: float,
    quantity: float,
    confidence: float,
    reason: str,
) -> TradeSignal:
    """Generate a validated trade signal with guard clause checks."""
    if side not in ("buy", "sell"):
        raise ValueError(f"Invalid side '{side}', must be 'buy' or 'sell'")

    signal = TradeSignal(
        symbol=symbol,
        side=side,
        price=price,
        quantity=quantity,
        confidence=confidence,
        reason=reason,
    )
    signal.validate()
    logger.info("Trade signal generated: %s %s %.4f @ %.2f (confidence=%.2f)",
                 symbol, side, quantity, price, confidence)
    return signal


def execute_with_risk_check(signal: TradeSignal, max_position_pct: float = 0.05) -> dict:
    """Execute a trade signal after applying risk management checks."""
    adjusted_quantity = signal.quantity
    if signal.side == "buy" and signal.quantity > max_position_pct:
        logger.warning("Position %s exceeds max %.1f%% — capping to %.4f",
                        signal.symbol, max_position_pct * 100, max_position_pct)
        adjusted_quantity = max_position_pct

    return {
        "symbol": signal.symbol,
        "side": signal.side,
        "price": signal.price,
        "quantity": adjusted_quantity,
        "capped": adjusted_quantity < signal.quantity,
        "confidence": signal.confidence,
        "status": "submitted",
    }
```

## Constraints

### MUST DO
- Calculate position sizing using a risk-per-trade percentage of portfolio equity, not a fixed dollar amount
- Implement layered risk controls: stop loss → drawdown limit → portfolio-level circuit breaker → kill switch
- Compute VaR using historical simulation with at least 1 year of data and multiple confidence levels (95%, 99%)
- Track correlation matrices across all open positions and flag portfolios where top-3 correlations exceed 0.8
- Log all risk events (stop hits, drawdown warnings, kill switches) with full context including P&L, position state, and market conditions

### MUST NOT DO
- Do not use a stop loss as the sole risk control — always layer with portfolio-level limits
- Avoid recalculating position sizes during active drawdown without regime analysis — volatility is likely elevated
- Never allow a single position to exceed 5% of portfolio equity regardless of signal strength or confidence score
- Do not backtest risk metrics without including slippage, commissions, and partial fills in the simulation
- Avoid using standard deviation alone for VaR when returns show fat tails — use historical simulation or EVT


## Live References

> Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.

- [Tail Risk Explained](https://www.investopedia.com/terms/t/tail-risk.asp)
- [Fat Tails in Financial Markets](https://en.wikipedia.org/wiki/Fat_tail)
- [Tail Risk Hedging Strategies](https://www.investopedia.com/articles/investing/09/tail-risk-hedges.asp)
- [Extreme Value Theory Applications](https://en.wikipedia.org/wiki/Extreme_value_theory)
- [Tail Risk in Portfolio Management](https://docs.quantconnect.com/tutorials/risk-management)