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

```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 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](https://www.investopedia.com/terms/c/correlation.asp)
- [Portfolio Correlation Analysis](https://en.wikipedia.org/wiki/Coefficient_of_correlation)
- [Correlation Risk Management](https://www.investopedia.com/terms/c/correlation-risk.asp)
- [Dynamic Correlation in Crisis Periods](https://papers.ssrn.com/sol3/papers.cfm?abstract_id=1495603)
- [Diversification and Correlation Benefits](https://www.investopedia.com/terms/d/diversification.asp)