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 typing import Any1112import pyarrow as pa13from findata_core.db import connection, transaction14from findata_market_firstrate import FirstRateMarketConnector15from findata_market_firstrate.client import FRDNoData16from findata_schema import Interval17from findata_schema.connector import FetchRequest1819from findata_ingest import queue20from findata_ingest.lake import Lake21from findata_ingest.tasks.common import (22 dataset_for_params,23 live_delta_paths_for_dates,24 record_anomalies,25 record_quality_report,26 record_raw_archive,27 upsert_coverage,28)293031def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:32 p = job.get("params") or {}33 asset = p.get("type", "stock")34 period = p.get("period", "day")35 timeframe = p.get("timeframe", "1min")36 adjustment = p.get("adjustment") or ("contin_UNadj" if asset == "futures" else ("UNADJUSTED" if asset in ("stock", "etf") else None))37 dataset_id = job.get("dataset_id") or dataset_for_params({"type": asset, "timeframe": timeframe, "adjustment": adjustment})38 run_id = job.get("run_id")39 conn_fr = FirstRateMarketConnector()40 vendor_last = conn_fr.client.last_update(asset)41 ck = queue.get_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}") or {}42 if ck.get("vendor_last_update") == vendor_last.isoformat() and not p.get("force"):43 log.info("incremental.skip", reason="vendor not updated since last run", vendor_last_update=str(vendor_last))44 return {"skipped": True, "vendor_last_update": vendor_last.isoformat()}4546 req = FetchRequest(dataset_id=dataset_id, params={k: v for k, v in {"endpoint": "data_file", "type": asset, "period": period,47 "timeframe": timeframe, "adjustment": adjustment}.items() if v is not None})48 queue.progress(job["id"], {"stage": "download"})49 try:50 res = conn_fr.fetch_range(req, lake.raw_dir("firstrate", dataset_id))51 except FRDNoData as e:52 queue.set_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}", {"vendor_last_update": vendor_last.isoformat(), "no_data": True})53 log.info("incremental.no_data", reason=str(e)[:200])54 return {"skipped": True, "reason": str(e)[:300], "vendor_last_update": vendor_last.isoformat()}55 rel_raw = lake.rel(res.path)56 with transaction() as conn:57 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)58 if not is_new:59 res.path.unlink(missing_ok=True) # identical bytes already stored immutably60 queue.set_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}", {"vendor_last_update": vendor_last.isoformat(), "sha256": res.sha256})61 log.info("incremental.duplicate_archive", sha256=res.sha256[:12])62 return {"skipped": True, "reason": "archive already ingested", "sha256": res.sha256}63 log.info("incremental.downloaded", bytes=res.bytes, files=res.file_count, vendor_last_update=str(vendor_last))6465 # normalize → group rows by session_date (across all symbols)66 queue.progress(job["id"], {"stage": "normalize", "bytes": res.bytes})67 by_date: dict[Any, list[pa.Table]] = defaultdict(list)68 anomalies: list[dict[str, Any]] = []69 symbols = 070 raw_rows = 071 dropped = 072 for batch in conn_fr.normalize(res):73 rep = conn_fr.validate(batch)74 anomalies.extend(rep.anomalies)75 symbols += 176 raw_rows += batch.raw_rows77 dropped += batch.raw_rows - batch.row_count78 t = batch.table79 for d in set(t.column("session_date").to_pylist()):80 import pyarrow.compute as pc81 by_date[d].append(t.filter(pc.equal(t.column("session_date"), pa.scalar(d, pa.date32()))))82 if symbols % 1000 == 0:83 queue.progress(job["id"], {"stage": "normalize", "symbols": symbols})84 if not by_date:85 raise RuntimeError("archive produced no rows")8687 # write one delta file per session_date, then publish atomically (supersede existing dates)88 queue.progress(job["id"], {"stage": "write", "sessions": len(by_date)})89 interval = Interval({"1min": "1m", "5min": "5m", "30min": "30m", "1hour": "1h", "1day": "1d"}[timeframe])90 written = []91 tables = {}92 for d in sorted(by_date):93 t = pa.concat_tables(by_date[d])94 tables[d] = t95 written.append(lake.write_table(dataset_id, t, "delta", {"session_date": d.isoformat()}))96 with connection() as conn:97 retire = live_delta_paths_for_dates(conn, dataset_id, list(tables))98 pub = lake.publish(dataset_id, written, retire, run_id, note=f"incremental {asset} {period} {timeframe} {adjustment}")99 with transaction() as conn:100 conn.execute("UPDATE raw_archives SET status='published' WHERE id=%s", (raw_id,))101 conn.execute("UPDATE datasets SET status='active', updated_at=now() WHERE id=%s AND status='planned'", (dataset_id,))102 cov = sum(upsert_coverage(conn, dataset_id, t, interval) for t in tables.values())103 record_anomalies(conn, dataset_id, anomalies, run_id)104 record_quality_report(conn, dataset_id, run_id, sum(t.num_rows for t in tables.values()), anomalies)105 queue.set_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}", {"vendor_last_update": vendor_last.isoformat(), "sha256": res.sha256})106 cp = conn_fr.checkpoint(res, len(written))107 queue.set_checkpoint(cp.scope, cp.key, cp.value)108 dates = [d.isoformat() for d in sorted(tables)]109 # downstream: derived 5-minute bars for these sessions110 if interval == Interval.m1 and asset in ("stock", "etf") and dataset_id.endswith(".raw"):111 queue.enqueue("derive", dataset_id.replace(".1m.", ".5m."), {"source_dataset": dataset_id, "target_interval": "5m", "dates": dates},112 priority=40, dedupe_key=f"derive:{dataset_id}:5m:{dates[0]}:{dates[-1]}:{pub['version']}")113 result = {"vendor_last_update": vendor_last.isoformat(), "archive_sha256": res.sha256, "archive_bytes": res.bytes,114 "symbols": symbols, "raw_rows": raw_rows, "rows_dropped_unparseable": dropped,115 "rows_published": sum(t.num_rows for t in tables.values()), "sessions": dates, "files_added": len(written),116 "files_retired": len(retire), "coverage_rows": cov, "anomalies": len(anomalies), **pub}117 log.info("incremental.done", **{k: v for k, v in result.items() if k != "sessions"})118 return result119