ρ(t) = ρ_normal if State = Normal
ρ(t) = ρ_crisis if State = Crisis
Transition Probability:
P(Crisis_t+1 | Normal_t) = λ_normal
P(Normal_t+1 | Crisis_t) = λ_crisis

Tail_ρ = P(R_X < q | R_Y < q) / (q²)

import pandas as pd
import numpy as np
from scipy import stats
import yfinance as yf
import matplotlib.pyplot as plt
from sklearn.covariance import EmpiricalCovariance, MinCovarianceDeterminant
class CorrelationBreakdownDetector:
    def __init__(self, returns_df):
        """
        returns_df: DataFrame of returns for multiple assets
        """
        self.returns = returns_df
        self.full_corr = returns_df.corr()
def rolling_correlation(self, window=252):
        """
        Calculate rolling correlation to detect breakdowns
        Sharp increases in correlation indicate crisis periods
        """
        rolling_corr = {}
        dates = []
for i in range(len(self.returns) - window):
            window_data = self.returns.iloc[i:i+window]
            corr = window_data.corr()
            dates.append(self.returns.index[i+window])
# Average pairwise correlation
            avg_corr = corr.values[np.triu_indices_from(corr.values, k=1)].mean()
            rolling_corr[dates[-1]] = avg_corr
return pd.Series(rolling_corr)
def tail_correlation(self, quantile=0.05):
        """
        Correlation during extreme market moves
        Compare to normal correlation to identify breakdown risk
        """
        tail_returns = self.returns[self.returns.min(axis=1) <= self.returns.quantile(quantile).min()]
        tail_corr = tail_returns.corr()
# Pairwise differences: tail corr vs normal corr
        differences = {}
        for col1 in self.full_corr.columns:
            for col2 in self.full_corr.columns:
                if col1 < col2:
                    key = f"{col1}-{col2}"
                    differences[key] = tail_corr.loc[col1, col2] - self.full_corr.loc[col1, col2]
return tail_corr, differences
def correlation_stress_factor(self):
        """
        Identify which asset pairs are vulnerable to breakdown
        Factor = tail_corr - normal_corr / normal_corr
        """
        _, differences = self.tail_correlation()
        stress_factors = {k: v / (self.full_corr[k.split('-')[0]][k.split('-')[1]] + 0.001)
                         for k, v in differences.items()}
        return stress_factors
Example: Analyze diversified portfolio
assets = pd.concat([
    yf.download('SPY', start='2015-01-01')['Adj Close'],
    yf.download('TLT', start='2015-01-01')['Adj Close'],
    yf.download('GLD', start='2015-01-01')['Adj Close'],
    yf.download('QQQ', start='2015-01-01')['Adj Close']
], axis=1).dropna()
returns = assets.pct_change().dropna()
detector = CorrelationBreakdownDetector(returns)
Rolling correlation
rolling_corr = detector.rolling_correlation(window=252)
plt.figure(figsize=(12, 6))
rolling_corr.plot()
plt.xlabel('Date')
plt.ylabel('Average Pairwise Correlation')
plt.title('Rolling Correlation Over Time')
plt.axvline(x=pd.Timestamp('2020-03-16'), color='r', linestyle='--', label='COVID Crisis')
plt.legend()
plt.show()
Tail correlations
tail_corr, differences = detector.tail_correlation()
print("Correlation breakdown during extremes:")
for pair, diff in sorted(differences.items(), key=lambda x: x[1], reverse=True)[:5]:
    print(f"{pair}: {diff:+.3f}")

from hmmlearn import hmm
from sklearn.preprocessing import StandardScaler
class RegimeSwitchingCorrelation:
    def __init__(self, returns_df, n_states=2):
        """
        Fit Hidden Markov Model to detect market regimes
        Regime 1: Normal market
        Regime 2: Crisis market
        """
        self.returns = returns_df
        self.n_states = n_states
# Fit HMM on volatility as regime indicator
        scaler = StandardScaler()
        volatility = returns_df.std(axis=1).values.reshape(-1, 1)
        vol_scaled = scaler.fit_transform(volatility)
self.model = hmm.GaussianHMM(n_components=n_states)
        self.model.fit(vol_scaled)
self.states = self.model.predict(vol_scaled)
def correlations_by_regime(self):
        """Calculate correlation matrix for each regime"""
        regime_corrs = {}
for regime in range(self.n_states):
            mask = self.states == regime
            regime_returns = self.returns[mask]
            regime_corrs[regime] = regime_returns.corr()
return regime_corrs
def current_regime(self):
        """Predict current market regime"""
        latest_vol = self.returns.std(axis=1).iloc[-252:].std()
        latest_vol_scaled = (latest_vol - self.returns.std(axis=1).mean()) / self.returns.std(axis=1).std()
        regime = self.model.predict([[latest_vol_scaled]])[0]
        return regime
Example
rsm = RegimeSwitchingCorrelation(returns, n_states=2)
regime_corrs = rsm.correlations_by_regime()
current_regime = rsm.current_regime()
print(f"Current Regime: {current_regime}")
print(f"\nNormal Market Correlations:")
print(regime_corrs[0])
print(f"\nCrisis Market Correlations:")
print(regime_corrs[1])

class CrisisProtectedPortfolio:
    def __init__(self, assets, normal_weights, crisis_weights):
        """
        Maintain two portfolios:
        normal_weights: For calm market conditions
        crisis_weights: Rebalance to if crisis detected
        """
        self.assets = assets
        self.normal_weights = normal_weights
        self.crisis_weights = crisis_weights
        self.current_weights = normal_weights.copy()
def var_historical(self, returns, confidence=0.05):
        """Value at Risk using historical simulation"""
        sorted_returns = np.sort(returns)
        index = int(len(returns) * confidence)
        return sorted_returns[index]
def portfolio_var(self, returns_df, weights, confidence=0.05):
        """Portfolio VaR accounting for correlation changes"""
        weighted_returns = (returns_df * weights).sum(axis=1)
        return self.var_historical(weighted_returns, confidence)
def adjust_for_crisis(self, current_correlation):
        """
        If correlation rises above threshold, reduce diversification benefits
        by shifting toward crisis-resistant weights
        """
        avg_corr = current_correlation.values[
            np.triu_indices_from(current_correlation.values, k=1)
        ].mean()
if avg_corr > 0.75:  # Crisis threshold
            # Smooth transition to crisis weights
            alpha = 0.7  # Weight shift intensity
            self.current_weights = (alpha * self.crisis_weights +
                                   (1 - alpha) * self.normal_weights)
            return True
        else:
            self.current_weights = self.normal_weights.copy()
            return False
def get_hedging_positions(self, returns_df):
        """
        What hedges protect during correlation breakdown?
        Test which assets stay uncorrelated during crises
        """
        detector = CorrelationBreakdownDetector(returns_df)
        _, stress_factors = detector.tail_correlation()
# Assets with negative stress factors are good hedges
        hedges = {k: v for k, v in stress_factors.items() if v < -0.2}
        return hedges
Example
normal_weights = np.array([0.60, 0.30, 0.05, 0.05])  # SPY, TLT, GLD, VXX
crisis_weights = np.array([0.40, 0.50, 0.05, 0.05])  # Shift from stocks to bonds
portfolio = CrisisProtectedPortfolio(
    assets.columns, normal_weights, crisis_weights
)
Check if rebalancing needed
detector = CorrelationBreakdownDetector(returns)
rolling_corr = detector.rolling_correlation(window=252)
current_corr = returns.iloc[-252:].corr()
is_crisis = portfolio.adjust_for_crisis(current_corr)
print(f"Crisis detected: {is_crisis}")
print(f"Current weights: {portfolio.current_weights}")