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 datetime import date
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). Address + registration date
    come straight off the cached payload (the register metadata) so the harness
    can dedupe re-lodgements and hold out a whole address."""
    cohort: list[Comparable] = []
    for cert in certs:
        path = CORPUS / postcode / f"{cert}.json"
        if not path.exists():
            continue
        raw = json.loads(path.read_text())
        try:
            epc = EpcPropertyDataMapper.from_api_response(raw)
        except Exception:  # noqa: BLE001 — a bad cert must not abort the sweep
            continue
        cohort.append(
            Comparable(
                epc=epc,
                certificate_number=cert,
                address=_address(raw),
                registration_date=_registration_date(raw),
            )
        )
    return cohort


def _address(raw: dict[str, object]) -> Optional[str]:
    value = raw.get("address_line_1")
    return str(value).strip().upper() if value else None


def _registration_date(raw: dict[str, object]) -> Optional[date]:
    value = raw.get("registration_date")
    return date.fromisoformat(str(value)) if value else None


def _ground_truth_properties(cohort: list[Comparable]) -> list[Comparable]:
    """Collapse a postcode's certs to one held-out property per address — the
    latest cert, the best ground truth. Comparables with no address each stand
    alone."""
    latest: dict[str, Comparable] = {}
    standalone: list[Comparable] = []
    for c in cohort:
        if c.address is None:
            standalone.append(c)
        elif c.address not in latest or _recency(c) > _recency(latest[c.address]):
            latest[c.address] = c
    return list(latest.values()) + standalone


def _recency(comparable: Comparable) -> tuple[date, str]:
    return (
        comparable.registration_date or date.min,
        comparable.certificate_number,
    )


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()

    # Classification: name -> [hits, applicable-total], populated from whatever
    # components compare_prediction reports (insertion order preserved). A None
    # hit (the actual lodges no value) is excluded from the denominator.
    categoricals: dict[str, list[int]] = {}
    floor_res: list[float] = []
    window_count_res: list[int] = []
    window_area_res: list[float] = []
    parts_res: list[int] = []
    door_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)
        targets = _ground_truth_properties(cohort)
        if len(targets) < 2:
            skipped_no_cohort += len(targets)
            continue
        for held_out in targets:
            # Exclude every cert of the held-out address (not just the held cert)
            # so a re-lodgement of the same property cannot leak into the cohort.
            others = [
                c
                for c in cohort
                if c.address is None or c.address != held_out.address
            ]
            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)
            for name, hit in cmp.categorical_hits.items():
                _tally(categoricals.setdefault(name, [0, 0]), hit)
            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)
            door_res.append(cmp.door_count_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")
    for name, (hits, total) in categoricals.items():
        if total:
            print(f"CLASSIFICATION  {name}: {hits}/{total} = {hits / total:.1%}")
    print()
    _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])
    _residual("door_count", [float(x) for x in door_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 _tally(counter: list[int], hit: Optional[bool]) -> None:
    """Record one classification outcome: a None hit (actual absent) is not
    applicable and skipped; else increment the applicable total and the hits."""
    if hit is None:
        return
    counter[1] += 1
    counter[0] += int(hit)


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()