Pairs harness: score plain vs historic-conditioned prediction per attribute 🟩

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 <noreply@anthropic.com>

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

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