#!/usr/bin/env python3 """AF-1: correlation-breakdown backtest, parametric on a (anchor, basket) pair. Drives the real CorrelationBreakdown feature on aligned, resampled mid-price grids. Validated scenarios: - LUNA-vs-BTC 2022-05-11 (collapse): corr should BREAK (LUNA->0, BTC flat). - LUNA-vs-BTC 2022-05-09 (calm pre): corr should stay high (no breakdown). - BTC-vs-ETH 2021-05-19 (market-wide crash): corr stays HIGH (both crashed). Usage: PYTHONPATH=ml python scripts/run_correlation_backtest.py \ --a LUNAUSDT --a-parq data/parquet/LUNAUSDT_2022-05-11.parquet \ --b BTCUSDT --b-parq data/parquet/BTCUSDT_2022-05-11.parquet \ --bin-s 60 --corr-s 7200 --floor 0.4 --sustain 180 --cooldown 600 """ import argparse import sys from pathlib import Path import numpy as np import pandas as pd sys.path.insert(0, "ml") from flash_crash_watchdog.features.correlation import CorrelationBreakdown, CorrelationConfig # noqa def load_binned(parquet: str, bin_s: int) -> pd.Series: df = pd.read_parquet(parquet, columns=["timestamp_ms", "mid_price"]).dropna(subset=["mid_price"]) df["bin"] = df.timestamp_ms // (bin_s * 1000) return df.groupby("bin").mid_price.mean() def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--a", default="LUNAUSDT") ap.add_argument("--a-parq", required=True) ap.add_argument("--b", default="BTCUSDT") ap.add_argument("--b-parq", required=True) ap.add_argument("--bin-s", type=int, default=60) ap.add_argument("--corr-s", type=int, default=7200, help="rolling corr window (seconds)") ap.add_argument("--floor", type=float, default=0.4) ap.add_argument("--sustain", type=int, default=180, help="seconds below floor to fire") ap.add_argument("--cooldown", type=int, default=600) ap.add_argument("--label", default="day") args = ap.parse_args() a = load_binned(args.a_parq, args.bin_s) b = load_binned(args.b_parq, args.bin_s) both = pd.concat([a, b], axis=1, keys=["a", "b"]).dropna().sort_index() if len(both) < 200: print("insufficient aligned samples:", len(both)) return 1 win_bins = max(2, args.corr_s // args.bin_s) sustain_bins = max(1, args.sustain // args.bin_s) cooldown_bins = max(1, args.cooldown // args.bin_s) corr = CorrelationBreakdown(CorrelationConfig( anchor=args.a, corr_window_bins=win_bins, baseline_bins=max(win_bins * 4, 600), warmup_bins=win_bins * 2, collapse_z=2.0, floor_corr=args.floor, sustain_s=sustain_bins)) # NOTE: module's sustain is in bins here (fed at bin cadence). alerts = [] last = -10 ** 9 trough_bin = int(both.a.idxmin()) n = len(both) for i, (bin_, row) in enumerate(both.iterrows()): corr.update(args.a, float(row.a), int(bin_) * args.bin_s * 1000) corr.update(args.b, float(row.b), int(bin_) * args.bin_s * 1000) z, score, fire = corr.evaluate() if fire and (i - last) > cooldown_bins: last = i alerts.append(int(bin_)) if i % 5000 == 0 and i: sys.stderr.write(".") # progress dots sys.stderr.write("\n") first = alerts[0] if alerts else None lead = (trough_bin - first) * args.bin_s if first is not None else None print("[%s] bins=%d corr-win=%ds floor=%.2f sustain=%ds cooldown=%ds" % ( args.label, n, args.corr_s, args.floor, args.sustain, args.cooldown)) print(" %s-vs-%s correlation-breakdown alerts=%d first=bin%d lead_to_trough=%s" % (args.a, args.b, len(alerts), first if first is not None else -1, ("%.0f s" % lead) if lead is not None else "n/a")) return 0 if __name__ == "__main__": raise SystemExit(main())