setup_data.py

#!/usr/bin/env python3
"""Pin Binance public futures data or create an explicit synthetic test fixture."""

from __future__ import annotations

import argparse
from concurrent.futures import ThreadPoolExecutor
import hashlib
import io
import json
import time
import zipfile
from pathlib import Path

import httpx
import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parent
DATA = ROOT / "data"
BASE = "https://data.binance.vision/data/futures/um/monthly"
DAILY = "https://data.binance.vision/data/futures/um/daily"
START = pd.Timestamp("2019-09-01", tz="UTC")
END = pd.Timestamp("2026-07-01", tz="UTC")


def sha256(path: Path) -> str:
    with path.open("rb") as handle:
        return hashlib.file_digest(handle, "sha256").hexdigest()


def parse_mixed_timestamp(values: pd.Series) -> pd.Series:
    """Parse seconds/ms/us/ns by magnitude for every row, not once per file."""
    numeric = pd.to_numeric(values, errors="coerce"); magnitude = numeric.abs()
    parsed = pd.Series(pd.NaT, index=values.index, dtype="datetime64[ns, UTC]")
    groups = ((magnitude < 1e11, "s"),
              ((magnitude >= 1e11) & (magnitude < 1e14), "ms"),
              ((magnitude >= 1e14) & (magnitude < 1e17), "us"),
              (magnitude >= 1e17, "ns"))
    for mask, unit in groups:
        if mask.any():
            parsed.loc[mask] = pd.to_datetime(numeric.loc[mask].round().astype("int64"), unit=unit,
                                              utc=True, errors="coerce").dt.as_unit("ns")
    return parsed


def _months(start: pd.Timestamp, end: pd.Timestamp):
    first = pd.Timestamp(year=start.year, month=start.month, day=1, tz="UTC")
    return pd.date_range(first, end, freq="MS", inclusive="left").strftime("%Y-%m")


def _zip_csv(client: httpx.Client, url: str) -> pd.DataFrame | None:
    for attempt in range(4):
        try:
            response = client.get(url)
            if response.status_code not in {429, 500, 502, 503, 504}: break
        except httpx.HTTPError:
            if attempt == 3: raise
        time.sleep(0.25 * 2**attempt)
    if response.status_code == 404:
        return None
    response.raise_for_status()
    with zipfile.ZipFile(io.BytesIO(response.content)) as archive:
        names = [name for name in archive.namelist() if name.endswith(".csv")]
        if len(names) != 1: raise ValueError(f"expected one CSV in {url}, found {names}")
        with archive.open(names[0]) as handle:
            payload = handle.read()
            first_cell = payload.split(b",", 1)[0].strip()
            has_header = not first_cell.replace(b".", b"", 1).isdigit()
            return pd.read_csv(io.BytesIO(payload), header=0 if has_header else None)


def _daily_metrics(client: httpx.Client, symbol: str, start: pd.Timestamp,
                   end: pd.Timestamp) -> list[pd.DataFrame]:
    days = pd.date_range(start.floor("D"), end.floor("D"), freq="D", inclusive="left")
    urls = [f"{DAILY}/metrics/{symbol}/{symbol}-metrics-{day:%Y-%m-%d}.zip" for day in days]
    with ThreadPoolExecutor(max_workers=16) as pool:
        frames = list(pool.map(lambda url: _zip_csv(client, url), urls))
    return [frame for frame in frames if frame is not None]


def _download_symbol(client: httpx.Client, symbol: str, listed: str, delisted: str | None):
    start = max(START, pd.Timestamp(listed, tz="UTC"))
    end = min(END, pd.Timestamp(delisted, tz="UTC") if delisted else END)
    bars_parts: list[pd.DataFrame] = []; funding_parts: list[pd.DataFrame] = []
    for month in _months(start, end):
        stem = f"{symbol}-1h-{month}"
        frame = _zip_csv(client, f"{BASE}/klines/{symbol}/1h/{stem}.zip")
        if frame is not None:
            if "open_time" not in frame.columns:
                frame.columns = ["open_time", "open", "high", "low", "close", "volume",
                                 "close_time", "quote_volume", "n_trades", "taker_base",
                                 "taker_quote", "ignore"][: len(frame.columns)]
            frame["symbol"] = symbol
            bars_parts.append(frame)
        fund = _zip_csv(client, f"{BASE}/fundingRate/{symbol}/{symbol}-fundingRate-{month}.zip")
        if fund is not None:
            funding_parts.append(fund.assign(symbol=symbol))
    if not bars_parts:
        return None, None
    bars = pd.concat(bars_parts, ignore_index=True)
    bars["ts"] = parse_mixed_timestamp(bars["open_time"])
    bars = bars[["ts", "symbol", "open", "high", "low", "close", "volume", "quote_volume"]]
    bars = bars[(bars["ts"] >= start) & (bars["ts"] < end)]
    for col in bars.columns.difference(["ts", "symbol"]):
        bars[col] = pd.to_numeric(bars[col], errors="coerce")
    funding = pd.DataFrame(columns=["ts", "symbol", "funding"])
    if funding_parts:
        raw = pd.concat(funding_parts, ignore_index=True)
        funding = pd.DataFrame({"ts": parse_mixed_timestamp(raw["calc_time"]), "symbol": symbol,
                                "funding": pd.to_numeric(raw["last_funding_rate"], errors="coerce")})
        funding = funding[(funding["ts"] >= start) & (funding["ts"] < end)]
    metrics_parts = _daily_metrics(client, symbol, start, end)
    if metrics_parts:
        raw = pd.concat(metrics_parts, ignore_index=True)
        oi_ts = pd.to_datetime(raw["create_time"], utc=True, errors="coerce").dt.as_unit("ns")
        oi = pd.DataFrame({"ts": oi_ts, "symbol": symbol,
                           "open_interest": pd.to_numeric(raw["sum_open_interest"], errors="coerce")})
        oi = oi.dropna().set_index("ts").resample("1h").last().ffill().reset_index()
        bars = pd.merge_asof(bars.sort_values("ts"), oi.sort_values("ts"), on="ts", by="symbol",
                             direction="backward", tolerance=pd.Timedelta("2h"))
    if "open_interest" not in bars:
        bars["open_interest"] = np.nan
    return bars, funding


def download_real() -> tuple[pd.DataFrame, pd.DataFrame]:
    universe = json.loads((DATA / "universe.json").read_text())
    bars: list[pd.DataFrame] = []; funding: list[pd.DataFrame] = []
    def fetch(item):
        with httpx.Client(timeout=60, follow_redirects=True, headers={"User-Agent": "autoquant/0.1"}) as client:
            print(f"downloading {item['symbol']}", flush=True)
            return _download_symbol(client, item["symbol"], item["listed_at"], item["delisted_at"])
    with ThreadPoolExecutor(max_workers=4) as pool:
        for b, f in pool.map(fetch, universe["symbols"]):
            if b is not None: bars.append(b)
            if f is not None and not f.empty: funding.append(f)
    if not bars: raise RuntimeError("no Binance data downloaded")
    return pd.concat(bars, ignore_index=True), pd.concat(funding, ignore_index=True)


def synthetic_data(step_hours: int = 12) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Small deterministic fixture. It is visibly marked and never a real-data fallback."""
    universe = json.loads((DATA / "universe.json").read_text())["symbols"]
    times = pd.date_range(START, END, freq=f"{step_hours}h", inclusive="left")
    rng = np.random.default_rng(20260808)
    n_t, n_s = len(times), len(universe)
    common = rng.normal(0.00012, 0.006, (n_t, 1))
    innovations = rng.normal(0.0, 0.009, (n_t, n_s))
    trend = np.zeros_like(innovations)
    for i in range(1, n_t):
        trend[i] = 0.97 * trend[i - 1] + innovations[i]
    # A weak, regime-varying continuation premium makes the seed useful but leaves room.
    returns = common + innovations + 0.002 * np.roll(trend, 1, axis=0)
    returns[0] = 0.0
    start_prices = np.geomspace(8.0, 20_000.0, n_s)
    close = start_prices * np.exp(np.cumsum(returns, axis=0))
    open_ = close * np.exp(rng.normal(0, 0.001, close.shape))
    high = np.maximum(open_, close) * (1 + rng.uniform(0, 0.004, close.shape))
    low = np.minimum(open_, close) * (1 - rng.uniform(0, 0.004, close.shape))
    volume = rng.lognormal(11.0, 0.8, close.shape) / np.maximum(close, 1e-6)
    funding_rate = np.clip(0.000006 * trend + rng.normal(0, 0.000008, close.shape), -0.0005, 0.0005)
    oi = rng.lognormal(13.0, 0.3, close.shape) * np.exp(np.cumsum(rng.normal(0, 0.002, close.shape), axis=0))
    records: list[pd.DataFrame] = []; fund_records: list[pd.DataFrame] = []
    for j, item in enumerate(universe):
        listed = pd.Timestamp(item["listed_at"], tz="UTC")
        delisted = pd.Timestamp(item["delisted_at"], tz="UTC") if item["delisted_at"] else END
        mask = (times >= listed) & (times < delisted)
        frame = pd.DataFrame({"ts": times[mask], "symbol": item["symbol"], "open": open_[mask, j],
                              "high": high[mask, j], "low": low[mask, j], "close": close[mask, j],
                              "volume": volume[mask, j], "quote_volume": volume[mask, j] * close[mask, j],
                              "open_interest": oi[mask, j]})
        records.append(frame)
        settlement = mask & (times.hour % 8 == 0)
        fund_records.append(pd.DataFrame({"ts": times[settlement], "symbol": item["symbol"],
                                          "funding": funding_rate[settlement, j]}))
    return pd.concat(records, ignore_index=True), pd.concat(fund_records, ignore_index=True)


def normalize_funding_clock(funding: pd.DataFrame) -> pd.DataFrame:
    funding = funding.copy()
    funding["ts"] = pd.to_datetime(funding["ts"], utc=True).dt.as_unit("ns")
    nearest = funding["ts"].dt.round("h")
    if ((funding["ts"] - nearest).abs() > pd.Timedelta("5min")).any():
        raise ValueError("funding timestamp is not within five minutes of a settlement hour")
    funding["ts"] = nearest
    return funding.sort_values(["ts", "symbol"]).drop_duplicates(["ts", "symbol"], keep="last")


def _write_parquet(bars: pd.DataFrame, funding: pd.DataFrame, synthetic: bool) -> None:
    DATA.mkdir(exist_ok=True); bars = bars.dropna(subset=["ts", "symbol"]).sort_values(["ts", "symbol"]).reset_index(drop=True)
    funding = normalize_funding_clock(funding.dropna(subset=["ts", "symbol"])).reset_index(drop=True)
    bars.to_parquet(DATA / "bars.parquet", index=False, compression="zstd")
    funding.to_parquet(DATA / "funding.parquet", index=False, compression="zstd")
    manifest = {
        "format": 1,
        "source": "SYNTHETIC TEST FIXTURE" if synthetic else "data.binance.vision",
        "bars.parquet": sha256(DATA / "bars.parquet"),
        "funding.parquet": sha256(DATA / "funding.parquet"),
        "universe.json": sha256(DATA / "universe.json"),
    }
    (DATA / "MANIFEST.sha256").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")


def verify(data_dir: Path = DATA, quiet: bool = False) -> None:
    manifest_path = data_dir / "MANIFEST.sha256"
    if not manifest_path.exists(): raise SystemExit("DATA MISSING: run setup_data.py first")
    manifest = json.loads(manifest_path.read_text())
    for name, expected in manifest.items():
        if not name.endswith((".parquet", ".json")): continue
        path = data_dir / name
        actual = sha256(path) if path.exists() else "MISSING"
        if actual != expected: raise SystemExit(f"DATA HASH MISMATCH: {name} expected={expected} actual={actual}")
    if not quiet: print(f"DATA OK ({manifest.get('source', 'unknown source')})")


def write_protected_hashes() -> None:
    entries = {}
    for relative in ("evaluate.py", "run.py", "data/MANIFEST.sha256"):
        path = ROOT / relative
        if not path.exists(): raise RuntimeError(f"cannot protect missing {relative}")
        entries[relative] = sha256(path)
    (DATA / "PROTECTED.sha256").write_text(json.dumps(entries, indent=2, sort_keys=True) + "\n")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--verify", action="store_true")
    parser.add_argument("--synthetic", action="store_true", help="explicit deterministic test fixture")
    parser.add_argument("--force", action="store_true")
    parser.add_argument("--step-hours", type=int, default=12, help=argparse.SUPPRESS)
    args = parser.parse_args()
    if args.verify: verify(); return
    if any((DATA / name).exists() for name in ("bars.parquet", "funding.parquet")) and not args.force:
        raise SystemExit("data exists; use --verify or explicitly pass --force")
    bars, funding = synthetic_data(args.step_hours) if args.synthetic else download_real()
    _write_parquet(bars, funding, args.synthetic)
    # Reference calibration is separate because it executes the strategy 41 times.
    write_protected_hashes()
    verify()
    print("SETUP COMPLETE; run `python evaluate.py --calibrate` before an arena")


if __name__ == "__main__":
    main()