tegwick
2c2a6073ff
T06: optimization-case schema, fail-closed evaluator, cadence and decision template. Every option including the baseline must present all ten decision fields; one unknown blocks the comparison. Validated on the storage case (Hetzner computes and loses to Scaleway by EUR 29.14/month on operator labour; Host Europe blocks on four named gaps) and on the non-storage reef-railiance k3s rightsizing case (low utilization is real, but nothing is costable while the railiance01 price is unknown). T07: portfolio report over coverage, lifecycle, utilization, cost, renewals, risks, open cases, and next actions, derived only from committed evidence. Portfolio spend is reported null rather than as a partial sum, unattributed cost is a named list rather than a spread, and unmeasurable resources are reported rather than dropped. RESOURCE-WP-0003 is finished; both cases remain blocked_on_evidence against live delegated records in other repositories. RESOURCE-WP-0002 is untouched. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
99 lines
4.7 KiB
Python
99 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Calculate comparable 12-month object-storage forecasts from JSON evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def quote_provider(provider: dict, scenario: dict, stored_gb: float, restore_gb: float) -> dict:
|
|
excess_storage = max(0, stored_gb - provider["included_storage_gb"])
|
|
excess_egress = max(0, restore_gb - provider["included_egress_gb"])
|
|
missing = []
|
|
if excess_storage and provider["storage_eur_per_gb_month"] is None:
|
|
missing.append("storage_eur_per_gb_month")
|
|
if excess_egress and provider["egress_eur_per_gb"] is None:
|
|
missing.append("egress_eur_per_gb")
|
|
for field in ("monthly_minimum_eur", "operations_eur_per_month", "support_eur_per_month"):
|
|
if provider[field] is None:
|
|
missing.append(field)
|
|
|
|
infrastructure = None
|
|
labor_hours = max(scenario["operator_hours_per_month"], provider["operator_hours_per_month"])
|
|
labor = labor_hours * scenario["operator_hourly_eur"]
|
|
recurring = None
|
|
exit_cost = None
|
|
if not missing:
|
|
usage = excess_storage * (provider["storage_eur_per_gb_month"] or 0)
|
|
usage += excess_egress * (provider["egress_eur_per_gb"] or 0)
|
|
usage += scenario.get("write_requests_per_month", 0) / 1000 * provider.get("write_eur_per_1000", 0)
|
|
usage += scenario.get("read_requests_per_month", 0) / 1000 * provider.get("read_eur_per_1000", 0)
|
|
service = max(provider["monthly_minimum_eur"], usage)
|
|
infrastructure = service + provider["operations_eur_per_month"] + provider["support_eur_per_month"]
|
|
recurring = infrastructure + labor
|
|
exit_excess = max(0, stored_gb - provider["included_egress_gb"])
|
|
if exit_excess and provider["egress_eur_per_gb"] is None:
|
|
missing.append("egress_eur_per_gb_for_exit")
|
|
else:
|
|
exit_cost = exit_excess * (provider["egress_eur_per_gb"] or 0) + 4 * scenario["operator_hourly_eur"]
|
|
|
|
setup_internal_hours = provider.get("setup_operator_hours", 0)
|
|
setup_internal = setup_internal_hours * scenario["operator_hourly_eur"]
|
|
setup_external = provider.get("setup_external_eur", None if "garage" in provider["id"] else 0)
|
|
return {
|
|
"monthly_infrastructure_eur": None if infrastructure is None else round(infrastructure, 2),
|
|
"monthly_internal_labor_hours": labor_hours,
|
|
"monthly_internal_labor_eur": round(labor, 2),
|
|
"recurring_total_eur": None if recurring is None else round(recurring, 2),
|
|
"setup_internal_labor_hours": setup_internal_hours,
|
|
"setup_internal_labor_eur": round(setup_internal, 2),
|
|
"setup_external_services_eur": setup_external,
|
|
"setup_known_total_eur": None if setup_external is None else round(setup_internal + setup_external, 2),
|
|
"exit_cost_eur": None if exit_cost is None else round(exit_cost, 2),
|
|
"missing_price_fields": sorted(set(missing)),
|
|
}
|
|
|
|
|
|
def forecast(demand: dict, catalog: dict) -> dict:
|
|
result = {"currency": demand["currency"], "months": 12, "scenarios": {}, "comparison_320gb": []}
|
|
retention = demand["retention_days"]
|
|
backups_per_day = demand["base_backups_per_day"]
|
|
for scenario_name, scenario in demand["scenarios"].items():
|
|
rows = []
|
|
db_gb = scenario["initial_database_gb"]
|
|
for month in range(1, 13):
|
|
stored_gb = db_gb * retention * backups_per_day + scenario["wal_gb_per_day"] * retention
|
|
restore_gb = scenario["restore_egress_gb"]
|
|
for provider in catalog["providers"]:
|
|
rows.append({
|
|
"month": month, "provider_id": provider["id"],
|
|
"database_gb": round(db_gb, 3), "stored_gb": round(stored_gb, 3),
|
|
"restore_egress_gb": restore_gb,
|
|
**quote_provider(provider, scenario, stored_gb, restore_gb),
|
|
})
|
|
db_gb *= 1 + scenario["monthly_database_growth_pct"] / 100
|
|
result["scenarios"][scenario_name] = rows
|
|
normalized = demand["scenarios"]["base"]
|
|
for provider in catalog["providers"]:
|
|
result["comparison_320gb"].append({
|
|
"provider_id": provider["id"], "stored_gb": 320,
|
|
"restore_egress_gb": normalized["restore_egress_gb"],
|
|
**quote_provider(provider, normalized, 320, normalized["restore_egress_gb"]),
|
|
})
|
|
return result
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) != 3:
|
|
print(f"usage: {sys.argv[0]} DEMAND.json PROVIDERS.json", file=sys.stderr)
|
|
return 2
|
|
demand = json.loads(Path(sys.argv[1]).read_text())
|
|
catalog = json.loads(Path(sys.argv[2]).read_text())
|
|
print(json.dumps(forecast(demand, catalog), indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|