Repository Public
1"""Download public, keyless OHLCV into local Parquet files so the lab works without the ingester.23 uv run python -m spq.data.fetch --symbols ETH,BTC,LINK,SOL --interval 1h --start 2024-11-10 --out data45Sources (both verified live on 2026-11-10):6- coinbase: GET https://api.exchange.coinbase.com/products/{SYM}-USD/candles?granularity=&start=&end=7 rows [time, low, high, open, close, volume], newest first, at most 300 per request, full history.8- kraken: GET https://api.kraken.com/0/public/OHLC?pair={SYM}USD&interval=<minutes>9 rows [time, open, high, low, close, vwap, volume, count], oldest first, ONLY the most recent 720 bars10 (no deep history): useful for a cross-source check of recent bars, not for a two-year backtest.1112Both return the bar in progress; it is dropped here (closed bars only). Files are written to13<out>/<source>/<SYMBOL>_<interval>.parquet and merged with what is already there.14This module only reads public market data. It sends no credentials and places no orders.15"""16from __future__ import annotations1718import argparse19import json20import sys21import time22import urllib.error23import urllib.parse24import urllib.request25from datetime import datetime, timezone26from pathlib import Path2728import pandas as pd2930from .core import COLUMNS, DataError, interval_seconds, normalise, to_utc3132USER_AGENT = "spq-research/0.1"33COINBASE_URL = "https://api.exchange.coinbase.com/products/{product}/candles"34KRAKEN_URL = "https://api.kraken.com/0/public/OHLC"35COINBASE_GRANULARITIES = {60, 300, 900, 3600, 21600, 86400}36KRAKEN_INTERVALS = {1, 5, 15, 30, 60, 240, 1440, 10080}37KRAKEN_ALIASES = {"BTC": "XBT", "DOGE": "XDG"}383940def _get_json(url: str, params: dict, retries: int = 5, timeout: float = 30.0):41 full = url + "?" + urllib.parse.urlencode(params)42 for attempt in range(retries):43 try:44 req = urllib.request.Request(full, headers={"User-Agent": USER_AGENT, "Accept": "application/json"})45 with urllib.request.urlopen(req, timeout=timeout) as resp:46 return json.loads(resp.read().decode("utf-8"))47 except urllib.error.HTTPError as e:48 if e.code not in (429, 500, 502, 503, 504) or attempt == retries - 1:49 raise DataError(f"HTTP {e.code} from {full}: {e.read()[:200]!r}") from e50 except (urllib.error.URLError, TimeoutError) as e:51 if attempt == retries - 1:52 raise DataError(f"network error for {full}: {e}") from e53 time.sleep(1.5 * (attempt + 1))54 raise DataError(f"no answer from {full}")555657def fetch_coinbase(symbol: str, interval: str, start, end=None, pause: float = 0.12) -> pd.DataFrame:58 step = interval_seconds(interval)59 if step not in COINBASE_GRANULARITIES:60 raise DataError(f"coinbase does not serve interval {interval}")61 t0 = int(to_utc(start).timestamp()) // step * step62 t1 = int((to_utc(end) if end is not None else pd.Timestamp(datetime.now(timezone.utc))).timestamp())63 rows: list = []64 cur = t065 while cur < t1:66 hi = min(cur + 299 * step, t1) # both bounds are inclusive on the Coinbase side, 300 bars at most67 chunk = _get_json(68 COINBASE_URL.format(product=f"{symbol.upper()}-USD"),69 {"granularity": step,70 "start": datetime.fromtimestamp(cur, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),71 "end": datetime.fromtimestamp(hi, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")},72 )73 if not isinstance(chunk, list):74 raise DataError(f"coinbase {symbol}: unexpected answer {str(chunk)[:200]}")75 rows.extend(chunk)76 cur = hi + step77 time.sleep(pause)78 df = pd.DataFrame(rows, columns=["ts", "low", "high", "open", "close", "volume"])79 df = df.drop_duplicates(subset="ts")80 return df[["ts", *COLUMNS]]818283def fetch_kraken(symbol: str, interval: str, start=None, end=None) -> pd.DataFrame:84 step = interval_seconds(interval)85 if step % 60 or step // 60 not in KRAKEN_INTERVALS:86 raise DataError(f"kraken does not serve interval {interval}")87 pair = KRAKEN_ALIASES.get(symbol.upper(), symbol.upper()) + "USD"88 out = _get_json(KRAKEN_URL, {"pair": pair, "interval": step // 60})89 if out.get("error"):90 raise DataError(f"kraken {symbol}: {out['error']}")91 key = next(k for k in out["result"] if k != "last")92 df = pd.DataFrame(out["result"][key], columns=["ts", "open", "high", "low", "close", "vwap", "volume", "count"])93 return df[["ts", *COLUMNS]].astype("float64")949596FETCHERS = {"coinbase": fetch_coinbase, "kraken": fetch_kraken}979899def download(symbol: str, interval: str, start, end=None, *, source: str = "coinbase", out="data"):100 """Fetch, keep closed bars, merge with the existing file, write Parquet. Returns (path, Quality)."""101 if source not in FETCHERS:102 raise DataError(f"unknown source {source!r} (known: {sorted(FETCHERS)})")103 raw = FETCHERS[source](symbol, interval, start, end)104 new, _ = normalise(raw, symbol=symbol.upper(), interval=interval, source=source, start=start, end=end)105 path = Path(out) / source / f"{symbol.upper()}_{interval}.parquet"106 if path.exists():107 old = pd.read_parquet(path).set_index("ts")108 old.index = pd.to_datetime(old.index, utc=True)109 both = pd.concat([old[COLUMNS], new])110 new = both[~both.index.duplicated(keep="last")].sort_index()111 path.parent.mkdir(parents=True, exist_ok=True)112 new.reset_index().to_parquet(path, index=False)113 _, quality = normalise(new.reset_index(), symbol=symbol.upper(), interval=interval, source=source)114 return path, quality115116117def main(argv=None) -> int:118 p = argparse.ArgumentParser(prog="python -m spq.data.fetch", description=__doc__.split("\n")[0])119 p.add_argument("--symbols", required=True, help="comma-separated, e.g. ETH,BTC,LINK,SOL")120 p.add_argument("--interval", default="1h")121 p.add_argument("--start", required=True, help="UTC, inclusive (ignored by kraken: last 720 bars only)")122 p.add_argument("--end", default=None, help="UTC, exclusive (default: now)")123 p.add_argument("--source", default="coinbase", choices=sorted(FETCHERS))124 p.add_argument("--out", default="data")125 a = p.parse_args(argv)126 status = 0127 for sym in [s.strip().upper() for s in a.symbols.split(",") if s.strip()]:128 try:129 path, q = download(sym, a.interval, a.start, a.end, source=a.source, out=a.out)130 print(json.dumps({"symbol": sym, "file": str(path), "bars": q.n_bars, "start": q.start, "end": q.end,131 "missing_bars": q.missing_bars, "max_gap_bars": q.max_gap_bars,132 "ohlc_violations": q.ohlc_violations, "ok": q.ok()}))133 except DataError as e:134 print(f"{sym}: {e}", file=sys.stderr)135 status = 1136 return status137138139if __name__ == "__main__":140 sys.exit(main())141