tegwick
17de8b831e
First delegated evidence from RESOURCE-WP-0003-T04 to land. railiance-platform delivered apps-pg capacity, utilization, consumers, and the apps-pg-dbbytes-v1 allocation driver, and correctly delivered no EUR. - data/resources/apps-pg.json: real capacity; allocation unattributed -> shared under apps-pg-dbbytes-v1; second consumer vergabe-teilnahme registered - data/control-cycle/apps-pg-2026-09-base.json: first operational control-cycle record in the repository - examples/control-cycle/apps-pg-*.json retired; the invented fixture collided with the real record's identifier - data/portfolio-coverage-2026-08-14.json: gap marked delivered with three residual unknowns still open The real evidence exposed a design gap in the T05 schema: v0.1 required a number for every cost field, so recording genuine usage without a booked cost meant inventing one. Schema 0.2 permits null costs, null unattributed_eur, a technical unattributed_share, and null measurements. Null is unknown, never zero; an unknown component makes the total null rather than the sum of the known parts; and the comparator classifies unknown amounts as data_quality instead of computing a variance. Existing 0.1 records are not rewritten. apps-pg is now measured (idle at 5.8% of volume) and attributed, and remains unpriced: delivered technical evidence does not create a booked cost. 86 tests pass. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
92 lines
3.7 KiB
Python
92 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Compare a generic immutable resource forecast with an actual observation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
COST_FIELDS = ("infrastructure", "internal_labor", "external_labor", "total")
|
|
ATTRIBUTIONS = {"demand", "provider_price", "allocation", "labor", "model", "data_quality"}
|
|
|
|
|
|
def delta(forecast: float | None, actual: float | None) -> dict:
|
|
# A missing amount is unknown, not zero: subtracting against it would
|
|
# manufacture a variance the evidence does not support.
|
|
if forecast is None or actual is None:
|
|
return {"forecast": forecast, "actual": actual, "status": "unknown"}
|
|
error = actual - forecast
|
|
return {
|
|
"forecast": forecast,
|
|
"actual": actual,
|
|
"error": round(error, 4),
|
|
"absolute_percentage_error": None if forecast == 0 else round(abs(error) / forecast * 100, 2),
|
|
}
|
|
|
|
|
|
def compare(forecast: dict, actual: dict) -> dict:
|
|
if forecast["record_type"] != "forecast" or actual["record_type"] != "actual":
|
|
raise ValueError("expected forecast and actual records")
|
|
for field in ("resource_id", "resource_class", "period"):
|
|
if forecast[field] != actual[field]:
|
|
raise ValueError(f"{field} mismatch")
|
|
if actual.get("forecast_ref") != forecast["record_id"]:
|
|
raise ValueError("actual forecast_ref must identify the immutable forecast")
|
|
|
|
attribution = actual.get("variance_attribution", {})
|
|
unknown = set(attribution.values()) - ATTRIBUTIONS
|
|
if unknown:
|
|
raise ValueError(f"unknown variance attribution: {sorted(unknown)}")
|
|
|
|
proxies = {}
|
|
all_proxies = sorted(set(forecast["usage_proxies"]) | set(actual["usage_proxies"]))
|
|
for name in all_proxies:
|
|
planned = forecast["usage_proxies"].get(name)
|
|
observed = actual["usage_proxies"].get(name)
|
|
if planned is None or observed is None:
|
|
proxies[name] = {"status": "missing", "category": "data_quality"}
|
|
elif planned["unit"] != observed["unit"]:
|
|
proxies[name] = {"status": "unit-mismatch", "category": "data_quality"}
|
|
else:
|
|
proxies[name] = {**delta(planned["value"], observed["value"]), "unit": planned["unit"], "category": attribution.get(name, "demand")}
|
|
|
|
costs = {}
|
|
for name in COST_FIELDS:
|
|
default_category = "labor" if "labor" in name else "provider_price"
|
|
result = delta(forecast["costs"][name], actual["costs"][name])
|
|
# An unknown amount is a data-quality gap, not a price or labour movement.
|
|
category = "data_quality" if result.get("status") == "unknown" else attribution.get(
|
|
f"costs.{name}", default_category
|
|
)
|
|
costs[name] = {**result, "currency": "EUR", "category": category}
|
|
|
|
if forecast["allocation"] != actual["allocation"]:
|
|
costs["allocation_method"] = {"status": "changed", "category": attribution.get("allocation", "allocation")}
|
|
|
|
return {
|
|
"forecast_ref": forecast["record_id"],
|
|
"actual_ref": actual["record_id"],
|
|
"resource_id": forecast["resource_id"],
|
|
"resource_class": forecast["resource_class"],
|
|
"period": forecast["period"],
|
|
"usage_proxies": proxies,
|
|
"costs": costs,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) != 3:
|
|
print(f"usage: {sys.argv[0]} FORECAST.json ACTUAL.json", file=sys.stderr)
|
|
return 2
|
|
try:
|
|
result = compare(json.loads(Path(sys.argv[1]).read_text()), json.loads(Path(sys.argv[2]).read_text()))
|
|
except (KeyError, ValueError) as exc:
|
|
print(f"control-cycle error: {exc}", file=sys.stderr)
|
|
return 1
|
|
print(json.dumps(result, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|