resource-control/tools/thresholds.py

#!/usr/bin/env python3
"""Evaluate a monthly observation against a resource's declared thresholds.

Fail-closed in both directions. An unknown measurement never passes a
threshold — a missing number is reported as `unmeasured`, because a threshold
that silently passes on absent evidence is worse than no threshold at all.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

import variance

# Verdicts, in the order a reviewer should read them.
BREACH = "breach"
UNMEASURED = "unmeasured"
NOT_APPLICABLE = "not_applicable"
WITHIN = "within"


def _row_for(payload: dict, period: str) -> dict | None:
    for row in payload["rows"]:
        if row["period"] == period:
            return row
    return None


def evaluate_threshold(threshold: dict, observed: dict, compared: dict | None) -> dict:
    metric = threshold["metric"]
    result = {
        "id": threshold["id"],
        "kind": threshold["kind"],
        "metric": metric,
        "limit": threshold["limit"],
        "action": threshold["action"],
    }

    if threshold.get("status") == "not_applicable":
        return {**result, "verdict": NOT_APPLICABLE, "detail": threshold["action"]}

    value = observed.get(metric)
    if value is None:
        gaps = [gap for gap in observed.get("measurement_gaps", []) if gap.startswith(f"{metric}:")]
        return {
            **result,
            "verdict": UNMEASURED,
            "detail": gaps[0] if gaps else f"{metric} is not present in the observation",
        }

    comparison = threshold["comparison"]
    if comparison == "minimum":
        breached = value < threshold["limit"]
        return {**result, "verdict": BREACH if breached else WITHIN, "observed": value}
    if comparison == "maximum":
        breached = value > threshold["limit"]
        return {**result, "verdict": BREACH if breached else WITHIN, "observed": value}
    if comparison == "unplanned":
        breached = value > threshold["limit"]
        return {**result, "verdict": BREACH if breached else WITHIN, "observed": value}

    # The remaining comparisons need a forecast to compare against.
    if compared is None or metric not in compared:
        return {
            **result,
            "verdict": UNMEASURED,
            "observed": value,
            "detail": "no forecast row exists for this period, so no variance can be computed",
        }
    entry = compared[metric]
    if entry.get("status") == "unknown":
        return {**result, "verdict": UNMEASURED, "observed": value, "detail": "forecast or actual amount is unknown"}
    if comparison == "absolute_percentage_error":
        measured = entry["absolute_percentage_error"]
        if measured is None:
            return {
                **result,
                "verdict": UNMEASURED,
                "observed": value,
                "detail": "forecast is zero, so percentage error is undefined; review the absolute error instead",
            }
    else:
        measured = abs(entry["error"])
    return {
        **result,
        "verdict": BREACH if measured > threshold["limit"] else WITHIN,
        "observed": value,
        "measured": measured,
    }


def evaluate(config: dict, observation: dict, forecast: dict | None, period: str) -> dict:
    observed = _row_for(observation, period)
    if observed is None:
        raise ValueError(f"observation has no row for {period}")

    compared = None
    if forecast is not None:
        report = variance.compare(forecast, observation)
        for row in report["rows"]:
            if row["period"] == period and row["status"] == "compared":
                compared = row["metrics"]

    results = [evaluate_threshold(t, observed, compared) for t in config["thresholds"]]
    counts: dict[str, int] = {}
    for result in results:
        counts[result["verdict"]] = counts.get(result["verdict"], 0) + 1
    return {
        "resource_id": config["resource_id"],
        "period": period,
        "forecast_available": compared is not None,
        "results": results,
        "summary": dict(sorted(counts.items())),
        "breaches": [r["id"] for r in results if r["verdict"] == BREACH],
        "unmeasured": [r["id"] for r in results if r["verdict"] == UNMEASURED],
        "known_gaps": config.get("known_gaps", []),
    }


def main() -> int:
    if len(sys.argv) < 4:
        print(f"usage: {sys.argv[0]} THRESHOLDS.json OBSERVATION.json PERIOD [FORECAST.json]", file=sys.stderr)
        return 2
    config = json.loads(Path(sys.argv[1]).read_text())
    observation = json.loads(Path(sys.argv[2]).read_text())
    period = sys.argv[3]
    forecast = json.loads(Path(sys.argv[4]).read_text()) if len(sys.argv) > 4 else None
    report = evaluate(config, observation, forecast, period)
    print(json.dumps(report, indent=2))
    # A breach is a non-zero exit so the monthly cadence can gate on it.
    return 1 if report["breaches"] else 0


if __name__ == "__main__":
    raise SystemExit(main())