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 dataset_for_params(params: dict[str, Any]) -> str:16 from findata_market_firstrate.catalog import dataset_id_for17 return dataset_id_for(params.get("type", "stock"), params.get("timeframe", "1min"), params.get("adjustment"),18 contracts=bool(params.get("contracts")))192021def record_raw_archive(conn, source_id: str, dataset_id: str, request: dict[str, Any], rel_path: str, sha256: str,22 nbytes: int, file_count: int | None, vendor_last_update: date | None, run_id: str | None) -> tuple[str, bool]:23 """Insert raw archive row; returns (id, is_new). Same sha256 → already known (idempotent re-fetch)."""24 r = conn.execute("SELECT id, status FROM raw_archives WHERE source_id=%s AND sha256=%s", (source_id, sha256)).fetchone()25 if r:26 return str(r["id"]), False27 r = conn.execute(28 "INSERT INTO raw_archives(source_id, dataset_id, request, path, sha256, bytes, file_count, vendor_last_update, ingest_run_id, status) "29 "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,'stored') RETURNING id",30 (source_id, dataset_id, json.dumps(request, default=str), rel_path, sha256, nbytes, file_count, vendor_last_update, run_id)).fetchone()31 return str(r["id"]), True323334COVERAGE_DATASETS = {"market_bars.stock.1m.raw", "market_bars.etf.1m.raw", "market_bars.index.1m.raw",35 "market_bars.stock.5m.raw", "market_bars.etf.5m.raw", "market_bars.stock.1d.raw", "market_bars.etf.1d.raw"}363738def upsert_coverage(conn, dataset_id: str, table: pa.Table, interval: Interval) -> int:39 if dataset_id not in COVERAGE_DATASETS:40 return 0 # adjusted/coarse variants share the raw datasets' sessions; keeping per-session rows for all 64 datasets cost 130 GB41 if table.num_rows and table.column("asset_class")[0].as_py() in ("fx", "crypto", "futures"):42 return 0 # no exchange-session expectation for 24h markets43 rows = session_coverage(table, interval)44 for r in rows:45 conn.execute(46 "INSERT INTO coverage(dataset_id, symbol, session_date, bar_count, first_ts, last_ts, expected_bars, flags) "47 "VALUES (%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT (dataset_id, symbol, session_date) DO UPDATE SET bar_count=EXCLUDED.bar_count, "48 "first_ts=EXCLUDED.first_ts, last_ts=EXCLUDED.last_ts, expected_bars=EXCLUDED.expected_bars, flags=EXCLUDED.flags, updated_at=now()",49 (dataset_id, r["symbol"], r["session_date"], r["bar_count"], r["first_ts"], r["last_ts"], r["expected_bars"], r["flags"]))50 return len(rows)515253def record_anomalies(conn, dataset_id: str, anomalies: list[dict[str, Any]], run_id: str | None, limit: int = 5000) -> int:54 n = 055 for a in anomalies[:limit]:56 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)",57 (dataset_id, a.get("symbol"), a.get("session_date"), a["code"], a["severity"], json.dumps(a.get("detail", {}), default=str), run_id))58 n += 159 return n606162def record_quality_report(conn, dataset_id: str, run_id: str | None, rows: int, anomalies: list[dict[str, Any]]) -> None:63 summary: dict[str, int] = {}64 for a in anomalies:65 summary[a["code"]] = summary.get(a["code"], 0) + int(a.get("detail", {}).get("rows", 1))66 conn.execute("INSERT INTO quality_reports(dataset_id, ingest_run_id, rows_checked, anomalies, summary) VALUES (%s,%s,%s,%s,%s)",67 (dataset_id, run_id, rows, len(anomalies), json.dumps(summary)))686970def live_delta_paths_for_dates(conn, dataset_id: str, dates: list[date]) -> list[str]:71 if not dates:72 return []73 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)",74 (dataset_id, dates)).fetchall()75 return [r["path"] for r in rows]767778def live_delta_dates(conn, dataset_id: str) -> set[date]:79 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()80 return {r["date_min"] for r in rows}818283def now() -> datetime:84 return datetime.now(UTC)85