1"""Anonymous overview metrics for the public landing page. No data rows, no paths, no secrets — only catalog-level2figures (what exists, how much, how fresh). Everything else stays behind authentication (FirstRate license:3raw bars/tables are private)."""4from __future__ import annotations56import threading7import time8from datetime import UTC, datetime9from typing import Any1011from fastapi import APIRouter12from findata_core.db import connection1314from findata_api.health import cached_evaluate1516router = APIRouter(prefix="/public", tags=["public"])17_cache: dict[str, Any] = {"at": 0.0, "value": None}18_lock = threading.Lock()192021def _compute() -> dict[str, Any]:22 with connection() as conn:23 totals = conn.execute("SELECT count(*) files, coalesce(sum(row_count),0) rows, coalesce(sum(bytes),0) bytes FROM lake_files WHERE version_removed IS NULL").fetchone()24 fam = conn.execute("""25 SELECT split_part(dataset_id,'.',1)||'.'||split_part(dataset_id,'.',2) AS family, count(DISTINCT dataset_id) datasets,26 sum(row_count) rows, sum(bytes) bytes27 FROM lake_files WHERE version_removed IS NULL GROUP BY 1 ORDER BY sum(bytes) DESC""").fetchall()28 ds = conn.execute("""29 SELECT d.id, d.name, d.kind, d.asset_class, d.interval, d.adjustment, d.status, d.license_class, d.current_version,30 v.row_count, v.bytes, v.file_count, v.created_at AS version_created_at31 FROM datasets d LEFT JOIN dataset_versions v ON v.dataset_id=d.id AND v.version=d.current_version ORDER BY d.id""").fetchall()32 cov = conn.execute("""33 SELECT dataset_id, count(DISTINCT symbol) symbols, min(session_date) first_session, max(session_date) last_session, sum(bar_count) bars34 FROM coverage WHERE dataset_id IN ('market_bars.stock.1m.raw','market_bars.etf.1m.raw','market_bars.index.1m.raw') GROUP BY 1""").fetchall()35 opt = conn.execute("SELECT count(DISTINCT symbol_min) underlyings, min(date_min) first_day, max(date_max) last_day, sum(row_count) rows FROM lake_files WHERE dataset_id='options.eod' AND version_removed IS NULL").fetchone()36 fund = conn.execute("SELECT count(DISTINCT entity_id) entities, count(*) facts, min(fiscal_period_end) first_period, max(fiscal_period_end) last_period FROM fundamentals_facts WHERE superseded_at IS NULL").fetchone()37 earn = conn.execute("SELECT count(*) n FROM earnings_events").fetchone()38 inst = conn.execute("SELECT count(*) n, count(*) FILTER (WHERE status='active') active, count(*) FILTER (WHERE status='delisted') delisted FROM instruments").fetchone()39 raw = conn.execute("SELECT count(*) n, coalesce(sum(bytes),0) bytes FROM raw_archives WHERE path IS NOT NULL").fetchone()40 jobs = conn.execute("SELECT status, count(*) n FROM ingest_jobs WHERE created_at > now() - interval '24 hours' GROUP BY 1").fetchall()41 lastjob = conn.execute("SELECT max(finished_at) t FROM ingest_jobs WHERE status='succeeded'").fetchone()42 ck = conn.execute("SELECT value FROM checkpoints WHERE scope='dataset:market_bars.stock.1m.raw' AND key='last_vendor_update:day'").fetchone()43 versions = conn.execute("SELECT count(*) n FROM dataset_versions").fetchone()44 backup = conn.execute("SELECT updated_at, value->>'verified' AS verified FROM settings_kv WHERE key='last_backup'").fetchone()45 h = cached_evaluate()46 lake = h.get("components", {}).get("lake", {}) if isinstance(h, dict) else {}47 return {48 "generated_at": datetime.now(UTC).isoformat(),49 "status": h.get("status"),50 "totals": {"files": int(totals["files"]), "rows": int(totals["rows"]), "bytes": int(totals["bytes"]),51 "datasets_active": sum(1 for d in ds if d["status"] == "active"), "dataset_versions": int(versions["n"]),52 "raw_archives": int(raw["n"]), "raw_bytes": int(raw["bytes"])},53 "families": [{"family": f["family"], "datasets": f["datasets"], "rows": int(f["rows"]), "bytes": int(f["bytes"])} for f in fam],54 "datasets": [{k: (v.isoformat() if hasattr(v, "isoformat") else v) for k, v in dict(d).items()} for d in ds],55 "coverage": [{k: (v.isoformat() if hasattr(v, "isoformat") else v) for k, v in dict(c).items()} for c in cov],56 "options": {k: (v.isoformat() if hasattr(v, "isoformat") else v) for k, v in dict(opt).items()},57 "fundamentals": {**{k: (v.isoformat() if hasattr(v, "isoformat") else v) for k, v in dict(fund).items()}, "earnings_events": int(earn["n"])},58 "instruments": dict(inst),59 "freshness": {"vendor_last_update": (ck["value"] or {}).get("vendor_last_update") if ck else None,60 "last_successful_job": lastjob["t"].isoformat() if lastjob and lastjob["t"] else None,61 "jobs_24h": {j["status"]: j["n"] for j in jobs}},62 "storage": {"hot_free_gb": lake.get("free_gb"), "cold_free_gb": lake.get("cold_free_gb"), "curated_gb": lake.get("curated_gb"),63 "raw_gb": lake.get("raw_gb")},64 "backup": {"last": backup["updated_at"].isoformat() if backup else None, "verified": (backup["verified"] == "true") if backup else None},65 "license": "FirstRate Data — private research use; raw bars and tables are not redistributed. Sign in for data access.",66 }676869_started = False707172def _refresh_loop(period_s: float) -> None:73 while True:74 try:75 v = _compute()76 _cache.update(at=time.monotonic(), value=v)77 except Exception as e:78 _cache.setdefault("error", str(e)[:200])79 time.sleep(period_s)808182def start_refresher(period_s: float = 300.0) -> None:83 global _started84 if _started:85 return86 _started = True87 threading.Thread(target=_refresh_loop, args=(period_s,), name="public-overview-refresher", daemon=True).start()888990@router.get("/overview")91def overview():92 """Served from a cache refreshed in the background every 5 minutes; the request path never runs the queries."""93 start_refresher()94 v = _cache["value"]95 if v is None:96 return {"generated_at": datetime.now(UTC).isoformat(), "status": "starting", "totals": None, "families": [], "datasets": [],97 "coverage": [], "options": {}, "fundamentals": {}, "instruments": {}, "freshness": {"jobs_24h": {}}, "storage": {},98 "backup": {}, "license": "FirstRate Data — private research use; raw bars and tables are not redistributed. Sign in for data access."}99 return v100