1from __future__ import annotations23import json4from datetime import UTC, date, datetime5from typing import Any67import pyarrow as pa8from findata_quality.validators import session_coverage9from findata_schema import Interval1011DATASET_ASSET = {"stock": "market_bars.stock.1m.raw", "etf": "market_bars.etf.1m.raw", "index": "market_bars.index.1m.raw",12 "fx": "market_bars.fx.1m.raw", "crypto": "market_bars.crypto.1m.raw", "futures": "market_bars.futures.1m.continuous"}131415def record_raw_archive(conn, source_id: str, dataset_id: str, request: dict[str, Any], rel_path: str, sha256: str,16 nbytes: int, file_count: int | None, vendor_last_update: date | None, run_id: str | None) -> tuple[str, bool]:17 """Insert raw archive row; returns (id, is_new). Same sha256 → already known (idempotent re-fetch)."""18 r = conn.execute("SELECT id, status FROM raw_archives WHERE source_id=%s AND sha256=%s", (source_id, sha256)).fetchone()19 if r:20 return str(r["id"]), False21 r = conn.execute(22 "INSERT INTO raw_archives(source_id, dataset_id, request, path, sha256, bytes, file_count, vendor_last_update, ingest_run_id, status) "23 "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,'stored') RETURNING id",24 (source_id, dataset_id, json.dumps(request, default=str), rel_path, sha256, nbytes, file_count, vendor_last_update, run_id)).fetchone()25 return str(r["id"]), True262728def upsert_coverage(conn, dataset_id: str, table: pa.Table, interval: Interval) -> int:29 rows = session_coverage(table, interval)30 for r in rows:31 conn.execute(32 "INSERT INTO coverage(dataset_id, symbol, session_date, bar_count, first_ts, last_ts, expected_bars, flags) "33 "VALUES (%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT (dataset_id, symbol, session_date) DO UPDATE SET bar_count=EXCLUDED.bar_count, "34 "first_ts=EXCLUDED.first_ts, last_ts=EXCLUDED.last_ts, expected_bars=EXCLUDED.expected_bars, flags=EXCLUDED.flags, updated_at=now()",35 (dataset_id, r["symbol"], r["session_date"], r["bar_count"], r["first_ts"], r["last_ts"], r["expected_bars"], r["flags"]))36 return len(rows)373839def record_anomalies(conn, dataset_id: str, anomalies: list[dict[str, Any]], run_id: str | None, limit: int = 5000) -> int:40 n = 041 for a in anomalies[:limit]:42 conn.execute("INSERT INTO quality_anomalies(dataset_id, symbol, session_date, code, severity, detail, ingest_run_id) VALUES (%s,%s,%s,%s,%s,%s,%s)",43 (dataset_id, a.get("symbol"), a.get("session_date"), a["code"], a["severity"], json.dumps(a.get("detail", {}), default=str), run_id))44 n += 145 return n464748def record_quality_report(conn, dataset_id: str, run_id: str | None, rows: int, anomalies: list[dict[str, Any]]) -> None:49 summary: dict[str, int] = {}50 for a in anomalies:51 summary[a["code"]] = summary.get(a["code"], 0) + int(a.get("detail", {}).get("rows", 1))52 conn.execute("INSERT INTO quality_reports(dataset_id, ingest_run_id, rows_checked, anomalies, summary) VALUES (%s,%s,%s,%s,%s)",53 (dataset_id, run_id, rows, len(anomalies), json.dumps(summary)))545556def live_delta_paths_for_dates(conn, dataset_id: str, dates: list[date]) -> list[str]:57 if not dates:58 return []59 rows = conn.execute("SELECT path FROM lake_files WHERE dataset_id=%s AND layout='delta' AND version_removed IS NULL AND date_min = ANY(%s)",60 (dataset_id, dates)).fetchall()61 return [r["path"] for r in rows]626364def live_delta_dates(conn, dataset_id: str) -> set[date]:65 rows = conn.execute("SELECT DISTINCT date_min FROM lake_files WHERE dataset_id=%s AND layout='delta' AND version_removed IS NULL", (dataset_id,)).fetchall()66 return {r["date_min"] for r in rows}676869def now() -> datetime:70 return datetime.now(UTC)71