1"""Bounded, idempotent FinData exports of raw 1-minute bars (<= 12 years per request, one symbol per request)."""2from __future__ import annotations34import datetime as dt5from pathlib import Path67import pyarrow.parquet as pq89from .api_client import FinDataClient, export_result_dict10from .provenance import append_jsonl, manifest_index1112BARS_FIELDS = ["instrument_id", "symbol_as_reported", "venue", "asset_class", "interval", "timestamp_utc", "session_date",13 "open", "high", "low", "close", "volume", "currency", "trade_count", "vwap", "adjustment_type", "source_id",14 "ingested_at", "quality_flags", "revision_id"]151617def chunk_ranges(start: dt.date, end: dt.date, max_years: int = 12) -> list[tuple[dt.date, dt.date]]:18 out, a = [], start19 while a <= end:20 b = min(end, dt.date(a.year + max_years - 1, 12, 31))21 out.append((a, b))22 a = dt.date(b.year + 1, 1, 1)23 return out242526def validate_bars_parquet(path: Path) -> dict:27 meta = pq.read_metadata(path)28 schema = pq.read_schema(path)29 names = schema.names30 if names != BARS_FIELDS:31 raise ValueError(f"unexpected bars schema in {path.name}: {names}")32 md = {k.decode(): v.decode() for k, v in (schema.metadata or {}).items()}33 if md.get("findata.schema") != "bars" or md.get("findata.timestamp_convention") != "bar_open":34 raise ValueError(f"missing/unexpected FinData schema metadata in {path.name}: {md}")35 return {"rows": meta.num_rows, "schema_version": md.get("findata.schema_version")}363738def export_symbol(client: FinDataClient, symbol: str, start: dt.date, end: dt.date, dataset_id: str, cache_dir: Path,39 manifest: Path, keep: bool = False) -> list[Path]:40 """Export all chunks of one symbol; skip chunks already in the manifest when the file is still present (or keep=False41 and the chunk was already consumed). Returns the list of Parquet paths available on disk."""42 idx = manifest_index(manifest, ["symbol", "start", "end", "dataset_id"])43 paths = []44 for a, b in chunk_ranges(start, end):45 dest = cache_dir / f"{symbol}__{a}__{b}.parquet"46 key = (symbol, str(a), str(b), dataset_id)47 if dest.exists():48 paths.append(dest)49 continue50 if key in idx and idx[key].get("consumed") and not keep:51 # already processed in a previous run; downstream will find the feature rows52 continue53 res = client.export(dataset_id, [symbol], str(a), str(b), dest)54 info = validate_bars_parquet(dest)55 rec = export_result_dict(res)56 rec.update({"symbol": symbol, "rows_in_file": info["rows"], "schema_version": info["schema_version"], "consumed": False})57 append_jsonl(manifest, rec)58 paths.append(dest)59 return paths606162def mark_consumed(manifest: Path, symbol: str, dataset_id: str) -> None:63 append_jsonl(manifest, {"symbol": symbol, "dataset_id": dataset_id, "consumed": True, "event": "features_written"})64