"""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] = [] 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) 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]) 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()