spb/anomaly-atlas Public License

1# =============================================================================2#  Project   : anomaly-atlas3#  File      : src/anomaly_atlas/stats/leadlag.py4#  Purpose   : Lead-lag tests: lagged cross-correlation and asymmetry5#  Author    : Simon-Pierre Boucher6#  Contact   : contact@spboucher.ai7#  Data src  : hfmarketdata.io (sole data source)8#  Created   : 2026-08-129#  Modified  : 2026-08-1210#  Platform  : macOS / Apple Silicon (arm64)11#  License   : All rights reserved (research code)12# =============================================================================13"""Lagged cross-correlation between two return series.1415Sign convention: lag k > 0 means x LEADS y by k bars — corr(x_{t-k}, y_t).16Validated on synthetic ground truth (charter §8.1) before real data.17"""1819from __future__ import annotations2021import numpy as np222324def lagged_xcorr(x: np.ndarray, y: np.ndarray, max_lag: int) -> dict[int, float]:25    """corr(x_{t-k}, y_t) for k in [-max_lag, +max_lag]; k>0 = x leads y."""26    x = np.asarray(x, dtype=float)27    y = np.asarray(y, dtype=float)28    n = min(len(x), len(y))29    x, y = x[:n], y[:n]30    out: dict[int, float] = {}31    for k in range(-max_lag, max_lag + 1):32        if k >= 0:33            a, b = x[: n - k] if k else x, y[k:] if k else y34        else:35            a, b = x[-k:], y[: n + k]36        if len(a) < 3 or a.std() == 0.0 or b.std() == 0.0:37            out[k] = float("nan")38            continue39        out[k] = float(np.corrcoef(a, b)[0, 1])40    return out414243def peak_lag(xc: dict[int, float]) -> int:44    """Lag with the largest |corr| (ties: smallest |lag|)."""45    finite = {k: v for k, v in xc.items() if np.isfinite(v)}46    if not finite:47        return 048    return min(finite, key=lambda k: (-abs(finite[k]), abs(k)))495051def leadlag_asymmetry(xc: dict[int, float]) -> float:52    """Sum of corr at positive lags minus sum at negative lags.5354    Zero (in expectation) for synchronous series; positive when x leads y.55    """56    pos = sum(v for k, v in xc.items() if k > 0 and np.isfinite(v))57    neg = sum(v for k, v in xc.items() if k < 0 and np.isfinite(v))58    return float(pos - neg)59