#!/usr/bin/env python3 """Validate the APLD production forecast-vs-actual ledger.""" from __future__ import annotations import csv import re import sys from pathlib import Path BASE = Path(__file__).resolve().parent ROWS = BASE / "apld_forecast_actual_rows.csv" MANIFEST = BASE / "apld_forecast_actual_manifest.csv" REPORT = BASE / "apld_forecast_actual_validation_report.md" REQUIRED = [ "company", "period", "metric_name", "raw_value", "normalized_value", "unit", "value_qualifier", "ownership_basis", "forecast_or_actual", "is_derived", "source_url", "source_location", "source_snippet", "snippet_support_status", "confidence", "caveat", ] ALLOWED_FORECAST_ACTUAL = {"forecast", "actual"} ALLOWED_SUPPORT = {"exact", "partial"} ALLOWED_CONFIDENCE = {"high", "medium", "low"} ALLOWED_UNITS = {"MW", "USD_billions"} MIN_ROWS = 20 def tokens(text: str) -> list[str]: return re.findall(r"\d+(?:\.\d+)?", str(text).replace(",", "")) def expected_normalized(raw_value: str, unit: str, snippet: str) -> float: raw = raw_value.replace(",", "").strip() if "+" in raw: return sum(float(part) for part in raw.split("+")) value = float(raw) if unit == "MW" and re.search(r"\bGW\b|gigawatt", snippet, flags=re.I): return value * 1000 return value def load_csv(path: Path) -> list[dict[str, str]]: if not path.exists(): raise FileNotFoundError(path) with path.open(newline="", encoding="utf-8") as f: return list(csv.DictReader(f)) def validate() -> tuple[list[str], list[dict[str, str]], list[dict[str, str]]]: errors: list[str] = [] rows = load_csv(ROWS) manifest = load_csv(MANIFEST) manifest_by_id = {row["source_id"]: row for row in manifest} manifest_urls = {row["source_url"] for row in manifest} if len(rows) < MIN_ROWS: errors.append(f"expected at least {MIN_ROWS} rows, found {len(rows)}") if rows: missing_cols = sorted(set(REQUIRED) - set(rows[0])) if missing_cols: errors.append(f"missing required columns: {missing_cols}") classes = set() units = set() periods = set() source_ids = set() for idx, row in enumerate(rows, start=2): for field in REQUIRED: if not row.get(field, "").strip(): errors.append(f"row {idx}: missing {field}") if row.get("company") != "APLD": errors.append(f"row {idx}: company must be APLD") if row.get("forecast_or_actual") not in ALLOWED_FORECAST_ACTUAL: errors.append(f"row {idx}: invalid forecast_or_actual {row.get('forecast_or_actual')}") if row.get("is_derived") not in {"true", "false"}: errors.append(f"row {idx}: is_derived must be true/false") if row.get("unit") not in ALLOWED_UNITS: errors.append(f"row {idx}: unexpected unit {row.get('unit')}") if row.get("snippet_support_status") not in ALLOWED_SUPPORT: errors.append(f"row {idx}: invalid snippet_support_status") if row.get("confidence") not in ALLOWED_CONFIDENCE: errors.append(f"row {idx}: invalid confidence") if not row.get("source_url", "").startswith("https://"): errors.append(f"row {idx}: source_url must be https") if row.get("source_url") not in manifest_urls: errors.append(f"row {idx}: source_url not found in manifest") sid = row.get("source_id", "") if sid and sid not in manifest_by_id: errors.append(f"row {idx}: source_id {sid} missing from manifest") if sid and row.get("source_url") != manifest_by_id.get(sid, {}).get("source_url"): errors.append(f"row {idx}: source_id/source_url mismatch") raw_tokens = tokens(row.get("raw_value", "")) snippet = row.get("source_snippet", "").replace(",", "") if raw_tokens and not any(tok in snippet for tok in raw_tokens): errors.append(f"row {idx}: source_snippet lacks raw numeric support for {row.get('raw_value')}") try: normalized = float(row.get("normalized_value", "")) expected = expected_normalized(row.get("raw_value", ""), row.get("unit", ""), row.get("source_snippet", "")) except ValueError: errors.append(f"row {idx}: normalized_value/raw_value is not numeric-normalizable") else: if abs(normalized - expected) > 0.001: errors.append(f"row {idx}: normalized_value {normalized} != expected {expected}") if len(row.get("ownership_basis", "")) < 12: errors.append(f"row {idx}: ownership_basis too terse") if len(row.get("caveat", "")) < 12: errors.append(f"row {idx}: caveat too terse") classes.add(row.get("forecast_or_actual")) units.add(row.get("unit")) periods.add(row.get("period")) source_ids.add(row.get("source_id", "")) if classes != {"forecast", "actual"}: errors.append(f"ledger must include both forecast and actual rows; found {sorted(classes)}") if "MW" not in units or "USD_billions" not in units: errors.append("ledger must include MW and USD_billions units") if len(periods) < 4: errors.append(f"expected at least 4 quarter periods, found {len(periods)}") if len(source_ids - {""}) < 8: errors.append(f"expected at least 8 primary sources represented, found {len(source_ids - {''})}") comparable_forecast = [r for r in rows if r["metric_name"] == "planned_ready_for_service_capacity" and r["campus"] == "Polaris Forge 1" and r["raw_value"] == "100" and r["forecast_or_actual"] == "forecast"] comparable_actual = [r for r in rows if r["metric_name"] in {"ready_for_service_capacity", "operating_hpc_capacity"} and r["campus"] == "Polaris Forge 1" and r["raw_value"] == "100" and r["forecast_or_actual"] == "actual"] if not comparable_forecast or not comparable_actual: errors.append("missing comparable PF1 100 MW forecast and actual rows") for m in manifest: for field in ["source_id", "company", "source_family", "source_type", "period", "source_date", "source_url", "expected_table_or_section", "access_caveats"]: if not m.get(field, "").strip(): errors.append(f"manifest {m.get('source_id', '')}: missing {field}") if m.get("source_family") != "PRIMARY": errors.append(f"manifest {m.get('source_id')}: non-primary source") if not m.get("source_url", "").startswith("https://"): errors.append(f"manifest {m.get('source_id')}: source_url must be https") return errors, rows, manifest def write_report(errors: list[str], rows: list[dict[str, str]], manifest: list[dict[str, str]]) -> None: status = "PASS" if not errors else "FAIL" lines = [ "# APLD Forecast-vs-Actual Validation Report", "", f"Status: {status}", "", "## Checks", f"- Required row fields: {'pass' if rows and set(REQUIRED).issubset(rows[0]) else 'fail'}", f"- Minimum row count ({MIN_ROWS}): {'pass' if len(rows) >= MIN_ROWS else 'fail'} ({len(rows)} rows)", f"- Primary-source manifest entries: {'pass' if manifest else 'fail'} ({len(manifest)} sources)", f"- Forecast/actual coverage: {sorted({r.get('forecast_or_actual') for r in rows})}", f"- Period coverage: {sorted({r.get('period') for r in rows})}", f"- Source URL provenance: {'pass' if not any('source_url' in e for e in errors) else 'fail'}", f"- Unit normalization: {'pass' if not any('normalized_value' in e for e in errors) else 'fail'}", f"- Snippet numeric support: {'pass' if not any('source_snippet lacks raw numeric support' in e for e in errors) else 'fail'}", "", "## Comparable Metric Gate", "PF1 100 MW ready-for-service is represented as a FY2025Q4 forecast and FY2026Q2 actual RFS/operating delivery.", "", "## Errors", ] if errors: lines.extend(f"- {err}" for err in errors) else: lines.append("- None") lines.append("") REPORT.write_text("\n".join(lines), encoding="utf-8") def main() -> int: try: errors, rows, manifest = validate() except Exception as exc: errors, rows, manifest = [f"fatal validation error: {exc}"], [], [] write_report(errors, rows, manifest) print(REPORT.read_text(encoding="utf-8")) return 1 if errors else 0 if __name__ == "__main__": sys.exit(main())