"""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()