backend/observability.py

"""Observability aggregations over inference_logs. Phase 7.

Pure functions (testable) that turn raw inference_logs rows into the three dashboard views
computed entirely from Supabase data — no external observability tool:
  - rolling mean prediction_set_size per day  -> drift proxy (rising = distribution shift)
  - latency p50 / p95 per day
  - class distribution: current 7-day window vs the prior 7-day baseline
"""
from __future__ import annotations

from collections import defaultdict
from datetime import date, timedelta


def _day(ts) -> str:
    return str(ts)[:10]


def _percentile(values: list[float], p: float) -> float:
    if not values:
        return 0.0
    v = sorted(values)
    k = (len(v) - 1) * p / 100.0
    f = int(k)
    if f + 1 < len(v):
        return round(v[f] + (v[f + 1] - v[f]) * (k - f), 1)
    return round(v[f], 1)


def rolling_set_size(logs: list[dict]) -> list[dict]:
    by: dict[str, list[float]] = defaultdict(list)
    for r in logs:
        by[_day(r["ts"])].append(float(r.get("prediction_set_size", 1)))
    return [{"date": d, "avg_set_size": round(sum(v) / len(v), 3), "n": len(v)}
            for d, v in sorted(by.items())]


def latency_percentiles(logs: list[dict]) -> list[dict]:
    by: dict[str, list[float]] = defaultdict(list)
    for r in logs:
        by[_day(r["ts"])].append(float(r.get("latency_ms", 0)))
    return [{"date": d, "p50": _percentile(v, 50), "p95": _percentile(v, 95), "n": len(v)}
            for d, v in sorted(by.items())]


def class_distribution(logs: list[dict], classes: tuple[str, ...]) -> dict:
    days = sorted({_day(r["ts"]) for r in logs})
    if not days:
        return {"current": {}, "baseline": {}, "current_window": None, "baseline_window": None}
    anchor = date.fromisoformat(days[-1])
    cur_lo = anchor - timedelta(days=6)
    base_hi = cur_lo - timedelta(days=1)
    base_lo = base_hi - timedelta(days=6)

    def frac(lo: date, hi: date) -> dict:
        counts = {c: 0 for c in classes}
        for r in logs:
            d = date.fromisoformat(_day(r["ts"]))
            if lo <= d <= hi:
                lab = r.get("predicted_label")
                if lab in counts:
                    counts[lab] += 1
        total = sum(counts.values()) or 1
        return {c: round(counts[c] / total * 100, 1) for c in classes}

    return {
        "current": frac(cur_lo, anchor),
        "baseline": frac(base_lo, base_hi),
        "current_window": [cur_lo.isoformat(), anchor.isoformat()],
        "baseline_window": [base_lo.isoformat(), base_hi.isoformat()],
    }


def build(logs: list[dict], classes: tuple[str, ...]) -> dict:
    return {
        "drift": rolling_set_size(logs),
        "latency": latency_percentiles(logs),
        "class_distribution": class_distribution(logs, classes),
        "total_logs": len(logs),
    }