Correlation Regime Detection

1. Compute rolling-window pairwise correlations of returns
2. Reduce each correlation matrix to one number: edge density
   = fraction of asset pairs with |ρ| ≥ edge_threshold
3. Smooth the density series with a TRAILING window (causal — see warning below)
4. Run a hysteresis (Schmitt-trigger) state machine over the smoothed series:
   enter FUSED when density ≥ enter_threshold, exit only when ≤ exit_threshold
5. Emit regime state + transition timestamps for monitoring / reporting

import numpy as np
import pandas as pd


def compute_edge_density(
    returns: pd.DataFrame,
    corr_window: int = 60,
    edge_threshold: float = 0.5,
) -> pd.Series:
    """Reduce rolling correlation matrices to an edge-density series.

    Edge density is the fraction of distinct asset pairs whose rolling
    |correlation| clears ``edge_threshold`` — a scalar "how fused is the
    market" gauge in [0, 1].

    Args:
        returns: Multi-asset return matrix, columns are symbols
        corr_window: Rolling window length (bars) for pairwise correlation
        edge_threshold: |ρ| level at which a pair counts as an "edge"

    Returns:
        Edge-density series aligned to ``returns.index`` (NaN during warmup)
    """
    n_assets = returns.shape[1]
    n_pairs = n_assets * (n_assets - 1) // 2
    upper_mask = np.triu(np.ones((n_assets, n_assets), dtype=bool), k=1)

    density = pd.Series(np.nan, index=returns.index)
    for i in range(corr_window, len(returns) + 1):
        corr = returns.iloc[i - corr_window:i].corr().abs().to_numpy()
        density.iloc[i - 1] = float((corr[upper_mask] >= edge_threshold).sum()) / n_pairs
    return density


def detect_regimes(
    density: pd.Series,
    smooth_window: int = 5,
    enter_threshold: float = 0.65,
    exit_threshold: float = 0.45,
) -> pd.DataFrame:
    """Hysteresis (Schmitt-trigger) regime state machine on smoothed density.

    The market is FUSED once smoothed density reaches ``enter_threshold`` and
    stays FUSED until it falls back to ``exit_threshold``. The dead band
    between the two thresholds is what suppresses chatter.

    Args:
        density: Edge-density series from :func:`compute_edge_density`
        smooth_window: Trailing smoothing window (causal; never centered)
        enter_threshold: Density level that opens a FUSED regime
        exit_threshold: Density level that closes it (must be < enter_threshold)

    Returns:
        DataFrame with columns ``density``, ``smoothed``, ``fused`` (0/1)
    """
    if exit_threshold >= enter_threshold:
        raise ValueError("exit_threshold must be below enter_threshold")

    # Trailing mean = causal. A centered window here silently reads the future.
    smoothed = density.rolling(smooth_window, min_periods=1).mean()

    fused = False
    states = np.zeros(len(smoothed), dtype=int)
    for i, value in enumerate(smoothed.to_numpy()):
        if np.isnan(value):
            states[i] = int(fused)
            continue
        if not fused and value >= enter_threshold:
            fused = True
        elif fused and value <= exit_threshold:
            fused = False
        states[i] = int(fused)

    return pd.DataFrame(
        {"density": density, "smoothed": smoothed, "fused": states},
        index=density.index,
    )

def regime_exposure_context(
    regimes: pd.DataFrame,
    base_gross: float = 1.0,
    fused_gross: float = 0.5,
) -> pd.Series:
    """Descriptive gross-exposure context per bar (NOT a trade signal).

    Args:
        regimes: Output of :func:`detect_regimes`
        base_gross: Reference gross exposure during defused (calm) regimes
        fused_gross: Reference gross exposure during fused regimes

    Returns:
        Per-bar reference gross-exposure series for risk reporting
    """
    return pd.Series(
        np.where(regimes["fused"] == 1, fused_gross, base_gross),
        index=regimes.index,
        name="reference_gross",
    )

1. For each asset, compute a short-horizon move-intensity series
   (rolling sum of |returns|)
2. Score it as a robust z-score against the asset's OWN calm baseline
   (median/MAD, baseline strictly prior to the bar being scored)
3. Watch tier (low z): informational watchlist only — never allowed to accuse
4. Alarm tier (very high z): record each asset's first crossing time
5. Apply the verdict rules: lead-gap ⇒ NAME, pack-together ⇒ MACRO,
   near-tie leaders ⇒ AMBIGUOUS, no crossings ⇒ ABSTAIN

def first_mover_attribution(
    returns: pd.DataFrame,
    baseline_window: int = 120,
    move_window: int = 3,
    alarm_z: float = 8.0,
    watch_z: float = 3.0,
    lead_gap: int = 2,
    macro_span: int = 1,
    macro_fraction: float = 0.6,
) -> dict:
    """First-mover crisis attribution with abstention.

    Each asset's short-horizon move intensity is scored as a robust z-score
    against its own trailing calm baseline (median / MAD, shifted so the move
    being scored never contaminates its own baseline). Verdicts follow the
    honesty protocol: NAME only on a clear solo lead, otherwise MACRO /
    AMBIGUOUS / ABSTAIN.

    Args:
        returns: Multi-asset return matrix, columns are symbols
        baseline_window: Trailing window (bars) for the per-asset calm baseline
        move_window: Short horizon (bars) of the move-intensity sum
        alarm_z: Robust z at which an asset counts as "in violent collapse"
        watch_z: Informational watch-tier level (never used for naming)
        lead_gap: Minimum lead (bars) of the first crosser over the second
            required to NAME it
        macro_span: If the pack crosses within this many bars of the first
            crossing, the episode is MACRO
        macro_fraction: Fraction of assets that must cross to call MACRO

    Returns:
        Dict with ``verdict`` (NAME / MACRO / AMBIGUOUS / ABSTAIN),
        ``named`` (symbol or None), ``candidates``, ``crossings``
        (symbol → first alarm timestamp), ``watchlist``, and ``z`` (the
        full z-score DataFrame for inspection)
    """
    intensity = returns.abs().rolling(move_window).sum()
    # Proper rolling MAD: each window's deviations from its OWN median.
    # (Nesting two full-length rolling medians instead stacks their warmups —
    # the score would silently stay NaN for 2x baseline_window bars.)
    # shift(1) keeps the bar being scored out of its own baseline.
    med = intensity.rolling(baseline_window).median().shift(1)
    mad = intensity.rolling(baseline_window).apply(
        lambda window: np.median(np.abs(window - np.median(window))), raw=True
    ).shift(1)
    z = (intensity - med) / (1.4826 * mad.replace(0.0, np.nan))

    crossings: dict[str, pd.Timestamp] = {}
    for symbol in z.columns:
        hits = z.index[z[symbol] >= alarm_z]
        if len(hits) > 0:
            crossings[symbol] = hits[0]

    watch_hits = (z >= watch_z).any()
    watchlist = sorted(watch_hits.index[watch_hits])

    if not crossings:
        return {
            "verdict": "ABSTAIN", "named": None, "candidates": [],
            "crossings": {}, "watchlist": watchlist, "z": z,
        }

    ordered = sorted(crossings.items(), key=lambda item: item[1])
    first_symbol, first_time = ordered[0]
    positions = {ts: i for i, ts in enumerate(z.index)}
    first_pos = positions[first_time]

    pack_size = sum(
        1 for _, ts in ordered if positions[ts] - first_pos <= macro_span
    )
    if pack_size >= max(2, int(np.ceil(macro_fraction * returns.shape[1]))):
        return {
            "verdict": "MACRO", "named": None,
            "candidates": [s for s, _ in ordered],
            "crossings": crossings, "watchlist": watchlist, "z": z,
        }

    leaders = [
        symbol for symbol, ts in ordered if positions[ts] - first_pos < lead_gap
    ]
    if len(leaders) == 1 and (
        len(ordered) == 1 or positions[ordered[1][1]] - first_pos >= lead_gap
    ):
        return {
            "verdict": "NAME", "named": first_symbol, "candidates": leaders,
            "crossings": crossings, "watchlist": watchlist, "z": z,
        }

    return {
        "verdict": "AMBIGUOUS", "named": None, "candidates": leaders,
        "crossings": crossings, "watchlist": watchlist, "z": z,
    }