#!/usr/bin/env python3 """Evaluate resource optimization cases. The evaluator is fail-closed: any decision field that is unknown makes the affected comparison unknown rather than optimistic. A case only reaches a recommendation when every field the decision template requires is present for the baseline and for the alternative being compared. """ from __future__ import annotations import json import sys from pathlib import Path CASE_TYPES = { "rightsizing", "consolidation", "commitment", "renewal", "migration", "retirement", "provider_switch", } DECISION_STATES = { "blocked_on_evidence", "proposed", "approved", "rejected", "superseded", } COST_FIELDS = ( "recurring_infrastructure_eur_month", "recurring_internal_labor_eur_month", "recurring_external_labor_eur_month", ) # Fields the decision template requires before any recommendation is made. REQUIRED_FOR_DECISION = COST_FIELDS + ("one_time_eur", "exit_path") # A saving below this monthly threshold is not material enough to act on. MATERIALITY_EUR_MONTH = 5.0 def recurring_total(option: dict) -> float | None: """Monthly recurring cost, or None when any component is unknown.""" values = [option[field] for field in COST_FIELDS] if any(value is None for value in values): return None return round(sum(values), 2) def missing_fields(option: dict) -> list[str]: missing = [field for field in REQUIRED_FOR_DECISION if option[field] is None] if not option["utilization"]: missing.append("utilization") if not option["service_constraints"]: missing.append("service_constraints") return sorted(missing) def utilization_ratios(option: dict) -> dict: """Used-over-provisioned per metric; None where either side is unknown.""" ratios = {} for metric, pair in option["utilization"].items(): provisioned = pair["provisioned"]["value"] used = pair["used"]["value"] if provisioned in (None, 0) or used is None: ratios[metric] = None else: ratios[metric] = round(used / provisioned, 4) return ratios def compare_option(baseline: dict, alternative: dict) -> dict: """Compare one alternative against the baseline on the decision fields.""" if alternative.get("excluded"): # The deciding authority set this option aside rather than wait for # evidence that was not going to arrive. It no longer blocks the case, # and the reason travels with the decision. return { "option_id": alternative["option_id"], "label": alternative["label"], "verdict": "excluded", "exclusion_reason": alternative.get("exclusion_reason"), "blocking_evidence": [], } blocking = sorted(set( [f"baseline.{field}" for field in missing_fields(baseline)] + [f"{alternative['option_id']}.{field}" for field in missing_fields(alternative)] + [f"baseline.unknown:{item}" for item in baseline["unknowns"]] + [f"{alternative['option_id']}.unknown:{item}" for item in alternative["unknowns"]] )) base_recurring = recurring_total(baseline) alt_recurring = recurring_total(alternative) monthly_delta = None monthly_saving = None if base_recurring is not None and alt_recurring is not None: monthly_delta = round(alt_recurring - base_recurring, 2) monthly_saving = round(-monthly_delta, 2) one_time = alternative["one_time_eur"] payback_months = None payback_note = None if monthly_saving is None or one_time is None: payback_note = "unknown: incomplete cost evidence" elif monthly_saving <= 0: payback_note = "never: the alternative does not reduce recurring cost" elif one_time == 0: payback_months = 0.0 payback_note = "immediate: no one-time cost" else: payback_months = round(one_time / monthly_saving, 1) # Failure domains the alternative removes, and ones it newly introduces. base_domains = set(baseline["failure_domains"]) alt_domains = set(alternative["failure_domains"]) if blocking: verdict = "blocked_on_evidence" elif monthly_saving is not None and monthly_saving > MATERIALITY_EUR_MONTH: verdict = "recommend" elif monthly_delta is not None and abs(monthly_delta) <= MATERIALITY_EUR_MONTH: verdict = "no_material_change" else: verdict = "reject" return { "option_id": alternative["option_id"], "label": alternative["label"], "verdict": verdict, "baseline_recurring_eur_month": base_recurring, "alternative_recurring_eur_month": alt_recurring, "monthly_delta_eur": monthly_delta, "monthly_saving_eur": monthly_saving, "one_time_eur": one_time, "payback_months": payback_months, "payback_note": payback_note, "baseline_utilization": utilization_ratios(baseline), "alternative_utilization": utilization_ratios(alternative), "uncertainty": alternative["uncertainty"]["level"], "failure_domains_removed": sorted(base_domains - alt_domains), "failure_domains_added": sorted(alt_domains - base_domains), "exit_path_known": alternative["exit_path"] is not None, "blocking_evidence": blocking, } def validate_case(case: dict) -> None: if case["schema_version"] != "0.1": raise ValueError("unsupported optimization-case schema version") if case["record_scope"] not in {"operational", "illustrative"}: raise ValueError("record_scope must be operational or illustrative") if case["case_type"] not in CASE_TYPES: raise ValueError(f"unknown case_type {case['case_type']}") if not case["case_id"].startswith("opt:"): raise ValueError("case_id must start with 'opt:'") if not case["resource_ids"]: raise ValueError("a case must name at least one resource") if any(not rid.startswith("resource:") for rid in case["resource_ids"]): raise ValueError("resource_ids must be portfolio resource identifiers") if not case["evidence"]: raise ValueError("a case must cite evidence") decision = case["decision"] if decision["state"] not in DECISION_STATES: raise ValueError(f"unknown decision state {decision['state']}") option_ids = [case["baseline"]["option_id"]] + [a["option_id"] for a in case["alternatives"]] if len(set(option_ids)) != len(option_ids): raise ValueError("option identifiers must be unique within a case") for alternative in case["alternatives"]: if alternative.get("excluded") and not alternative.get("exclusion_reason"): raise ValueError( f"excluded option {alternative['option_id']} must record an exclusion_reason" ) report = evaluate(case) blocked = any(r["verdict"] == "blocked_on_evidence" for r in report["comparisons"]) # A case cannot be approved while its own evidence is incomplete, and an # approval must name the authority that gave it. if decision["state"] in {"approved", "rejected"}: if decision["approver"] is None or decision["approved_on"] is None: raise ValueError("a decided case must record approver and approved_on") if decision["state"] == "approved": if blocked: raise ValueError("cannot approve a case with blocking evidence gaps") if decision["recommended_option_id"] not in option_ids: raise ValueError("approved case must recommend a known option") if decision["state"] == "blocked_on_evidence" and not blocked: raise ValueError("case is marked blocked but every decision field is known") if decision["state"] == "proposed" and blocked: raise ValueError("case has blocking evidence gaps and cannot be proposed") def evaluate(case: dict) -> dict: baseline = case["baseline"] comparisons = [compare_option(baseline, alt) for alt in case["alternatives"]] recommended = [c for c in comparisons if c["verdict"] == "recommend"] recommended.sort(key=lambda c: (c["payback_months"] is None, c["payback_months"])) return { "case_id": case["case_id"], "case_type": case["case_type"], "record_scope": case["record_scope"], "review_period": case["review_period"], "baseline": { "option_id": baseline["option_id"], "label": baseline["label"], "recurring_eur_month": recurring_total(baseline), "utilization": utilization_ratios(baseline), }, "comparisons": comparisons, "best_option_id": recommended[0]["option_id"] if recommended else None, "decision_state": case["decision"]["state"], } def main() -> int: paths = sys.argv[1:] or sorted(str(p) for p in Path("data/optimization").glob("*.json")) if not paths: print("no optimization cases found", file=sys.stderr) return 2 reports = [] for path in paths: case = json.loads(Path(path).read_text()) validate_case(case) reports.append(evaluate(case)) print(json.dumps(reports, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())