1"""Incremental sync of an update archive (period=day|week|month) into session_date delta files.23Idempotent: the same archive (sha256) is never published twice; a session_date already live is *superseded*4(old file retired, new file added, dataset version +1) so re-fetching cannot create duplicates.5Checkpoints: dataset:<id> / last_archive, last_vendor_update.6"""7from __future__ import annotations89from collections import defaultdict10from datetime import UTC, datetime11from typing import Any1213import pyarrow as pa1415from findata_core.db import connection, transaction16from findata_ingest import queue17from findata_ingest.lake import Lake18from findata_ingest.tasks.common import (DATASET_ASSET, live_delta_paths_for_dates, record_anomalies, record_quality_report,19 record_raw_archive, upsert_coverage)20from findata_market_firstrate import FirstRateMarketConnector21from findata_schema import Interval22from findata_schema.connector import FetchRequest232425def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:26 p = job.get("params") or {}27 asset = p.get("type", "stock")28 period = p.get("period", "day")29 timeframe = p.get("timeframe", "1min")30 adjustment = p.get("adjustment", "UNADJUSTED")31 dataset_id = job.get("dataset_id") or DATASET_ASSET[asset]32 run_id = job.get("run_id")33 conn_fr = FirstRateMarketConnector()34 vendor_last = conn_fr.client.last_update(asset)35 ck = queue.get_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}") or {}36 if ck.get("vendor_last_update") == vendor_last.isoformat() and not p.get("force"):37 log.info("incremental.skip", reason="vendor not updated since last run", vendor_last_update=str(vendor_last))38 return {"skipped": True, "vendor_last_update": vendor_last.isoformat()}3940 req = FetchRequest(dataset_id=dataset_id, params={"endpoint": "data_file", "type": asset, "period": period,41 "timeframe": timeframe, "adjustment": adjustment})42 queue.progress(job["id"], {"stage": "download"})43 res = conn_fr.fetch_range(req, lake.raw_dir("firstrate", dataset_id))44 rel_raw = str(res.path.relative_to(lake.root))45 with transaction() as conn:46 raw_id, is_new = record_raw_archive(conn, "firstrate", dataset_id, req.params, rel_raw, res.sha256, res.bytes, res.file_count, res.vendor_last_update, run_id)47 if not is_new:48 res.path.unlink(missing_ok=True) # identical bytes already stored immutably49 queue.set_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}", {"vendor_last_update": vendor_last.isoformat(), "sha256": res.sha256})50 log.info("incremental.duplicate_archive", sha256=res.sha256[:12])51 return {"skipped": True, "reason": "archive already ingested", "sha256": res.sha256}52 log.info("incremental.downloaded", bytes=res.bytes, files=res.file_count, vendor_last_update=str(vendor_last))5354 # normalize → group rows by session_date (across all symbols)55 queue.progress(job["id"], {"stage": "normalize", "bytes": res.bytes})56 by_date: dict[Any, list[pa.Table]] = defaultdict(list)57 anomalies: list[dict[str, Any]] = []58 symbols = 059 raw_rows = 060 dropped = 061 for batch in conn_fr.normalize(res):62 rep = conn_fr.validate(batch)63 anomalies.extend(rep.anomalies)64 symbols += 165 raw_rows += batch.raw_rows66 dropped += batch.raw_rows - batch.row_count67 t = batch.table68 for d in set(t.column("session_date").to_pylist()):69 import pyarrow.compute as pc70 by_date[d].append(t.filter(pc.equal(t.column("session_date"), pa.scalar(d, pa.date32()))))71 if symbols % 1000 == 0:72 queue.progress(job["id"], {"stage": "normalize", "symbols": symbols})73 if not by_date:74 raise RuntimeError("archive produced no rows")7576 # write one delta file per session_date, then publish atomically (supersede existing dates)77 queue.progress(job["id"], {"stage": "write", "sessions": len(by_date)})78 interval = Interval({"1min": "1m", "5min": "5m", "30min": "30m", "1hour": "1h", "1day": "1d"}[timeframe])79 written = []80 tables = {}81 for d in sorted(by_date):82 t = pa.concat_tables(by_date[d])83 tables[d] = t84 written.append(lake.write_table(dataset_id, t, "delta", {"session_date": d.isoformat()}))85 with connection() as conn:86 retire = live_delta_paths_for_dates(conn, dataset_id, list(tables))87 pub = lake.publish(dataset_id, written, retire, run_id, note=f"incremental {asset} {period} {timeframe} {adjustment}")88 with transaction() as conn:89 conn.execute("UPDATE raw_archives SET status='published' WHERE id=%s", (raw_id,))90 cov = sum(upsert_coverage(conn, dataset_id, t, interval) for t in tables.values())91 record_anomalies(conn, dataset_id, anomalies, run_id)92 record_quality_report(conn, dataset_id, run_id, sum(t.num_rows for t in tables.values()), anomalies)93 queue.set_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}", {"vendor_last_update": vendor_last.isoformat(), "sha256": res.sha256})94 cp = conn_fr.checkpoint(res, len(written))95 queue.set_checkpoint(cp.scope, cp.key, cp.value)96 dates = [d.isoformat() for d in sorted(tables)]97 # downstream: derived 5-minute bars for these sessions98 if interval == Interval.m1 and asset in ("stock", "etf"):99 queue.enqueue("derive", dataset_id.replace(".1m.", ".5m."), {"source_dataset": dataset_id, "target_interval": "5m", "dates": dates},100 priority=40, dedupe_key=f"derive:{dataset_id}:5m:{dates[0]}:{dates[-1]}:{pub['version']}")101 result = {"vendor_last_update": vendor_last.isoformat(), "archive_sha256": res.sha256, "archive_bytes": res.bytes,102 "symbols": symbols, "raw_rows": raw_rows, "rows_dropped_unparseable": dropped,103 "rows_published": sum(t.num_rows for t in tables.values()), "sessions": dates, "files_added": len(written),104 "files_retired": len(retire), "coverage_rows": cov, "anomalies": len(anomalies), **pub}105 log.info("incremental.done", **{k: v for k, v in result.items() if k != "sessions"})106 return result107