1#!/usr/bin/env python32"""Phase 2 — ingest 1-minute and daily bars for the frozen universe into the private cache (resumable).3usage: ingest_bars.py [--config configs/instruments.yaml] [--classes etf,stock,...] [--symbols SPY,QQQ] [--workers 3]4Writes results/manifests/ingest_manifest.json (dataset versions, sha256, row counts) — no bar values."""5from __future__ import annotations67import argparse8import json9import sys10from concurrent.futures import ThreadPoolExecutor, as_completed11from pathlib import Path1213import yaml1415ROOT = Path(__file__).resolve().parents[1]16sys.path.insert(0, str(ROOT / "src"))17from wp46.findata_client import FinDataClient # noqa: E40218from wp46.findata_client.ingest import DAILY_FOR, DATASET_FOR, ingest_symbol # noqa: E402192021def main() -> None:22 ap = argparse.ArgumentParser()23 ap.add_argument("--config", default=str(ROOT / "configs" / "instruments.yaml"))24 ap.add_argument("--classes", default="etf,stock,futures,fx,crypto")25 ap.add_argument("--symbols", default="")26 ap.add_argument("--start", default="2000-01-01")27 ap.add_argument("--end", default="2026-10-30")28 ap.add_argument("--workers", type=int, default=3)29 ap.add_argument("--no-daily", action="store_true")30 a = ap.parse_args()31 cfg = yaml.safe_load(open(a.config))32 uni = cfg.get("universe") or cfg["candidates"]33 want = set(a.symbols.split(",")) if a.symbols else None34 jobs = []35 for cls in a.classes.split(","):36 for s in uni.get(cls, []):37 if want and s not in want:38 continue39 jobs.append((cls, s))40 fd = FinDataClient(log_path=ROOT / "results" / "logs" / "ingest_provenance.jsonl")41 man_path = ROOT / "results" / "manifests" / "ingest_manifest.json"42 manifest = json.loads(man_path.read_text()) if man_path.exists() else {}4344 def work(cls, s):45 recs = {"minute": ingest_symbol(fd, DATASET_FOR[cls], s, a.start, a.end)}46 if not a.no_daily and cls in DAILY_FOR:47 recs["daily"] = ingest_symbol(fd, DAILY_FOR[cls], s, a.start, a.end)48 return cls, s, recs4950 with ThreadPoolExecutor(a.workers) as ex:51 futs = [ex.submit(work, c, s) for c, s in jobs]52 for f in as_completed(futs):53 cls, s, recs = f.result()54 manifest[f"{cls}:{s}"] = recs55 n = sum(r.get("row_count") or 0 for r in recs["minute"])56 print(f"{cls:8s} {s:8s} 1m rows={n:,} chunks={len(recs['minute'])} daily_chunks={len(recs.get('daily', []))}", flush=True)57 man_path.parent.mkdir(parents=True, exist_ok=True)58 man_path.write_text(json.dumps(manifest, indent=1))59 # regime variables (index daily)60 for s in cfg.get("regime_variables", {}).get("index", []):61 manifest[f"index:{s}"] = {"daily": ingest_symbol(fd, DAILY_FOR["index"], s, a.start, a.end)}62 man_path.write_text(json.dumps(manifest, indent=1))63 print("manifest", man_path)646566if __name__ == "__main__":67 main()68