From 1d83637afa189caebb98ed15289363b88c480d05 Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Mon, 6 Jul 2026 09:05:59 +0000 Subject: [PATCH] =?UTF-8?q?Pairs=20harness=20reports=20the=20full=20Compon?= =?UTF-8?q?ent=20Accuracy=20suite=20per=20arm=20=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Classification hit-rates for every compare_prediction component, all five numeric residuals, and the secondary calculator-floored SAP residual (calc(predicted) − lodged), plain vs conditioned side by side — the same metric shape as the prediction-corpus gate (ADR-0030). Pair-check now precedes the cohort fetch so a national postcode sweep only pays the expensive search-by-postcode for shards that actually hold a pair. Co-Authored-By: Claude Opus 4.8 --- scripts/expired_prediction_pairs_harness.py | 187 +++++++++++++----- .../test_expired_prediction_pairs_harness.py | 61 +++--- 2 files changed, 178 insertions(+), 70 deletions(-) diff --git a/scripts/expired_prediction_pairs_harness.py b/scripts/expired_prediction_pairs_harness.py index e384103bc..bb619b469 100644 --- a/scripts/expired_prediction_pairs_harness.py +++ b/scripts/expired_prediction_pairs_harness.py @@ -1,22 +1,29 @@ -"""Pre-2012 x post-June-2025 pairs harness for Expired-Enhanced Prediction (ADR-0054). +"""Pre-2012 x SAP-10.2 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: +Accuracy, ADR-0030). 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. +Both arms are scored against the lodged SAP-10.2 cert with the SAME metric +suite as the prediction corpus (ADR-0030 Component Accuracy): per-component +classification hit-rates, mean-absolute numeric residuals, plus the secondary +calculator-floored SAP residual (calc(predicted) vs the lodged score). The +per-attribute breakdown is the whitelist evidence: an attribute whose +conditioned hit-rate is WORSE than plain is stale and gets demoted. + +The expensive cohort fetch (search-by-postcode + per-cert fetch) happens only +for postcodes where a pair actually exists, so the script can sweep hundreds +of postcodes cheaply. Usage: python scripts/expired_prediction_pairs_harness.py "B93 8SY" "LS6 1AA" - python scripts/expired_prediction_pairs_harness.py --postcodes-file pcs.txt + python scripts/expired_prediction_pairs_harness.py --postcodes-file pcs.txt --out report.md Env: OPEN_EPC_API_TOKEN, DATA_BUCKET; ambient AWS credentials for S3. """ @@ -59,6 +66,14 @@ from domain.property.property import PropertyIdentity # noqa: E402 _PRE_2012 = "2012-01-01" _VALIDATION_SAP_VERSION = 10.2 +_RESIDUAL_COMPONENTS = ( + "floor_area_m2", + "building_parts", + "window_count", + "total_window_area_m2", + "door_count", +) + def latest_pre_2012_by_uprn(records: list[HistoricEpc]) -> dict[str, HistoricEpc]: """One historic cert per UPRN: the latest lodgement strictly before 2012 @@ -76,54 +91,102 @@ def latest_pre_2012_by_uprn(records: list[HistoricEpc]) -> dict[str, HistoricEpc return by_uprn +@dataclass(frozen=True) +class PairScore: + """One arm's score for one pair: the component comparison plus the + calculator-floored SAP residual (calc(predicted) − lodged), None when the + calculator could not score the predicted picture.""" + + comparison: PredictionComparison + sap_residual: Optional[float] + + @dataclass(frozen=True) class ArmScores: - """Aggregated categorical hit counts for one arm: component -> (hits, scored).""" + """One arm aggregated in the ComponentAccuracy shape (ADR-0030): + classification maps component -> (hits, applicable-total); residuals maps a + numeric component -> signed values; sap_residuals are calc − lodged.""" - hits: dict[str, tuple[int, int]] - floor_area_abs_residuals: list[float] + classification: dict[str, tuple[int, int]] + residuals: dict[str, list[float]] + sap_residuals: list[float] -def aggregate(comparisons: list[PredictionComparison]) -> ArmScores: +def aggregate(scores: list[PairScore]) -> ArmScores: counts: dict[str, list[int]] = defaultdict(lambda: [0, 0]) - residuals: list[float] = [] - for comparison in comparisons: + residuals: dict[str, list[float]] = defaultdict(list) + sap_residuals: list[float] = [] + for score in scores: + comparison = score.comparison 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)) + residuals["floor_area_m2"].append(comparison.floor_area_residual) + residuals["building_parts"].append(float(comparison.building_parts_residual)) + residuals["window_count"].append(float(comparison.window_count_residual)) + residuals["total_window_area_m2"].append( + comparison.total_window_area_residual + ) + residuals["door_count"].append(float(comparison.door_count_residual)) + if score.sap_residual is not None: + sap_residuals.append(score.sap_residual) return ArmScores( - hits={k: (v[0], v[1]) for k, v in counts.items()}, - floor_area_abs_residuals=residuals, + classification={k: (v[0], v[1]) for k, v in counts.items()}, + residuals=dict(residuals), + sap_residuals=sap_residuals, ) +def _mean_abs(values: list[float]) -> str: + return f"{sum(abs(v) for v in values) / len(values):.1f}" if values else "n/a" + + +def _rate_cell(hits: tuple[int, int]) -> str: + hit, total = hits + return f"{hit}/{total} ({hit / total:.0%})" if total else "n/a" + + def format_report(plain: ArmScores, conditioned: ArmScores, pairs: int) -> str: - components = sorted(set(plain.hits) | set(conditioned.hits)) + """The two arms side by side, in the corpus gate's shape: classification + hit-rates per component, then mean-abs residuals, then the secondary SAP + residual.""" + components = sorted(set(plain.classification) | set(conditioned.classification)) lines = [ f"# Expired-Enhanced Prediction pairs report ({pairs} pairs)", "", + "## Classification hit-rates (hits/applicable)", + "", "| 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" + p = _rate_cell(plain.classification.get(component, (0, 0))) + c = _rate_cell(conditioned.classification.get(component, (0, 0))) 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)} |" - ) + lines += [ + "", + "## Mean absolute residuals (predicted − actual)", + "", + "| component | plain | conditioned |", + "|---|---|---|", + ] + for component in _RESIDUAL_COMPONENTS: + p = _mean_abs(plain.residuals.get(component, [])) + c = _mean_abs(conditioned.residuals.get(component, [])) + lines.append(f"| {component} | {p} | {c} |") + lines += [ + "", + "## SAP residual — secondary, calculator-floored (calc(predicted) − lodged)", + "", + "| metric | plain | conditioned |", + "|---|---|---|", + f"| mean abs | {_mean_abs(plain.sap_residuals)} " + f"| {_mean_abs(conditioned.sap_residuals)} |", + f"| scored | {len(plain.sap_residuals)} | {len(conditioned.sap_residuals)} |", + ] return "\n".join(lines) @@ -141,6 +204,7 @@ def _predict_arm( def run(postcodes: list[str]) -> str: # pragma: no cover - live IO composition + from domain.sap10_calculator.calculator import Sap10Calculator from infrastructure.epc_client.epc_client_service import EpcClientService from repositories.comparable_properties.epc_comparable_properties_repository import ( EpcComparablePropertiesRepository, @@ -154,29 +218,50 @@ def run(postcodes: list[str]) -> str: # pragma: no cover - live IO composition 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() + calculator = Sap10Calculator() - plain_comparisons: list[PredictionComparison] = [] - conditioned_comparisons: list[PredictionComparison] = [] + def sap_residual( + predicted: Optional[EpcPropertyData], actual: EpcPropertyData + ) -> Optional[float]: + if predicted is None or actual.energy_rating_current is None: + return None + try: + calculated: float = calculator.calculate(predicted).sap_score_continuous + except Exception as error: # the calculator strict-raises on gaps + print(f" calculator raised: {error}", file=sys.stderr) + return None + return calculated - float(actual.energy_rating_current) + + plain_scores: list[PairScore] = [] + conditioned_scores: list[PairScore] = [] pairs = 0 - for raw_postcode in postcodes: + for index, raw_postcode in enumerate(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) + # Pair-check first: only a postcode with a SAP-10.2 relodgement of a + # pre-2012 UPRN pays for the cohort fetch. + paired: list[tuple[str, HistoricEpc, EpcPropertyData]] = [] 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 + if actual is not None and actual.sap_version == _VALIDATION_SAP_VERSION: + paired.append((uprn, record, actual)) + print( + f"[{index + 1}/{len(postcodes)}] {postcode}: " + f"{len(historic)} pre-2012 UPRNs, {len(paired)} pairs", + file=sys.stderr, + flush=True, + ) + if not paired: + continue + cohort = comparables_repo.candidates_for(postcode) + for uprn, record, actual in paired: pairs += 1 loo_cohort = [c for c in cohort if c.epc.uprn != int(uprn)] identity = PropertyIdentity( @@ -193,20 +278,25 @@ def run(postcodes: list[str]) -> str: # pragma: no cover - live IO composition 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)) + plain_scores.append( + PairScore(compare_prediction(plain, actual), sap_residual(plain, actual)) + ) if conditioned is not None: - conditioned_comparisons.append(compare_prediction(conditioned, actual)) - print(f"{postcode} {uprn}: pair scored", file=sys.stderr) + conditioned_scores.append( + PairScore( + compare_prediction(conditioned, actual), + sap_residual(conditioned, actual), + ) + ) - return format_report( - aggregate(plain_comparisons), aggregate(conditioned_comparisons), pairs - ) + return format_report(aggregate(plain_scores), aggregate(conditioned_scores), 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) + parser.add_argument("--out", type=Path, default=None, help="write the report here") args = parser.parse_args() postcodes: list[str] = list(args.postcodes) if args.postcodes_file is not None: @@ -217,7 +307,10 @@ def main() -> None: # pragma: no cover - CLI entry ] if not postcodes: parser.error("no postcodes given") - print(run(postcodes)) + report = run(postcodes) + if args.out is not None: + args.out.write_text(report + "\n") + print(report) if __name__ == "__main__": diff --git a/tests/scripts/test_expired_prediction_pairs_harness.py b/tests/scripts/test_expired_prediction_pairs_harness.py index 98de0c007..2781dc566 100644 --- a/tests/scripts/test_expired_prediction_pairs_harness.py +++ b/tests/scripts/test_expired_prediction_pairs_harness.py @@ -3,10 +3,12 @@ from __future__ import annotations import dataclasses +from typing import Optional from datatypes.epc.domain.historic_epc import HistoricEpc from domain.epc_prediction.prediction_comparison import PredictionComparison from scripts.expired_prediction_pairs_harness import ( + PairScore, aggregate, format_report, latest_pre_2012_by_uprn, @@ -20,14 +22,19 @@ def _hist(uprn: str, lodgement_date: str) -> HistoricEpc: 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 _score( + hits: dict[str, Optional[bool]], sap_residual: Optional[float] = None +) -> PairScore: + return PairScore( + comparison=PredictionComparison( + categorical_hits=hits, + floor_area_residual=-4.0, + building_parts_residual=1, + window_count_residual=-2, + total_window_area_residual=3.5, + door_count_residual=0, + ), + sap_residual=sap_residual, ) @@ -48,31 +55,39 @@ def test_pairs_keep_only_the_latest_pre_2012_cert_per_uprn(): 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}), +def test_aggregate_matches_the_component_accuracy_shape(): + # Arrange — two pairs; wall scored twice (1 hit), roof scored once (None + # means the actual lodges no value — out of the denominator, per ADR-0030). + scores = [ + _score({"wall_construction": True, "roof_construction": None}, sap_residual=6.0), + _score({"wall_construction": False, "roof_construction": True}), ] # Act - scores = aggregate(comparisons) + arm = aggregate(scores) - # 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] + # Assert — classification (hits, applicable), all five residual components, + # and the SAP residual list (only the scored pair contributes). + assert arm.classification["wall_construction"] == (1, 2) + assert arm.classification["roof_construction"] == (1, 1) + assert arm.residuals["floor_area_m2"] == [-4.0, -4.0] + assert arm.residuals["building_parts"] == [1.0, 1.0] + assert arm.residuals["window_count"] == [-2.0, -2.0] + assert arm.residuals["total_window_area_m2"] == [3.5, 3.5] + assert arm.residuals["door_count"] == [0.0, 0.0] + assert arm.sap_residuals == [6.0] def test_report_prints_both_arms_side_by_side(): # Arrange - plain = aggregate([_comparison({"wall_construction": False})]) - conditioned = aggregate([_comparison({"wall_construction": True})]) + plain = aggregate([_score({"wall_construction": False}, sap_residual=-8.0)]) + conditioned = aggregate([_score({"wall_construction": True}, sap_residual=2.0)]) # Act report = format_report(plain, conditioned, pairs=1) - # Assert - assert "| wall_construction | 0/1 | 1/1 |" in report + # Assert — hit-rates, residuals and the SAP arm all present, side by side. + assert "| wall_construction | 0/1 (0%) | 1/1 (100%) |" in report + assert "| floor_area_m2 | 4.0 | 4.0 |" in report + assert "| mean abs | 8.0 | 2.0 |" in report assert "1 pairs" in report