Raw-JSON EPC disk cache + roof_construction co-occurrence sweep 🟩

JsonCachingEpcClient caches at the raw layer (cert payloads verbatim,
search rows as JSON) so warm re-runs replay through retry+mapper+parsing
in minutes AND the cache doubles as integration-test fixture material —
the ADR-0030 corpus pattern (anonymised raw JSON through the mapper),
which pickled domain objects could never be. Wired into the pairs
harness (--cache-dir). roof_construction_code_sweep pins the RdSAP roof
code table the basement-code-6 way: single-building-part certs only,
code x description-prefix purity table. Also fixes the tfa telemetry key
rename that left the agreement row empty.

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

scripts/epc_disk_cache.py

@@ -0,0 +1,81 @@
+"""A raw-JSON disk cache around the EPC-API client, for harness scripts.
+
+Caches at the RAW layer — certificate payloads exactly as the API returned
+them, search results as plain JSON rows — so that:
+
+1. **Iteration is fast**: the pairs harness re-reads the same certs on every
+   run (pairs and cohorts don't change), so each re-run was re-paying 10-20k
+   sequential HTTP calls; a warm cache replays in minutes.
+2. **The cache is fixture material**: a committed integration-test corpus
+   should hold anonymised raw API JSON loaded through
+   ``EpcPropertyDataMapper.from_api_response`` (the ADR-0030 corpus pattern),
+   so replays keep exercising the mapper and survive domain-dataclass changes.
+   Pickles of internal dataclasses would do neither.
+
+Successful responses only — errors propagate uncached. Script-support code,
+not a repository: production ingestion must keep reading live (a re-lodgement
+must be seen), so this stays out of the DDD layers.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import json
+import re
+from pathlib import Path
+from typing import Any, Optional, cast
+
+from datatypes.epc.search.epc_search_result import EpcSearchResult
+from infrastructure.epc_client.epc_client_service import EpcClientService
+
+
+def _safe(key: str) -> str:
+    return re.sub(r"[^A-Za-z0-9_-]", "_", key)
+
+
+class JsonCachingEpcClient(EpcClientService):
+    """An ``EpcClientService`` whose raw HTTP layer is disk-cached as JSON.
+
+    Overrides the two private fetchers, so retry, mapping and search parsing
+    all still run on every call — a warm replay exercises the exact production
+    code path minus the network.
+    """
+
+    def __init__(self, auth_token: str, cache_dir: Path) -> None:
+        super().__init__(auth_token)
+        self._cache_dir = cache_dir
+        cache_dir.mkdir(parents=True, exist_ok=True)
+
+    def _path(self, key: str) -> Path:
+        return self._cache_dir / f"{_safe(key)}.json"
+
+    def _read(self, key: str) -> Optional[Any]:
+        path = self._path(key)
+        if not path.exists():
+            return None
+        return json.loads(path.read_text())
+
+    def _write(self, key: str, value: Any) -> None:
+        self._path(key).write_text(json.dumps(value))
+
+    def _fetch_certificate(self, cert_num: str) -> dict[str, Any]:
+        cached = self._read(f"cert_{cert_num}")
+        if cached is not None:
+            return cast(dict[str, Any], cached)
+        raw = super()._fetch_certificate(cert_num)
+        self._write(f"cert_{cert_num}", raw)
+        return raw
+
+    def _search(
+        self,
+        postcode: Optional[str] = None,
+        uprn: Optional[int] = None,
+    ) -> list[EpcSearchResult]:
+        key = f"search_pc_{postcode}" if postcode else f"search_uprn_{uprn}"
+        cached = self._read(key)
+        if cached is not None:
+            rows = cast(list[dict[str, Any]], cached)
+            return [EpcSearchResult(**row) for row in rows]
+        results = super()._search(postcode=postcode, uprn=uprn)
+        self._write(key, [dataclasses.asdict(r) for r in results])
+        return results

scripts/expired_prediction_pairs_harness.py

@@ -358,7 +358,9 @@ def format_diagnosis(rows: list[dict[str, object]]) -> str:
 
 
 def run(  # pragma: no cover - live IO composition
-    postcodes: list[str], telemetry_path: Optional[Path] = None
+    postcodes: list[str],
+    telemetry_path: Optional[Path] = None,
+    cache_dir: Optional[Path] = None,
 ) -> str:
     from domain.sap10_calculator.calculator import Sap10Calculator
     from infrastructure.epc_client.epc_client_service import EpcClientService
@@ -374,7 +376,14 @@ def run(  # 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"])
+    if cache_dir is not None:
+        from scripts.epc_disk_cache import JsonCachingEpcClient
+
+        epc_client: EpcClientService = JsonCachingEpcClient(
+            os.environ["OPEN_EPC_API_TOKEN"], cache_dir
+        )
+    else:
+        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()
@@ -511,7 +520,7 @@ def run(  # pragma: no cover - live IO composition
                             if conditioning.main_fuel is None
                             else conditioning.main_fuel == _actual_fuel(actual)
                         ),
-                        "agrees_tfa_within_5pct": _tfa_within_band(
+                        "agrees_tfa_within_band": _tfa_within_band(
                             actual.total_floor_area_m2,
                             conditioning.total_floor_area_m2,
                         ),
@@ -543,6 +552,12 @@ def main() -> None:  # pragma: no cover - CLI entry
     parser.add_argument(
         "--telemetry", type=Path, default=None, help="write per-pair JSONL here"
     )
+    parser.add_argument(
+        "--cache-dir",
+        type=Path,
+        default=None,
+        help="raw-JSON disk cache for API responses (fast re-runs; fixture material)",
+    )
     args = parser.parse_args()
     postcodes: list[str] = list(args.postcodes)
     if args.postcodes_file is not None:
@@ -553,7 +568,7 @@ def main() -> None:  # pragma: no cover - CLI entry
         ]
     if not postcodes:
         parser.error("no postcodes given")
-    report = run(postcodes, telemetry_path=args.telemetry)
+    report = run(postcodes, telemetry_path=args.telemetry, cache_dir=args.cache_dir)
     if args.out is not None:
         args.out.write_text(report + "\n")
     print(report)

scripts/roof_construction_code_sweep.py

@@ -0,0 +1,114 @@
+"""Pin the RdSAP ``roof_construction`` code table empirically (ADR-0054 gap).
+
+The gov-EPC API lodges ``sap_building_parts[].roof_construction`` as an int
+whose enumeration is published nowhere we hold; the historic dump lodges roofs
+as display text. The wall equivalent was pinned the same way this script works
+(the basement wall code 6: a bulk co-occurrence sweep). Method:
+
+- fetch every cert in the given postcodes' cohorts;
+- keep only certs with EXACTLY one building part and one roofs element, so the
+  (code, description) pairing is unambiguous;
+- cross-tabulate code x description-prefix (text before the first comma) and
+  report each code's dominant prefix with its purity.
+
+Usage:
+    python scripts/roof_construction_code_sweep.py --postcodes-file pcs.txt \
+        --cache-dir .epc_cache --out roof_codes.md
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+from collections import Counter, defaultdict
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from datatypes.epc.domain.epc_property_data import EpcPropertyData  # noqa: E402
+
+
+def prefix_of(description: str) -> str:
+    """The roof description's construction/form half — the text before the
+    first comma ("Pitched, 100 mm loft insulation" -> "Pitched")."""
+    return description.split(",", 1)[0].strip()
+
+
+def tabulate(certs: list[EpcPropertyData]) -> dict[int, Counter[str]]:
+    """code -> Counter(description prefix), over unambiguous certs only."""
+    table: dict[int, Counter[str]] = defaultdict(Counter)
+    for cert in certs:
+        if len(cert.sap_building_parts) != 1 or len(cert.roofs) != 1:
+            continue
+        code = cert.sap_building_parts[0].roof_construction
+        if code is None:
+            continue
+        table[code][prefix_of(cert.roofs[0].description)] += 1
+    return dict(table)
+
+
+def format_table(table: dict[int, Counter[str]]) -> str:
+    lines = [
+        "# roof_construction code x description-prefix co-occurrence",
+        "",
+        "| code | n | dominant prefix | purity | runners-up |",
+        "|---|---|---|---|---|",
+    ]
+    for code in sorted(table):
+        counts = table[code]
+        total = sum(counts.values())
+        (top, top_n), *rest = counts.most_common()
+        runners = ", ".join(f"{p} ({n})" for p, n in rest[:3]) or "—"
+        lines.append(
+            f"| {code} | {total} | {top} | {top_n / total:.0%} | {runners} |"
+        )
+    return "\n".join(lines)
+
+
+def main() -> None:  # pragma: no cover - CLI/IO composition
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--postcodes-file", type=Path, required=True)
+    parser.add_argument("--cache-dir", type=Path, required=True)
+    parser.add_argument("--out", type=Path, default=None)
+    args = parser.parse_args()
+
+    from scripts.e2e_common import load_env
+    from scripts.epc_disk_cache import JsonCachingEpcClient
+
+    load_env()
+    client = JsonCachingEpcClient(os.environ["OPEN_EPC_API_TOKEN"], args.cache_dir)
+
+    certs: list[EpcPropertyData] = []
+    postcodes = [
+        line.strip()
+        for line in args.postcodes_file.read_text().splitlines()
+        if line.strip()
+    ]
+    for index, postcode in enumerate(postcodes):
+        try:
+            results = client.search_by_postcode(postcode)
+        except Exception as error:
+            print(f"{postcode}: search failed: {error}", file=sys.stderr)
+            continue
+        for result in results:
+            try:
+                certs.append(
+                    client.get_by_certificate_number(result.certificate_number)
+                )
+            except Exception:
+                continue  # unmappable cert — not usable evidence
+        print(
+            f"[{index + 1}/{len(postcodes)}] {postcode}: {len(certs)} certs so far",
+            file=sys.stderr,
+            flush=True,
+        )
+
+    report = format_table(tabulate(certs))
+    if args.out is not None:
+        args.out.write_text(report + "\n")
+    print(report)
+
+
+if __name__ == "__main__":
+    main()