#!/usr/bin/env python3 """ Permutation test: do Aave's liquidations cluster beyond independence? Uses price data (Yahoo Finance) + Aave's published liquidation thresholds. No RPC key, no on-chain data, no API key needed. Method: 1. Download daily prices for the major Aave V3 collateral assets. 2. For each asset on each day, compute the daily return. 3. Mark a day as a "liquidation trigger" for that asset if the drop would push a position from max LTV to liquidation (health < 1). 4. Run the column-shuffle permutation test on the resulting binary matrix. 5. Also run on the raw worst-5% tail events (same as The Rogue Wave). What this tests: Aave sets each asset's liquidation threshold independently. If the prices were independent, the probability that M assets all trigger liquidations on the same day would be the product of their individual rates. The ratio R = real / shuffled measures how much worse it actually is. R >> 1 means the independent-threshold design is structurally wrong. This is Aave's own risk model tested against Aave's own parameters with public market data. """ import sys import numpy as np from collections import defaultdict from datetime import datetime try: import yfinance as yf except ImportError: print("pip install yfinance") sys.exit(1) # ─── Aave V3 Ethereum parameters (from aave.com/docs) ──────────────── # Format: (ticker, name, max_LTV, liquidation_threshold) # LTV and threshold are fractions. A position at max LTV liquidates # when the price drops by (1 - LTV/threshold). # Source: https://aave.com/docs (risk parameters, Ethereum V3 main market) AAVE_ASSETS = [ ("ETH-USD", "ETH", 0.8050, 0.8300), # WETH ("BTC-USD", "BTC", 0.7300, 0.7800), # WBTC ("LINK-USD", "LINK", 0.6800, 0.7600), # Chainlink ("AAVE-USD", "AAVE", 0.6600, 0.7300), # Aave ("MKR-USD", "MKR", 0.6500, 0.7000), # Maker ("UNI-USD", "UNI", 0.6500, 0.7700), # Uniswap ("SNX-USD", "SNX", 0.4900, 0.6500), # Synthetix ("CRV-USD", "CRV", 0.3500, 0.4200), # Curve ("BAL-USD", "BAL", 0.5700, 0.6200), # Balancer ("LDO-USD", "LDO", 0.4000, 0.5000), # Lido DAO ] # Start date: earliest where all tokens have data (post DeFi-summer 2020) START = "2020-09-01" B = 2000 SEED = 42 # ─── Data ───────────────────────────────────────────────────────────── def fetch_prices() -> tuple: """Fetch daily close prices from Yahoo Finance.""" tickers = [a[0] for a in AAVE_ASSETS] names = [a[1] for a in AAVE_ASSETS] print(f"Downloading {len(tickers)} assets from Yahoo Finance...") data = yf.download(tickers, start=START, auto_adjust=True, progress=False) # Handle multi-level columns if isinstance(data.columns, __import__('pandas').MultiIndex): prices = data["Close"] else: prices = data # Drop assets with too many NaNs good = [] for i, t in enumerate(tickers): col = prices[t] if t in prices.columns else None if col is not None: pct_valid = col.notna().mean() if pct_valid > 0.8: good.append(i) print(f" {names[i]:>6s}: {col.notna().sum()} days ({pct_valid:.0%})") else: print(f" {names[i]:>6s}: {col.notna().sum()} days ({pct_valid:.0%}) — SKIPPED") else: print(f" {names[i]:>6s}: not found — SKIPPED") assets = [AAVE_ASSETS[i] for i in good] price_df = prices[[a[0] for a in assets]].dropna() print(f"\n{len(price_df)} trading days, {len(assets)} assets") print(f"Range: {price_df.index[0].date()} to {price_df.index[-1].date()}") return price_df, assets def build_matrices(price_df, assets) -> tuple: """Build two binary matrices: 1. Liquidation-trigger matrix (asset-specific thresholds from Aave) 2. Tail-event matrix (each asset's own 5th percentile, as in The Rogue Wave) """ returns = price_df.pct_change().iloc[1:] # daily returns, drop first NaN row T = len(returns) N = len(assets) names = [a[1] for a in assets] # Matrix 1: Aave liquidation triggers # A position at max LTV triggers liquidation when price drops by: # drop_threshold = 1 - (LTV / liquidation_threshold) # e.g. ETH: 1 - 0.805/0.83 = 3.0% — a 3% daily drop liquidates liq_matrix = np.zeros((T, N), dtype=np.int8) print(f"\nAave liquidation triggers (position at max LTV):") for j, (ticker, name, ltv, liq_thresh) in enumerate(assets): drop = 1.0 - ltv / liq_thresh col = returns[ticker].values liq_matrix[:, j] = (col <= -drop).astype(np.int8) rate = liq_matrix[:, j].mean() days = liq_matrix[:, j].sum() print(f" {name:>6s}: LTV={ltv:.0%} thresh={liq_thresh:.0%} " f"→ drop ≥ {drop:.1%} → {days} days ({rate:.1%})") # Matrix 2: Tail events (5th percentile of each asset's own returns) tail_matrix = np.zeros((T, N), dtype=np.int8) print(f"\nTail events (each asset's own worst 5%):") for j, (ticker, name, _, _) in enumerate(assets): col = returns[ticker].values thresh = np.nanpercentile(col, 5) tail_matrix[:, j] = (col <= thresh).astype(np.int8) rate = tail_matrix[:, j].mean() days = tail_matrix[:, j].sum() print(f" {name:>6s}: threshold = {thresh:.2%} → {days} days ({rate:.1%})") dates = [str(d.date()) for d in returns.index] return liq_matrix, tail_matrix, names, dates, returns # ─── Permutation test ───────────────────────────────────────────────── def count_co(matrix, M): return int(np.sum(np.sum(matrix, axis=1) >= M)) def shuffle_test(matrix, names, label, B=2000): """Column-shuffle permutation test.""" T, N = matrix.shape rng = np.random.default_rng(SEED) M_values = list(range(2, min(N + 1, 8))) print(f"\n{'='*65}") print(f" {label}") print(f" {T} days × {N} assets, B = {B}") print(f"{'='*65}") print(f"{'M':>3s} {'Real':>6s} {'Null':>9s} {'Std':>8s} " f"{'R':>8s} {'p':>8s} {'d':>6s}") print("-" * 65) results = {} for M in M_values: real = count_co(matrix, M) null = np.zeros(B) for b in range(B): shuf = matrix.copy() for j in range(N): rng.shuffle(shuf[:, j]) null[b] = count_co(shuf, M) null_mean = null.mean() null_std = null.std() R = real / null_mean if null_mean > 0 else float("inf") p = (np.sum(null >= real) + 1) / (B + 1) d = R ** (1.0 / M) if R > 1 and M > 1 else 1.0 sig = "***" if p < 0.001 else "**" if p < 0.01 else "*" if p < 0.05 else "" print(f"{M:3d} {real:6d} {null_mean:9.1f} {null_std:8.1f} " f"{R:7.1f}× {p:8.4f} {d:5.2f} {sig}") results[M] = {"real": real, "null_mean": null_mean, "R": R, "p": p, "d": d} return results def pairwise(matrix, names, label, B=1000): """Pairwise synchronisation ratios.""" T, N = matrix.shape rng = np.random.default_rng(SEED + 1) print(f"\nPairwise synchronisation — {label}:") print(f"{'A':>6s} {'B':>6s} {'Both':>5s} {'Exp':>7s} {'R':>7s} {'p':>7s}") print("-" * 50) pairs = [] for i in range(N): for j in range(i + 1, N): real = int(np.sum((matrix[:, i] == 1) & (matrix[:, j] == 1))) null_counts = np.zeros(B) for b in range(B): s = matrix[:, j].copy() rng.shuffle(s) null_counts[b] = np.sum((matrix[:, i] == 1) & (s == 1)) null_mean = null_counts.mean() R = real / null_mean if null_mean > 0 else float("inf") p = (np.sum(null_counts >= real) + 1) / (B + 1) pairs.append((names[i], names[j], real, null_mean, R, p)) pairs.sort(key=lambda x: -x[4]) for a, b, real, exp, R, p in pairs: flag = " ***" if p < 0.001 else " **" if p < 0.01 else " *" if p < 0.05 else "" print(f"{a:>6s} {b:>6s} {real:5d} {exp:7.1f} {R:6.1f}× {p:7.4f}{flag}") sig = sum(1 for *_, p in pairs if p < 0.05) total = len(pairs) print(f"\n{sig}/{total} pairs significant at p < 0.05 " f"(chance: {total * 0.05:.1f})") def cascade_days(matrix, names, returns_df, assets): """Show the worst co-liquidation days with actual returns.""" T, N = matrix.shape co_count = np.sum(matrix, axis=1) worst = np.argsort(-co_count) print(f"\nWorst co-liquidation days:") print(f"{'Date':>12s} {'Assets hit':>10s} {'Returns':>50s}") print("-" * 80) shown = 0 for idx in worst: n_hit = co_count[idx] if n_hit < 3: break if shown >= 15: break # Get the date date = returns_df.index[idx] day_str = str(date.date()) # Get returns for hit assets hit = [] for j in range(N): if matrix[idx, j] == 1: ret = returns_df.iloc[idx][assets[j][0]] hit.append(f"{names[j]} {ret:+.1%}") print(f"{day_str:>12s} {n_hit:10d} {', '.join(hit)}") shown += 1 # ─── Main ───────────────────────────────────────────────────────────── def main(): price_df, assets = fetch_prices() liq_matrix, tail_matrix, names, dates, returns = build_matrices(price_df, assets) # Test 1: Aave liquidation triggers liq_results = shuffle_test( liq_matrix, names, "AAVE LIQUIDATION TRIGGERS (position at max LTV)", B=B, ) # Test 2: Tail events (5th percentile) — same as The Rogue Wave tail_results = shuffle_test( tail_matrix, names, "TAIL EVENTS (each asset's own worst 5%)", B=B, ) # Pairwise for both pairwise(liq_matrix, names, "Aave liquidation triggers") pairwise(tail_matrix, names, "Tail events (5th percentile)") # Show worst days cascade_days(liq_matrix, names, returns, assets) # Summary print("\n" + "=" * 65) print("SUMMARY") print("=" * 65) print(f"\nAave treats each asset's liquidation threshold independently.") print(f"If prices were independent, co-liquidations would match the shuffle.\n") for label, results in [("Liquidation triggers", liq_results), ("Tail events (5%)", tail_results)]: print(f" {label}:") for M, r in results.items(): if r["p"] < 0.05: print(f" M={M}: R = {r['R']:.1f}× (p = {r['p']:.4f}), " f"implied d = {r['d']:.2f}") best_M = max((r for r in results.values() if r["p"] < 0.05), key=lambda r: r["R"], default=None) if best_M: print(f" Peak mispricing: {best_M['R']:.0f}×") print() print(" Comparison (from The Rogue Wave, same test on daily returns):") print(" 8 crypto tokens (BTC/ETH/6 DeFi): d = 3.31, R = 120× at M=4") print(" 10 US equity sector ETFs: d = 2.44, R = 35× at M=4") print() print(" If Aave's liquidation-trigger d matches the token-price d,") print(" then the per-asset threshold design is wrong by construction:") print(" the assets crash together, the liquidations fire together,") print(" and no single-asset risk parameter can account for it.") if __name__ == "__main__": main()