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_schema import Interval16from findata_schema.connector import FetchRequest1718from findata_ingest import queue19from findata_ingest.lake import Lake20from findata_ingest.tasks.common import (21 DATASET_ASSET,22 live_delta_paths_for_dates,23 record_anomalies,24 record_quality_report,25 record_raw_archive,26 upsert_coverage,27)282930def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:31 p = job.get("params") or {}32 asset = p.get("type", "stock")33 period = p.get("period", "day")34 timeframe = p.get("timeframe", "1min")35 adjustment = p.get("adjustment", "UNADJUSTED")36 dataset_id = job.get("dataset_id") or DATASET_ASSET[asset]37 run_id = job.get("run_id")38 conn_fr = FirstRateMarketConnector()39 vendor_last = conn_fr.client.last_update(asset)40 ck = queue.get_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}") or {}41 if ck.get("vendor_last_update") == vendor_last.isoformat() and not p.get("force"):42 log.info("incremental.skip", reason="vendor not updated since last run", vendor_last_update=str(vendor_last))43 return {"skipped": True, "vendor_last_update": vendor_last.isoformat()}4445 req = FetchRequest(dataset_id=dataset_id, params={"endpoint": "data_file", "type": asset, "period": period,46 "timeframe": timeframe, "adjustment": adjustment})47 queue.progress(job["id"], {"stage": "download"})48 res = conn_fr.fetch_range(req, lake.raw_dir("firstrate", dataset_id))49 rel_raw = str(res.path.relative_to(lake.root))50 with transaction() as conn:51 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)52 if not is_new:53 res.path.unlink(missing_ok=True) # identical bytes already stored immutably54 queue.set_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}", {"vendor_last_update": vendor_last.isoformat(), "sha256": res.sha256})55 log.info("incremental.duplicate_archive", sha256=res.sha256[:12])56 return {"skipped": True, "reason": "archive already ingested", "sha256": res.sha256}57 log.info("incremental.downloaded", bytes=res.bytes, files=res.file_count, vendor_last_update=str(vendor_last))5859 # normalize → group rows by session_date (across all symbols)60 queue.progress(job["id"], {"stage": "normalize", "bytes": res.bytes})61 by_date: dict[Any, list[pa.Table]] = defaultdict(list)62 anomalies: list[dict[str, Any]] = []63 symbols = 064 raw_rows = 065 dropped = 066 for batch in conn_fr.normalize(res):67 rep = conn_fr.validate(batch)68 anomalies.extend(rep.anomalies)69 symbols += 170 raw_rows += batch.raw_rows71 dropped += batch.raw_rows - batch.row_count72 t = batch.table73 for d in set(t.column("session_date").to_pylist()):74 import pyarrow.compute as pc75 by_date[d].append(t.filter(pc.equal(t.column("session_date"), pa.scalar(d, pa.date32()))))76 if symbols % 1000 == 0:77 queue.progress(job["id"], {"stage": "normalize", "symbols": symbols})78 if not by_date:79 raise RuntimeError("archive produced no rows")8081 # write one delta file per session_date, then publish atomically (supersede existing dates)82 queue.progress(job["id"], {"stage": "write", "sessions": len(by_date)})83 interval = Interval({"1min": "1m", "5min": "5m", "30min": "30m", "1hour": "1h", "1day": "1d"}[timeframe])84 written = []85 tables = {}86 for d in sorted(by_date):87 t = pa.concat_tables(by_date[d])88 tables[d] = t89 written.append(lake.write_table(dataset_id, t, "delta", {"session_date": d.isoformat()}))90 with connection() as conn:91 retire = live_delta_paths_for_dates(conn, dataset_id, list(tables))92 pub = lake.publish(dataset_id, written, retire, run_id, note=f"incremental {asset} {period} {timeframe} {adjustment}")93 with transaction() as conn:94 conn.execute("UPDATE raw_archives SET status='published' WHERE id=%s", (raw_id,))95 cov = sum(upsert_coverage(conn, dataset_id, t, interval) for t in tables.values())96 record_anomalies(conn, dataset_id, anomalies, run_id)97 record_quality_report(conn, dataset_id, run_id, sum(t.num_rows for t in tables.values()), anomalies)98 queue.set_checkpoint(f"dataset:{dataset_id}", f"last_vendor_update:{period}", {"vendor_last_update": vendor_last.isoformat(), "sha256": res.sha256})99 cp = conn_fr.checkpoint(res, len(written))100 queue.set_checkpoint(cp.scope, cp.key, cp.value)101 dates = [d.isoformat() for d in sorted(tables)]102 # downstream: derived 5-minute bars for these sessions103 if interval == Interval.m1 and asset in ("stock", "etf"):104 queue.enqueue("derive", dataset_id.replace(".1m.", ".5m."), {"source_dataset": dataset_id, "target_interval": "5m", "dates": dates},105 priority=40, dedupe_key=f"derive:{dataset_id}:5m:{dates[0]}:{dates[-1]}:{pub['version']}")106 result = {"vendor_last_update": vendor_last.isoformat(), "archive_sha256": res.sha256, "archive_bytes": res.bytes,107 "symbols": symbols, "raw_rows": raw_rows, "rows_dropped_unparseable": dropped,108 "rows_published": sum(t.num_rows for t in tables.values()), "sessions": dates, "files_added": len(written),109 "files_retired": len(retire), "coverage_rows": cov, "anomalies": len(anomalies), **pub}110 log.info("incremental.done", **{k: v for k, v in result.items() if k != "sessions"})111 return result112