"""Dukascopy XAU/USD tick download + bid/ask spread study.

Validates ForexBot2's $0.30 round-trip cost assumption (config:
SignalConfig.round_trip_cost / PAIR_INFO default_spread). Round-trip
cost of a market entry+exit = one ask-bid spread width (buy at ask,
sell at bid), so the assumption holds iff typical spread ~ $0.30.

bi5 format: LZMA 'alone' stream of 20-byte big-endian records
(ms_in_hour uint32, ask uint32, bid uint32, ask_vol f32, bid_vol f32);
XAUUSD prices scaled x1000. URL months are 0-indexed.
"""
import lzma
import struct
import sys
from concurrent.futures import ThreadPoolExecutor
from datetime import date, timedelta
from pathlib import Path

import numpy as np
import pandas as pd
import requests

SCALE = 1000.0
CACHE = Path(__file__).parent / "tick_cache"
CACHE.mkdir(exist_ok=True)
URL = "https://datafeed.dukascopy.com/datafeed/XAUUSD/{y}/{m:02d}/{d:02d}/{h:02d}h_ticks.bi5"

# Bot's tradable bar-close hours: session 03:00-23:00 minus XAU blocked
# hours [0,1,2,10,11,13,16,21] (M5 additionally blocks 9 — reported
# separately below).
BLOCKED = {0, 1, 2, 10, 11, 13, 16, 21}
TRADABLE = sorted(set(range(3, 23)) - BLOCKED)

# ── Day sample ────────────────────────────────────────────────────────
recent = [date(2026, 7, 6) + timedelta(days=i) for i in range(17)]
recent = [d for d in recent if d.weekday() < 5]              # Mon-Fri Jul 6-22
monthly = [date(2025, 8, 13), date(2025, 9, 10), date(2025, 10, 15),
           date(2025, 11, 12), date(2025, 12, 10), date(2026, 1, 14),
           date(2026, 2, 11), date(2026, 3, 11), date(2026, 4, 15),
           date(2026, 5, 13), date(2026, 6, 10)]             # mid-month Wednesdays
nfp = [date(2026, 6, 5), date(2026, 7, 3)]                   # NFP Fridays
sundays = [date(2026, 7, 12), date(2026, 7, 19)]             # Sun 22-24 open
DAYS = sorted(set(recent + monthly + nfp + sundays))
assert all(d.weekday() < 5 for d in recent + monthly + nfp)


_tls = __import__("threading").local()


def _session() -> requests.Session:
    if not hasattr(_tls, "s"):
        _tls.s = requests.Session()
        _tls.s.headers["User-Agent"] = "Mozilla/5.0 (research; spread-study)"
    return _tls.s


def fetch(day: date, hour: int) -> Path:
    p = CACHE / f"{day.isoformat()}_{hour:02d}.bi5"
    if p.exists():
        return p
    u = URL.format(y=day.year, m=day.month - 1, d=day.day, h=hour)
    last = None
    for attempt in range(4):
        try:
            r = _session().get(u, timeout=30)
            if r.status_code != 200:
                raise RuntimeError(f"{u} -> HTTP {r.status_code}")
            p.write_bytes(r.content)
            return p
        except (requests.ConnectionError, requests.Timeout, RuntimeError) as exc:
            last = exc
            __import__("time").sleep(1.5 * (attempt + 1))
            if hasattr(_tls, "s"):
                _tls.s.close()
                del _tls.s
    raise RuntimeError(f"giving up on {u}: {last}")


def decode(p: Path, day: date, hour: int) -> pd.DataFrame:
    blob = p.read_bytes()
    if not blob:
        return pd.DataFrame()
    raw = lzma.decompress(blob)
    n = len(raw) // 20
    a = np.frombuffer(raw[: n * 20], dtype=">u4").reshape(n, 5)
    f = np.frombuffer(raw[: n * 20], dtype=">f4").reshape(n, 5)
    df = pd.DataFrame({
        "ms": a[:, 0].astype(np.int64),
        "ask": a[:, 1] / SCALE,
        "bid": a[:, 2] / SCALE,
    })
    df["spread"] = df["ask"] - df["bid"]
    df["date"], df["hour"] = day.isoformat(), hour
    df["weekday"] = day.weekday()
    return df


def load_all() -> pd.DataFrame:
    jobs = [(d, h) for d in DAYS for h in range(24)]
    with ThreadPoolExecutor(max_workers=3) as ex:
        paths = list(ex.map(lambda j: fetch(*j), jobs))
    frames = [decode(p, d, h) for p, (d, h) in zip(paths, jobs)]
    df = pd.concat([f for f in frames if not f.empty], ignore_index=True)
    return df


def wstats(g: pd.DataFrame) -> pd.Series:
    """Tick stats + time-weighted mean (each tick's spread held until the
    next tick; last tick held to hour end)."""
    tw_num = tw_den = 0.0
    keys = [k for k in ("date", "hour") if k in g.columns]
    for _, grp in g.groupby(keys, sort=False):
        ms = grp["ms"].to_numpy()
        sp = grp["spread"].to_numpy()
        w = np.diff(np.append(ms, 3_600_000)).clip(min=0)
        tw_num += float((sp * w).sum())
        tw_den += float(w.sum())
    s = g["spread"]
    return pd.Series({
        "ticks": len(s),
        "tw_mean": tw_num / tw_den if tw_den else np.nan,
        "mean": s.mean(), "median": s.median(),
        "p75": s.quantile(0.75), "p90": s.quantile(0.90),
        "p99": s.quantile(0.99), "max": s.max(),
        "pct_le_030": (s <= 0.30).mean() * 100,
        "pct_ge_060": (s >= 0.60).mean() * 100,
    })


def m5_boundary_spreads(df: pd.DataFrame) -> pd.DataFrame:
    """Spread of the FIRST tick at/after each M5 boundary — what a
    next-open fill actually pays."""
    rows = []
    for (day, hour), grp in df.groupby(["date", "hour"], sort=False):
        ms = grp["ms"].to_numpy()
        sp = grp["spread"].to_numpy()
        wd = grp["weekday"].iat[0]
        for b in range(0, 3_600_000, 300_000):
            i = np.searchsorted(ms, b)
            if i < len(ms) and ms[i] - b <= 30_000:   # tick within 30 s
                rows.append((day, wd, hour, b // 60_000, sp[i]))
    return pd.DataFrame(rows, columns=["date", "weekday", "hour", "minute", "spread"])


def main() -> None:
    df = load_all()
    print(f"days={df['date'].nunique()}  ticks={len(df):,}  "
          f"price range {df['bid'].min():.0f}-{df['ask'].max():.0f}")

    print("\n── Overall (all sampled hours) ──")
    print(wstats(df).round(3).to_string())

    trad = df[df["hour"].isin(TRADABLE) & (df["weekday"] < 5)]
    print("\n── Bot-tradable hours (session 03-23 minus blocked, Mon-Fri) ──")
    print(wstats(trad).round(3).to_string())

    roll = df[df["hour"].isin([21, 22, 23])]
    print("\n── Rollover window 21-24 UTC ──")
    print(wstats(roll).round(3).to_string())

    print("\n── Per UTC hour ──")
    per_hour = df[df["weekday"] < 5].groupby("hour").apply(wstats, include_groups=False).round(3)
    per_hour["tradable"] = ["Y" if h in TRADABLE else "-" for h in per_hour.index]
    print(per_hour.to_string())
    per_hour.to_csv(Path(__file__).parent / "duka_spread_by_hour.csv")

    print("\n── M5 next-open boundary fills ──")
    mb = m5_boundary_spreads(df)
    mb_trad = mb[mb["hour"].isin(TRADABLE) & (mb["weekday"] < 5)]
    for name, d in [("all hours", mb), ("tradable hours", mb_trad)]:
        s = d["spread"]
        print(f"{name}: n={len(s):,} mean={s.mean():.3f} median={s.median():.3f} "
              f"p90={s.quantile(0.9):.3f} p99={s.quantile(0.99):.3f} "
              f"%<=0.30={(s <= 0.30).mean()*100:.1f} %>=0.60={(s >= 0.60).mean()*100:.1f}")
    mb.to_csv(Path(__file__).parent / "duka_m5_boundary_spreads.csv", index=False)

    print("\n── By month (tradable hours, tick median / tw_mean) ──")
    trad2 = trad.copy()
    trad2["month"] = trad2["date"].str[:7]
    bym = trad2.groupby("month").apply(wstats, include_groups=False).round(3)
    print(bym[["ticks", "tw_mean", "median", "p90", "pct_le_030", "pct_ge_060"]].to_string())

    print("\n── NFP Fridays 12:00-15:00 UTC vs same hours other days ──")
    nfp_days = {"2026-06-05", "2026-07-03"}
    win = df[df["hour"].isin([12, 13, 14])]
    for name, d in [("NFP", win[win["date"].isin(nfp_days)]),
                    ("non-NFP", win[~win["date"].isin(nfp_days) & (win["weekday"] < 5)])]:
        s = d["spread"]
        print(f"{name}: n={len(s):,} tick-median={s.median():.3f} p90={s.quantile(0.9):.3f} "
              f"p99={s.quantile(0.99):.3f} max={s.max():.2f}")


if __name__ == "__main__":
    sys.exit(main())