Model/scripts/validate_epc_prediction.py

"""Leave-one-out accuracy harness for EPC Prediction (ADR-0029).

Runs entirely against the frozen postcode-clustered corpus
(`fetch_epc_prediction_corpus.py`). For every cert that has neighbours, it
drops that cert from its postcode cohort, predicts it from the rest using only
its *guaranteed* inputs (property type + built form), and compares the predicted
`EpcPropertyData` to the actual one.

Reports the ADR-0029 metrics:
  - classification rate: main wall construction (extend as coverage grows);
  - geometry residuals: floor area, window count + total window area, building
    parts (mean signed + mean absolute);
  - SAP reported three ways — predicted-then-calculated vs (a) the actual lodged
    SAP, (b) the calculator on the actual components, (c) the neighbour-mean SAP
    baseline (the number predict-then-calculate must beat).

USAGE
-----
    PYTHONPATH=. python scripts/validate_epc_prediction.py

Corpus dir: $EPC_PREDICTION_CORPUS (default /tmp/epc_prediction_corpus).
"""

from __future__ import annotations

import json
import os
import statistics
from pathlib import Path
from typing import Optional

from datatypes.epc.domain.epc_property_data import EpcPropertyData
from datatypes.epc.domain.mapper import EpcPropertyDataMapper
from domain.epc_prediction.comparable_properties import (
    Comparable,
    PredictionTarget,
    select_comparables,
)
from domain.epc_prediction.epc_prediction import EpcPrediction
from domain.epc_prediction.prediction_comparison import compare_prediction
from domain.sap10_calculator.calculator import Sap10Calculator

CORPUS = Path(os.environ.get("EPC_PREDICTION_CORPUS", "/tmp/epc_prediction_corpus"))


def _load_cohort(postcode: str, certs: list[str]) -> list[Comparable]:
    """Map a postcode's cached cert payloads to Comparables, skipping any the
    mapper rejects (unsupported schema, malformed)."""
    cohort: list[Comparable] = []
    for cert in certs:
        path = CORPUS / postcode / f"{cert}.json"
        if not path.exists():
            continue
        try:
            epc = EpcPropertyDataMapper.from_api_response(json.loads(path.read_text()))
        except Exception:  # noqa: BLE001 — a bad cert must not abort the sweep
            continue
        cohort.append(Comparable(epc=epc, certificate_number=cert))
    return cohort


def _sap(calculator: Sap10Calculator, epc: EpcPropertyData) -> Optional[float]:
    try:
        return calculator.calculate(epc).sap_score_continuous
    except Exception:  # noqa: BLE001 — some pictures don't score; count as misses
        return None


def main() -> None:
    index_path = CORPUS / "_index.json"
    if not index_path.exists():
        raise SystemExit(f"no corpus at {CORPUS} — run fetch_epc_prediction_corpus.py")
    index: dict[str, list[str]] = json.loads(index_path.read_text())

    calculator = Sap10Calculator()
    predictor = EpcPrediction()

    wall_hits = wall_total = 0
    floor_res: list[float] = []
    window_count_res: list[int] = []
    window_area_res: list[float] = []
    parts_res: list[int] = []
    sap_vs_lodged: list[float] = []
    sap_vs_calc_actual: list[float] = []
    sap_vs_neighbour_mean: list[float] = []
    predicted_n = skipped_no_cohort = 0

    for postcode, certs in index.items():
        cohort = _load_cohort(postcode, certs)
        if len(cohort) < 2:
            skipped_no_cohort += len(cohort)
            continue
        for i, held_out in enumerate(cohort):
            others = [c for j, c in enumerate(cohort) if j != i]
            actual = held_out.epc
            target = PredictionTarget(
                postcode=postcode,
                property_type=actual.property_type or "",
                built_form=actual.built_form,
            )
            comparables = select_comparables(target, others)
            if not comparables.members:
                continue
            predicted = predictor.predict(target, comparables)
            predicted_n += 1

            cmp = compare_prediction(predicted, actual)
            wall_total += 1
            wall_hits += int(cmp.wall_construction_correct)
            floor_res.append(cmp.floor_area_residual)
            window_count_res.append(cmp.window_count_residual)
            window_area_res.append(cmp.total_window_area_residual)
            parts_res.append(cmp.building_parts_residual)

            sap_pred = _sap(calculator, predicted)
            lodged = actual.energy_rating_current
            if sap_pred is not None and lodged is not None:
                sap_vs_lodged.append(abs(sap_pred - lodged))
            sap_actual = _sap(calculator, actual)
            if sap_pred is not None and sap_actual is not None:
                sap_vs_calc_actual.append(abs(sap_pred - sap_actual))
            neighbour_lodged = [
                c.epc.energy_rating_current
                for c in comparables.members
                if c.epc.energy_rating_current is not None
            ]
            if neighbour_lodged and lodged is not None:
                baseline = statistics.mean(neighbour_lodged)
                sap_vs_neighbour_mean.append(abs(baseline - lodged))

    print(f"corpus: {CORPUS}")
    print(f"predicted {predicted_n} held-out certs ({skipped_no_cohort} had no cohort)\n")
    if wall_total:
        print(f"CLASSIFICATION  wall_construction: {wall_hits}/{wall_total} = "
              f"{wall_hits / wall_total:.1%}")
    _residual("floor_area (m2)", floor_res)
    _residual("window_count", [float(x) for x in window_count_res])
    _residual("total_window_area (m2)", window_area_res)
    _residual("building_parts", [float(x) for x in parts_res])
    print()
    _sap_line("SAP |pred-calc − lodged|", sap_vs_lodged)
    _sap_line("SAP |pred-calc − calc(actual)|", sap_vs_calc_actual)
    _sap_line("SAP |neighbour-mean − lodged| (baseline)", sap_vs_neighbour_mean)


def _residual(label: str, values: list[float]) -> None:
    if not values:
        print(f"RESIDUAL  {label}: (none)")
        return
    mean_signed = statistics.mean(values)
    mean_abs = statistics.mean(abs(v) for v in values)
    print(f"RESIDUAL  {label}: mean {mean_signed:+.2f} | mean|·| {mean_abs:.2f} "
          f"(n={len(values)})")


def _sap_line(label: str, values: list[float]) -> None:
    if not values:
        print(f"{label}: (none)")
        return
    print(f"{label}: MAE {statistics.mean(values):.2f} | "
          f"median {statistics.median(values):.2f} (n={len(values)})")


if __name__ == "__main__":
    main()