File size: 2,765 Bytes
d699ea7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 | #!/usr/bin/env python3
"""AF-1 core-signal MVP: cross-asset correlation-breakdown detection.
Aligns two symbols' mid-prices on a common time grid and measures the rolling
pairwise correlation. Test: on a CRASH day (BTC/ETH 2021-05-19), does the
correlation COLLAPSE around the crash trough vs the calm baseline of the same
day? If yes, a rolling-correlation-drop rule is a viable early-warning signal
that the (disabled) Stage-4 was supposed to provide.
Usage: PYTHONPATH=ml python scripts/correlation_breakdown.py
"""
import sys
from pathlib import Path
import numpy as np
import pandas as pd
sys.path.insert(0, "ml")
def load_mid(parquet, start_ts=None, end_ts=None, step=1):
df = pd.read_parquet(parquet, columns=["timestamp_ms", "mid_price"])
if start_ts is not None:
df = df[(df.timestamp_ms >= start_ts) & (df.timestamp_ms <= end_ts)]
df = df.dropna(subset=["mid_price"])
# resample to 1s grid via mean mid per second
df["sec"] = df.timestamp_ms // 1000
s = df.groupby("sec").mid_price.mean()
return s
def main() -> int:
btc = load_mid("data/parquet/BTCUSDT_2021-05-19.parquet")
eth = load_mid("data/parquet/ETHUSDT_2021-05-19.parquet")
# inner join on the 1s grid
both = pd.concat([btc, eth], axis=1, keys=["btc", "eth"]).dropna()
b = both.btc; e = both.eth
print("aligned 1s samples:", len(b))
print("BTC day range %.2f -> %.2f | ETH %.2f -> %.2f"
% (b.min(), b.max(), e.min(), e.max()))
# crash trough in this slice
trough = int(e.idxmin())
print("ETH trough at sec", trough)
# rolling correlation (log-returns over a ~120s window)
rb = np.log(b / b.shift(1)); re = np.log(e / e.shift(1))
corr = rb.rolling(120).corr(re)
# z-score vs the day's calm (pre-crash) baseline
pre = corr.loc[: trough - 600].dropna()
calm_mean, calm_std = pre.mean(), pre.std()
print("pre-crash (calm) rolling-corr: mean=%.3f std=%.3f" % (calm_mean, calm_std))
crash_win = corr.loc[trough - 300: trough + 60].dropna()
if len(crash_win) and calm_std > 1e-6:
z = (crash_win.mean() - calm_mean) / calm_std
print("crash-window rolling-corr mean=%.3f (z=%.2f vs calm)" % (crash_win.mean(), z))
print("correlation COLLAPSE" if z <= -2.0 else
("correlation weakens (z=%.1f)" % z))
else:
print("insufficient data for crash-window corr")
# how early does the corr drop? find first time corr < calm_mean - 2*std
thr = calm_mean - 2.0 * calm_std
below = corr[corr < thr]
if len(below):
first = below.index.min()
print("corr first < calm-2std at sec %d (%.0f s before ETH trough)" % (first, trough - first))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|