#!/usr/bin/env python3
"""MFT horizon sweep — does a SLOWER (lower-turnover) intraday trend edge clear costs?

Prong A (mft_quality_test) showed: at 15-min holds the net ≈ -1 tick = the cost itself (gross ≈ 0).
This sweeps the hold horizon UP (15m → full session) on clean ES OHLCV-1m. As the hold grows the
forward move (in ticks) grows ~√time while the 1-tick cost stays fixed, so cost/gross shrinks — this
isolates the question the user posed: keep costs low by trading SLOWER. If gross edge grows with
horizon (trend persistence), there's an MFT sweet spot; if gross stays ~0 everywhere, the price signal
is dead regardless of speed.

Pre-registered (NOT tuned): trend-follow, lookback = hold = H bars (5-min each); direction = sign(ret_H);
conviction = |t-stat| = |ret_H| / (per-bar vol · √H); entry close[t], exit close[t+H], intraday only
(same-day, flat-by-close); cost = 1.0 tick round-trip (crossing). Quality curve by conviction decile,
IS(first 70%) / OOS(last 30%). PASS bar: OOS top-decile net > 0 ticks & hit > 50% & |t| ≥ 2, at a
horizon with low trades/day (genuinely MFT).
"""
import glob
import math

import numpy as np
import torch

DEV = "cuda" if torch.cuda.is_available() else None
TICK = 0.25
COST_TICKS = 1.0
BAR_NS = 5 * 60 * 10**9
DAY_NS = 86_400 * 10**9
HOLDS = [3, 12, 24, 48, 78]   # 15m, 1h, 2h, 4h, ~full session (5-min bars)


def load_es_5min():
    import databento as db
    es1m = sorted(glob.glob("data/surfer/es1m/*.dbn"))
    ts_all, c_all = [], []
    if es1m:
        for p in es1m:
            try:
                arr = db.DBNStore.from_file(p).to_ndarray()
            except Exception:
                continue
            if len(arr) == 0 or "close" not in arr.dtype.names:
                continue
            ts = arr["ts_event"].astype(np.int64)
            c = arr["close"].astype(np.float64) / 1e9
            m = c > 0
            ts_all.append(ts[m]); c_all.append(c[m])
    else:
        for p in sorted(glob.glob("test_data/futures-baseline/ES.FUT/*.dbn.zst")):
            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]
            if df.empty:
                continue
            dom = df["instrument_id"].value_counts().idxmax()
            df = df[df["instrument_id"] == dom].sort_values("ts_event")
            ts_all.append(df["ts_event"].astype("int64").to_numpy())
            c_all.append(df["close"].to_numpy(np.float64))
    ts = np.concatenate(ts_all); c = np.concatenate(c_all)
    _u, _ui = np.unique(ts, return_index=True)
    ts, c = ts[_ui], c[_ui]
    o = np.argsort(ts, kind="stable"); ts, c = ts[o], c[o]
    b5 = ts // BAR_NS
    _, first = np.unique(b5, return_index=True)
    last = np.append(first[1:] - 1, len(b5) - 1)
    return ts[last], c[last]


def rolling_std(x, w):
    z = torch.zeros(1, dtype=x.dtype, device=x.device)
    cs = torch.cat([z, torch.cumsum(x, 0)])
    cs2 = torch.cat([z, torch.cumsum(x * x, 0)])
    s = cs[w:] - cs[:-w]
    s2 = cs2[w:] - cs2[:-w]
    mean = s / w
    var = (s2 / w - mean * mean).clamp_min(0.0)
    out = torch.full_like(x, float("nan"))
    out[w - 1:] = var.sqrt()
    return out


def run():
    ts5, close5 = load_es_5min()
    day = ts5 // DAY_NS
    n_days = int(day.max() - day.min())
    lc = torch.tensor(np.log(close5), device=DEV, dtype=torch.float64)
    px = torch.tensor(close5, device=DEV, dtype=torch.float64)
    dy = torch.tensor(day, device=DEV)
    T = len(lc)
    t = torch.arange(T, device=DEV)
    split = int(0.7 * T)

    def sameday(lag):
        m = torch.zeros(T, dtype=torch.bool, device=DEV)
        m[lag:] = dy[lag:] == dy[:-lag]
        return m

    r1 = torch.zeros_like(lc); r1[1:] = lc[1:] - lc[:-1]
    r1 = torch.where(sameday(1), r1, torch.zeros_like(r1))
    vol1 = rolling_std(r1, 12)   # per-bar 5-min vol (fixed 1h window, pre-registered)

    print("\n================ MFT HORIZON SWEEP (GPU, intraday trend, flat-by-close) ================")
    print(f"device={DEV}  5-min bars={T}  span days≈{n_days}  cost={COST_TICKS} tick round-trip")
    print("trend-follow: dir=sign(ret_H), conv=|ret_H|/(vol·√H). PASS: OOS top-decile net>0 & hit>50% & |t|≥2\n")
    print(f"{'hold':>6} {'top-q':>6} {'n_OOS':>7} {'gross_t':>8} {'net_t':>8} {'cost/gr':>8} {'hit%':>6} {'t-stat':>7} {'trd/day':>8}")
    print("-" * 78)

    for H in HOLDS:
        retH = torch.full_like(lc, float("nan")); retH[H:] = lc[H:] - lc[:-H]
        tstat = retH / (vol1 * math.sqrt(H)).clamp_min(1e-9)
        fwd_ticks = torch.full_like(px, float("nan"))
        fwd_ticks[:-H] = (px[H:] - px[:-H]) / TICK
        fwd_valid = torch.zeros(T, dtype=torch.bool, device=DEV)
        fwd_valid[:-H] = dy[H:] == dy[:-H]                       # exit same day (flat-by-close)
        valid = fwd_valid & sameday(H) & torch.isfinite(tstat) & torch.isfinite(vol1) & (vol1 > 0)
        direction = torch.sign(retH)
        net = direction * fwd_ticks - COST_TICKS
        gross = direction * fwd_ticks
        cand = valid & (direction != 0)
        idx = torch.nonzero(cand, as_tuple=True)[0]
        if len(idx) < 100:
            print(f"{H:>6} (too few candidates: {len(idx)})"); continue
        conv = tstat.abs()[idx]; tt = t[idx]
        order = torch.argsort(conv, descending=True)
        hold_hours = H * 5 / 60.0
        for q in [0.05, 0.10, 1.00]:
            k = max(int(q * len(idx)), 1)
            sel = idx[order[:k]]
            oos = sel[tt[order[:k]] >= split]
            no = int(len(oos))
            if no < 20:
                print(f"{H:>6} {q:>6.2f}  (OOS n<20)"); continue
            g = gross[oos]; nt = net[oos]
            gm = float(g.mean()); nm = float(nt.mean())
            tval = nm / (float(nt.std()) / math.sqrt(no) + 1e-12)
            hit = 100.0 * float((nt > 0).float().mean())
            cost_gr = (COST_TICKS / gm * 100.0) if gm > 0 else float("nan")
            trd_day = no / (0.3 * n_days)                        # OOS is last 30% of days
            tag = f"{H}({hold_hours:.1f}h)" if q == 0.05 else ""
            print(f"{tag:>6} {q:>6.2f} {no:>7} {gm:>+8.3f} {nm:>+8.3f} {cost_gr:>7.0f}% {hit:>6.1f} {tval:>+7.2f} {trd_day:>8.2f}")
        print("-" * 78)


def main():
    if DEV is None:
        raise SystemExit("CUDA not available")
    run()


if __name__ == "__main__":
    main()