coinmetrics daily panel ingest

ingest2.py

@@ -0,0 +1,94 @@
+"""Ingest daily price panel for 6-8 major crypto assets from coinmetrics.
+
+Source: coinmetrics-community CSVs on GitHub (PriceUSD = volume-weighted
+index across exchanges). Long daily history (btc 2010+, ltc/xrp 2013+,
+eth 2015+) — this is the statistical-power lever for the DSR gate.
+
+Output (uploaded to Avi-L/crypto-quant-eval-data):
+  panel_daily_close.parquet   wide close-price panel, index=date, cols=symbols
+  panel_daily_meta.json       per-asset start dates, obs counts, ADV assumptions
+  manifest2.json              sha256 outputs + sources
+"""
+import hashlib
+import io
+import json
+import tempfile
+import urllib.request
+from pathlib import Path
+
+import pandas as pd
+from huggingface_hub import HfApi
+
+DEST = "Avi-L/crypto-quant-eval-data"
+BASE = "https://raw.githubusercontent.com/coinmetrics/data/master/csv/{}.csv"
+# majors with long daily history; ADV = realistic combined spot+perp
+# notional/day used by the impact model (conservative, 2025-26 orderbook)
+ASSETS = {
+    "BTC": 20e9, "ETH": 10e9, "LTC": 1.0e9, "XRP": 1.5e9,
+    "ADA": 0.5e9, "DOGE": 1.0e9, "ETC": 0.3e9, "DASH": 0.1e9,
+}
+
+MANIFEST = {"sources": {}, "outputs": {}, "assets": {}}
+
+
+def sha256_file(p: Path) -> str:
+    h = hashlib.sha256()
+    with open(p, "rb") as f:
+        for chunk in iter(lambda: f.read(1 << 20), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def load_asset(sym: str) -> pd.Series:
+    url = BASE.format(sym.lower())
+    print(f"[ingest] fetching {url}", flush=True)
+    with urllib.request.urlopen(url, timeout=120) as r:
+        df = pd.read_csv(io.BytesIO(r.read()))
+    if "time" not in df or "PriceUSD" not in df:
+        raise ValueError(f"{sym}: missing time/PriceUSD columns")
+    s = pd.Series(df["PriceUSD"].values,
+                  index=pd.to_datetime(df["time"], utc=True),
+                  name=sym).dropna()
+    return s[s.index >= "2013-01-01"]
+
+
+def main() -> None:
+    with tempfile.TemporaryDirectory() as tmp:
+        out = Path(tmp)
+        panel = {}
+        for sym, adv in ASSETS.items():
+            try:
+                s = load_asset(sym)
+            except Exception as e:
+                print(f"[ingest] {sym} FAILED: {e}", flush=True)
+                continue
+            panel[sym] = s
+            MANIFEST["assets"][sym] = {
+                "start": str(s.index[0].date()), "obs": int(len(s)), "adv_usd": adv,
+            }
+            print(f"[ingest] {sym}: {len(s):,} days from {s.index[0].date()}",
+                  flush=True)
+        if "BTC" not in panel:
+            raise RuntimeError("no BTC data — aborting")
+        close = pd.DataFrame(panel).sort_index()
+        close.index.name = "date"
+        p = out / "panel_daily_close.parquet"
+        close.to_parquet(p)
+        MANIFEST["outputs"][p.name] = {"sha256": sha256_file(p),
+                                       "rows": len(close), "cols": close.shape[1]}
+        meta = out / "panel_daily_meta.json"
+        meta.write_text(json.dumps(MANIFEST, indent=2))
+        MANIFEST["sources"]["coinmetrics-community CSVs"] = BASE.format("btc")
+        meta.write_text(json.dumps(MANIFEST, indent=2))
+
+        api = HfApi()
+        for f in (p, meta):
+            api.upload_file(path_or_fileobj=str(f), path_in_repo=f.name,
+                            repo_id=DEST, repo_type="dataset",
+                            commit_message="daily multi-asset price panel")
+        print(f"[ingest] panel: {close.shape[0]:,} days x "
+              f"{close.shape[1]} assets, uploaded", flush=True)
+
+
+if __name__ == "__main__":
+    main()