7495eaa286
Built survivorship-free universe: 135 perps incl 28 known-dead (LUNA/SRM/MATIC...), universe rebuilt daily as top-K by trailing dollar-volume (dead coins in while trading, drop out after crash), no lookahead. mom_20 TOPK=30: full +0.76, IS+0.85/OOS+0.56, CPCV-med +0.73, DSR 0.86, POSITIVE EVERY YEAR 2020-2026 incl 2022 +0.72. Counterintuitive: survivorship was HIDING the edge (survivor-only 2022 -0.40 -> PIT +0.72) because momentum shorts the dying coins and profits from crashes. Decisive survivorship confirmation. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
89 lines
3.8 KiB
Python
89 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Point-in-time survivorship test for crypto cross-sectional momentum.
|
|
|
|
Universe is rebuilt EACH DAY: among coins with active trailing-30d dollar-volume (>0),
|
|
take the top-K by that volume. Dead coins (LUNA, SRM, ...) are IN the cross-section while
|
|
they trade and DROP OUT (after carrying their crash) when volume dies. No lookahead:
|
|
volume & momentum use only past data; weights apply to next-day return. If momentum's
|
|
edge survives here — especially 2022 with LUNA included — it is real beyond the bootstrap proxy.
|
|
"""
|
|
import glob
|
|
import math
|
|
import os
|
|
import sys
|
|
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from signal_sweep import xs_weights, pnl_w, validate, sharpe_t # noqa: E402
|
|
import torch # noqa: E402
|
|
|
|
DEV = "cuda" if torch.cuda.is_available() else "cpu"
|
|
|
|
|
|
def load():
|
|
syms, data = [], {}
|
|
for p in sorted(glob.glob("data/surfer/crypto_pit/*.npz")):
|
|
d = np.load(p); s = p.split("/")[-1][:-4]
|
|
data[s] = (d["day"].astype(np.int64), d["close"].astype(float), d["qvol"].astype(float), d["funding"].astype(float))
|
|
syms.append(s)
|
|
syms = sorted(syms)
|
|
days = np.array(sorted(set().union(*[set(data[s][0].tolist()) for s in syms])))
|
|
di = {int(v): i for i, v in enumerate(days.tolist())}
|
|
T, N = len(days), len(syms)
|
|
close = np.full((T, N), np.nan); qv = np.full((T, N), np.nan); fund = np.full((T, N), np.nan)
|
|
for j, s in enumerate(syms):
|
|
dd, cc, vv, ff = data[s]
|
|
for k in range(len(dd)):
|
|
r = di[int(dd[k])]; close[r, j] = cc[k]; qv[r, j] = vv[k]; fund[r, j] = ff[k]
|
|
return syms, days, close, qv, fund
|
|
|
|
|
|
def trailing(lc, L):
|
|
out = np.full_like(lc, np.nan); out[L:] = lc[L:] - lc[:-L]; return out
|
|
|
|
|
|
def main():
|
|
syms, days, close, qv, fund = load()
|
|
T, N = close.shape
|
|
lc = np.log(close)
|
|
R = np.zeros((T, N)); R[1:] = lc[1:] - lc[:-1]; R = np.where(np.isfinite(R), R, 0.0)
|
|
Reff = R - np.where(np.isfinite(fund), fund, 0.0)
|
|
qv = np.nan_to_num(qv)
|
|
dv30 = np.full((T, N), 0.0) # trailing 30d mean dollar-volume
|
|
for t in range(30, T):
|
|
dv30[t] = qv[t - 30:t].mean(axis=0)
|
|
dead = sum(1 for j in range(N) if qv[-30:, j].mean() == 0)
|
|
year = (1970 + days / 365.25).astype(int)
|
|
|
|
print(f"\n===== POINT-IN-TIME MOMENTUM — {N} coins ({dead} dead/delisted), {T}d =====")
|
|
for TOPK in [30, 50]:
|
|
# daily universe = top-K by trailing dollar-vol among actively-trading coins
|
|
univ = np.zeros((T, N), bool)
|
|
for t in range(T):
|
|
elig = np.where((dv30[t] > 0) & np.isfinite(close[t]))[0]
|
|
if len(elig) == 0:
|
|
continue
|
|
k = min(TOPK, len(elig))
|
|
top = elig[np.argsort(-dv30[t, elig])[:k]]
|
|
univ[t, top] = True
|
|
print(f"\n--- TOPK={TOPK} (avg coins/day in-universe = {univ.sum(1).mean():.0f}) ---")
|
|
print(f"{'lookback':>9} {'full':>6} {'IS':>6} {'OOS':>6} {'CPCVmed':>8} {'DSR':>5} | per-year (2020..2026)")
|
|
for L in [10, 20, 30]:
|
|
sig = trailing(lc, L).copy()
|
|
sig[~univ] = np.nan # restrict signal to PIT universe
|
|
w = xs_weights(sig)
|
|
pnl = pnl_w(w, Reff, cost_bp=10)
|
|
v = validate(pnl, days, 6)
|
|
py = []
|
|
for y in range(2020, 2027):
|
|
m = year[1:] == y
|
|
py.append(sharpe_t(torch.tensor(pnl[m], device=DEV, dtype=torch.float64)) if m.sum() > 30 else float("nan"))
|
|
pys = " ".join(f"{p:>+5.2f}" if not math.isnan(p) else " n/a" for p in py)
|
|
print(f"{L:>9} {v['full']:>+6.2f} {v['is_']:>+6.2f} {v['oos']:>+6.2f} {v['med']:>+8.2f} {v['dsr']:>5.2f} | {pys}")
|
|
print("\nVERDICT: if mom_20-30 stays positive (esp. 2022 with LUNA in-universe) => survivorship-robust REAL edge.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|