From d9576db42965938ed4ccd179670d02cbf2b9ab8f Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Mon, 6 Jul 2026 08:46:16 +0000 Subject: [PATCH] =?UTF-8?q?Pairs=20harness:=20score=20plain=20vs=20histori?= =?UTF-8?q?c-conditioned=20prediction=20per=20attribute=20=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Finds pre-2012 (S3 backup) x SAP-10.2 (new API) cert pairs per postcode, predicts each from its leave-one-out cohort with and without ADR-0054 conditioning, and reports compare_prediction hit rates side by side. Evidence for the stable-attribute whitelist, not a CI gate. Co-Authored-By: Claude Opus 4.8 --- scripts/expired_prediction_pairs_harness.py | 224 ++++++++++++++++++ .../test_expired_prediction_pairs_harness.py | 78 ++++++ 2 files changed, 302 insertions(+) create mode 100644 scripts/expired_prediction_pairs_harness.py create mode 100644 tests/scripts/test_expired_prediction_pairs_harness.py diff --git a/scripts/expired_prediction_pairs_harness.py b/scripts/expired_prediction_pairs_harness.py new file mode 100644 index 000000000..e384103bc --- /dev/null +++ b/scripts/expired_prediction_pairs_harness.py @@ -0,0 +1,224 @@ +"""Pre-2012 x post-June-2025 pairs harness for Expired-Enhanced Prediction (ADR-0054). + +For each postcode, find properties holding BOTH a pre-2012 cert in the historic +S3 backup AND a SAP-10.2 cert on the new gov API (RdSAP 10 went live June 2025; +only a same-spec lodged figure is a valid validation target — see Component +Accuracy). Each pair is predicted twice from its leave-one-out postcode cohort: + +- PLAIN arm: property type + built form only (what a blind prediction sees); +- CONDITIONED arm: the historic cert's stable attributes conditioning cohort + selection (ADR-0054). + +Both arms are scored against the lodged SAP-10.2 components with +`compare_prediction`; the report prints per-attribute hit rates side by side, +so any whitelist member (main fuel is the judgement call) can be promoted or +demoted with evidence. A report, not a CI gate. + +Usage: + python scripts/expired_prediction_pairs_harness.py "B93 8SY" "LS6 1AA" + python scripts/expired_prediction_pairs_harness.py --postcodes-file pcs.txt + +Env: OPEN_EPC_API_TOKEN, DATA_BUCKET; ambient AWS credentials for S3. +""" + +from __future__ import annotations + +import argparse +import os +import sys +from collections import defaultdict +from dataclasses import dataclass +from pathlib import Path +from typing import Optional + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from datatypes.epc.domain.epc_property_data import EpcPropertyData # noqa: E402 +from datatypes.epc.domain.historic_epc import HistoricEpc # noqa: E402 +from domain.epc_prediction.comparable_properties import ( # noqa: E402 + ComparableProperty, + select_comparables, +) +from domain.epc_prediction.epc_prediction import EpcPrediction # noqa: E402 +from domain.epc_prediction.historic_conditioning import ( # noqa: E402 + attributes_with_historic_fallback, + conditioning_from_historic, + target_with_conditioning, +) +from domain.epc_prediction.prediction_comparison import ( # noqa: E402 + PredictionComparison, + compare_prediction, +) +from domain.epc_prediction.prediction_target import ( # noqa: E402 + PredictionTarget, + build_prediction_target, +) +from domain.postcode import Postcode # noqa: E402 +from domain.property.property import PropertyIdentity # noqa: E402 + +_PRE_2012 = "2012-01-01" +_VALIDATION_SAP_VERSION = 10.2 + + +def latest_pre_2012_by_uprn(records: list[HistoricEpc]) -> dict[str, HistoricEpc]: + """One historic cert per UPRN: the latest lodgement strictly before 2012 + (ISO date strings compare lexicographically). UPRN-less rows are dropped — + without a UPRN there is nothing to pair.""" + by_uprn: dict[str, HistoricEpc] = {} + for record in records: + if not record.uprn or not record.lodgement_date: + continue + if record.lodgement_date >= _PRE_2012: + continue + current = by_uprn.get(record.uprn) + if current is None or record.lodgement_date > current.lodgement_date: + by_uprn[record.uprn] = record + return by_uprn + + +@dataclass(frozen=True) +class ArmScores: + """Aggregated categorical hit counts for one arm: component -> (hits, scored).""" + + hits: dict[str, tuple[int, int]] + floor_area_abs_residuals: list[float] + + +def aggregate(comparisons: list[PredictionComparison]) -> ArmScores: + counts: dict[str, list[int]] = defaultdict(lambda: [0, 0]) + residuals: list[float] = [] + for comparison in comparisons: + for component, hit in comparison.categorical_hits.items(): + if hit is None: + continue + counts[component][1] += 1 + if hit: + counts[component][0] += 1 + residuals.append(abs(comparison.floor_area_residual)) + return ArmScores( + hits={k: (v[0], v[1]) for k, v in counts.items()}, + floor_area_abs_residuals=residuals, + ) + + +def format_report(plain: ArmScores, conditioned: ArmScores, pairs: int) -> str: + components = sorted(set(plain.hits) | set(conditioned.hits)) + lines = [ + f"# Expired-Enhanced Prediction pairs report ({pairs} pairs)", + "", + "| component | plain | conditioned |", + "|---|---|---|", + ] + for component in components: + p_hit, p_all = plain.hits.get(component, (0, 0)) + c_hit, c_all = conditioned.hits.get(component, (0, 0)) + p = f"{p_hit}/{p_all}" if p_all else "n/a" + c = f"{c_hit}/{c_all}" if c_all else "n/a" + lines.append(f"| {component} | {p} | {c} |") + + def _mean(values: list[float]) -> str: + return f"{sum(values) / len(values):.1f}" if values else "n/a" + + lines.append( + f"| floor_area mean abs residual (m²) | " + f"{_mean(plain.floor_area_abs_residuals)} | " + f"{_mean(conditioned.floor_area_abs_residuals)} |" + ) + return "\n".join(lines) + + +def _predict_arm( + target: Optional[PredictionTarget], + cohort: list[ComparableProperty], + predictor: EpcPrediction, +) -> Optional[EpcPropertyData]: + if target is None: + return None + comparables = select_comparables(target, cohort) + if not comparables.members: + return None + return predictor.predict(target, comparables) + + +def run(postcodes: list[str]) -> str: # pragma: no cover - live IO composition + from infrastructure.epc_client.epc_client_service import EpcClientService + from repositories.comparable_properties.epc_comparable_properties_repository import ( + EpcComparablePropertiesRepository, + ) + from repositories.geospatial.geospatial_s3_repository import ( + GeospatialS3Repository, + ) + from repositories.historic_epc.historic_epc_s3_repository import ( + HistoricEpcS3Repository, + ) + from scripts.e2e_common import load_env, s3_parquet_reader + + load_env() + + epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"]) + geospatial = GeospatialS3Repository(s3_parquet_reader(os.environ["DATA_BUCKET"])) + comparables_repo = EpcComparablePropertiesRepository(epc_client, geospatial) + historic_repo = HistoricEpcS3Repository.with_default_s3_client() + predictor = EpcPrediction() + + plain_comparisons: list[PredictionComparison] = [] + conditioned_comparisons: list[PredictionComparison] = [] + pairs = 0 + for raw_postcode in postcodes: + postcode = str(Postcode(raw_postcode)) + historic = latest_pre_2012_by_uprn( + historic_repo.get_for_postcode(Postcode(raw_postcode)) + ) + if not historic: + print(f"{postcode}: no pre-2012 historic certs", file=sys.stderr) + continue + cohort = comparables_repo.candidates_for(postcode) + for uprn, record in historic.items(): + actual = epc_client.get_by_uprn(int(uprn)) + if actual is None or actual.sap_version != _VALIDATION_SAP_VERSION: + continue + pairs += 1 + loo_cohort = [c for c in cohort if c.epc.uprn != int(uprn)] + identity = PropertyIdentity( + portfolio_id=0, postcode=postcode, address=record.address, uprn=int(uprn) + ) + conditioning = conditioning_from_historic(record) + attributes = attributes_with_historic_fallback(None, conditioning) + plain_target = build_prediction_target(identity, None, attributes) + conditioned_target = ( + target_with_conditioning(plain_target, conditioning) + if plain_target is not None + else None + ) + plain = _predict_arm(plain_target, loo_cohort, predictor) + conditioned = _predict_arm(conditioned_target, loo_cohort, predictor) + if plain is not None: + plain_comparisons.append(compare_prediction(plain, actual)) + if conditioned is not None: + conditioned_comparisons.append(compare_prediction(conditioned, actual)) + print(f"{postcode} {uprn}: pair scored", file=sys.stderr) + + return format_report( + aggregate(plain_comparisons), aggregate(conditioned_comparisons), pairs + ) + + +def main() -> None: # pragma: no cover - CLI entry + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("postcodes", nargs="*", help="postcodes to scan") + parser.add_argument("--postcodes-file", type=Path, default=None) + args = parser.parse_args() + postcodes: list[str] = list(args.postcodes) + if args.postcodes_file is not None: + postcodes += [ + line.strip() + for line in args.postcodes_file.read_text().splitlines() + if line.strip() + ] + if not postcodes: + parser.error("no postcodes given") + print(run(postcodes)) + + +if __name__ == "__main__": + main() diff --git a/tests/scripts/test_expired_prediction_pairs_harness.py b/tests/scripts/test_expired_prediction_pairs_harness.py new file mode 100644 index 000000000..98de0c007 --- /dev/null +++ b/tests/scripts/test_expired_prediction_pairs_harness.py @@ -0,0 +1,78 @@ +"""Pure pair-selection and aggregation logic of the ADR-0054 pairs harness.""" + +from __future__ import annotations + +import dataclasses + +from datatypes.epc.domain.historic_epc import HistoricEpc +from domain.epc_prediction.prediction_comparison import PredictionComparison +from scripts.expired_prediction_pairs_harness import ( + aggregate, + format_report, + latest_pre_2012_by_uprn, +) + + +def _hist(uprn: str, lodgement_date: str) -> HistoricEpc: + fields = {f.name: "" for f in dataclasses.fields(HistoricEpc)} + fields["uprn"] = uprn + fields["lodgement_date"] = lodgement_date + return HistoricEpc(**fields) + + +def _comparison(hits: dict[str, bool | None]) -> PredictionComparison: + return PredictionComparison( + categorical_hits=hits, + floor_area_residual=-4.0, + building_parts_residual=0, + window_count_residual=0, + total_window_area_residual=0.0, + door_count_residual=0, + ) + + +def test_pairs_keep_only_the_latest_pre_2012_cert_per_uprn(): + # Arrange — one UPRN lodged twice pre-2012, once post-2012; one UPRN-less row. + records = [ + _hist("100", "2008-05-01"), + _hist("100", "2010-11-30"), + _hist("100", "2013-01-01"), + _hist("", "2009-01-01"), + ] + + # Act + by_uprn = latest_pre_2012_by_uprn(records) + + # Assert — the 2010 cert wins; the 2013 one can never seed an expired pair. + assert set(by_uprn) == {"100"} + assert by_uprn["100"].lodgement_date == "2010-11-30" + + +def test_aggregate_counts_hits_and_skips_not_applicable(): + # Arrange — two comparisons; wall scored twice (1 hit), roof scored once + # (None means the actual lodged no value — out of the denominator). + comparisons = [ + _comparison({"wall_construction": True, "roof_construction": None}), + _comparison({"wall_construction": False, "roof_construction": True}), + ] + + # Act + scores = aggregate(comparisons) + + # Assert + assert scores.hits["wall_construction"] == (1, 2) + assert scores.hits["roof_construction"] == (1, 1) + assert scores.floor_area_abs_residuals == [4.0, 4.0] + + +def test_report_prints_both_arms_side_by_side(): + # Arrange + plain = aggregate([_comparison({"wall_construction": False})]) + conditioned = aggregate([_comparison({"wall_construction": True})]) + + # Act + report = format_report(plain, conditioned, pairs=1) + + # Assert + assert "| wall_construction | 0/1 | 1/1 |" in report + assert "1 pairs" in report