#!/usr/bin/env python3
"""Deflated signal sweep — a factor zoo on data in hand, judged against the FULL search.

Wisdom: breadth without multiple-testing control is the fastest way to fool yourself.
So every signal's Deflated Sharpe is deflated by N_trials = the number of signals tested.
Batch 1 (instant, free, data already downloaded): 22-futures daily panel — cross-sectional
& time-series momentum, short-term reversal, low-vol — plus structural SEASONALITY
(overnight-drift, intraday, day-of-week). Each signal → daily PnL → full/IS/OOS Sharpe,
CPCV(45-path) median+5th-pct, Deflated Sharpe. Ranked by OOS. Survivor = DSR>0.95 AND
OOS>0 AND CPCV-median>0 (consistent + multiple-testing-honest).

Validation on GPU (torch); panel/signal staging on CPU (small). Roll-zeroed close-close
returns (screening; survivors get proper roll handling in deeper validation).
"""
import glob
import math
import re
from itertools import combinations
from statistics import NormalDist

import numpy as np
import torch

ND = NormalDist()
DEV = "cuda" if torch.cuda.is_available() else "cpu"
MONTHS = "FGHJKMNQUVXZ"
SPLIT_DAY = 19723  # 2024-01-01 in days-since-epoch


def load_panel():
    import databento as db
    series = {}
    for p in sorted(glob.glob("data/surfer/*.dbn")):
        root = p.split("/")[-1][:-4]
        pat = re.compile("^" + re.escape(root) + "[" + MONTHS + r"]\d{1,2}$")
        try:
            df = db.DBNStore.from_file(p).to_df().reset_index()
        except Exception:
            continue
        if df.empty or "close" not in df.columns:
            continue
        df = df[df["close"] > 0]
        df = df[df["symbol"].astype(str).str.match(pat)]
        if df.empty:
            continue
        df["day"] = (df["ts_event"].astype("int64") // (86_400 * 10**9))
        idx = df.groupby("day")["volume"].idxmax()
        f = df.loc[idx, ["day", "instrument_id", "open", "close"]].sort_values("day")
        series[root] = (f["day"].to_numpy(np.int64), f["instrument_id"].to_numpy(np.int64),
                        f["open"].to_numpy(np.float64), f["close"].to_numpy(np.float64))
    roots = sorted(series)
    days = np.array(sorted(set().union(*[set(series[r][0].tolist()) for r in roots])))
    di = {d: i for i, d in enumerate(days.tolist())}
    T, N = len(days), len(roots)
    close = np.full((T, N), np.nan); op = np.full((T, N), np.nan); inst = np.full((T, N), -1, np.int64)
    for j, r in enumerate(roots):
        d, ii, oo, cc = series[r]
        for k in range(len(d)):
            row = di[int(d[k])]; close[row, j] = cc[k]; op[row, j] = oo[k]; inst[row, j] = ii[k]
    weekday = (days + 3) % 7  # epoch day0 = Thursday → 0=Mon..6=Sun
    return roots, days, close, op, inst, weekday


def build_returns(close, op, inst):
    T, N = close.shape
    lc = np.log(close)
    R = np.zeros((T, N)); R[1:] = lc[1:] - lc[:-1]
    same = np.zeros((T, N), bool); same[1:] = inst[1:] == inst[:-1]
    R = np.where(same & np.isfinite(R), R, 0.0)
    ov = np.zeros((T, N)); ov[1:] = np.log(op[1:] / close[:-1])
    ov = np.where(same & np.isfinite(ov), ov, 0.0)
    intr = np.log(close / op); intr = np.where(np.isfinite(intr), intr, 0.0)
    return lc, R, ov, intr


def trailing(lc, L):
    out = np.full_like(lc, np.nan); out[L:] = lc[L:] - lc[:-L]; return out


def xs_weights(sig):  # cross-sectional z-score, market-neutral, unit gross
    mu = np.nanmean(sig, axis=1, keepdims=True); sd = np.nanstd(sig, axis=1, keepdims=True)
    z = np.nan_to_num((sig - mu) / np.where(sd > 0, sd, 1))
    g = np.sum(np.abs(z), axis=1, keepdims=True); g[g == 0] = 1
    return z / g


def pnl_w(w, R, cost_bp=2.0):
    p = np.sum(w[:-1] * R[1:], axis=1)
    turn = np.sum(np.abs(w[1:] - w[:-1]), axis=1)
    return p - turn * (cost_bp / 1e4)


def sharpe_t(x):
    x = x[torch.isfinite(x)]
    if len(x) < 10 or float(x.std()) == 0:
        return float("nan")
    return float(x.mean() / x.std() * math.sqrt(252))


def validate(pnl, days, n_trials):
    x = torch.tensor(np.asarray(pnl), device=DEV, dtype=torch.float64)
    n = len(x); full = sharpe_t(x)
    dd = days[-n:]
    ism = torch.tensor(dd < SPLIT_DAY, device=DEV)
    sis = sharpe_t(x[ism])
    soos = sharpe_t(x[~ism]) if int((~ism).sum()) > 30 else float("nan")
    nb, k = 10, 2
    blk = [torch.arange(i * n // nb, (i + 1) * n // nb, device=DEV) for i in range(nb)]
    oos = [sharpe_t(x[torch.cat([blk[b] for b in c])]) for c in combinations(range(nb), k)]
    oos = np.array([o for o in oos if not math.isnan(o)])
    med = float(np.median(oos)) if len(oos) else float("nan")
    p5 = float(np.percentile(oos, 5)) if len(oos) else float("nan")
    srd = full / math.sqrt(252)
    if n_trials > 1:
        sr0 = (1 / math.sqrt(n)) * ((1 - 0.5772) * ND.inv_cdf(1 - 1.0 / n_trials)
                                    + 0.5772 * ND.inv_cdf(1 - 1.0 / (n_trials * math.e)))
    else:
        sr0 = 0.0
    dsr = ND.cdf((srd - sr0) * math.sqrt(n - 1)) if not math.isnan(full) else float("nan")
    return dict(full=full, is_=sis, oos=soos, med=med, p5=p5, dsr=dsr)


def main():
    roots, days, close, op, inst, weekday = load_panel()
    lc, R, ov, intr = build_returns(close, op, inst)
    T, N = close.shape
    vol63 = np.full_like(lc, np.nan)
    for t in range(63, T):
        vol63[t] = np.nanstd(R[t - 63:t], axis=0)
    invvol = 1.0 / np.where(vol63 > 0, vol63, np.nan)

    sig = {}
    for L in [21, 63, 126, 252]:
        sig[f"XS_mom_{L}"] = pnl_w(xs_weights(trailing(lc, L)), R)
    for L in [1, 5]:
        sig[f"XS_rev_{L}"] = pnl_w(xs_weights(-trailing(lc, L)), R)
    sig["XS_lowvol"] = pnl_w(xs_weights(-vol63), R)
    for L in [21, 63, 252]:
        w = np.nan_to_num(np.sign(trailing(lc, L)) * invvol)
        g = np.sum(np.abs(w), axis=1, keepdims=True); g[g == 0] = 1
        sig[f"TS_mom_{L}"] = pnl_w(w / g, R)
    sig["SEAS_overnight"] = np.nanmean(ov[1:], axis=1) - 1.0 / 1e4   # EW long overnight, ~1bp cost
    sig["SEAS_intraday"] = np.nanmean(intr[1:], axis=1) - 1.0 / 1e4  # EW long intraday
    ewR = np.nanmean(R, axis=1)
    for d, nm in enumerate(["Mon", "Tue", "Wed", "Thu", "Fri"]):
        sig[f"SEAS_{nm}"] = np.where(weekday[1:] == d, ewR[1:], 0.0)

    NT = len(sig)
    rows = [(nm, validate(p, days, NT)) for nm, p in sig.items()]
    rows.sort(key=lambda r: -(r[1]["oos"] if not math.isnan(r[1]["oos"]) else -9))
    print(f"\n===== DEFLATED SIGNAL SWEEP — Batch 1 (22 futures daily + seasonality) =====")
    print(f"device={DEV}  roots={len(roots)}  days={T}  N_trials(deflation)={NT}")
    print(f"{'signal':>16} {'full':>7} {'IS':>7} {'OOS':>7} {'CPCVmed':>8} {'CPCV5%':>8} {'DSR':>6}")
    for nm, v in rows:
        print(f"{nm:>16} {v['full']:>+7.2f} {v['is_']:>+7.2f} {v['oos']:>+7.2f} "
              f"{v['med']:>+8.2f} {v['p5']:>+8.2f} {v['dsr']:>6.2f}")
    surv = [nm for nm, v in rows if v["dsr"] > 0.95 and v["oos"] > 0 and v["med"] > 0]
    print(f"\nSURVIVORS (DSR>0.95 & OOS>0 & CPCVmed>0): {surv if surv else 'NONE'}")
    print("DSR is deflated by N_trials → multiple-testing-honest. Survivors get deeper validation + crypto/COT batches.")


if __name__ == "__main__":
    main()