1"""Bounded historical backfill from `period=full&ticker_range=<letter>` archives → base files per (symbol, year).23Streams one vendor file (one ticker, full history) at a time so memory stays flat. Dates that already have a live4delta file are excluded (they are compacted later) so the no-duplicate invariant holds. Resumable: progress5checkpoint `job:<id>/done_symbols`; a restarted job skips symbols already published.6"""7from __future__ import annotations89from typing import Any1011import pyarrow as pa12import pyarrow.compute as pc13from findata_core.db import connection, transaction14from findata_market_firstrate import FirstRateMarketConnector, FRDNoData15from findata_schema import Interval16from findata_schema.connector import FetchRequest1718from findata_ingest import queue19from findata_ingest.lake import Lake20from findata_ingest.tasks.common import (21 dataset_for_params,22 live_delta_dates,23 record_anomalies,24 record_quality_report,25 record_raw_archive,26 upsert_coverage,27)2829PUBLISH_BATCH = 20303132def _release_memory() -> None:33 """Return freed Arrow/Polars buffers to the OS between symbols (RSS otherwise grows to several GB per worker)."""34 import contextlib35 import gc36 gc.collect()37 with contextlib.suppress(Exception):38 pa.default_memory_pool().release_unused()394041def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:42 p = job.get("params") or {}43 asset = p.get("type", "stock")44 letter = p.get("ticker_range") or p.get("archive_number") or p.get("update") or p.get("contract_files") or p.get("period", "full")45 timeframe = p.get("timeframe", "1min")46 adjustment = p.get("adjustment")47 only = set(s.upper() for s in (p.get("symbols") or []))48 max_symbols = int(p.get("max_symbols") or 0)49 dataset_id = job.get("dataset_id") or dataset_for_params(p)50 run_id = job.get("run_id")51 interval = Interval({"1min": "1m", "5min": "5m", "30min": "30m", "1hour": "1h", "1day": "1d"}[timeframe])52 conn_fr = FirstRateMarketConnector()53 vendor_params = {k: v for k, v in p.items() if k in ("endpoint", "type", "period", "ticker_range", "timeframe", "adjustment",54 "archive_number", "update", "contract_files", "contracts", "year", "quarter") and v is not None}55 vendor_params.setdefault("endpoint", "data_file")56 if vendor_params["endpoint"] == "data_file":57 vendor_params.setdefault("period", "full")5859 ck = queue.get_checkpoint(f"job:{job['id']}", "state") or {}60 done = set(ck.get("done_symbols") or [])61 raw_rel = ck.get("raw_path")62 if raw_rel and (lake.root / raw_rel).exists():63 from datetime import datetime6465 from findata_schema.connector import FetchResult66 res = FetchResult(request=FetchRequest(dataset_id, vendor_params),67 path=lake.abs(raw_rel), sha256=ck["sha256"], bytes=ck["bytes"], fetched_at=datetime.fromisoformat(ck["fetched_at"]),68 vendor_last_update=None)69 log.info("backfill.resume", done=len(done), archive=raw_rel)70 else:71 req = FetchRequest(dataset_id=dataset_id, params=vendor_params)72 queue.progress(job["id"], {"stage": "download", "archive": letter})73 try:74 res = conn_fr.fetch_range(req, lake.raw_dir("firstrate", dataset_id))75 except FRDNoData as e:76 log.info("backfill.no_data", reason=str(e)[:200])77 return {"skipped": True, "reason": str(e)[:300]}78 raw_rel = lake.rel(res.path)79 with transaction() as conn:80 record_raw_archive(conn, "firstrate", dataset_id, req.params, raw_rel, res.sha256, res.bytes, res.file_count, res.vendor_last_update, run_id)81 queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes,82 "fetched_at": res.fetched_at.isoformat(), "done_symbols": sorted(done)})83 log.info("backfill.downloaded", bytes=res.bytes, files=res.file_count)8485 with connection() as conn:86 delta_dates = live_delta_dates(conn, dataset_id)87 stats = {"symbols": 0, "rows": 0, "files": 0, "skipped_rows_in_delta": 0, "anomalies": 0}88 pending_written: list = []89 pending_retire: list[str] = []90 pending_symbols: list[str] = []91 pending_tables: list = []9293 def _flush() -> None:94 """Publish the accumulated symbols in one version (one transaction), then checkpoint."""95 if pending_written:96 lake.publish(dataset_id, list(pending_written), sorted(set(pending_retire)), run_id,97 note=f"backfill {asset} {letter} {timeframe} {adjustment} {len(pending_symbols)} symbols")98 with transaction() as conn:99 for t_, anomalies_ in pending_tables:100 upsert_coverage(conn, dataset_id, t_, interval)101 record_anomalies(conn, dataset_id, anomalies_, run_id, limit=40)102 for sym in pending_symbols:103 done.add(sym)104 if pending_symbols:105 queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes,106 "fetched_at": res.fetched_at.isoformat(), "done_symbols": sorted(done)})107 queue.progress(job["id"], {"stage": "publish", **stats, "last_symbol": pending_symbols[-1]})108 for lst in (pending_written, pending_retire, pending_symbols, pending_tables):109 lst.clear()110111 for batch in conn_fr.normalize(res, symbols=list(only) if only else None):112 if batch.symbol in done:113 continue114 if max_symbols and stats["symbols"] >= max_symbols:115 break116 rep = conn_fr.validate(batch)117 t = batch.table118 if delta_dates:119 mask = pc.invert(pc.is_in(t.column("session_date"), value_set=pa.array(sorted(delta_dates), pa.date32())))120 kept = t.filter(mask)121 stats["skipped_rows_in_delta"] += t.num_rows - kept.num_rows122 t = kept123 if t.num_rows == 0:124 done.add(batch.symbol)125 queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes,126 "fetched_at": res.fetched_at.isoformat(), "done_symbols": sorted(done)})127 continue128 years = pc.year(t.column("session_date"))129 written = []130 for y in sorted(set(years.to_pylist())):131 ty = t.filter(pc.equal(years, y))132 written.append(lake.write_table(dataset_id, ty, "base", {"symbol": batch.symbol, "year": str(y)}))133 # retire previous base files of this symbol (re-backfill / full_update replaces the whole symbol history)134 with connection() as conn:135 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",136 (dataset_id, batch.symbol, batch.symbol)).fetchall()137 pending_written.extend(written)138 pending_retire.extend(o["path"] for o in old)139 pending_symbols.append(batch.symbol)140 pending_tables.append((t, rep.anomalies))141 stats["symbols"] += 1142 stats["rows"] += t.num_rows143 stats["files"] += len(written)144 stats["anomalies"] += len(rep.anomalies)145 del t, rep, written146 if len(pending_symbols) >= PUBLISH_BATCH:147 _flush()148 if stats["symbols"] % 10 == 0:149 _release_memory()150 _flush()151 with transaction() as conn:152 conn.execute("UPDATE raw_archives SET status='published' WHERE sha256=%s", (res.sha256,))153 if stats["rows"]:154 conn.execute("UPDATE datasets SET status='active', updated_at=now() WHERE id=%s AND status='planned'", (dataset_id,))155 record_quality_report(conn, dataset_id, run_id, stats["rows"], [])156 if p.get("delete_raw_after_publish", False):157 # full archives are re-downloadable; sha256 stays in raw_archives158 res.path.unlink(missing_ok=True)159 with transaction() as conn:160 conn.execute("UPDATE raw_archives SET status='deleted_after_publish', path=NULL WHERE sha256=%s", (res.sha256,))161 if interval == Interval.m1 and stats["rows"] and dataset_id in ("market_bars.stock.1m.raw", "market_bars.etf.1m.raw"):162 queue.enqueue("derive", dataset_id.replace(".1m.", ".5m."), {"source_dataset": dataset_id, "target_interval": "5m",163 "symbols": sorted(done)}, priority=45, dedupe_key=f"derive:{dataset_id}:5m:backfill:{job['id']}")164 if conn_fr.skipped_files:165 stats["files_skipped_unparseable"] = len(conn_fr.skipped_files)166 stats["skipped_examples"] = conn_fr.skipped_files[:5]167 log.warning("backfill.skipped_files", n=len(conn_fr.skipped_files), examples=conn_fr.skipped_files[:3])168 log.info("backfill.done", **{k: v for k, v in stats.items() if k != "skipped_examples"})169 return stats170