1"""Derive 5-minute (or other) bars from validated 1-minute bars, per session_date (delta layout, supersede)."""2from __future__ import annotations34from datetime import date5from typing import Any67from findata_core.db import connection, transaction8from findata_quality import aggregate_bars9from findata_schema import Interval1011from findata_ingest.lake import Lake, live_files12from findata_ingest.reader import read_bars13from findata_ingest.tasks.common import live_delta_paths_for_dates, upsert_coverage141516def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:17 p = job.get("params") or {}18 src = p["source_dataset"]19 dst = job.get("dataset_id") or src.replace(".1m.", ".5m.")20 target = Interval(p.get("target_interval", "5m"))21 dates = [date.fromisoformat(d) for d in p.get("dates", [])]22 symbols = p.get("symbols")23 run_id = job.get("run_id")24 with connection() as conn:25 if not dates:26 # all live source dates for requested symbols27 fl = live_files(conn, src, symbols)28 ds = set()29 for f in fl:30 d = f["date_min"]31 while d <= f["date_max"]:32 ds.add(d)33 d = d.fromordinal(d.toordinal() + 1)34 dates = sorted(ds)35 all_syms = symbols or [r["symbol"] for r in conn.execute("SELECT DISTINCT symbol FROM coverage WHERE dataset_id=%s AND session_date = ANY(%s)", (src, dates)).fetchall()]36 written, rows = [], 037 tables = {}38 for d in dates:39 t, _ = read_bars(src, all_syms, d, d, lake=lake)40 if t.num_rows == 0:41 continue42 agg = aggregate_bars(t, target, Interval.m1)43 tables[d] = agg44 rows += agg.num_rows45 written.append(lake.write_table(dst, agg, "delta", {"session_date": d.isoformat()}))46 if not written:47 return {"rows": 0, "dates": len(dates)}48 with connection() as conn:49 retire = live_delta_paths_for_dates(conn, dst, list(tables)) if not symbols else []50 pub = lake.publish(dst, written, retire, run_id, note=f"derive {target.value} from {src}")51 with transaction() as conn:52 for t in tables.values():53 upsert_coverage(conn, dst, t, target)54 log.info("derive.done", rows=rows, files=len(written), **pub)55 return {"rows": rows, "files": len(written), "dates": len(tables), **pub}56