cross-asset-relationships

> **Skill:** Cross Asset Relationships | **Domain:** trading | **Category:** fundamentals | **Level:** advanced > **Tags:** `trading`, `fundamentals`, `correlation`, `intermarket`, `carry-trade`, `cross-asset` --- ## Pair Correlation Engine # Pair Correlation Engine ## Overview Analyzes statistical relationships between financial instruments across multiple timeframes and historical periods. Detects regime shifts, divergences, lead-lag relationships, and provides actionable correlation intelligence for trading decisions. ## Architecture ``` ┌───────────────────────────────────────────────────┐ │ Pair Correlation Engine │ ├────────────┬─────────────┬────────────────────────┤ │ Correlation│ History vs │ Regime Detection & │ │ Matrix │ Current │ Divergence Scanner │ └────────────┴─────────────┴────────────────────────┘ ``` --- ## 1. Core Correlation Computation ```python import pandas as pd import numpy as np from scipy import stats from scipy.cluster.hierarchy import linkage, fcluster, dendrogram from typing import Optional from datetime import datetime, timedelta def compute_returns(prices: pd.DataFrame, method: str = "log") -> pd.DataFrame: """Convert price DataFrame to returns. Columns = symbols.""" if method == "log": return np.log(prices / prices.shift(1)).dropna() return prices.pct_change().dropna() def correlation_matrix( prices: pd.DataFrame, method: str = "pearson", window: Optional[int] = None ) -> pd.DataFrame: """ Full correlation matrix. method: pearson, spearman, kendall window: if set, uses last N bars only """ returns = compute_returns(prices) if window: returns = returns.tail(window) return returns.corr(method=method) def rolling_correlation( series_a: pd.Series, series_b: pd.Series, window: int = 60, method: str = "pearson" ) -> pd.Series: """Rolling window correlation between two series.""" ret_a = np.log(series_a / series_a.shift(1)).dropna() ret_b = np.log(series_b / series_b.shift(1)).dropna() aligned = pd.concat([ret_a, ret_b], axis=1).dropna() aligned.columns = ["a", "b"] return aligned["a"].rolling(window).corr(aligned["b"]) ``` --- ## 2. Historical vs Current Correlation (Core Feature) ```python def historical_vs_current( prices: pd.DataFrame, current_window: int = 30, historical_windows: list[int] = [90, 180, 365], method: str = "pearson" ) -> dict: """ Compare current correlation regime vs historical norms. Returns: {pair: {current, hist_90d, hist_180d, hist_365d, deviation, regime_shift}} """ returns = compute_returns(prices) symbols = returns.columns.tolist() results = {} for i, sym_a in enumerate(symbols): for sym_b in symbols[i + 1:]: pair_key = f"{sym_a}/{sym_b}" pair_data = returns[[sym_a, sym_b]].dropna() current_corr = pair_data.tail(current_window).corr().iloc[0, 1] hist_corrs = {} for w in historical_windows: if len(pair_data) >= w: hist_corrs[f"hist_{w}d"] = pair_data.tail(w).corr().iloc[0, 1] else: hist_corrs[f"hist_{w}d"] = np.nan # Compute deviation from longest available history longest_hist = next((hist_corrs[f"hist_{w}d"] for w in sorted(historical_windows, reverse=True) if not np.isnan(hist_corrs.get(f"hist_{w}d", np.nan))), np.nan) deviation = current_corr - longest_hist if not np.isnan(longest_hist) else 0 regime_shift = abs(deviation) > 0.3 # >0.3 = significant regime change results[pair_key] = { "current": round(current_corr, 4), **{k: round(v, 4) for k, v in hist_corrs.items()}, "deviation": round(deviation, 4), "regime_shift": regime_shift, "signal": _interpret_deviation(deviation), } return results def _interpret_deviation(dev: float) -> str: """Interpret correlation deviation into actionable signal.""" if abs(dev) < 0.1: return "STABLE — correlations normal" if dev > 0.3: return "CONVERGENCE — pairs moving together unusually strongly" if dev < -0.3: return "DIVERGENCE — pairs decoupling, watch for mean reversion" if dev > 0.1: return "STRENGTHENING — correlation increasing" return "WEAKENING — correlation decreasing" ``` --- ## 3. Divergence Detection ```python def detect_divergences( prices: pd.DataFrame, lookback: int = 60, threshold: float = 0.3 ) -> list[dict]: """ Find pairs that have recently diverged from their historical correlation. These are potential mean-reversion or trend-change signals. """ hvc = historical_vs_current(prices, current_window=lookback // 2) divergences = [] for pair, data in hvc.items(): if data["regime_shift"]: divergences.append({ "pair": pair, "current_corr": data["current"], "historical_corr": data.get("hist_365d", data.get("hist_180d", np.nan)), "deviation": data["deviation"], "type": "convergence_anomaly" if data["deviation"] > 0 else "divergence_anomaly", "signal": data["signal"], "priority": abs(data["deviation"]), }) return sorted(divergences, key=lambda x: x["priority"], reverse=True) def spread_analysis( price_a: pd.Series, price_b: pd.Series, window: int = 60 ) -> pd.DataFrame: """ Compute normalized spread between two series for pair trading. Z-score indicates mean-reversion opportunity. """ # Normalize both series to start at 1.0 norm_a = price_a / price_a.iloc[0] norm_b = price_b / price_b.iloc[0] spread = norm_a - norm_b z_score = (spread - spread.rolling(window).mean()) / spread.rolling(window).std() return pd.DataFrame({ "spread": spread, "z_score": z_score, "upper_band": spread.rolling(window).mean() + 2 * spread.rolling(window).std(), "lower_band": spread.rolling(window).mean() - 2 * spread.rolling(window).std(), }) ``` --- ## 4. Lead-Lag Analysis ```python def lead_lag_analysis( series_a: pd.Series, series_b: pd.Series, max_lag: int = 10 ) -> pd.DataFrame: """ Cross-correlation at different lags to detect if one pair leads another. Positive lag = A leads B. Negative lag = B leads A. """ ret_a = np.log(series_a / series_a.shift(1)).dropna() ret_b = np.log(series_b / series_b.shift(1)).dropna() aligned = pd.concat([ret_a, ret_b], axis=1).dropna() aligned.columns = ["a", "b"] results = [] for lag in range(-max_lag, max_lag + 1): if lag >= 0: corr = aligned["a"].iloc[lag:].reset_index(drop=True).corr(aligned["b"].iloc[:len(aligned) - lag].reset_index(drop=True)) else: corr = aligned["a"].iloc[:len(aligned) + lag].reset_index(drop=True).corr(aligned["b"].iloc[-lag:].reset_index(drop=True)) results.append({"lag": lag, "correlation": round(corr, 4)}) df = pd.DataFrame(results) best = df.loc[df["correlation"].abs().idxmax()] df.attrs["best_lag"] = int(best["lag"]) df.attrs["best_corr"] = float(best["correlation"]) df.attrs["interpretation"] = ( f"{'A leads B' if best['lag'] > 0 else 'B leads A' if best['lag'] < 0 else 'Synchronous'} " f"by {abs(int(best['lag']))} bars (r={best['correlation']:.4f})" ) return df ``` --- ## 5. Cluster Analysis — Group Correlated Pairs ```python def cluster_pairs( prices: pd.DataFrame, n_clusters: int = 4, method: str = "ward" ) -> dict: """ Hierarchical clustering of instruments by correlation. Returns cluster assignments and cluster statistics. """ corr = correlation_matrix(prices) distance = np.sqrt(2 * (1 - corr)) np.fill_diagonal(distance.values, 0) condensed = distance.values[np.triu_indices_from(distance.values, k=1)] Z = linkage(condensed, method=method) labels = fcluster(Z, t=n_clusters, criterion="maxclust") clusters = {} for sym, cluster_id in zip(corr.columns, labels): cid = int(cluster_id) if cid not in clusters: clusters[cid] = [] clusters[cid].append(sym) return { "n_clusters": n_clusters, "clusters": clusters, "linkage": Z, "assignments": dict(zip(corr.columns.tolist(), [int(l) for l in labels])), } def find_hedge_pairs( prices: pd.DataFrame, target_symbol: str, min_negative_corr: float = -0.5 ) -> list[dict]: """Find instruments negatively correlated to target — potential hedges.""" corr = correlation_matrix(prices) if target_symbol not in corr.columns: return [] target_corr = corr[target_symbol].drop(target_symbol).sort_values() hedges = [] for sym, c in target_corr.items(): if c <= min_negative_corr: hedges.append({"symbol": sym, "correlation": round(c, 4), "hedge_quality": "strong" if c < -0.7 else "moderate"}) return hedges ``` --- ## 6. Correlation Report Generator ```python def full_correlation_report( prices: pd.DataFrame, current_window: int = 30 ) -> dict: """ Complete correlation intelligence report. Pipe this to trading-brain for decision-making. """ return { "timestamp": datetime.utcnow().isoformat(), "symbols_analyzed": prices.columns.tolist(), "n_bars": len(prices), "current_matrix": correlation_matrix(prices, window=current_window).to_dict(), "historical_matrix": correlation_matrix(prices).to_dict(), "historical_vs_current": historical_vs_current(prices, current_window), "divergences": detect_divergences(prices), "clusters": cluster_pairs(prices), } ``` --- ## Usage Conventions 1. **Always use log returns** for correlation — more statistically stable than simple returns 2. **Check sample size** — need minimum 30 bars for meaningful correlation 3. **Multi-timeframe** — compute correlations on H1, H4, and D1 for robust signals 4. **Regime shifts > 0.3** are significant and should trigger alerts 5. **Lead-lag > 2 bars** with |r| > 0.3 is a potential predictive signal 6. **Correlation ≠ causation** — always note this in reports --- ## Correlation Heatmap Visualizer # Correlation Heatmap Visualizer ```python import pandas as pd, numpy as np, matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import io, base64 class CorrelationHeatmapVisualizer: @staticmethod def render(prices: pd.DataFrame, method: str = "pearson", window: int = None, save_path: str = None) -> str: returns = np.log(prices / prices.shift(1)).dropna() if window: returns = returns.tail(window) corr = returns.corr(method=method) fig, ax = plt.subplots(figsize=(12, 10), facecolor="#131722") ax.set_facecolor("#131722") im = ax.imshow(corr.values, cmap="RdYlGn", vmin=-1, vmax=1, aspect="auto") ax.set_xticks(range(len(corr.columns))); ax.set_yticks(range(len(corr.columns)))