1"""P1/P4: download 1-minute bars for a symbol set through /exports (resumable, quota-checked) and build dense grids.2Usage: download_pilot.py --config configs/pilot.yaml [--symbols-from pilot|feasibility|all] [--workers 3]"""3from __future__ import annotations45import argparse6import json7import sys8import time9from concurrent.futures import ProcessPoolExecutor, as_completed10from pathlib import Path1112import polars as pl1314sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))15from chronos_x.config import load_config # noqa: E40216from chronos_x.data.client import FinDataClient # noqa: E40217from chronos_x.data.exports import download_universe, fetch_instrument # noqa: E40218from chronos_x.data.features import build_grid, save_grid # noqa: E4021920ROOT = Path(__file__).resolve().parents[1]212223def symbols_for(man: dict, which: str) -> list[dict]:24 inst = {r["symbol"]: r for r in man["instruments"]}25 if which == "pilot":26 return [inst[s] for s in man["pilot_symbols"]]27 if which == "feasibility":28 return [inst[s] for s in man["feasibility_symbols"]]29 return list(inst.values())303132def build_one(args) -> dict:33 sym, ac, cache, start, end, version = args34 t0 = time.time()35 raw = pl.read_parquet(Path(cache) / "bars" / f"{sym}.parquet")36 inst = json.loads((Path(cache) / "instruments" / f"{sym}.json").read_text())37 arrays, meta = build_grid(raw, inst.get("corporate_actions", []), start, end, sym, ac, version)38 save_grid(arrays, meta, Path(cache) / "grids" / f"{start}_{end}" / sym)39 return {"symbol": sym, "seconds": round(time.time() - t0, 1), **meta.counts, "qa": meta.qa}404142def main() -> None:43 ap = argparse.ArgumentParser()44 ap.add_argument("--config", default="configs/pilot.yaml")45 ap.add_argument("--symbols-from", default="pilot", choices=["pilot", "feasibility", "all"])46 ap.add_argument("--workers", type=int, default=3)47 ap.add_argument("--skip-download", action="store_true")48 a = ap.parse_args()49 cfg = load_config(a.config)50 man = json.loads((ROOT / cfg["universe"]["manifest"]).read_text())51 rows = symbols_for(man, a.symbols_from)52 cache = Path(cfg["paths"]["cache"])53 start, end = cfg["data"]["start"], cfg["data"]["end"]54 c = FinDataClient(log_path=ROOT / "results" / "logs" / f"provenance_download_{cfg['name']}.jsonl")55 out_dir = ROOT / "results" / "metrics"; out_dir.mkdir(parents=True, exist_ok=True)56 if not a.skip_download:57 t0 = time.time()58 ds_for = {r["symbol"]: cfg["data"]["dataset_etf"] if r["asset_class"] == "etf" else cfg["data"]["dataset_stock"]59 for r in rows}60 res = download_universe(c, [r["symbol"] for r in rows], start, end, cache, ds_for, workers=a.workers)61 summary = {"config": cfg["name"], "symbols_from": a.symbols_from, "n": len(res), "start": start, "end": end,62 "downloaded": sum(r["status"] == "downloaded" for r in res), "cached": sum(r["status"] == "cached" for r in res),63 "failed": [r for r in res if r["status"] == "failed"], "rows": sum(r.get("row_count", 0) for r in res),64 "dataset_versions": sorted({r.get("dataset_version") for r in res if r.get("dataset_version")}),65 "wall_s": round(time.time() - t0, 1)}66 (out_dir / f"download_{cfg['name']}_{a.symbols_from}.json").write_text(json.dumps(summary, indent=1))67 print(json.dumps({k: v for k, v in summary.items() if k != "failed"}), "failed:", len(summary["failed"]))68 # grids69 t0 = time.time()70 jobs = []71 for r in rows:72 p = cache / "bars" / f"{r['symbol']}.parquet"73 if p.exists():74 fetch_instrument(c, r["symbol"], cache)75 jobs.append((r["symbol"], r["asset_class"], str(cache), start, end, None))76 results = []77 with ProcessPoolExecutor(max_workers=max(1, a.workers * 2)) as ex:78 for f in as_completed([ex.submit(build_one, j) for j in jobs]):79 results.append(f.result())80 results.sort(key=lambda r: r["symbol"])81 cov = [r["coverage"] for r in results]82 rep = {"config": cfg["name"], "symbols_from": a.symbols_from, "n_grids": len(results), "wall_s": round(time.time() - t0, 1),83 "coverage_mean": sum(cov) / max(1, len(cov)), "coverage_min": min(cov) if cov else None,84 "rows_regular": sum(r["rows_regular"] for r in results), "rows_extended": sum(r["rows_extended_hours"] for r in results),85 "rows_dropped_integrity": sum(r["rows_dropped_integrity"] for r in results),86 "duplicates_in_raw": sum(r["qa"].get("duplicates", 0) for r in results), "per_symbol": results}87 (out_dir / f"grids_{cfg['name']}_{a.symbols_from}.json").write_text(json.dumps(rep, indent=1))88 print(json.dumps({k: v for k, v in rep.items() if k != "per_symbol"}))899091if __name__ == "__main__":92 main()93