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