ingest.py

"""Build the curated eval dataset repo: Avi-L/crypto-quant-eval-data.

Sources (all verified to resolve on the HF Hub, 2026-10-23):
  bars  : vaquum/binance_btcusdt_1m_klines  (BTCUSDT spot 1m, 2020->present)
  fund  : Torch-Trade/btcusdt_perp_funding_8h_05_2021_to_02_2026   (MIT)
          Torch-Trade/ethusdt_perp_funding_8h_05_2021_to_02_2026   (MIT)
  chain : coinmetrics-community btc.csv (GitHub, no usable HF on-chain repo)

ETH spot 1m is NOT ingested: the only verified ETH 1m repo
(HaianCao/Binance-Public-OHLCV) is 34.6 GB — out of scope for this job;
ETH enters via funding. Documented in the dataset README.

Outputs (uploaded to the dataset repo):
  bars_BTCUSDT_1h.parquet, bars_BTCUSDT_1d.parquet  (datetime index, OHLCV
    + taker_volume_share when available, funding_rate_last, funding_tick)
  funding_BTCUSDT.parquet, funding_ETHUSDT.parquet  (funding_rate, 8h)
  onchain_btc_daily.parquet
  manifest.json  (sha256 of every output + source file hashes)
"""

from __future__ import annotations

import hashlib
import json
import tempfile
from pathlib import Path

import pandas as pd
import pyarrow.parquet as pq
from huggingface_hub import HfApi, snapshot_download

DEST_REPO = "Avi-L/crypto-quant-eval-data"
BARS_REPO = "vaquum/binance_btcusdt_1m_klines"
FUNDING_REPOS = {
    "BTCUSDT": "Torch-Trade/btcusdt_perp_funding_8h_05_2021_to_02_2026",
    "ETHUSDT": "Torch-Trade/ethusdt_perp_funding_8h_05_2021_to_02_2026",
}
ONCHAIN_URL = (
    "https://raw.githubusercontent.com/coinmetrics/data/master/csv/btc.csv"
)

MANIFEST: dict = {"generated": "2026-10-23", "sources": {}, "outputs": {}}


def sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def load_bars_streaming(cache: Path) -> pd.DataFrame:
    """Read the vaquum 1m parquet files one at a time, resample to 1h/1d."""
    d = Path(snapshot_download(repo_id=BARS_REPO, repo_type="dataset",
                               local_dir=cache / "bars"))
    files = sorted(d.rglob("*.parquet"))
    if not files:
        raise ValueError(f"no parquet files under {d}")
    daily, hourly = [], []
    for f in files:
        df = pd.read_parquet(f)
        # vaquum schema: datetime (ts[ms,UTC]) + OHLCV-ish columns
        tcol = "datetime" if "datetime" in df.columns else df.columns[0]
        ts = pd.to_datetime(df[tcol], utc=True)
        df = df.set_index(ts).sort_index()
        agg = {"open": "first", "high": "max", "low": "min", "close": "last",
               "volume": "sum"}
        # optional columns present in some schemas
        for col, how in (("quote_volume", "sum"), ("count", "sum"),
                         ("no_of_trades", "sum"),
                         ("taker_buy_volume", "sum")):
            if col in df.columns:
                agg[col] = how
        agg_df = df[list(agg)].resample("1h").agg(agg).dropna(
            subset=["open", "close"])
        hourly.append(agg_df)
        daily.append(df[list(agg)].resample("1d").agg(agg).dropna(
            subset=["open", "close"]))
    h = pd.concat(hourly)
    dd = pd.concat(daily)
    return h[~h.index.duplicated()].sort_index(), dd[~dd.index.duplicated()].sort_index()


def load_funding(cache: Path, repo_id: str) -> pd.DataFrame:
    d = Path(snapshot_download(repo_id=repo_id, repo_type="dataset",
                               local_dir=cache / repo_id.replace("/", "_")))
    csvs = sorted(d.rglob("*.csv"))
    pqs = sorted(d.rglob("*.parquet"))
    if csvs:
        df = pd.concat(pd.read_csv(f) for f in csvs)
    elif pqs:
        df = pd.concat(pd.read_parquet(f) for f in pqs)
    else:
        raise ValueError(f"no csv/parquet under {d}")
    if "timestamp" in df.columns:
        df.index = pd.to_datetime(df["timestamp"], utc=True, unit="ms")
    elif "datetime" in df.columns:
        df.index = pd.to_datetime(df["datetime"], utc=True)
    else:
        raise ValueError(f"funding has no time column: {list(df.columns)}")
    if "funding_rate" not in df.columns:
        raise ValueError(f"funding has no funding_rate: {list(df.columns)}")
    return df[["funding_rate"]].sort_index()


def merge_funding(bars: pd.DataFrame, funding: pd.DataFrame) -> pd.DataFrame:
    out = bars.copy()
    f = funding.reindex(out.index, method="ffill")["funding_rate"]
    out["funding_rate_last"] = f
    out["funding_tick"] = out.index.isin(funding.index)
    return out


def onchain_daily(out: Path) -> None:
    df = pd.read_csv(ONCHAIN_URL)
    df["time"] = pd.to_datetime(df["time"], utc=True)
    want = [c for c in ("time", "AdrActCnt", "BlkCnt", "CapMrktCurUSD",
                        "FeeMeanNtv", "IssuanceNtv", "TxCnt", "NVTAdj")
            if c in df.columns]
    df[want].set_index("time").to_parquet(out / "onchain_btc_daily.parquet")
    MANIFEST["sources"]["coinmetrics-community btc.csv"] = ONCHAIN_URL


def main() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        cache = Path(tmp)
        out = cache / "curated"
        out.mkdir()

        print("[ingest] bars: streaming + resampling", flush=True)
        hourly, daily = load_bars_streaming(cache)
        funding_btc = load_funding(cache, FUNDING_REPOS["BTCUSDT"])
        funding_eth = load_funding(cache, FUNDING_REPOS["ETHUSDT"])

        for freq, bars in (("1h", hourly), ("1d", daily)):
            merged = merge_funding(bars, funding_btc)
            p = out / f"bars_BTCUSDT_{freq}.parquet"
            merged.reset_index(names="datetime").to_parquet(p)
            MANIFEST["outputs"][p.name] = {"sha256": sha256_file(p),
                                           "rows": len(merged), "freq": freq}
            print(f"[ingest] {p.name}: {len(merged):,} rows", flush=True)

        for sym, df in (("BTCUSDT", funding_btc), ("ETHUSDT", funding_eth)):
            p = out / f"funding_{sym}.parquet"
            df.to_parquet(p)
            MANIFEST["outputs"][p.name] = {"sha256": sha256_file(p),
                                           "rows": len(df), "freq": "8h"}
            print(f"[ingest] {p.name}: {len(df):,} ticks", flush=True)

        print("[ingest] on-chain metrics", flush=True)
        onchain_daily(out)
        p = out / "onchain_btc_daily.parquet"
        MANIFEST["outputs"][p.name] = {"sha256": sha256_file(p),
                                       "rows": pq.ParquetFile(p).metadata.num_rows,
                                       "freq": "1d"}

        (out / "manifest.json").write_text(json.dumps(MANIFEST, indent=2))
        api = HfApi()
        api.upload_folder(
            repo_id=DEST_REPO, repo_type="dataset",
            folder_path=str(out), commit_message="curated eval dataset v1",
        )
        print(f"[ingest] uploaded {len(MANIFEST['outputs'])} files + manifest "
              f"to {DEST_REPO}", flush=True)


if __name__ == "__main__":
    main()