dataset ingest script

ingest.py

@@ -0,0 +1,169 @@
+"""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()