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

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

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
• Fat Tails in Financial Markets
• Tail Risk Hedging Strategies
• Extreme Value Theory Applications
• Tail Risk in Portfolio Management