#!/usr/bin/env python3
"""Data layer for forex / gold backtests (first user: the topg-audit project).

Generalises tools/ib_data.py (which stays untouched for the umar-audit backtest) from one index CFD
to any Dukascopy instrument that trades ~24/5: spot gold (XAUUSD) and the major FX pairs.

Source: Dukascopy Bank's free historical datafeed (datafeed.dukascopy.com), BID and ASK 1-minute
candles, one LZMA .bi5 file per UTC calendar day and side. Raw files are cached byte-for-byte under
projects/<project>/data_raw/dukascopy/<SYMBOL>/YYYY/MM/DD/{BID,ASK}_candles_min_1.bi5 (the month in
the path is 0-based, exactly as in the feed). Prices are never edited.

Record layout (verified on XAUUSD/EURUSD/USDJPY files, May 2024): big-endian int32
[seconds from 00:00 UTC, open, close, low, high] + float32 volume; prices in integer units of
1/PRICE_SCALE (XAUUSD and JPY pairs 1e3, other majors 1e5). Dukascopy writes 1,440 rows for every
day, including weekends and the gold daily break; minutes without any tick are flat filler rows with
zero volume.

Time handling (explicit):
  * Everything is stored and processed in UTC (minute bars labelled by their START time).
  * The FX "trading day" is the New York 17:00 -> 17:00 day (the market convention used by MT4/MT5
    brokers on NY-close server time and by TradingView's FX daily candles). `fx_day` = the NY date of
    (bar time + 7 h), so Sunday-evening bars belong to Monday's day.
  * Hourly candles are the same in UTC, New York, broker server time (GMT+2/+3) or IST display,
    because those offsets are whole hours or (IST) only relabel TradingView's UTC-aligned bars.
    4-hour and daily candles DO depend on the alignment: `resample()` aligns them to 17:00 New York.
  * Weekends: Saturday UTC files are pure filler and are not fetched. Minutes with zero volume on
    both sides (no ticks: weekend, gold's daily 21:00/22:00 UTC break, holidays, feed gaps) are
    dropped from the processed data, so resampled candles look like a broker chart (no flat candles).
    Every trading day's active-minute count and every intraday gap > 15 min is written to the gap
    report, so nothing is dropped silently.

Usage:
  .venv/bin/python tools/fx_data.py fetch  --symbols XAUUSD,EURUSD [--start 2015-01-01 --end 2026-09-30 --threads 16]
  .venv/bin/python tools/fx_data.py status --symbols XAUUSD
  .venv/bin/python tools/fx_data.py build  --symbols XAUUSD      # -> data_raw/<SYM>_m1.npz + <SYM>_gaps.json
  .venv/bin/python tools/fx_data.py sizes                        # disk used under data_raw

In code:
  from fx_data import load_m1, resample
  m1 = load_m1("XAUUSD")            # DataFrame, UTC index, float prices: bo bh bl bc ao ah al ac vb va
  h1 = resample(m1, "1h")           # hourly BID+ASK candles from active minutes
"""
from __future__ import annotations

import argparse
import concurrent.futures as cf
import json
import lzma
import os
import random
import sys
import threading
import time
import urllib.request
from datetime import date, datetime, timedelta, timezone
from pathlib import Path

import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parents[1]
PROJECT = os.environ.get("FX_PROJECT", "topg-audit")
RAW = ROOT / "projects" / PROJECT / "data_raw"
DUKA_ROOT = RAW / "dukascopy"
DUKA_URL = "https://datafeed.dukascopy.com/datafeed"
NY = "America/New_York"
DEFAULT_START, DEFAULT_END = "2015-01-01", "2026-09-30"

# integer price units per 1.0 of price in the .bi5 files (checked against known levels)
PRICE_SCALE = {
    "XAUUSD": 1e3, "XAGUSD": 1e3,
    "EURUSD": 1e5, "GBPUSD": 1e5, "AUDUSD": 1e5, "NZDUSD": 1e5, "USDCAD": 1e5, "USDCHF": 1e5,
    "EURGBP": 1e5, "EURCHF": 1e5, "EURAUD": 1e5,
    "USDJPY": 1e3, "EURJPY": 1e3, "GBPJPY": 1e3, "AUDJPY": 1e3,
}
# one pip in price units (for reporting costs in pips; gold: $0.10 is often called a "pip" by retail,
# but we report gold in dollars per ounce and keep pip = 0.1 only as a label)
PIP = {s: (0.01 if s.endswith("JPY") else 0.1 if s.startswith("XA") else 0.0001) for s in PRICE_SCALE}


def raw_dir(symbol: str) -> Path:
    return DUKA_ROOT / symbol


def rel_path(d: date, side: str) -> str:
    return f"{d.year}/{d.month - 1:02d}/{d.day:02d}/{side}_candles_min_1.bi5"


def fetch_days(start: str, end: str) -> list[date]:
    """Every UTC calendar day except Saturdays (FX/gold are shut from Fri ~21:00 to Sun ~21:00 UTC)."""
    d0, d1 = pd.Timestamp(start).date(), pd.Timestamp(end).date()
    out, d = [], d0
    while d <= d1:
        if d.weekday() != 5:
            out.append(d)
        d += timedelta(days=1)
    return out


# ------------------------------------------------------------------ fetch ---
_lock = threading.Lock()
_stats = {"ok": 0, "cached": 0, "fail": 0, "retries": 0}


def fetch_one(symbol: str, rel: str, max_attempts: int = 40) -> bool:
    path = raw_dir(symbol) / rel
    if path.exists():
        with _lock:
            _stats["cached"] += 1
        return True
    url = f"{DUKA_URL}/{symbol}/{rel}"
    for attempt in range(max_attempts):
        try:
            req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
            with urllib.request.urlopen(req, timeout=120) as r:
                body = r.read()
            if body:
                lzma.decompress(body)  # validate before caching
            path.parent.mkdir(parents=True, exist_ok=True)
            tmp = path.with_suffix(f".part{os.getpid()}_{threading.get_ident()}")
            tmp.write_bytes(body)
            tmp.rename(path)
            with _lock:
                _stats["ok"] += 1
            return True
        except urllib.error.HTTPError as e:
            if e.code == 404:  # no file for that day (e.g. before the instrument existed)
                path.parent.mkdir(parents=True, exist_ok=True)
                (path.parent / (path.name + ".404")).write_text("404\n")
                with _lock:
                    _stats["fail"] += 1
                return False
            with _lock:
                _stats["retries"] += 1
            time.sleep(min(60, 2 + attempt * 2) * random.uniform(0.5, 1.5))
        except Exception:  # 503 "no server available" / timeouts are common on the free feed
            with _lock:
                _stats["retries"] += 1
            time.sleep(min(60, 2 + attempt * 2) * random.uniform(0.5, 1.5))
    with _lock:
        _stats["fail"] += 1
    print(f"FAILED {url}", flush=True)
    return False


def symbols_arg(a) -> list[str]:
    syms = [s.strip().upper() for s in a.symbols.split(",") if s.strip()]
    for s in syms:
        if s not in PRICE_SCALE:
            raise SystemExit(f"unknown symbol {s}; add its price scale to PRICE_SCALE after checking a file")
    return syms


def cmd_fetch(a) -> None:
    days = fetch_days(a.start, a.end)
    for sym in symbols_arg(a):
        # all BID files first, then ASK (BID alone already allows a run with a modelled spread)
        jobs = [rel_path(d, s) for s in ("BID", "ASK") for d in days]
        todo = [j for j in jobs if not (raw_dir(sym) / j).exists()]
        if a.reverse:
            todo = [j for j in reversed(todo) if "BID_" in j] + [j for j in reversed(todo) if "ASK_" in j]
        if a.shuffle:  # a 3rd worker: random order, so it rarely collides with the forward/reverse ones
            random.Random(a.shuffle).shuffle(todo)
        print(f"[{sym}] {len(jobs)} files for {len(days)} days, {len(todo)} to download", flush=True)
        t0 = time.time()
        with cf.ThreadPoolExecutor(a.threads) as ex:
            futs = [ex.submit(fetch_one, sym, j) for j in todo]
            for i, _ in enumerate(cf.as_completed(futs), 1):
                if i % 200 == 0 or i == len(todo):
                    el = time.time() - t0
                    print(f"[{sym}] {i}/{len(todo)} in {el/60:.1f} min ({i/max(el,1e-9):.2f}/s) {_stats}", flush=True)
        print(f"[{sym}] fetch finished {_stats}", flush=True)


def cmd_status(a) -> None:
    days = fetch_days(a.start, a.end)
    for sym in symbols_arg(a):
        n = miss = 0
        for d in days:
            for s in ("BID", "ASK"):
                n += 1
                miss += not (raw_dir(sym) / rel_path(d, s)).exists()
        print(f"{sym}: {n - miss}/{n} cached, {miss} missing")


def cmd_sizes(a) -> None:
    tot = 0
    for p in sorted(RAW.glob("*")):
        if p.is_dir():
            for q in sorted(p.glob("*")):
                sz = sum(f.stat().st_size for f in q.rglob("*") if f.is_file())
                tot += sz
                print(f"{sz/1e6:9.1f} MB  {q.relative_to(RAW)}")
        else:
            tot += p.stat().st_size
            print(f"{p.stat().st_size/1e6:9.1f} MB  {p.name}")
    print(f"{tot/1e6:9.1f} MB  total under {RAW.relative_to(ROOT)}")


# ------------------------------------------------------------------ parse ---
def read_day_raw(symbol: str, d: date, side: str) -> np.ndarray | None:
    """(n, 6) int64 array [unix_seconds, o, h, l, c, vol*1e6] or None; prices stay integer units."""
    path = raw_dir(symbol) / rel_path(d, side)
    if not path.exists():
        return None
    body = path.read_bytes()
    if not body:
        return np.zeros((0, 6), dtype=np.int64)
    a = np.frombuffer(lzma.decompress(body), dtype=">i4").reshape(-1, 6)
    base = int(datetime(d.year, d.month, d.day, tzinfo=timezone.utc).timestamp())
    out = np.empty((len(a), 6), dtype=np.int64)
    out[:, 0] = base + a[:, 0].astype(np.int64)
    out[:, 1] = a[:, 1]  # open
    out[:, 2] = a[:, 4]  # high
    out[:, 3] = a[:, 3]  # low
    out[:, 4] = a[:, 2]  # close
    out[:, 5] = np.round(a[:, 5].copy().view(">f4").astype(np.float64) * 1e6).astype(np.int64)
    return out


def m1_path(symbol: str) -> Path:
    return RAW / f"{symbol}_m1.npz"


def gaps_path(symbol: str) -> Path:
    return RAW / f"{symbol}_gaps.json"


def cmd_build(a) -> None:
    days = fetch_days(a.start, a.end)
    for sym in symbols_arg(a):
        rows, missing_bid, missing_ask = [], [], []
        for d in days:
            b = read_day_raw(sym, d, "BID")
            k = read_day_raw(sym, d, "ASK")
            if b is None:
                missing_bid.append(str(d))
                continue
            if k is None:
                missing_ask.append(str(d))
                continue
            if len(b) != len(k) or (len(b) and not np.array_equal(b[:, 0], k[:, 0])):
                # align on timestamps (never seen so far, but be safe)
                kb = pd.DataFrame(k, columns=list("tohlcv")).set_index("t")
                bb = pd.DataFrame(b, columns=list("tohlcv")).set_index("t")
                j = bb.join(kb, lsuffix="b", rsuffix="a", how="inner")
                b = np.column_stack([j.index.to_numpy(), j[["ob", "hb", "lb", "cb", "vb"]].to_numpy()])
                k = np.column_stack([j.index.to_numpy(), j[["oa", "ha", "la", "ca", "va"]].to_numpy()])
            active = (b[:, 5] > 0) | (k[:, 5] > 0)
            if active.any():
                rows.append(np.column_stack([b[active], k[active][:, 1:]]))
        if not rows:
            print(f"[{sym}] nothing cached")
            continue
        A = np.concatenate(rows)
        order = np.argsort(A[:, 0], kind="stable")
        A = A[order]
        t = A[:, 0]
        assert (np.diff(t) > 0).all(), "duplicate minutes"
        # columns: t, bo bh bl bc bv, ao ah al ac av
        np.savez_compressed(
            m1_path(sym), t=t.astype(np.int64),
            bid=A[:, 1:5].astype(np.int32), ask=A[:, 6:10].astype(np.int32),
            vb=(A[:, 5] / 1e6).astype(np.float32), va=(A[:, 10] / 1e6).astype(np.float32),
            scale=np.array([PRICE_SCALE[sym]]),
        )
        info = gap_report(sym, t, A)
        info.update({"missing_bid_files": missing_bid, "missing_ask_files": missing_ask})
        gaps_path(sym).write_text(json.dumps(info, indent=1))
        print(f"[{sym}] {len(t):,} active minutes, {t[0]} .. {t[-1]} -> {m1_path(sym).relative_to(ROOT)} "
              f"({m1_path(sym).stat().st_size/1e6:.1f} MB); missing BID files {len(missing_bid)}, ASK {len(missing_ask)}")
        print(json.dumps({k: v for k, v in info.items() if not isinstance(v, list)}, indent=1))


def gap_report(sym: str, t: np.ndarray, A: np.ndarray) -> dict:
    ts = pd.to_datetime(t, unit="s", utc=True)
    fxday = (ts.tz_convert(NY) + pd.Timedelta(hours=7)).normalize().tz_localize(None)
    per_day = pd.Series(1, index=fxday).groupby(level=0).sum()
    per_day = per_day[per_day.index.dayofweek < 5]  # Mon-Fri FX days (Sun evening belongs to Monday)
    dt = np.diff(t) // 60
    big = np.where(dt > 15)[0]
    gaps = []
    for i in big:
        a_, b_ = ts[i], ts[i + 1]
        an = a_.tz_convert(NY)
        weekend = an.dayofweek == 4 and an.hour >= 16 and dt[i] > 60 * 40
        # the daily maintenance break (gold ~17:00-18:00 NY, FX a few minutes around 17:00 NY)
        daily_break = (an.hour == 16 and an.minute >= 50 or an.hour == 17 and an.minute <= 10) and dt[i] <= 75
        if weekend or daily_break:
            continue
        gaps.append({"from_utc": str(a_), "to_utc": str(b_), "minutes": int(dt[i] - 1)})
    # median spread per year in price units
    spread = (A[:, 9] - A[:, 4]) / PRICE_SCALE[sym]
    yr = ts.year
    sp = pd.Series(spread).groupby(np.asarray(yr)).median()
    px = pd.Series(A[:, 4] / PRICE_SCALE[sym]).groupby(np.asarray(yr)).median()
    short_days = per_day[per_day < 0.8 * per_day.median()]
    return {
        "symbol": sym, "source": "Dukascopy Bank historical datafeed (datafeed.dukascopy.com), BID+ASK 1-minute",
        "first_utc": str(ts[0]), "last_utc": str(ts[-1]), "active_minutes": int(len(t)),
        "fx_days": int(len(per_day)), "median_active_minutes_per_fx_day": int(per_day.median()),
        "median_spread_by_year": {str(k): round(float(v), 5) for k, v in sp.items()},
        "median_close_by_year": {str(k): round(float(v), 5) for k, v in px.items()},
        "intraday_gaps_over_15min_count": len(gaps),
        "short_fx_days_count": int(len(short_days)),
        "_note": "fx_day = New York 17:00->17:00 day labelled by its NY end date. Gold's daily break (~1 h at "
                 "17:00 NY) and weekend closes are excluded from 'gaps'. Short days = under 80% of the median "
                 "active minutes (holidays, half days, feed outages).",
        "short_fx_days": [{"day": str(d.date()), "active_minutes": int(n)} for d, n in short_days.items()],
        "intraday_gaps_over_15min": gaps,
    }


# ------------------------------------------------------------------ load ---
def load_m1(symbol: str, start: str | None = None, end: str | None = None) -> pd.DataFrame:
    """Active 1-minute bars, UTC index (bar start), float prices bo bh bl bc ao ah al ac, volumes vb va."""
    z = np.load(m1_path(symbol))
    scale = float(z["scale"][0])
    idx = pd.to_datetime(z["t"], unit="s", utc=True)
    bid, ask = z["bid"].astype(np.float64) / scale, z["ask"].astype(np.float64) / scale
    df = pd.DataFrame({"bo": bid[:, 0], "bh": bid[:, 1], "bl": bid[:, 2], "bc": bid[:, 3],
                       "ao": ask[:, 0], "ah": ask[:, 1], "al": ask[:, 2], "ac": ask[:, 3],
                       "vb": z["vb"], "va": z["va"]}, index=idx)
    df.index.name = "t"
    # round away float noise from the integer->float division
    dec = int(round(np.log10(scale)))
    df[["bo", "bh", "bl", "bc", "ao", "ah", "al", "ac"]] = df[["bo", "bh", "bl", "bc", "ao", "ah", "al", "ac"]].round(dec)
    if start:
        df = df[df.index >= pd.Timestamp(start, tz="UTC")]
    if end:
        df = df[df.index < pd.Timestamp(end, tz="UTC") + pd.Timedelta(days=1)]
    return df


def fx_day(idx: pd.DatetimeIndex) -> pd.DatetimeIndex:
    """NY 17:00 -> 17:00 trading day, labelled by the NY date it ends on."""
    return (idx.tz_convert(NY) + pd.Timedelta(hours=7)).normalize().tz_localize(None)


def resample(m1: pd.DataFrame, rule: str) -> pd.DataFrame:
    """OHLC candles (BID and ASK) from active minutes, labelled by start time (UTC).

    rule '15min', '1h' etc. are aligned to the UTC clock (identical in NY / server time).
    rule '4h' and '1D' are aligned to 17:00 New York (NY-close), DST-aware.
    Candles with no active minute are not created (no flat filler candles).
    """
    agg = {"bo": "first", "bh": "max", "bl": "min", "bc": "last",
           "ao": "first", "ah": "max", "al": "min", "ac": "last", "vb": "sum", "va": "sum"}
    if rule in ("4h", "1D", "D"):
        ny = m1.index.tz_convert(NY).tz_localize(None)        # NY wall time, naive
        shifted = ny + pd.Timedelta(hours=7)                  # 17:00 NY -> 00:00
        hours = 24 if rule in ("1D", "D") else 4
        floor = shifted.floor("D") + pd.to_timedelta((shifted.hour // hours) * hours, unit="h")
        key = floor - pd.Timedelta(hours=7)                   # NY wall time of the candle start
        g = m1.groupby(key)
        out = g.agg(agg)
        out["n_min"] = g.size()
        # 17:00 NY (daily) is never ambiguous; a 4h 01:00 start on the autumn DST night is read as standard time
        idx = pd.DatetimeIndex(out.index)
        out.index = idx.tz_localize(NY, ambiguous=np.zeros(len(idx), dtype=bool),
                                    nonexistent="shift_forward").tz_convert("UTC")
        out.index.name = "t"
        return out
    out = m1.resample(rule, label="left", closed="left").agg(agg)
    out["n_min"] = m1["bc"].resample(rule, label="left", closed="left").count()
    return out[out["n_min"] > 0]


# ----------------------------------------------------------------- hourly ---
# Dukascopy also publishes one file of HOURLY BID / ASK candles per month
# (<SYM>/YYYY/MM/{BID,ASK}_candles_hour_1.bi5, month 0-based, seconds from the month start). 282 files per
# symbol for 2015-2026 instead of ~7,400 minute files: used for (a) the hourly spread profile that models the
# ASK side when only BID minute data (HistData) is available, and (b) checking our own hourly candles.
def hour_rel(y: int, m: int, side: str) -> str:
    return f"{y}/{m - 1:02d}/{side}_candles_hour_1.bi5"


def cmd_hourly(a) -> None:
    y0, m0 = int(a.start[:4]), int(a.start[5:7])
    y1, m1 = int(a.end[:4]), int(a.end[5:7])
    months = [(y, m) for y in range(y0, y1 + 1) for m in range(1, 13)
              if (y, m) >= (y0, m0) and (y, m) <= (y1, m1)]
    if a.months:  # e.g. "1,7": only these calendar months (enough for an hourly spread profile)
        keep = {int(x) for x in a.months.split(",")}
        months = [(y, m) for (y, m) in months if m in keep]
    for sym in symbols_arg(a):
        jobs = [hour_rel(y, m, s) for s in ("BID", "ASK") for (y, m) in months]
        with cf.ThreadPoolExecutor(min(a.threads, 8)) as ex:
            list(ex.map(lambda j: fetch_one(sym, j), jobs))
        rows = []
        sc = PRICE_SCALE[sym]
        for (y, m) in months:
            parts = []
            for s in ("BID", "ASK"):
                p = raw_dir(sym) / hour_rel(y, m, s)
                if not p.exists() or p.stat().st_size == 0:
                    parts = None
                    break
                arr = np.frombuffer(lzma.decompress(p.read_bytes()), dtype=">i4").reshape(-1, 6)
                parts.append(arr)
            if parts is None:
                print(f"[{sym}] missing hourly file {y}-{m:02d}")
                continue
            b, k = parts
            base = int(datetime(y, m, 1, tzinfo=timezone.utc).timestamp())
            n = min(len(b), len(k))
            vb = b[:n, 5].copy().view(">f4")
            va = k[:n, 5].copy().view(">f4")
            act = (vb > 0) | (va > 0)
            t = base + b[:n, 0].astype(np.int64)
            rows.append(np.column_stack([t, b[:n, 1], b[:n, 4], b[:n, 3], b[:n, 2],
                                         k[:n, 1], k[:n, 4], k[:n, 3], k[:n, 2]])[act])
        A = np.concatenate(rows)
        np.savez_compressed(RAW / f"{sym}_h1_duka.npz", t=A[:, 0].astype(np.int64),
                            bid=A[:, 1:5].astype(np.int32), ask=A[:, 5:9].astype(np.int32), scale=np.array([sc]))
        sp = (A[:, 5] - A[:, 1]) / sc
        print(f"[{sym}] hourly candles {len(A):,} -> {sym}_h1_duka.npz; median open spread {np.median(sp):.5f}")


def spread_profile_m1(symbol: str, max_days: int = 4000) -> pd.DataFrame:
    """Median BID->ASK close spread by (year, UTC hour) from whatever Dukascopy MINUTE days are cached with
    both sides (active minutes only). Cached in data_raw/<SYM>_spread_profile_m1.csv."""
    out = RAW / f"{symbol}_spread_profile_m1.csv"
    rows = []
    for bid in sorted(raw_dir(symbol).rglob("BID_candles_min_1.bi5"))[:max_days]:
        ask = bid.with_name("ASK_candles_min_1.bi5")
        if not ask.exists():
            continue
        y, mo, d = int(bid.parts[-4]), int(bid.parts[-3]) + 1, int(bid.parts[-2])
        b, k = read_day_raw(symbol, date(y, mo, d), "BID"), read_day_raw(symbol, date(y, mo, d), "ASK")
        if b is None or k is None or len(b) == 0 or len(b) != len(k):
            continue
        act = (b[:, 5] > 0) & (k[:, 5] > 0)
        if act.sum() < 60:
            continue
        hrs = (b[act, 0] // 3600) % 24
        sp = (k[act, 4] - b[act, 4]) / PRICE_SCALE[symbol]
        df = pd.DataFrame({"hour": hrs, "spread": sp}).groupby("hour")["spread"].median().reset_index()
        df["year"] = y
        rows.append(df)
    prof = pd.concat(rows).groupby(["year", "hour"])["spread"].median().reset_index()
    prof.to_csv(out, index=False)
    return prof


def spread_profile(symbol: str) -> pd.DataFrame:
    """Median BID->ASK spread at the hour's open and close, by (year, UTC hour of day), from the Dukascopy
    hourly candles (or, if those are not cached, from the cached Dukascopy minute days). Used to model the
    ASK of BID-only (HistData) minute bars."""
    if not (RAW / f"{symbol}_h1_duka.npz").exists():
        cached = RAW / f"{symbol}_spread_profile_m1.csv"
        return pd.read_csv(cached) if cached.exists() else spread_profile_m1(symbol)
    z = np.load(RAW / f"{symbol}_h1_duka.npz")
    sc = float(z["scale"][0])
    t = pd.to_datetime(z["t"], unit="s", utc=True)
    sp = ((z["ask"][:, 0] - z["bid"][:, 0]) + (z["ask"][:, 3] - z["bid"][:, 3])) / 2 / sc
    df = pd.DataFrame({"year": t.year, "hour": t.hour, "spread": sp})
    return df.groupby(["year", "hour"])["spread"].median().rename("spread").reset_index()


# --------------------------------------------------------------- histdata ---
# Second source for cross-checks: HistData.com free 1-minute ASCII bars (BID only), one zip per year
# (per month for the current year). Documented as "EST without daylight saving" (fixed UTC-5); this is
# verified against Dukascopy in the backtest's cross-check before use (see BACKTEST.md).
HD_PAGE = "https://www.histdata.com/download-free-forex-historical-data/?/ascii/1-minute-bar-quotes/{sym}/"


def _histdata_fetch(sym: str, period: str) -> Path:
    import re
    import http.cookiejar
    import urllib.parse
    path = RAW / "histdata" / f"{sym}_M1_{period.replace('/', '_')}.zip"
    if path.exists():
        return path
    page_url = HD_PAGE.format(sym=sym.lower()) + period
    cj = http.cookiejar.CookieJar()
    op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
    op.addheaders = [("User-Agent", "Mozilla/5.0")]
    page = op.open(page_url, timeout=60).read().decode("utf-8", "replace")
    form = {k: re.search(rf'name="{k}" id="{k}" value="([^"]*)"', page).group(1)
            for k in ("tk", "date", "datemonth", "platform", "timeframe", "fxpair")}
    req = urllib.request.Request("https://www.histdata.com/get.php", data=urllib.parse.urlencode(form).encode(),
                                 headers={"Referer": page_url})
    body = op.open(req, timeout=180).read()
    if body[:2] != b"PK":
        raise OSError(f"HistData {sym} {period}: not a zip ({body[:80]!r})")
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(body)
    time.sleep(2)
    return path


def histdata_fetch(sym: str, period: str) -> Path | None:
    for attempt in range(6):
        try:
            return _histdata_fetch(sym, period)
        except (OSError, AttributeError) as e:
            print(f"  histdata {sym} {period}: {e!r}, retrying", flush=True)
            time.sleep(5 * (attempt + 1))
    print(f"  histdata {sym} {period}: FAILED", flush=True)
    return None


def cmd_histdata(a) -> None:
    import zipfile
    y0, y1 = int(a.start[:4]), int(a.end[:4])
    last_m = int(a.end[5:7])
    for sym in symbols_arg(a):
        periods = [str(y) for y in range(y0, y1)] + [f"{y1}/{m}" for m in range(1, last_m + 1)]
        frames = []
        for per in periods:
            p = histdata_fetch(sym, per)
            if p is None:
                continue
            z = zipfile.ZipFile(p)
            name = next(n for n in z.namelist() if n.lower().endswith(".csv"))
            df = pd.read_csv(z.open(name), sep=";", header=None, names=["ts", "o", "h", "l", "c", "v"])
            frames.append(df)
            print(sym, per, len(df), flush=True)
        df = pd.concat(frames)
        # HistData documents "EST without DST" (fixed UTC-5), but for XAUUSD the minute returns line up with
        # Dukascopy only when the stamps are read as New York LOCAL time (UTC-5 winter / UTC-4 summer):
        # checked on 117 random days (topg BACKTEST_PLAN.md), as the umar audit found for NSXUSD. Stamps in
        # the DST switch hour (Sunday night, market closed) are dropped.
        # Caveat found in the check: in the 2-3 weeks a year when the US and Europe are on different clocks
        # (March, late Oct), some years' stamps follow the European switch instead (e.g. 2024-03-13, 2025-03-12
        # are 60 min off under the NY reading, while March 2015 fits it). HistData is only the cross-check
        # source, and the cross-check aligns those days empirically (see topg_backtest.py crosscheck).
        t = pd.DatetimeIndex(pd.to_datetime(df["ts"], format="%Y%m%d %H%M%S"))
        t = t.tz_localize(NY, ambiguous="NaT", nonexistent="NaT").tz_convert("UTC").tz_localize(None)
        ok = ~pd.isna(t)
        df, t = df[ok], pd.Series(t[ok])
        sc = PRICE_SCALE[sym]
        out = RAW / f"{sym}_histdata_m1.npz"
        tt = ((t - pd.Timestamp("1970-01-01")) // pd.Timedelta(seconds=1)).to_numpy().astype(np.int64)
        order = np.argsort(tt, kind="stable")
        tt = tt[order]
        keep = np.concatenate([[True], np.diff(tt) > 0])
        px = np.round(df[["o", "h", "l", "c"]].to_numpy()[order][keep] * sc).astype(np.int64)
        np.savez_compressed(out, t=tt[keep], bid=px.astype(np.int32), scale=np.array([sc]))
        print(f"[{sym}] histdata {keep.sum():,} minutes -> {out.relative_to(ROOT)} ({out.stat().st_size/1e6:.1f} MB)")


def load_histdata_m1(symbol: str, utc_shift_hours: float = 0.0) -> pd.DataFrame:
    """HistData BID bars as a load_m1-like frame (ASK = BID, so no spread). utc_shift_hours lets the
    cross-check test the time-zone reading (0 = documented fixed UTC-5 already applied)."""
    z = np.load(RAW / f"{symbol}_histdata_m1.npz")
    sc = float(z["scale"][0])
    idx = pd.to_datetime(z["t"] + int(utc_shift_hours * 3600), unit="s", utc=True)
    b = z["bid"].astype(np.float64) / sc
    df = pd.DataFrame({"bo": b[:, 0], "bh": b[:, 1], "bl": b[:, 2], "bc": b[:, 3]}, index=idx)
    for k in "ohlc":
        df[f"a{k}"] = df[f"b{k}"]
    df["vb"] = df["va"] = 1.0
    df.index.name = "t"
    return df


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("cmd", choices=["fetch", "build", "status", "sizes", "histdata", "hourly"])
    ap.add_argument("--symbols", default="XAUUSD")
    ap.add_argument("--start", default=DEFAULT_START)
    ap.add_argument("--end", default=DEFAULT_END)
    ap.add_argument("--threads", type=int, default=16)
    ap.add_argument("--reverse", action="store_true", help="download newest first (to run a 2nd worker)")
    ap.add_argument("--shuffle", type=int, default=0, help="random order with this seed (to run a 3rd worker)")
    ap.add_argument("--months", default="", help='hourly only: calendar months to fetch, e.g. "1,7"')
    a = ap.parse_args()
    {"fetch": cmd_fetch, "build": cmd_build, "status": cmd_status, "sizes": cmd_sizes,
     "histdata": cmd_histdata, "hourly": cmd_hourly}[a.cmd](a)


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