src/inferscale/validation.py

from __future__ import annotations

import math
from statistics import mean, median
from typing import Any

from .simulator import run_simulation


SUPPORTED_METRICS = {
    "goodput_rps": ("summary", "goodput_rps"),
    "request_throughput_rps": ("summary", "request_throughput_rps"),
    "slo_attainment": ("summary", "slo_attainment"),
    "p95_ttft_ms": ("latency", "ttft_ms", "p95"),
    "p99_ttft_ms": ("latency", "ttft_ms", "p99"),
    "p95_e2e_ms": ("latency", "e2e_ms", "p95"),
    "p99_e2e_ms": ("latency", "e2e_ms", "p99"),
}


def _predicted(result: dict[str, Any], metric: str) -> float:
    path = SUPPORTED_METRICS[metric]
    value: Any = result
    for key in path:
        value = value[key]
    return float(value)


def validate_cases(cases: list[dict[str, Any]]) -> dict[str, Any]:
    """Compare simulator predictions with externally measured serving cases.

    Each case contains a normal SimulationConfig dictionary and a `measured`
    mapping. No measured data ships as benchmark truth with InferScale; this
    function is the explicit integration point for future empirical validation.
    """
    rows: list[dict[str, Any]] = []
    absolute_percentage_errors: list[float] = []

    for index, case in enumerate(cases):
        if "config" not in case or "measured" not in case:
            raise ValueError(f"Validation case {index} requires config and measured fields")
        result = run_simulation(case["config"])
        name = str(case.get("name", f"case-{index + 1}"))
        for metric, measured_raw in case["measured"].items():
            if metric not in SUPPORTED_METRICS:
                raise ValueError(f"Unsupported validation metric: {metric}")
            measured = float(measured_raw)
            predicted = _predicted(result, metric)
            error = predicted - measured
            ape = abs(error) / abs(measured) * 100.0 if abs(measured) > 1e-12 else math.nan
            if math.isfinite(ape):
                absolute_percentage_errors.append(ape)
            rows.append(
                {
                    "case": name,
                    "metric": metric,
                    "measured": measured,
                    "predicted": predicted,
                    "error": error,
                    "absolute_error": abs(error),
                    "absolute_percentage_error": ape,
                }
            )

    return {
        "case_count": len(cases),
        "observation_count": len(rows),
        "mape_pct": mean(absolute_percentage_errors) if absolute_percentage_errors else 0.0,
        "median_ape_pct": median(absolute_percentage_errors) if absolute_percentage_errors else 0.0,
        "max_ape_pct": max(absolute_percentage_errors, default=0.0),
        "rows": rows,
        "provenance": "external-measurements-vs-analytical-reference",
    }