#!/usr/bin/env python
"""Same late-day momentum study on ES / YM / RTY vs NQ.

Signal: |log(p1530/p1500)| > threshold -> sign, exit 16:00 close.
Thresholds: fixed 0.30% AND percentile-matched (same fire rate as NQ's 0.30% ~ 13.7%).
Costs (round trip, pts, incl. spread+slip, conservative): ES 0.50, NQ 1.00, YM 2.0, RTY 0.30.
$/pt: ES 50, NQ 20, YM 5, RTY 50.
"""
import glob, os
import numpy as np
import pandas as pd

SRCROOT = "<data-root>/fp-data/glbx"
HERE = os.path.dirname(os.path.abspath(__file__))
ET = "America/New_York"
SYMS = {"NQ": (1.00, 20), "ES": (0.50, 50), "YM": (2.00, 5), "RTY": (0.30, 50)}
OOS_START = 2020


def build(sym):
    cache = os.path.join(HERE, f"sessions_{sym}.parquet")
    if os.path.exists(cache):
        return pd.read_parquet(cache)
    frames = []
    for f in sorted(glob.glob(os.path.join(SRCROOT, sym, "1m", "*.csv"))):
        if os.path.getsize(f) < 1000:
            continue
        df = pd.read_csv(f, usecols=["ts_event", "open", "high", "low", "close", "volume"],
                         parse_dates=["ts_event"])
        et = df["ts_event"].dt.tz_convert(ET)
        df["date"] = et.dt.date
        df["tod"] = et.dt.hour * 60 + et.dt.minute
        frames.append(df[(df["tod"] >= 570) & (df["tod"] < 960)])
    allb = pd.concat(frames, ignore_index=True)
    rows = []
    for date, day in allb.groupby("date", sort=True):
        day = day.set_index("tod").sort_index()
        if len(day) < 330:
            continue
        def px(tod):
            sel = day.loc[day.index <= tod - 1]
            return sel["close"].iloc[-1] if len(sel) else np.nan
        rows.append(dict(date=pd.Timestamp(date), p1500=px(900), p1530=px(930), p1600=px(960),
                         volume_last30=day.loc[(day.index >= 930)]["volume"].sum()))
    out = pd.DataFrame(rows).sort_values("date").reset_index(drop=True)
    out.to_parquet(cache)
    return out


def line(pl_r, pl_p, cost, name):
    n = len(pl_r)
    if n < 30:
        return f"    {name:34s} n={n} (too few)"
    m = pl_r.mean(); t = m / (pl_r.std(ddof=1) / np.sqrt(n))
    return (f"    {name:34s} n={n:4d} {m*1e4:+6.2f}bps t={t:+5.2f} hit={(pl_r>0).mean():.3f} "
            f"net={pl_p.mean()-cost:+6.2f}pt")


nq_fire_rate = None
summary = []
for sym, (cost, dollar_pt) in SYMS.items():
    s = build(sym).dropna(subset=["p1500", "p1530", "p1600"]).copy()
    s["year"] = s["date"].dt.year
    r_sig = np.log(s.p1530 / s.p1500)
    label = np.log(s.p1600 / s.p1530)
    pts = s.p1600 - s.p1530
    side = np.sign(r_sig)

    print(f"\n=== {sym}  (sessions={len(s)} {s.date.min().date()}->{s.date.max().date()}, "
          f"cost={cost}pt, ${dollar_pt}/pt) ===")
    masks = {"fixed |r|>0.30%": np.abs(r_sig) > 0.003}
    if sym == "NQ":
        nq_fire_rate = masks["fixed |r|>0.30%"].mean()
    else:
        thr = np.abs(r_sig).quantile(1 - nq_fire_rate)
        masks[f"matched-rate |r|>{thr:.3%}"] = np.abs(r_sig) > thr
    for nm, mask in masks.items():
        pl_r, pl_p = (side * label)[mask], (side * pts)[mask]
        print(line(pl_r, pl_p, cost, nm + " ALL"))
        for eralab, em in (("IS<2020", mask & (s.year < OOS_START)),
                           ("OOS>=2020", mask & (s.year >= OOS_START)),
                           ("2023-2026", mask & (s.year >= 2023))):
            print(line((side * label)[em], (side * pts)[em], cost, "  " + eralab))
        if "matched" in nm or sym == "NQ":
            d = pd.DataFrame({"y": s.year, "r": side * label})[mask]
            yr = d.groupby("y")["r"].mean()
            net_bps = (pl_p.mean() - cost) / s.p1530[mask].mean() * 1e4
            usd = (pl_p.mean() - cost) * dollar_pt
            summary.append((sym, nm, len(pl_r), pl_r.mean()*1e4,
                            pl_r.mean()/(pl_r.std(ddof=1)/np.sqrt(len(pl_r))),
                            net_bps, usd, f"{(yr>0).sum()}/{len(yr)}"))

print("\n=== SUMMARY (net of per-symbol costs) ===")
print(f"{'sym':4s} {'rule':26s} {'n':>5s} {'gross bps':>9s} {'t':>6s} {'net bps':>8s} {'net $/trade':>11s} {'yrs+':>6s}")
for sym, nm, n, bps, t, netb, usd, yrs in summary:
    print(f"{sym:4s} {nm:26s} {n:5d} {bps:+9.2f} {t:+6.2f} {netb:+8.2f} {usd:+11.0f} {yrs:>6s}")

# signal-day overlap NQ vs ES
nq = build("NQ").dropna(subset=["p1500","p1530","p1600"]).set_index("date")
es = build("ES").dropna(subset=["p1500","p1530","p1600"]).set_index("date")
common = nq.index.intersection(es.index)
rn = np.log(nq.p1530/nq.p1500)[common]; re = np.log(es.p1530/es.p1500)[common]
mn = np.abs(rn) > 0.003
print(f"\nNQ-signal days where ES same-sign moved: {(np.sign(rn)[mn]==np.sign(re)[mn]).mean()*100:.0f}%  "
      f"(signal corr {rn.corr(re):.2f})")
# NQ signal traded on ES (cross)
lab_e = np.log(es.p1600/es.p1530)[common]; pts_e = (es.p1600-es.p1530)[common]
pl = (np.sign(rn)*lab_e)[mn]; plp = (np.sign(rn)*pts_e)[mn]
t = pl.mean()/(pl.std(ddof=1)/np.sqrt(len(pl)))
print(f"NQ signal -> trade ES: n={len(pl)} {pl.mean()*1e4:+.2f}bps t={t:+.2f} net={plp.mean()-0.5:+.2f}pt (${(plp.mean()-0.5)*50:+.0f}/trade)")
EOF_MARKER_NOT_USED = None