risk-correlation-risk

---

name: risk-correlation-risk compatibility: opencode completeness: 95 content-types:

• code
• guidance
• config
• do-dont description: '"Implements correlation breakdown and portfolio diversification risk for risk management and algorithmic trading execution."' license: MIT maturity: stable metadata: domain: trading output-format: code related-skills: ai-multi-asset-model, fundamentals-risk-management-basics, fundamentals-trading-edge, fundamentals-trading-plan role: implementation scope: implementation triggers: breakdown, diversification, portfolio, risk correlation risk, risk-correlation-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: Monitor and manage correlation risk in multi-asset portfolios

Philosophy: Correlations break down in crises; true diversification requires assets with stable low correlations

Key Principles

1. Correlation Stability: Correlations change over time and regime
2. Regime-Specific Correlations: Low volatility vs high volatility regimes
3. Diversification Benefit: Portfolio risk reduction from low correlations
4. Breakdown Detection: Alerts when correlations approach 1.0
5. Effective Correlation: Actual portfolio correlation vs nominal

Implementation Guidelines

Structure

• Core logic: risk_engine/correlation.py
• Helper functions: risk_engine/risk_decomposition.py
• Tests: tests/test_correlation.py

Patterns to Follow

• Track rolling correlations with multiple windows
• Identify correlation clusters
• Calculate effective correlation for risk

Adherence Checklist

Before completing your task, verify:

• Rolling correlations calculated for multiple windows
• Correlation breakdown alerts triggered
• Regime-specific correlations tracked
• Diversification benefit quantified
• Effective correlation used for portfolio VaR

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 CorrelationRegime:
    """Current correlation regime."""
    regime: str  # 'low', 'normal', 'high', 'breakdown'
    avg_correlation: float
    correlation_vol: float
    detected_breakdown: bool

class CorrelationRiskManager:
    """Manages correlation risk in multi-asset portfolios."""
    
    def __init__(self, windows: List[int] = [20, 50, 100]):
        self.windows = windows
    
    def calculate_rolling_correlation(
        self, returns1: pd.Series, returns2: pd.Series, window: int = 20
    ) -> pd.Series:
        """Calculate rolling correlation between two assets."""
        return returns1.rolling(window=window).corr(returns2)
    
    def calculate_correlation_matrix(
        self, returns_df: pd.DataFrame, window: int = 50
    ) -> pd.DataFrame:
        """Calculate rolling correlation matrix."""
        return returns_df.rolling(window=window).corr()
    
    def detect_correlation_breakdown(
        self, correlation_series: pd.Series, threshold: float = 0.8
    ) -> Dict:
        """Detect when correlations approach breakdown levels."""
        current_corr = correlation_series.iloc[-1] if len(correlation_series) > 0 else 0
        
        # Check for breakdown
        breakdown_detected = current_corr > threshold
        
        # Calculate time since last significant change
        diff = correlation_series.diff()
        change_points = (diff.abs() > 0.1).sum()
        
        return {
            'current_correlation': float(current_corr),
            'breakdown_detected': breakdown_detected,
            'threshold': threshold,
            'change_points': int(change_points),
            'warning': 'High correlation detected' if breakdown_detected else ''
        }
    
    def calculate_effective_correlation(
        self, weights: np.ndarray, correlation_matrix: np.ndarray
    ) -> float:
        """Calculate effective correlation of portfolio."""
        if len(weights) != correlation_matrix.shape[0]:
            return 0.0
        
        portfolio_var = weights @ correlation_matrix @ weights
        weighted_var = np.sum(weights ** 2)
        
        if weighted_var == 0:
            return 0.0
        
        # Effective correlation
        eff_corr = (portfolio_var - weighted_var) / (2 * np.sum(
            np.outer(weights, weights) * (1 - np.eye(len(weights)))
        ) + 1e-8)
        
        return float(eff_corr)
    
    def calculate_diversification_benefit(
        self, individual_vols: np.ndarray, portfolio_vol: float
    ) -> float:
        """Calculate diversification benefit from correlations."""
        if portfolio_vol <= 0:
            return 1.0
        
        # Unweighted average volatility
        avg_vol = np.mean(individual_vols)
        
        if avg_vol <= 0:
            return 1.0
        
        # Diversification benefit = avg vol / portfolio vol
        return avg_vol / portfolio_vol
    
    def regime_correlation_analysis(
        self, returns_df: pd.DataFrame, volatility_series: pd.Series
    ) -> Dict:
        """Analyze correlations in different volatility regimes."""
        # Split by volatility regime
        vol_mean = volatility_series.mean()
        vol_std = volatility_series.std()
        
        low_vol_mask = volatility_series < vol_mean - vol_std
        high_vol_mask = volatility_series > vol_mean + vol_std
        
        low_vol_corr = returns_df[low_vol_mask].corr()
        high_vol_corr = returns_df[high_vol_mask].corr()
        
        # Compare
        diff = high_vol_corr - low_vol_corr
        avg_diff = diff.values[np.triu_indices_from(diff.values, k=1)].mean()
        
        return {
            'low_vol_correlation': float(low_vol_corr.values[np.triu_indices_from(low_vol_corr.values, k=1)].mean()),
            'high_vol_correlation': float(high_vol_corr.values[np.triu_indices_from(high_vol_corr.values, k=1)].mean()),
            'correlation_change': float(avg_diff),
            'breakdown_warning': avg_diff > 0.3  # Large increase indicates breakdown
        }
    
    def cluster_correlations(
        self, correlation_matrix: pd.DataFrame, n_clusters: int = 3
    ) -> Dict:
        """Cluster assets by correlation similarity."""
        # Convert to distance matrix
        dist_matrix = 1 - correlation_matrix.values
        np.fill_diagonal(dist_matrix, 0)
        
        # Simple clustering based on correlation similarity
        assets = correlation_matrix.columns.tolist()
        
        # Sort assets by correlation to first asset
        sorted_assets = [assets[0]]
        remaining = assets[1:]
        
        while remaining:
            last = sorted_assets[-1]
            last_idx = assets.index(last)
            
            # Find most correlated to last added
            correlations = correlation_matrix.iloc[last_idx, remaining]
            next_asset = correlations.idxmax()
            sorted_assets.append(next_asset)
            remaining.remove(next_asset)
        
        # Divide into clusters
        cluster_size = len(sorted_assets) // n_clusters
        clusters = {}
        
        for i in range(n_clusters):
            start = i * cluster_size
            end = start + cluster_size if i < n_clusters - 1 else len(sorted_assets)
            clusters[f'cluster_{i}'] = sorted_assets[start:end]
        
        return clusters
    
    def correlation_shock_scenarios(
        self, correlation_matrix: pd.DataFrame, shock_amount: float = 0.3
    ) -> Dict[str, float]:
        """Simulate correlation shock scenarios."""
        # Increase all correlations by shock amount
        new_corr = correlation_matrix + shock_amount
        new_corr = new_corr.clip(upper=1.0)
        
        # Calculate new portfolio risk metrics
        return {
            'shock_amount': shock_amount,
            'avg_correlation_increase': float(shock_amount),
            'max_correlation': float(new_corr.max().max())
        }

---

---

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.

• Correlation in Finance Explained
• Portfolio Correlation Analysis
• Correlation Risk Management
• Dynamic Correlation in Crisis Periods
• Diversification and Correlation Benefits