"""Bounded historical backfill from `period=full&ticker_range=` archives → base files per (symbol, year). Streams one vendor file (one ticker, full history) at a time so memory stays flat. Dates that already have a live delta file are excluded (they are compacted later) so the no-duplicate invariant holds. Resumable: progress checkpoint `job:/done_symbols`; a restarted job skips symbols already published. """ from __future__ import annotations from typing import Any import pyarrow as pa import pyarrow.compute as pc from findata_core.db import connection, transaction from findata_market_firstrate import FirstRateMarketConnector from findata_schema import Interval from findata_schema.connector import FetchRequest from findata_ingest import queue from findata_ingest.lake import Lake from findata_ingest.tasks.common import ( DATASET_ASSET, live_delta_dates, record_anomalies, record_quality_report, record_raw_archive, upsert_coverage, ) def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: p = job.get("params") or {} asset = p.get("type", "stock") letter = p["ticker_range"] timeframe = p.get("timeframe", "1min") adjustment = p.get("adjustment", "UNADJUSTED") only = set(s.upper() for s in (p.get("symbols") or [])) max_symbols = int(p.get("max_symbols") or 0) dataset_id = job.get("dataset_id") or DATASET_ASSET[asset] run_id = job.get("run_id") interval = Interval({"1min": "1m", "5min": "5m", "30min": "30m", "1hour": "1h", "1day": "1d"}[timeframe]) conn_fr = FirstRateMarketConnector() ck = queue.get_checkpoint(f"job:{job['id']}", "state") or {} done = set(ck.get("done_symbols") or []) raw_rel = ck.get("raw_path") if raw_rel and (lake.root / raw_rel).exists(): from datetime import datetime from findata_schema.connector import FetchResult res = FetchResult(request=FetchRequest(dataset_id, {"type": asset, "period": "full", "ticker_range": letter, "timeframe": timeframe, "adjustment": adjustment}), path=lake.root / raw_rel, sha256=ck["sha256"], bytes=ck["bytes"], fetched_at=datetime.fromisoformat(ck["fetched_at"]), vendor_last_update=None) log.info("backfill.resume", done=len(done), archive=raw_rel) else: req = FetchRequest(dataset_id=dataset_id, params={"endpoint": "data_file", "type": asset, "period": "full", "ticker_range": letter, "timeframe": timeframe, "adjustment": adjustment}) queue.progress(job["id"], {"stage": "download", "ticker_range": letter}) res = conn_fr.fetch_range(req, lake.raw_dir("firstrate", dataset_id)) raw_rel = str(res.path.relative_to(lake.root)) with transaction() as conn: record_raw_archive(conn, "firstrate", dataset_id, req.params, raw_rel, res.sha256, res.bytes, res.file_count, res.vendor_last_update, run_id) queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes, "fetched_at": res.fetched_at.isoformat(), "done_symbols": sorted(done)}) log.info("backfill.downloaded", bytes=res.bytes, files=res.file_count) with connection() as conn: delta_dates = live_delta_dates(conn, dataset_id) stats = {"symbols": 0, "rows": 0, "files": 0, "skipped_rows_in_delta": 0, "anomalies": 0} for batch in conn_fr.normalize(res, symbols=list(only) if only else None): if batch.symbol in done: continue if max_symbols and stats["symbols"] >= max_symbols: break rep = conn_fr.validate(batch) t = batch.table if delta_dates: mask = pc.invert(pc.is_in(t.column("session_date"), value_set=pa.array(sorted(delta_dates), pa.date32()))) kept = t.filter(mask) stats["skipped_rows_in_delta"] += t.num_rows - kept.num_rows t = kept if t.num_rows == 0: done.add(batch.symbol) queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes, "fetched_at": res.fetched_at.isoformat(), "done_symbols": sorted(done)}) continue years = pc.year(t.column("session_date")) written = [] for y in sorted(set(years.to_pylist())): ty = t.filter(pc.equal(years, y)) written.append(lake.write_table(dataset_id, ty, "base", {"symbol": batch.symbol, "year": str(y)})) # retire previous base files of this symbol/year (re-backfill replaces) with connection() as conn: old = conn.execute("SELECT path FROM lake_files WHERE dataset_id=%s AND layout='base' AND version_removed IS NULL AND symbol_min=%s AND symbol_max=%s", (dataset_id, batch.symbol, batch.symbol)).fetchall() lake.publish(dataset_id, written, [o["path"] for o in old], run_id, note=f"backfill {asset} {letter} {batch.symbol}") with transaction() as conn: upsert_coverage(conn, dataset_id, t, interval) record_anomalies(conn, dataset_id, rep.anomalies, run_id, limit=500) stats["symbols"] += 1 stats["rows"] += t.num_rows stats["files"] += len(written) stats["anomalies"] += len(rep.anomalies) done.add(batch.symbol) queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes, "fetched_at": res.fetched_at.isoformat(), "done_symbols": sorted(done)}) queue.progress(job["id"], {"stage": "publish", **stats, "last_symbol": batch.symbol}) with transaction() as conn: conn.execute("UPDATE raw_archives SET status='published' WHERE sha256=%s", (res.sha256,)) record_quality_report(conn, dataset_id, run_id, stats["rows"], []) if not p.get("keep_raw", False) and p.get("delete_raw_after_publish", True): # full archives are re-downloadable; sha256 stays in raw_archives res.path.unlink(missing_ok=True) with transaction() as conn: conn.execute("UPDATE raw_archives SET status='deleted_after_publish', path=NULL WHERE sha256=%s", (res.sha256,)) if interval == Interval.m1 and stats["rows"]: queue.enqueue("derive", dataset_id.replace(".1m.", ".5m."), {"source_dataset": dataset_id, "target_interval": "5m", "symbols": sorted(done)}, priority=45, dedupe_key=f"derive:{dataset_id}:5m:backfill:{job['id']}") log.info("backfill.done", **stats) return stats