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