Expired-pairs integration gate: frozen single-file corpus + ratcheting floors 🟩

30 pairs (28 deterministically scoreable) from the 2,000-postcode sweep,
frozen as ONE anonymised raw-payload JSON (pairs + cohorts + actuals — a
thousand per-cert files would drown the PR diff). The gate replays the
whole conditioning path offline — mapper, conditioning, selection,
synthesis, comparison — in ~9s; floors are the measured values, tighten-
only. comparable_from_payload is extracted from the corpus loader so both
fixture formats share one payload->ComparableProperty path; the builder
(build_expired_pairs_corpus.py) refreezes from the raw-JSON disk cache.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

harness/epc_prediction_corpus.py

@@ -56,23 +56,34 @@ def _load_cohort(
         if not path.exists():
             continue
         raw: dict[str, Any] = 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
-        uprn = _uprn(raw)
-        cohort.append(
-            ComparableProperty(
-                epc=epc,
-                certificate_number=cert,
-                address=_address(raw),
-                registration_date=_registration_date(raw),
-                coordinates=coordinates.get(uprn) if uprn is not None else None,
-            )
-        )
+        comparable = comparable_from_payload(cert, raw, coordinates)
+        if comparable is not None:
+            cohort.append(comparable)
     return cohort
 
 
+def comparable_from_payload(
+    cert: str,
+    raw: dict[str, Any],
+    coordinates: dict[int, Coordinates],
+) -> Optional[ComparableProperty]:
+    """One frozen cert payload -> a ComparableProperty, or None when the mapper
+    can't take it (a bad cert must never abort a corpus load). Shared by the
+    per-file corpus above and the single-file expired-pairs corpus."""
+    try:
+        epc = EpcPropertyDataMapper.from_api_response(raw)
+    except Exception:  # noqa: BLE001
+        return None
+    uprn = _uprn(raw)
+    return ComparableProperty(
+        epc=epc,
+        certificate_number=cert,
+        address=_address(raw),
+        registration_date=_registration_date(raw),
+        coordinates=coordinates.get(uprn) if uprn is not None else None,
+    )
+
+
 def load_coordinates(corpus_dir: Path) -> dict[int, Coordinates]:
     """The optional `_coordinates.json` sidecar (`{uprn: [lon, lat]}`), resolved
     from the OS Open-UPRN data by `fetch_corpus_coordinates.py`. Absent for a

scripts/build_expired_pairs_corpus.py

@@ -0,0 +1,140 @@
+"""Freeze a subsample of harness pairs into the committed integration fixture.
+
+Turns the pairs harness's live evidence into a deterministic, offline
+regression gate (the ADR-0030 corpus pattern): anonymised RAW API payloads
+loaded through ``EpcPropertyDataMapper``, so the gate keeps exercising the
+mapper and survives domain-dataclass changes. Layout, extending the
+``tests/fixtures/epc_prediction`` conventions:
+
+    ONE json file (a thousand per-cert files would drown a PR diff):
+      pairs:   [{postcode, uprn, actual, historic: {...}}]
+      cohorts: {postcode: {token: anonymised payload}}
+      actuals: {token: anonymised payload}   (the lodged SAP-10.2 certs)
+
+Reads the raw-JSON disk cache (scripts/epc_disk_cache.py) for everything the
+API served, and the historic S3 backup for the pre-2012 records. Pairs are
+subsampled deterministically (sorted, strided) from the harness telemetry.
+
+Usage:
+    python scripts/build_expired_pairs_corpus.py \
+        --telemetry pairs_telemetry.jsonl --cache-dir .epc_cache \
+        --out tests/fixtures/expired_prediction_pairs.json --sample 30
+"""
+
+from __future__ import annotations
+
+import argparse
+import dataclasses
+import json
+import sys
+from pathlib import Path
+from typing import Any, Optional
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from datatypes.epc.domain.historic_epc import HistoricEpc  # noqa: E402
+from domain.postcode import Postcode  # noqa: E402
+from harness.epc_prediction_corpus import anonymise_payload, stable_hash  # noqa: E402
+
+# Free-text fields blanked on the frozen historic record; the postcode is kept
+# (coarse open data, the shard key) and the address becomes a stable token so
+# nothing joins back to a household.
+_HISTORIC_PII_BLANK = ("address1", "address2", "address3", "posttown")
+
+
+def anonymise_historic(record: HistoricEpc) -> dict[str, str]:
+    row = dataclasses.asdict(record)
+    for field in _HISTORIC_PII_BLANK:
+        row[field] = ""
+    row["address"] = stable_hash("addr", record.address) if record.address else ""
+    row["lmk_key"] = stable_hash("lmk", record.lmk_key) if record.lmk_key else ""
+    return row
+
+
+def _read_cache(cache_dir: Path, key: str) -> Optional[Any]:
+    path = cache_dir / f"{key}.json"
+    return json.loads(path.read_text()) if path.exists() else None
+
+
+def subsample(rows: list[dict[str, Any]], count: int) -> list[dict[str, Any]]:
+    """A deterministic spread across the telemetry: sort, stride."""
+    ordered = sorted(rows, key=lambda r: (str(r["postcode"]), str(r["uprn"])))
+    if len(ordered) <= count:
+        return ordered
+    stride = len(ordered) // count
+    return ordered[::stride][:count]
+
+
+def main() -> None:  # pragma: no cover - IO composition
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--telemetry", type=Path, required=True)
+    parser.add_argument("--cache-dir", type=Path, required=True)
+    parser.add_argument("--out", type=Path, required=True)
+    parser.add_argument("--sample", type=int, default=30)
+    args = parser.parse_args()
+
+    from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver
+    from repositories.historic_epc.historic_epc_s3_repository import (
+        HistoricEpcS3Repository,
+    )
+
+    resolver = HistoricEpcResolver(HistoricEpcS3Repository.with_default_s3_client())
+
+    telemetry = [json.loads(line) for line in args.telemetry.read_text().splitlines()]
+    chosen = subsample(telemetry, args.sample)
+
+    cohorts: dict[str, dict[str, Any]] = {}
+    actuals: dict[str, Any] = {}
+    pairs: list[dict[str, Any]] = []
+    for row in chosen:
+        postcode, uprn = str(row["postcode"]), str(row["uprn"])
+        historic = resolver.record_for_uprn(uprn, postcode)
+        if historic is None:
+            print(f"{postcode} {uprn}: historic record gone — skipped", file=sys.stderr)
+            continue
+        search = _read_cache(args.cache_dir, f"search_uprn_{uprn}")
+        if not search:
+            print(f"{postcode} {uprn}: uprn search not cached — skipped", file=sys.stderr)
+            continue
+        latest = max(search, key=lambda r: str(r["registration_date"]))
+        actual_raw = _read_cache(args.cache_dir, f"cert_{latest['certificate_number']}")
+        cohort_search = _read_cache(args.cache_dir, f"search_pc_{postcode}")
+        if actual_raw is None or cohort_search is None:
+            print(f"{postcode} {uprn}: cert/cohort not cached — skipped", file=sys.stderr)
+            continue
+
+        if postcode not in cohorts:
+            payloads: dict[str, Any] = {}
+            for result in cohort_search:
+                raw = _read_cache(
+                    args.cache_dir, f"cert_{result['certificate_number']}"
+                )
+                if raw is None:
+                    continue
+                token = stable_hash("cert", str(result["certificate_number"]))
+                payloads[token] = anonymise_payload(raw)
+            cohorts[postcode] = payloads
+
+        actual_token = stable_hash("cert", str(latest["certificate_number"]))
+        actuals[actual_token] = anonymise_payload(actual_raw)
+        pairs.append(
+            {
+                "postcode": str(Postcode(postcode)),
+                "uprn": uprn,
+                "actual": actual_token,
+                "historic": anonymise_historic(historic),
+            }
+        )
+        print(f"{postcode} {uprn}: frozen", file=sys.stderr)
+
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(
+        json.dumps(
+            {"pairs": pairs, "cohorts": cohorts, "actuals": actuals}, indent=1
+        )
+    )
+    print(f"{len(pairs)} pairs frozen across {len(cohorts)} postcodes -> {args.out}")
+
+
+if __name__ == "__main__":
+    main()

tests/domain/epc_prediction/test_expired_pairs_gate.py

@@ -0,0 +1,165 @@
+"""Tier-1 ratcheting gate for Expired-Enhanced Prediction (ADR-0054).
+
+Replays the pairs harness OFFLINE over the committed, anonymised fixture
+(`tests/fixtures/expired_prediction_pairs` — pre-2012 historic records paired
+with their lodged SAP-10.2 certs and full postcode cohorts, frozen from the
+2,000-postcode national sweep). Both arms run the real production path minus
+the network: the raw payloads go through `EpcPropertyDataMapper`, conditioning
+through `conditioning_from_historic`, selection through `select_comparables`,
+synthesis through `EpcPrediction`. Deterministic, so every run reproduces the
+same numbers exactly — a failure is a real regression in the conditioning
+path, never sample noise.
+
+Floors are the measured values over the frozen fixture and only ever
+**tighten** (the repo's no-tolerance-widening ethos), exactly like the
+Component Accuracy gate this extends.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Optional
+
+import pytest
+
+from datatypes.epc.domain.epc_property_data import EpcPropertyData
+from datatypes.epc.domain.historic_epc import HistoricEpc
+from datatypes.epc.domain.mapper import EpcPropertyDataMapper
+from domain.epc_prediction.comparable_properties import (
+    ComparableProperty,
+    select_comparables,
+)
+from domain.epc_prediction.epc_prediction import EpcPrediction
+from domain.epc_prediction.historic_conditioning import (
+    attributes_with_historic_fallback,
+    conditioning_from_historic,
+    target_with_conditioning,
+)
+from domain.epc_prediction.prediction_comparison import (
+    PredictionComparison,
+    compare_prediction,
+)
+from domain.epc_prediction.prediction_target import build_prediction_target
+from domain.property.property import PropertyIdentity
+from harness.epc_prediction_corpus import comparable_from_payload
+
+_FIXTURE = (
+    Path(__file__).parents[3]
+    / "tests"
+    / "fixtures"
+    / "expired_prediction_pairs.json"
+)
+
+# Conditioned-arm classification floors (hit-rate over the frozen 30-pair /
+# 28-scoreable fixture) and residual ceilings — the measured values; tighten,
+# never loosen. The general (unconditioned) prediction floors live in
+# test_component_accuracy_gate.py; this gate guards the CONDITIONING path.
+_CLASSIFICATION_FLOORS: dict[str, float] = {
+    "construction_age_band": 0.5357,
+    "construction_age_band_pm1": 0.8214,
+    "cylinder_insulation_type": 0.8000,
+    "floor_construction": 0.8636,
+    "floor_insulation": 1.0000,
+    "has_hot_water_cylinder": 0.8214,
+    "has_pv": 0.8929,
+    "has_room_in_roof": 0.8571,
+    "heating_main_category": 1.0000,
+    "heating_main_control": 0.6071,
+    "heating_main_fuel": 1.0000,
+    "modal_glazing_type": 0.5000,
+    "roof_construction": 0.7143,
+    "roof_insulation_thickness": 0.3333,
+    "roof_insulation_thickness_pm1": 0.4815,
+    "secondary_heating_type": 0.1667,
+    "solar_water_heating": 1.0000,
+    "wall_construction": 0.8929,
+    "wall_insulation_type": 0.7500,
+    "water_heating_code": 1.0000,
+    "water_heating_fuel": 1.0000,
+}
+_FLOOR_AREA_MAE_CEILING: Optional[float] = 21.121
+
+
+def _pairs() -> list[tuple[HistoricEpc, EpcPropertyData, list[ComparableProperty]]]:
+    corpus = json.loads(_FIXTURE.read_text())
+    cohorts: dict[str, list[ComparableProperty]] = {
+        postcode: [
+            comparable
+            for token, payload in payloads.items()
+            if (comparable := comparable_from_payload(token, payload, {})) is not None
+        ]
+        for postcode, payloads in corpus["cohorts"].items()
+    }
+    return [
+        (
+            HistoricEpc(**pair["historic"]),
+            EpcPropertyDataMapper.from_api_response(corpus["actuals"][pair["actual"]]),
+            cohorts[pair["postcode"]],
+        )
+        for pair in corpus["pairs"]
+    ]
+
+
+def _conditioned_comparisons() -> list[PredictionComparison]:
+    predictor = EpcPrediction()
+    comparisons: list[PredictionComparison] = []
+    for historic, actual, cohort in _pairs():
+        conditioning = conditioning_from_historic(historic)
+        attributes = attributes_with_historic_fallback(None, conditioning)
+        identity = PropertyIdentity(
+            portfolio_id=0,
+            postcode=historic.postcode,
+            address=historic.address,
+            uprn=int(historic.uprn),
+        )
+        target = build_prediction_target(identity, None, attributes)
+        if target is None:
+            continue
+        target = target_with_conditioning(target, conditioning)
+        loo = [c for c in cohort if c.epc.uprn != int(historic.uprn)]
+        comparables = select_comparables(target, loo)
+        if not comparables.members:
+            continue
+        predicted = predictor.predict(target, comparables)
+        comparisons.append(compare_prediction(predicted, actual))
+    return comparisons
+
+
+@pytest.fixture(scope="module")
+def comparisons() -> list[PredictionComparison]:
+    if not _FIXTURE.exists():
+        pytest.skip("expired-pairs fixture not present")
+    return _conditioned_comparisons()
+
+
+def test_fixture_yields_the_expected_pair_count(
+    comparisons: list[PredictionComparison],
+) -> None:
+    # The frozen fixture must keep producing its full set of scoreable pairs —
+    # a drop means the fixture, the conditioning gate, or selection changed.
+    # (30 frozen; 2 are gated out / find no comparables, deterministically.)
+    assert len(comparisons) == 28
+
+
+@pytest.mark.parametrize("component,floor", sorted(_CLASSIFICATION_FLOORS.items()))
+def test_conditioned_classification_rate_does_not_regress(
+    comparisons: list[PredictionComparison], component: str, floor: float
+) -> None:
+    applicable = [
+        hit
+        for comparison in comparisons
+        if (hit := comparison.categorical_hits.get(component)) is not None
+    ]
+    assert applicable, component
+    rate = sum(applicable) / len(applicable)
+    assert rate >= floor - 1e-3, f"{component}: {rate:.4f} < floor {floor:.4f}"
+
+
+def test_conditioned_floor_area_mae_does_not_regress(
+    comparisons: list[PredictionComparison],
+) -> None:
+    if _FLOOR_AREA_MAE_CEILING is None:
+        pytest.skip("ceiling not yet pinned")
+    mae = sum(abs(c.floor_area_residual) for c in comparisons) / len(comparisons)
+    assert mae <= _FLOOR_AREA_MAE_CEILING + 1e-3