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

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"])