1"""Bounded, resumable, license-safe download of 1-minute bars through POST /exports (docs/DATA_CARD.md §ingestion).23Layout of the private cache (never in git):4 <cache>/bars/<SYMBOL>.parquet one Parquet per instrument for the whole sample window (raw, split later)5 <cache>/instruments/<SYMBOL>.json instrument detail (corporate actions, coverage) at retrieval time6 <cache>/manifests/bars_manifest.jsonl one line per export: dataset version, sha256, request id, rows, time7A download is skipped when the Parquet exists and its sha256 matches the manifest line. Disk quota is checked before8each export; the process stops cleanly when free space drops below `min_free_gb`.9"""10from __future__ import annotations1112import json13import shutil14import time15from concurrent.futures import ThreadPoolExecutor, as_completed16from dataclasses import asdict17from pathlib import Path1819from .client import FinDataClient, FinDataError, sha256_file202122def _manifest_index(manifest: Path) -> dict[str, dict]:23 idx = {}24 if manifest.exists():25 for line in manifest.read_text().splitlines():26 if line.strip():27 rec = json.loads(line)28 idx[rec["symbols"][0]] = rec29 return idx303132def _free_gb(path: Path) -> float:33 return shutil.disk_usage(path).free / 1e9343536def fetch_instrument(client: FinDataClient, symbol: str, cache: Path) -> dict:37 p = cache / "instruments" / f"{symbol}.json"38 if p.exists():39 return json.loads(p.read_text())40 d = client.instrument(symbol)41 d["_retrieved_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())42 p.parent.mkdir(parents=True, exist_ok=True)43 p.write_text(json.dumps(d))44 return d454647def download_symbol(client: FinDataClient, symbol: str, start: str, end: str, cache: Path, dataset_id: str,48 manifest: Path, min_free_gb: float = 40.0) -> dict:49 dest = cache / "bars" / f"{symbol}.parquet"50 idx = _manifest_index(manifest)51 if dest.exists() and symbol in idx and idx[symbol]["sha256"] == sha256_file(dest) \52 and idx[symbol]["start"] == start and idx[symbol]["end"] == end:53 return {"symbol": symbol, "status": "cached", **{k: idx[symbol][k] for k in ("row_count", "dataset_version")}}54 if _free_gb(cache) < min_free_gb:55 raise RuntimeError(f"disk quota: {_free_gb(cache):.1f} GB free < {min_free_gb} GB")56 res = client.export([symbol], start, end, dest, dataset_id=dataset_id, adjustment="raw")57 with open(manifest, "a") as fh:58 fh.write(json.dumps(asdict(res)) + "\n")59 fetch_instrument(client, symbol, cache)60 return {"symbol": symbol, "status": "downloaded", "row_count": res.row_count, "dataset_version": res.dataset_version}616263def download_universe(client: FinDataClient, symbols: list[str], start: str, end: str, cache: Path,64 dataset_for: dict[str, str], workers: int = 3, min_free_gb: float = 40.0) -> list[dict]:65 cache.mkdir(parents=True, exist_ok=True)66 (cache / "manifests").mkdir(exist_ok=True)67 manifest = cache / "manifests" / "bars_manifest.jsonl"68 out = []69 with ThreadPoolExecutor(max_workers=workers) as ex:70 futs = {ex.submit(download_symbol, client, s, start, end, cache, dataset_for[s], manifest, min_free_gb): s71 for s in symbols}72 for f in as_completed(futs):73 s = futs[f]74 try:75 out.append(f.result())76 except (FinDataError, RuntimeError) as exc:77 out.append({"symbol": s, "status": "failed", "error": str(exc)})78 return out79