derive: base layout per symbol/year for backfill-driven 5m aggregation
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 changed file +33 −1
modified
services/ingest/src/findata_ingest/tasks/derive.py
+33 −1
@@ -4,13 +4,14 @@ from __future__ import annotations | ||
| 4 | 4 | from datetime import date |
| 5 | 5 | from typing import Any |
| 6 | 6 | |
| 7 | +import pyarrow as pa | |
| 7 | 8 | from findata_core.db import connection, transaction |
| 8 | 9 | from findata_quality import aggregate_bars |
| 9 | 10 | from findata_schema import Interval |
| 10 | 11 | |
| 11 | 12 | from findata_ingest.lake import Lake, live_files |
| 12 | 13 | from findata_ingest.reader import read_bars |
| 13 | −from findata_ingest.tasks.common import live_delta_paths_for_dates, upsert_coverage | |
| 14 | +from findata_ingest.tasks.common import live_delta_dates, live_delta_paths_for_dates, upsert_coverage | |
| 14 | 15 | |
| 15 | 16 | |
| 16 | 17 | def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: |
@@ -35,6 +36,37 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | ||
| 35 | 36 | 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 | 37 | written, rows = [], 0 |
| 37 | 38 | tables = {} |
| 39 | + if symbols: | |
| 40 | + # backfill path: aggregate per symbol over all its sessions and write base files per (symbol, year) | |
| 41 | + import pyarrow.compute as pc | |
| 42 | + retire: list[str] = [] | |
| 43 | + for s in sorted(set(x.upper() for x in symbols)): | |
| 44 | + t, _ = read_bars(src, [s], None, None, lake=lake) | |
| 45 | + if t.num_rows == 0: | |
| 46 | + continue | |
| 47 | + agg = aggregate_bars(t, target, Interval.m1) | |
| 48 | + years = pc.year(agg.column("session_date")) | |
| 49 | + with connection() as conn: | |
| 50 | + 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", | |
| 51 | + (dst, s, s)).fetchall() | |
| 52 | + delta_dates = live_delta_dates(conn, dst) | |
| 53 | + if delta_dates: | |
| 54 | + agg = agg.filter(pc.invert(pc.is_in(agg.column("session_date"), value_set=pa.array(sorted(delta_dates), pa.date32())))) | |
| 55 | + years = pc.year(agg.column("session_date")) | |
| 56 | + for y in sorted(set(years.to_pylist())): | |
| 57 | + ty = agg.filter(pc.equal(years, y)) | |
| 58 | + written.append(lake.write_table(dst, ty, "base", {"symbol": s, "year": str(y)})) | |
| 59 | + retire.extend(o["path"] for o in old) | |
| 60 | + rows += agg.num_rows | |
| 61 | + tables[s] = agg | |
| 62 | + if not written: | |
| 63 | + return {"rows": 0, "symbols": len(symbols)} | |
| 64 | + pub = lake.publish(dst, written, sorted(set(retire)), run_id, note=f"derive {target.value} (base) from {src} for {len(symbols)} symbols") | |
| 65 | + with transaction() as conn: | |
| 66 | + for t in tables.values(): | |
| 67 | + upsert_coverage(conn, dst, t, target) | |
| 68 | + log.info("derive.done", rows=rows, files=len(written), **pub) | |
| 69 | + return {"rows": rows, "files": len(written), "symbols": len(tables), **pub} | |
| 38 | 70 | for d in dates: |
| 39 | 71 | t, _ = read_bars(src, all_syms, d, d, lake=lake) |
| 40 | 72 | if t.num_rows == 0: |
| 41 | 73 | |