1from __future__ import annotations23import json4from datetime import UTC, date, datetime5from typing import Any67import pyarrow as pa89from findata_core.db import transaction10from findata_ingest.lake import Lake, WrittenFile11from findata_quality.validators import session_coverage12from findata_schema import Interval1314DATASET_ASSET = {"stock": "market_bars.stock.1m.raw", "etf": "market_bars.etf.1m.raw", "index": "market_bars.index.1m.raw",15 "fx": "market_bars.fx.1m.raw", "crypto": "market_bars.crypto.1m.raw", "futures": "market_bars.futures.1m.continuous"}161718def record_raw_archive(conn, source_id: str, dataset_id: str, request: dict[str, Any], rel_path: str, sha256: str,19 nbytes: int, file_count: int | None, vendor_last_update: date | None, run_id: str | None) -> tuple[str, bool]:20 """Insert raw archive row; returns (id, is_new). Same sha256 → already known (idempotent re-fetch)."""21 r = conn.execute("SELECT id, status FROM raw_archives WHERE source_id=%s AND sha256=%s", (source_id, sha256)).fetchone()22 if r:23 return str(r["id"]), False24 r = conn.execute(25 "INSERT INTO raw_archives(source_id, dataset_id, request, path, sha256, bytes, file_count, vendor_last_update, ingest_run_id, status) "26 "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,'stored') RETURNING id",27 (source_id, dataset_id, json.dumps(request, default=str), rel_path, sha256, nbytes, file_count, vendor_last_update, run_id)).fetchone()28 return str(r["id"]), True293031def upsert_coverage(conn, dataset_id: str, table: pa.Table, interval: Interval) -> int:32 rows = session_coverage(table, interval)33 for r in rows:34 conn.execute(35 "INSERT INTO coverage(dataset_id, symbol, session_date, bar_count, first_ts, last_ts, expected_bars, flags) "36 "VALUES (%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT (dataset_id, symbol, session_date) DO UPDATE SET bar_count=EXCLUDED.bar_count, "37 "first_ts=EXCLUDED.first_ts, last_ts=EXCLUDED.last_ts, expected_bars=EXCLUDED.expected_bars, flags=EXCLUDED.flags, updated_at=now()",38 (dataset_id, r["symbol"], r["session_date"], r["bar_count"], r["first_ts"], r["last_ts"], r["expected_bars"], r["flags"]))39 return len(rows)404142def record_anomalies(conn, dataset_id: str, anomalies: list[dict[str, Any]], run_id: str | None, limit: int = 5000) -> int:43 n = 044 for a in anomalies[:limit]:45 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)",46 (dataset_id, a.get("symbol"), a.get("session_date"), a["code"], a["severity"], json.dumps(a.get("detail", {}), default=str), run_id))47 n += 148 return n495051def record_quality_report(conn, dataset_id: str, run_id: str | None, rows: int, anomalies: list[dict[str, Any]]) -> None:52 summary: dict[str, int] = {}53 for a in anomalies:54 summary[a["code"]] = summary.get(a["code"], 0) + int(a.get("detail", {}).get("rows", 1))55 conn.execute("INSERT INTO quality_reports(dataset_id, ingest_run_id, rows_checked, anomalies, summary) VALUES (%s,%s,%s,%s,%s)",56 (dataset_id, run_id, rows, len(anomalies), json.dumps(summary)))575859def live_delta_paths_for_dates(conn, dataset_id: str, dates: list[date]) -> list[str]:60 if not dates:61 return []62 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)",63 (dataset_id, dates)).fetchall()64 return [r["path"] for r in rows]656667def live_delta_dates(conn, dataset_id: str) -> set[date]:68 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()69 return {r["date_min"] for r in rows}707172def now() -> datetime:73 return datetime.now(UTC)74