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, SapResult

# Target-cert spec gate: only SAP 10.2 certs (schema 21.0.x) carry full-fidelity
# lodged components + a same-spec lodged figure to check against (ADR-0030). The
# source cohort keeps all vintages — components are methodology-agnostic.
_SAP_10_2: float = 10.2
_KG_PER_TONNE: float = 1000.0

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 _result(
    calculator: Sap10Calculator, epc: EpcPropertyData
) -> Optional[SapResult]:
    try:
        return calculator.calculate(epc)
    except Exception:  # noqa: BLE001 — some pictures don't score; count as misses
        return None


def _co2_tonnes(result: SapResult) -> float:
    """Calculated annual CO2 in tonnes, matching the lodged `co2_emissions_current`
    scale (see domain/property_baseline/performance.py)."""
    return result.co2_kg_per_yr / _KG_PER_TONNE


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] = []
    # End-to-end (calculator-FLOORED) vs API-lodged — secondary guard, ADR-0030.
    sap_vs_lodged: list[float] = []
    co2_vs_lodged: list[float] = []
    pei_vs_lodged: list[float] = []
    # Attribution readout: how far the calculator alone is from lodged on the
    # ACTUAL components — the floor the end-to-end numbers can reach.
    sap_calc_actual_vs_lodged: list[float] = []
    predicted_n = skipped_non_102 = 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:
            # Only SAP 10.2 certs are valid validation targets (ADR-0030); the
            # source cohort (`others`) keeps every vintage.
            if held_out.epc.sap_version != _SAP_10_2:
                skipped_non_102 += 1
                continue
            # 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)

            pred_result = _result(calculator, predicted)
            actual_result = _result(calculator, actual)
            lodged_sap = actual.energy_rating_current
            lodged_co2 = actual.co2_emissions_current
            lodged_pei = actual.energy_consumption_current
            if pred_result is not None:
                if lodged_sap is not None:
                    sap_vs_lodged.append(
                        abs(pred_result.sap_score_continuous - lodged_sap)
                    )
                if lodged_co2 is not None:
                    co2_vs_lodged.append(
                        abs(_co2_tonnes(pred_result) - lodged_co2)
                    )
                if lodged_pei is not None:
                    pei_vs_lodged.append(
                        abs(pred_result.primary_energy_kwh_per_m2 - lodged_pei)
                    )
            if actual_result is not None and lodged_sap is not None:
                sap_calc_actual_vs_lodged.append(
                    abs(actual_result.sap_score_continuous - lodged_sap)
                )

    print(f"corpus: {CORPUS}")
    print(
        f"predicted {predicted_n} SAP-10.2 held-out targets "
        f"({skipped_non_102} non-10.2 targets skipped, "
        f"{skipped_no_cohort} had no cohort)\n"
    )
    print("--- Component Accuracy (PRIMARY, calculator-independent) ---")
    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()
    print("--- End-to-end vs API-lodged (SECONDARY, calculator-FLOORED) ---")
    _sap_line("SAP |pred − lodged|", sap_vs_lodged)
    _sap_line("CO2 (t) |pred − lodged|", co2_vs_lodged)
    _sap_line("PEI (kWh/m2) |pred − lodged|", pei_vs_lodged)
    _sap_line("  floor: SAP |calc(actual) − lodged|", sap_calc_actual_vs_lodged)


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