Merge pull request #1356 from Hestia-Homes/feature/historic-epc-repository

Historic EPC repository: DDD port + resolver for address→UPRN
This commit is contained in:
Jun-te Kim 2026-07-07 10:56:33 +01:00 committed by GitHub
commit bc1cca77db
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
40 changed files with 367762 additions and 175 deletions

View file

@ -27,7 +27,7 @@ pytest-postgresql
moto[s3,sqs]==5.0.28 # mock_aws (moto 5.x) for S3/SQS in orchestration tests moto[s3,sqs]==5.0.28 # mock_aws (moto 5.x) for S3/SQS in orchestration tests
# Formatting # Formatting
black==26.1.0 black==26.1.0
boto3-stubs boto3-stubs[s3] # typed boto3.client("s3") for the S3 repositories
openai openai
# Type checking — strict pyright gate (CLAUDE.md). The pip `pyright` wrapper uses # Type checking — strict pyright gate (CLAUDE.md). The pip `pyright` wrapper uses
# the container's Node. pandas-stubs lets pandas-typed modules check cleanly # the container's Node. pandas-stubs lets pandas-typed modules check cleanly

View file

@ -34,4 +34,7 @@ All new code must pass `pyright` with zero errors under `typeCheckingMode = stri
Use Optional over | None Use Optional over | None
Annotate all function return types. Use `dict[str, Any]` for untyped external API Annotate all function return types. Use `dict[str, Any]` for untyped external API
payloads — never bare `dict`. Add `pandas-stubs` when introducing pandas to a module. payloads — never bare `dict`. Add `pandas-stubs` when introducing pandas to a module.
Annotate locals assigned from cross-module calls (e.g. `matches: list[ScoredHistoricEpc]
= rank_historic_epc(...)`) — the reader shouldn't need the callee's signature to follow
the flow; inference-only locals are fine within a module's own helpers.

View file

@ -81,6 +81,14 @@ _Avoid_: neighbours, similar properties, peer set
Producing a Property's `EpcPropertyData` picture from its **Comparable Properties** when it has no EPC (~30% of UK homes, typically long-tenure). **Deterministic** neighbour synthesis (k-NN-style — *not* ML; no trained model): take the cohort **mode** for the homogeneous categoricals (wall / roof / floor construction + insulation, construction age band), copy a single representative comparable's **structure** wholesale (building parts, per-window dimensions + orientations, floor dimensions) so the picture stays internally consistent for the calculator, then apply **Landlord Overrides** and the known inputs on top. The result is scored through **SAP10 Calculation** like any other **Effective EPC**, so a predicted Property flows through Rebaselining, Bill Derivation, and Modelling unchanged — held in a **distinct predicted-EPC slot** that coexists with any lodged EPC (so provenance is structural and the UI can flag it; see ADR-0031). A **known property type is required** — the hard cohort filter (a flat is never sized from houses) — supplied by a **Landlord Override** (or, later, an Ordnance Survey lookup); a Property whose property type is genuinely unknown is **gated out**, never predicted from a mixed-type cohort and never given a national default. The same cohort machinery also produces **EPC Anomaly Flags** for Properties that *do* have an EPC. A future learned-weighting refinement is possible but separate, as with the calculator's ML residual head. Producing a Property's `EpcPropertyData` picture from its **Comparable Properties** when it has no EPC (~30% of UK homes, typically long-tenure). **Deterministic** neighbour synthesis (k-NN-style — *not* ML; no trained model): take the cohort **mode** for the homogeneous categoricals (wall / roof / floor construction + insulation, construction age band), copy a single representative comparable's **structure** wholesale (building parts, per-window dimensions + orientations, floor dimensions) so the picture stays internally consistent for the calculator, then apply **Landlord Overrides** and the known inputs on top. The result is scored through **SAP10 Calculation** like any other **Effective EPC**, so a predicted Property flows through Rebaselining, Bill Derivation, and Modelling unchanged — held in a **distinct predicted-EPC slot** that coexists with any lodged EPC (so provenance is structural and the UI can flag it; see ADR-0031). A **known property type is required** — the hard cohort filter (a flat is never sized from houses) — supplied by a **Landlord Override** (or, later, an Ordnance Survey lookup); a Property whose property type is genuinely unknown is **gated out**, never predicted from a mixed-type cohort and never given a national default. The same cohort machinery also produces **EPC Anomaly Flags** for Properties that *do* have an EPC. A future learned-weighting refinement is possible but separate, as with the calculator's ML residual head.
_Avoid_: EpcPredictionService (no "service" suffix — name the operation), ML prediction (it is deterministic), EPC estimation _Avoid_: EpcPredictionService (no "service" suffix — name the operation), ML prediction (it is deterministic), EPC estimation
**Historic EPC**:
One certificate row from the final data dump of the shut-down old EPC register, held at `s3://retrofit-data-dev/historical_epc/{POSTCODE}/data.csv.gz` and read through the `HistoricEpcRepository` port. Partial, tabular, display-text data (the old API never exposed full SAP inputs), covering certificates the new gov API (registered ≥ 1 Jan 2012) cannot see. Consumed by `address2UPRN` (fuzzy address→UPRN resolution) and by **Expired-Enhanced Prediction** (exact-UPRN lookup only).
_Avoid_: old EPC (ambiguous with a pre-SAP10 cert from the new API), historical EPC API (the API is gone; only the backup exists)
**Expired-Enhanced Prediction**:
**EPC Prediction** for a Property whose only certificate predates 2012: the expired **Historic EPC**, found by exact UPRN in its postcode shard, **conditions** cohort selection with its *stable* attributes (property type, built form, wall material, roof construction, the age band's ±1-band neighbourhood, main fuel, and a ±20% floor-area band) exactly as a Landlord Override would. Band widths are harness-evidenced, not guessed (ADR-0054 amendment: age band agrees with a relodged cert 52% exactly / 90% within one band; TFA 45% within ±5% / 82% within ±20%). Volatile attributes (heating, hot water, glazing, PV, insulation, lighting) are excluded — 14+ years stale — and stay neighbour-predicted; the historic cert is **never copied into the Effective EPC as current state**. Persisted to the predicted slot with `source="expired"`. Scoped to the historic backup only; post-2012 expired certs from the new API keep their existing treatment (ADR-0054).
_Avoid_: historic override (it is conditioning, not an override the effective picture trusts), expired EPC path (names the input, not the operation)
### Survey documents ### Survey documents
**Ventilation Audit**: **Ventilation Audit**:

View file

@ -20,6 +20,7 @@ from orchestration.ara_first_run_pipeline import AraFirstRunPipeline
from orchestration.ingestion_orchestrator import ( from orchestration.ingestion_orchestrator import (
ComparablesRepo, ComparablesRepo,
EpcFetcher, EpcFetcher,
HistoricEpcReader,
IngestionOrchestrator, IngestionOrchestrator,
PredictionAttributesReader, PredictionAttributesReader,
SolarFetcher, SolarFetcher,
@ -70,6 +71,7 @@ def build_first_run_pipeline(
solar_fetcher: SolarFetcher, solar_fetcher: SolarFetcher,
comparables_repo: Optional[ComparablesRepo] = None, comparables_repo: Optional[ComparablesRepo] = None,
prediction_attributes_reader: Optional[PredictionAttributesReader] = None, prediction_attributes_reader: Optional[PredictionAttributesReader] = None,
historic_epc_reader: Optional[HistoricEpcReader] = None,
) -> AraFirstRunPipeline: ) -> AraFirstRunPipeline:
"""Compose the real three-stage pipeline on a Unit-of-Work factory. """Compose the real three-stage pipeline on a Unit-of-Work factory.
@ -95,6 +97,9 @@ def build_first_run_pipeline(
comparables_repo=comparables_repo, comparables_repo=comparables_repo,
prediction_attributes_reader=prediction_attributes_reader, prediction_attributes_reader=prediction_attributes_reader,
epc_prediction=EpcPrediction(), epc_prediction=EpcPrediction(),
# Expired-Enhanced Prediction (ADR-0054): off until a resolver over
# the historic S3 backup is supplied, like the two readers above.
historic_epc_reader=historic_epc_reader,
), ),
baseline=PropertyBaselineOrchestrator( baseline=PropertyBaselineOrchestrator(
unit_of_work=unit_of_work, unit_of_work=unit_of_work,

View file

@ -39,6 +39,10 @@ COPY infrastructure/ infrastructure/
# EpcClientService -> datatypes.epc.domain.mapper -> domain.sap10_calculator; # EpcClientService -> datatypes.epc.domain.mapper -> domain.sap10_calculator;
# without this the lambda fails at init with "No module named 'domain'". # without this the lambda fails at init with "No module named 'domain'".
COPY domain/ domain/ COPY domain/ domain/
# main.py resolves historic-EPC UPRNs via repositories.historic_epc.* (the
# HistoricEpcResolver + S3 repository); without this the lambda fails at init
# with "No module named 'repositories'".
COPY repositories/ repositories/
# Copy the handler # Copy the handler
COPY backend/address2UPRN/main.py . COPY backend/address2UPRN/main.py .

View file

@ -16,11 +16,11 @@ from datetime import datetime
from backend.utils.addressMatch import AddressMatch from backend.utils.addressMatch import AddressMatch
from backend.address2UPRN.scoring import all_uprns_match, rank_address_similarity from backend.address2UPRN.scoring import all_uprns_match, rank_address_similarity
from datatypes.epc.domain.historic_epc_matching import (
match_addresses_for_postcode,
)
from infrastructure.epc_client.epc_client_service import EpcClientService from infrastructure.epc_client.epc_client_service import EpcClientService
from datatypes.epc.domain.historic_epc_matching import ScoredHistoricEpc from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver
from repositories.historic_epc.historic_epc_s3_repository import (
HistoricEpcS3Repository,
)
logger = setup_logger() logger = setup_logger()
@ -45,26 +45,14 @@ def get_uprn_from_historic_epc(
"""Resolve a UPRN via historic EPC S3 data. """Resolve a UPRN via historic EPC S3 data.
Returns (uprn, address, lexiscore) when the historic dataset agrees on a Returns (uprn, address, lexiscore) when the historic dataset agrees on a
single rank-1 UPRN, None otherwise (missing postcode file, zero score, single rank-1 UPRN, None otherwise (no stored data, zero score, or
or ambiguous top rank). The score gate is `unambiguous_uprn`'s own ambiguous top rank). The score gate is `unambiguous_uprn`'s own (score > 0);
(score > 0); the 0.7 heuristic used for the new-EPC source isn't applied the 0.7 heuristic used for the new-EPC source isn't applied here because
here because historic addresses use a more verbose format that historic addresses use a more verbose format that systematically depresses
systematically depresses lexiscores. lexiscores.
""" """
repo = HistoricEpcS3Repository.with_default_s3_client()
try: return HistoricEpcResolver(repo).resolve_uprn(user_inputed_address, postcode)
result = match_addresses_for_postcode(user_inputed_address, postcode)
except FileNotFoundError:
return None
uprn: Optional[str] = result.unambiguous_uprn()
if not uprn or uprn == "nan":
return None
top: Optional[ScoredHistoricEpc] = result.top()
if top is None:
return None
return uprn, top.record.address, top.lexiscore
def get_uprn_with_epc_df( def get_uprn_with_epc_df(

View file

@ -1,27 +1,127 @@
from collections.abc import Hashable, Mapping
from dataclasses import dataclass from dataclasses import dataclass
from typing import Optional from typing import Any, Optional
import pandas as pd import pandas as pd
from botocore.exceptions import ClientError
from backend.address2UPRN.scoring import rank_address_similarity from backend.address2UPRN.scoring import rank_address_similarity
from backend.utils.addressMatch import AddressMatch
from datatypes.epc.domain.historic_epc import HistoricEpc from datatypes.epc.domain.historic_epc import HistoricEpc
from utils.pandas_utils import pandas_cell_to_str from utils.pandas_utils import pandas_cell_to_str
from utils.s3 import parse_s3_uri, read_csv_gz_from_s3
DEFAULT_S3_ROOT = "s3://retrofit-data-dev/historical_epc" DEFAULT_S3_ROOT = "s3://retrofit-data-dev/historical_epc"
_EXTRA_COLS = {"lexiscore", "lexirank"}
def map_historic_epc_row_to_domain(row: Mapping[Hashable, Any]) -> HistoricEpc:
"""Map one historic-EPC shard row (upper-cased CSV columns) to the domain.
def _map_historic_epc_pandas_row_to_domain(row: pd.Series) -> HistoricEpc: Field-by-field so pyright checks every constructor argument: a missing or
kwargs = { renamed CSV column fails loudly here (KeyError) rather than surfacing as a
col.lower(): pandas_cell_to_str(val) half-built record, and columns the domain type doesn't know are ignored.
for col, val in row.items() Takes a plain mapping (a ``DataFrame.to_dict("records")`` row), not a
if col.lower() not in _EXTRA_COLS pandas Series, so the signature carries no pandas types.
} """
return HistoricEpc(**kwargs)
def cell(column: str) -> str:
return pandas_cell_to_str(row[column])
# pandas reads an all-integer UPRN column as float, so the cell stringifies
# to "151020766.0"; the domain UPRN is the bare integer string.
uprn = cell("UPRN")
return HistoricEpc(
lmk_key=cell("LMK_KEY"),
address1=cell("ADDRESS1"),
address2=cell("ADDRESS2"),
address3=cell("ADDRESS3"),
postcode=cell("POSTCODE"),
building_reference_number=cell("BUILDING_REFERENCE_NUMBER"),
current_energy_rating=cell("CURRENT_ENERGY_RATING"),
potential_energy_rating=cell("POTENTIAL_ENERGY_RATING"),
current_energy_efficiency=cell("CURRENT_ENERGY_EFFICIENCY"),
potential_energy_efficiency=cell("POTENTIAL_ENERGY_EFFICIENCY"),
property_type=cell("PROPERTY_TYPE"),
built_form=cell("BUILT_FORM"),
inspection_date=cell("INSPECTION_DATE"),
local_authority=cell("LOCAL_AUTHORITY"),
constituency=cell("CONSTITUENCY"),
county=cell("COUNTY"),
lodgement_date=cell("LODGEMENT_DATE"),
transaction_type=cell("TRANSACTION_TYPE"),
environment_impact_current=cell("ENVIRONMENT_IMPACT_CURRENT"),
environment_impact_potential=cell("ENVIRONMENT_IMPACT_POTENTIAL"),
energy_consumption_current=cell("ENERGY_CONSUMPTION_CURRENT"),
energy_consumption_potential=cell("ENERGY_CONSUMPTION_POTENTIAL"),
co2_emissions_current=cell("CO2_EMISSIONS_CURRENT"),
co2_emiss_curr_per_floor_area=cell("CO2_EMISS_CURR_PER_FLOOR_AREA"),
co2_emissions_potential=cell("CO2_EMISSIONS_POTENTIAL"),
lighting_cost_current=cell("LIGHTING_COST_CURRENT"),
lighting_cost_potential=cell("LIGHTING_COST_POTENTIAL"),
heating_cost_current=cell("HEATING_COST_CURRENT"),
heating_cost_potential=cell("HEATING_COST_POTENTIAL"),
hot_water_cost_current=cell("HOT_WATER_COST_CURRENT"),
hot_water_cost_potential=cell("HOT_WATER_COST_POTENTIAL"),
total_floor_area=cell("TOTAL_FLOOR_AREA"),
energy_tariff=cell("ENERGY_TARIFF"),
mains_gas_flag=cell("MAINS_GAS_FLAG"),
floor_level=cell("FLOOR_LEVEL"),
flat_top_storey=cell("FLAT_TOP_STOREY"),
flat_storey_count=cell("FLAT_STOREY_COUNT"),
main_heating_controls=cell("MAIN_HEATING_CONTROLS"),
multi_glaze_proportion=cell("MULTI_GLAZE_PROPORTION"),
glazed_type=cell("GLAZED_TYPE"),
glazed_area=cell("GLAZED_AREA"),
extension_count=cell("EXTENSION_COUNT"),
number_habitable_rooms=cell("NUMBER_HABITABLE_ROOMS"),
number_heated_rooms=cell("NUMBER_HEATED_ROOMS"),
low_energy_lighting=cell("LOW_ENERGY_LIGHTING"),
number_open_fireplaces=cell("NUMBER_OPEN_FIREPLACES"),
hotwater_description=cell("HOTWATER_DESCRIPTION"),
hot_water_energy_eff=cell("HOT_WATER_ENERGY_EFF"),
hot_water_env_eff=cell("HOT_WATER_ENV_EFF"),
floor_description=cell("FLOOR_DESCRIPTION"),
floor_energy_eff=cell("FLOOR_ENERGY_EFF"),
floor_env_eff=cell("FLOOR_ENV_EFF"),
windows_description=cell("WINDOWS_DESCRIPTION"),
windows_energy_eff=cell("WINDOWS_ENERGY_EFF"),
windows_env_eff=cell("WINDOWS_ENV_EFF"),
walls_description=cell("WALLS_DESCRIPTION"),
walls_energy_eff=cell("WALLS_ENERGY_EFF"),
walls_env_eff=cell("WALLS_ENV_EFF"),
secondheat_description=cell("SECONDHEAT_DESCRIPTION"),
sheating_energy_eff=cell("SHEATING_ENERGY_EFF"),
sheating_env_eff=cell("SHEATING_ENV_EFF"),
roof_description=cell("ROOF_DESCRIPTION"),
roof_energy_eff=cell("ROOF_ENERGY_EFF"),
roof_env_eff=cell("ROOF_ENV_EFF"),
mainheat_description=cell("MAINHEAT_DESCRIPTION"),
mainheat_energy_eff=cell("MAINHEAT_ENERGY_EFF"),
mainheat_env_eff=cell("MAINHEAT_ENV_EFF"),
mainheatcont_description=cell("MAINHEATCONT_DESCRIPTION"),
mainheatc_energy_eff=cell("MAINHEATC_ENERGY_EFF"),
mainheatc_env_eff=cell("MAINHEATC_ENV_EFF"),
lighting_description=cell("LIGHTING_DESCRIPTION"),
lighting_energy_eff=cell("LIGHTING_ENERGY_EFF"),
lighting_env_eff=cell("LIGHTING_ENV_EFF"),
main_fuel=cell("MAIN_FUEL"),
wind_turbine_count=cell("WIND_TURBINE_COUNT"),
heat_loss_corridor=cell("HEAT_LOSS_CORRIDOR"),
unheated_corridor_length=cell("UNHEATED_CORRIDOR_LENGTH"),
floor_height=cell("FLOOR_HEIGHT"),
photo_supply=cell("PHOTO_SUPPLY"),
solar_water_heating_flag=cell("SOLAR_WATER_HEATING_FLAG"),
mechanical_ventilation=cell("MECHANICAL_VENTILATION"),
address=cell("ADDRESS"),
local_authority_label=cell("LOCAL_AUTHORITY_LABEL"),
constituency_label=cell("CONSTITUENCY_LABEL"),
posttown=cell("POSTTOWN"),
construction_age_band=cell("CONSTRUCTION_AGE_BAND"),
lodgement_datetime=cell("LODGEMENT_DATETIME"),
tenure=cell("TENURE"),
fixed_lighting_outlets_count=cell("FIXED_LIGHTING_OUTLETS_COUNT"),
low_energy_fixed_light_count=cell("LOW_ENERGY_FIXED_LIGHT_COUNT"),
uprn=uprn[:-2] if uprn.endswith(".0") else uprn,
uprn_source=cell("UPRN_SOURCE"),
report_type=cell("REPORT_TYPE"),
)
@dataclass(frozen=True) @dataclass(frozen=True)
@ -52,13 +152,48 @@ class HistoricEpcMatches:
return next(iter(uprns)) if len(uprns) == 1 else None return next(iter(uprns)) if len(uprns) == 1 else None
def _sanitise_postcode(postcode: str) -> str: def rank_historic_epc(
cleaned = (postcode or "").upper().replace(" ", "") records: list[HistoricEpc],
if not cleaned: user_address: str,
raise ValueError("postcode must contain non-whitespace characters") *,
if not AddressMatch.is_valid_postcode(cleaned): address_column: str = "ADDRESS",
raise ValueError(f"postcode {cleaned!r} is not a valid UK postcode") uprn_column: str = "UPRN",
return cleaned ) -> list[ScoredHistoricEpc]:
"""Score ``records`` against ``user_address`` (best first), keeping every
record including hard-zero non-matches. The pure scoring half of the
historic-EPC lookup: no I/O, so it is unit-testable without S3."""
if not user_address:
raise ValueError("user_address must be non-empty")
if not records:
return []
df = pd.DataFrame(
{
"_pos": range(len(records)),
address_column: [r.address for r in records],
uprn_column: [r.uprn for r in records],
}
)
scored = rank_address_similarity(
df,
user_address=user_address,
address_column=address_column,
uprn_column=uprn_column,
)
# pandas-stubs' to_dict overloads carry bare generics of their own, so
# strict mode flags the member access; the rows annotation keeps the
# comprehension below fully typed.
scored_rows: list[dict[Hashable, Any]] = scored.to_dict( # pyright: ignore[reportUnknownMemberType]
orient="records"
)
return [
ScoredHistoricEpc(
record=records[int(scored_row["_pos"])],
lexiscore=float(scored_row["lexiscore"]),
lexirank=int(scored_row["lexirank"]),
)
for scored_row in scored_rows
]
def match_addresses_for_postcode( def match_addresses_for_postcode(
@ -66,39 +201,19 @@ def match_addresses_for_postcode(
postcode: str, postcode: str,
*, *,
s3_root: str = DEFAULT_S3_ROOT, s3_root: str = DEFAULT_S3_ROOT,
address_column: str = "ADDRESS",
uprn_column: str = "UPRN",
) -> HistoricEpcMatches: ) -> HistoricEpcMatches:
if not user_address: """Score a postcode's historic EPCs against ``user_address``.
raise ValueError("user_address must be non-empty")
pc = _sanitise_postcode(postcode) A thin composition seam over the historic-EPC repository + resolver; the S3
bucket, root_prefix = parse_s3_uri(s3_root) read and scoring live there. A postcode with no stored shard yields empty
key = f"{root_prefix.rstrip('/')}/{pc}/data.csv.gz" matches (not FileNotFoundError); an unusable postcode raises PostcodeNotFound.
Imported lazily so the domain layer doesn't import the repositories layer at
try: module load (the repository depends on this module).
df = read_csv_gz_from_s3(bucket, key) """
except ClientError as e: from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver
if e.response.get("Error", {}).get("Code") in ("NoSuchKey", "404"): from repositories.historic_epc.historic_epc_s3_repository import (
raise FileNotFoundError( HistoricEpcS3Repository,
f"No historic EPC data at s3://{bucket}/{key}"
) from e
raise
scored = rank_address_similarity(
df,
user_address=user_address,
address_column=address_column,
uprn_column=uprn_column,
) )
matches = [ repo = HistoricEpcS3Repository.with_default_s3_client(s3_root)
ScoredHistoricEpc( return HistoricEpcResolver(repo).match(user_address, postcode)
record=_map_historic_epc_pandas_row_to_domain(row),
lexiscore=float(row["lexiscore"]),
lexirank=int(row["lexirank"]),
)
for _, row in scored.iterrows()
]
return HistoricEpcMatches(user_address=user_address, postcode=pc, matches=matches)

View file

@ -6,13 +6,12 @@ import pandas as pd
import pytest import pytest
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
from datatypes.epc.domain import historic_epc_matching as matcher_mod
from datatypes.epc.domain.historic_epc_matching import ( from datatypes.epc.domain.historic_epc_matching import (
HistoricEpcMatches, HistoricEpcMatches,
ScoredHistoricEpc, ScoredHistoricEpc,
_sanitise_postcode,
match_addresses_for_postcode, match_addresses_for_postcode,
) )
from infrastructure.s3.gzip_csv_s3_client import GzipCsvS3Client
# Columns required by the HistoricEpc dataclass (lower-cased CSV columns). # Columns required by the HistoricEpc dataclass (lower-cased CSV columns).
# The matcher only reads ADDRESS + UPRN to score; everything else is filled # The matcher only reads ADDRESS + UPRN to score; everything else is filled
@ -125,52 +124,23 @@ def _build_df(rows: list[dict]) -> pd.DataFrame:
return pd.DataFrame(rows, columns=_FULL_COLUMN_FIELDS) return pd.DataFrame(rows, columns=_FULL_COLUMN_FIELDS)
@pytest.fixture
def patch_postcode_valid():
with patch.object(
matcher_mod.AddressMatch, "is_valid_postcode", return_value=True
) as m:
yield m
@pytest.fixture @pytest.fixture
def patch_read(): def patch_read():
with patch.object(matcher_mod, "read_csv_gz_from_s3") as m: # match_addresses_for_postcode now reads through GzipCsvS3Client
# (infrastructure/s3) — not the old utils.s3 free function. Patch the
# client's read (it is called with the per-postcode key only; the bucket
# lives in the client) and stub boto3.client so the seam runs with no S3 and
# no AWS environment.
with patch("boto3.client"), patch.object(GzipCsvS3Client, "read_csv_gz") as m:
yield m yield m
# ---------- _sanitise_postcode ----------
class TestSanitisePostcode:
def test_uppercases_and_strips_spaces(self, patch_postcode_valid):
assert _sanitise_postcode("ab33 8al") == "AB338AL"
def test_empty_raises(self, patch_postcode_valid):
with pytest.raises(ValueError, match="non-whitespace"):
_sanitise_postcode("")
def test_whitespace_only_raises(self, patch_postcode_valid):
with pytest.raises(ValueError, match="non-whitespace"):
_sanitise_postcode(" ")
def test_invalid_postcode_raises(self):
with patch.object(
matcher_mod.AddressMatch, "is_valid_postcode", return_value=False
):
with pytest.raises(ValueError, match="not a valid UK postcode"):
_sanitise_postcode("NONSENSE")
# ---------- match_addresses_for_postcode ---------- # ---------- match_addresses_for_postcode ----------
class TestMatchAddressesForPostcode: class TestMatchAddressesForPostcode:
def test_preserves_row_count_including_zero_score_rows( def test_preserves_row_count_including_zero_score_rows(self, patch_read):
self, patch_read, patch_postcode_valid
):
# Disjoint number sets => hard zero. Still kept in matches. # Disjoint number sets => hard zero. Still kept in matches.
patch_read.return_value = _build_df( patch_read.return_value = _build_df(
[ [
@ -182,9 +152,7 @@ class TestMatchAddressesForPostcode:
assert isinstance(result, HistoricEpcMatches) assert isinstance(result, HistoricEpcMatches)
assert len(result.matches) == 2 assert len(result.matches) == 2
def test_top_has_lexirank_one_and_lexiscore_monotone( def test_top_has_lexirank_one_and_lexiscore_monotone(self, patch_read):
self, patch_read, patch_postcode_valid
):
patch_read.return_value = _build_df( patch_read.return_value = _build_df(
[ [
_row("48 GORDON ROAD", "200"), # near miss _row("48 GORDON ROAD", "200"), # near miss
@ -192,47 +160,46 @@ class TestMatchAddressesForPostcode:
] ]
) )
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL") result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
assert result.top().lexirank == 1 top = result.top()
assert top is not None
assert top.lexirank == 1
scores = [m.lexiscore for m in result.matches] scores = [m.lexiscore for m in result.matches]
assert scores == sorted(scores, reverse=True) assert scores == sorted(scores, reverse=True)
def test_s3_key_built_from_default_root(self, patch_read, patch_postcode_valid): def test_s3_key_built_from_default_root(self, patch_read):
# The default root's prefix threads into the per-postcode key; the
# bucket parsing is covered by HistoricEpcS3Repository's factory test.
patch_read.return_value = _build_df([_row("47 GORDON ROAD", "100")]) patch_read.return_value = _build_df([_row("47 GORDON ROAD", "100")])
match_addresses_for_postcode("47 Gordon Road", "AB33 8AL") match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
patch_read.assert_called_once_with( patch_read.assert_called_once_with("historical_epc/AB338AL/data.csv.gz")
"retrofit-data-dev", "historical_epc/AB338AL/data.csv.gz"
)
def test_s3_key_respects_custom_root_with_trailing_slash( def test_s3_key_respects_custom_root_with_trailing_slash(self, patch_read):
self, patch_read, patch_postcode_valid
):
patch_read.return_value = _build_df([_row("47 GORDON ROAD", "100")]) patch_read.return_value = _build_df([_row("47 GORDON ROAD", "100")])
match_addresses_for_postcode( match_addresses_for_postcode(
"47 Gordon Road", "47 Gordon Road",
"AB33 8AL", "AB33 8AL",
s3_root="s3://my-bucket/some/prefix/", s3_root="s3://my-bucket/some/prefix/",
) )
patch_read.assert_called_once_with( patch_read.assert_called_once_with("some/prefix/AB338AL/data.csv.gz")
"my-bucket", "some/prefix/AB338AL/data.csv.gz"
)
def test_no_such_key_translates_to_filenotfound( def test_missing_postcode_object_yields_empty_matches(self, patch_read):
self, patch_read, patch_postcode_valid # A valid postcode with no stored shard is a normal miss, not an error:
): # empty matches, not FileNotFoundError.
patch_read.side_effect = ClientError( patch_read.side_effect = ClientError(
{"Error": {"Code": "NoSuchKey", "Message": "missing"}}, "GetObject" {"Error": {"Code": "NoSuchKey", "Message": "missing"}}, "GetObject"
) )
with pytest.raises(FileNotFoundError): result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
match_addresses_for_postcode("47 Gordon Road", "AB33 8AL") assert result.matches == []
assert result.unambiguous_uprn() is None
def test_other_client_error_propagates(self, patch_read, patch_postcode_valid): def test_other_client_error_propagates(self, patch_read):
patch_read.side_effect = ClientError( patch_read.side_effect = ClientError(
{"Error": {"Code": "AccessDenied", "Message": "nope"}}, "GetObject" {"Error": {"Code": "AccessDenied", "Message": "nope"}}, "GetObject"
) )
with pytest.raises(ClientError): with pytest.raises(ClientError):
match_addresses_for_postcode("47 Gordon Road", "AB33 8AL") match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
def test_empty_user_address_raises(self, patch_postcode_valid): def test_empty_user_address_raises(self):
with pytest.raises(ValueError, match="user_address"): with pytest.raises(ValueError, match="user_address"):
match_addresses_for_postcode("", "AB33 8AL") match_addresses_for_postcode("", "AB33 8AL")
@ -242,7 +209,7 @@ class TestMatchAddressesForPostcode:
class TestUnambiguousUprn: class TestUnambiguousUprn:
def test_exact_match_returns_uprn(self, patch_read, patch_postcode_valid): def test_exact_match_returns_uprn(self, patch_read):
patch_read.return_value = _build_df( patch_read.return_value = _build_df(
[ [
_row("47 GORDON ROAD", "100"), _row("47 GORDON ROAD", "100"),
@ -252,7 +219,7 @@ class TestUnambiguousUprn:
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL") result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
assert result.unambiguous_uprn() == "100" assert result.unambiguous_uprn() == "100"
def test_ambiguous_tie_returns_none(self, patch_read, patch_postcode_valid): def test_ambiguous_tie_returns_none(self, patch_read):
# Two duplicate addresses with different UPRNs share rank-1. # Two duplicate addresses with different UPRNs share rank-1.
patch_read.return_value = _build_df( patch_read.return_value = _build_df(
[ [
@ -263,9 +230,7 @@ class TestUnambiguousUprn:
result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL") result = match_addresses_for_postcode("47 Gordon Road", "AB33 8AL")
assert result.unambiguous_uprn() is None assert result.unambiguous_uprn() is None
def test_all_zero_score_returns_none_even_when_uprn_unique( def test_all_zero_score_returns_none_even_when_uprn_unique(self, patch_read):
self, patch_read, patch_postcode_valid
):
# User address has building number 47; no row has 47 -> all hard-zero. # User address has building number 47; no row has 47 -> all hard-zero.
patch_read.return_value = _build_df( patch_read.return_value = _build_df(
[ [
@ -277,9 +242,7 @@ class TestUnambiguousUprn:
assert all(m.lexiscore == 0.0 for m in result.matches) assert all(m.lexiscore == 0.0 for m in result.matches)
assert result.unambiguous_uprn() is None assert result.unambiguous_uprn() is None
def test_nan_uprn_becomes_empty_string_not_nan( def test_nan_uprn_becomes_empty_string_not_nan(self, patch_read):
self, patch_read, patch_postcode_valid
):
# Use a real NaN in the UPRN cell. # Use a real NaN in the UPRN cell.
patch_read.return_value = _build_df( patch_read.return_value = _build_df(
[ [
@ -304,7 +267,7 @@ class TestUnambiguousUprn:
class TestTopHelpers: class TestTopHelpers:
def test_top_n_returns_first_k(self, patch_read, patch_postcode_valid): def test_top_n_returns_first_k(self, patch_read):
patch_read.return_value = _build_df( patch_read.return_value = _build_df(
[ [
_row("47 GORDON ROAD", "100"), _row("47 GORDON ROAD", "100"),

View file

@ -0,0 +1,55 @@
"""rank_historic_epc scores already-fetched HistoricEpc records by address.
The pure scoring half of the historic-EPC lookup it takes records (not a
postcode) so it runs with no S3, and is the piece the HistoricEpcResolver plugs
on top of the repository.
"""
from __future__ import annotations
import dataclasses
import pytest
from datatypes.epc.domain.historic_epc import HistoricEpc
from datatypes.epc.domain.historic_epc_matching import (
ScoredHistoricEpc,
rank_historic_epc,
)
def _hist(address: str, uprn: str) -> HistoricEpc:
fields = {f.name: "" for f in dataclasses.fields(HistoricEpc)}
fields["address"] = address
fields["uprn"] = uprn
return HistoricEpc(**fields)
def test_ranks_records_best_first_keeping_zero_score_rows():
# Arrange — a near-disjoint non-match (kept) and the exact match (second).
records = [
_hist("999 SOMEWHERE ELSE", "200"),
_hist("47 GORDON ROAD", "100"),
]
# Act
result = rank_historic_epc(records, "47 Gordon Road")
# Assert
assert all(isinstance(s, ScoredHistoricEpc) for s in result)
assert result[0].record.address == "47 GORDON ROAD"
assert result[0].lexirank == 1
assert len(result) == 2 # zero-score row is kept, not dropped
scores = [s.lexiscore for s in result]
assert scores == sorted(scores, reverse=True)
def test_empty_records_returns_empty_list():
# Act / Assert
assert rank_historic_epc([], "47 Gordon Road") == []
def test_empty_user_address_raises():
# Act / Assert
with pytest.raises(ValueError, match="user_address"):
rank_historic_epc([_hist("47 GORDON ROAD", "100")], "")

View file

@ -0,0 +1,111 @@
# The expired historic EPC conditions prediction with its stable attributes; it is never trusted as current state
## Status
accepted
## Context
The old EPC register API has been shut down. Before it went, we captured its
final data dump to `s3://retrofit-data-dev/historical_epc/{POSTCODE}/data.csv.gz`
— one flat `HistoricEpc` row per certificate, partial tabular data (the old API
never exposed full SAP inputs). PR #1356 lifted that backup into a DDD stack:
`HistoricEpcRepository` (port) / `HistoricEpcS3Repository` (adapter) /
`HistoricEpcResolver`, so far consumed only by `address2UPRN`.
The new gov EPC API (get-energy-performance-data.communities.gov.uk) only
covers certificates registered **since 1 January 2012**. A Property whose only
certificate predates 2012 is EPC-less to Ingestion today, so **EPC Prediction**
synthesises its picture blind from `property_type` + `built_form` (+ any
Landlord Overrides). But the historic dump holds *observed* attributes for that
exact dwelling — wall, roof, floor area, fuel, age band, heating, glazing, PV.
The tension: a pre-2012 certificate is 14+ years stale. Construction and
geometry do not change (wall material, built form, age band, roughly the floor
area); the rest very plausibly has (heating system, hot water, glazing, PV,
insulation levels, lighting). Trusting a 2009 "back boiler, single glazed"
observation as override-grade current truth could make the prediction *worse*
than the neighbour-based default.
## Decision
An expired historic EPC **conditions** EPC Prediction the way a Landlord
Override does — it narrows and enriches the cohort — and is **never copied
into the Effective EPC as current state**.
1. **Stable attributes only.** The historic certificate contributes:
`property_type` (the hard cohort filter, as today), `built_form`, wall
construction **material** (the RdSAP `wall_construction` code resolved from
the description's material prefix, per `wall_type_overlay.py`), roof
construction, `construction_age_band`, and `main_fuel`. Volatile attributes
— heating system, hot water, glazing, PV, insulation states, lighting — are
excluded and stay neighbour-predicted.
2. **Floor area is a tolerance band, not an override.** Comparables are
soft-filtered to within **±20%** of the historic certificate's total floor
area — a coarse dwelling-size filter, not a precision match. The predicted
floor area remains the cohort's geo-weighted median. *(Amended from ±5% —
see Amendment below.)*
3. **Every historic filter rides the existing filter-then-relax ladder**
(ADR-0029): an attribute that cannot be resolved into the cohort's code
space maps to `None` and its filter is simply inactive; a filter that would
starve the cohort below the minimum is relaxed. Degradation is graceful by
construction.
4. **Exact-UPRN lookup.** The prediction path fetches the Property's postcode
shard and matches on UPRN equality (multiple rows for one UPRN → latest
lodgement date). Fuzzy address matching stays quarantined in `address2UPRN`;
a fuzzy hit will never import a neighbour's attributes as if observed.
5. **Provenance: `EpcSource` gains `"expired"`.** The enhanced prediction is
persisted to the predicted slot with `source="expired"` — enhanced by an
expired observation of this dwelling, rather than totally predicted.
`Property.source_path` / Effective-EPC precedence are unchanged.
6. **Scoped to the historic source.** Post-2012 certificates from the new API
— including expired or replaced ones — keep today's treatment. The
epistemic inconsistency (an expired 2014 cert is trusted as current; an
expired 2009 one only conditions) is acknowledged and deliberate: changing
the post-2012 path is a separate, larger-blast-radius decision.
7. **Validated by a pairs harness, not reasoning alone.** A repeatable script
finds properties holding a pre-2012 historic certificate *and* a
post-June-2025 (RdSAP 10 / SAP 10.2) lodged certificate, predicts from the
historic attributes, and reports Component Accuracy against the lodged
truth **per attribute** — so any whitelist member (main fuel is the
judgement call) can be promoted or demoted with evidence. A report, not a
CI gate.
## Consequences
- The lookup chain at Ingestion becomes: new EPC API → historic backup by UPRN
→ plain prediction. A historic miss costs one S3 GET.
- `PredictionTarget` grows optional stable-attribute fields and
`select_comparables` gains their soft filters (including the first
numeric-tolerance filter); all default to `None`/inactive, so existing
callers and behaviour are untouched.
- `EpcSource` widens to `Literal["lodged", "predicted", "expired"]`; the
`epc_property.source` column is already TEXT, so no migration.
- Downstream consumers that branch on `source == "predicted"` treat
`"expired"` the same unless they opt into the distinction (it lives in the
predicted slot; `source_path` is unchanged).
- The whitelist is a hypothesis until the pairs harness reports; attribute
membership changes are cheap (one resolver each).
## Amendment (2026-07-06): band widths set by the pairs harness
The harness ran at scale (2,000 stride-sampled postcodes → **439 pairs**, 419
scored per arm; tables on PR #1466) and two parameters changed with evidence:
- **Age band conditions on the ±1-band neighbourhood, not equality.** The
historic band agrees with the newly lodged one only **52%** exactly but
**90%** within one band — assessors re-band constantly (skewing newer: 31%
of new certs band later). Equality-conditioning steered engaged cohorts
toward a coin flip.
- **The floor-area band widens ±5% → ±20%.** Historic-vs-new TFA agreement:
45% within ±5%, 63% within ±10%, 72% within ±15%, **82% within ±20%**
(remeasurement + extensions). At ±5% the filter engaged on only 19% of
pairs and worsened the floor-area residual.
Also established by the same evidence: main fuel is **95%** stable (after
resolving the register's legacy "backwards compatibility" fuel descriptions),
property type **91%**, wall construction **79%** — the whitelist's core holds.
Known collateral to watch: `roof_insulation_thickness` transfers slightly
worse from age-conditioned cohorts (negative in all three harness rounds) —
candidate follow-up: exclude the roof-insulation mode when age conditioning
engaged.

View file

@ -21,6 +21,47 @@ from domain.geospatial.coordinates import Coordinates
# else it is relaxed (ADR-0029 filter-then-relax ladder). # else it is relaxed (ADR-0029 filter-then-relax ladder).
_DEFAULT_MINIMUM_COHORT = 5 _DEFAULT_MINIMUM_COHORT = 5
# Half-width of the floor-area conditioning band: an expired Historic EPC's
# observed floor area keeps comparables within ±20% of it — a coarse
# dwelling-size filter, not a precision match. Evidence (ADR-0054 amendment,
# 439-pair harness): the historic TFA agrees with the newly lodged one within
# ±5% only 45% of the time (remeasurement + extensions) but within ±20% 82%.
FLOOR_AREA_TOLERANCE = 0.20
# RdSAP Table S1 band letters in chronological order, for band-distance checks.
_AGE_BAND_ORDER = "ABCDEFGHIJKLM"
# roof_construction codes by FORM family. Pinned empirically — a 7,974-cert
# co-occurrence sweep of code x roofs[0].description over single-building-part
# certs (scripts/roof_construction_code_sweep.py): 1=Flat 98%, 4=Pitched 99%,
# 5/8=Pitched 88%, 3="(another dwelling above)" 100% — plus 7/9 = "(another
# premises above)", established by the #1452 heat-loss suppression fix.
ROOF_FORM_BY_CONSTRUCTION: dict[int, str] = {
1: "flat",
3: "dwelling_above",
4: "pitched",
5: "pitched",
7: "premises_above",
8: "pitched",
9: "premises_above",
}
def roof_form_of_construction(code: object) -> Optional[str]:
"""The form family of a roof_construction code, or None when unknown."""
return ROOF_FORM_BY_CONSTRUCTION.get(code) if isinstance(code, int) else None
def age_bands_within_one(candidate: object, target_band: object) -> bool:
"""Whether two Table S1 band letters are at most one band apart. Assessors
re-band constantly (harness: 52% exact agreement historic-vs-new, 90%
within one band), so the age filter keeps the band NEIGHBOURHOOD."""
if not (isinstance(candidate, str) and isinstance(target_band, str)):
return False
if candidate not in _AGE_BAND_ORDER or target_band not in _AGE_BAND_ORDER:
return False
return abs(_AGE_BAND_ORDER.index(candidate) - _AGE_BAND_ORDER.index(target_band)) <= 1
@dataclass(frozen=True) @dataclass(frozen=True)
class ComparableProperty: class ComparableProperty:
@ -88,6 +129,35 @@ def select_comparables(
active=target.wall_construction is not None, active=target.wall_construction is not None,
minimum_cohort=minimum_cohort, minimum_cohort=minimum_cohort,
) )
cohort = _maybe_filter(
cohort,
lambda c: _main_roof_form(c) == target.roof_form,
active=target.roof_form is not None,
minimum_cohort=minimum_cohort,
)
cohort = _maybe_filter(
cohort,
lambda c: age_bands_within_one(
_main_construction_age_band(c), target.construction_age_band
),
active=target.construction_age_band is not None,
minimum_cohort=minimum_cohort,
)
cohort = _maybe_filter(
cohort,
lambda c: _main_fuel_type(c) == target.main_fuel,
active=target.main_fuel is not None,
minimum_cohort=minimum_cohort,
)
target_area = target.total_floor_area_m2
cohort = _maybe_filter(
cohort,
lambda c: target_area is not None
and abs(c.epc.total_floor_area_m2 - target_area)
<= FLOOR_AREA_TOLERANCE * target_area,
active=target_area is not None,
minimum_cohort=minimum_cohort,
)
return ComparableProperties(members=tuple(cohort)) return ComparableProperties(members=tuple(cohort))
@ -124,3 +194,21 @@ def _main_wall_construction(comparable: ComparableProperty) -> object:
"""The main building part's wall construction, or None when no part lodged.""" """The main building part's wall construction, or None when no part lodged."""
parts = comparable.epc.sap_building_parts parts = comparable.epc.sap_building_parts
return parts[0].wall_construction if parts else None return parts[0].wall_construction if parts else None
def _main_roof_form(comparable: ComparableProperty) -> Optional[str]:
"""The main building part's roof-form family, or None when unresolvable."""
parts = comparable.epc.sap_building_parts
return roof_form_of_construction(parts[0].roof_construction) if parts else None
def _main_construction_age_band(comparable: ComparableProperty) -> object:
"""The main building part's Table S1 band letter, or None when no part lodged."""
parts = comparable.epc.sap_building_parts
return parts[0].construction_age_band if parts else None
def _main_fuel_type(comparable: ComparableProperty) -> object:
"""The primary heating fuel code, or None when no main heating lodged."""
details = comparable.epc.sap_heating.main_heating_details
return details[0].main_fuel_type if details else None

View file

@ -0,0 +1,193 @@
"""Resolve an expired Historic EPC's *stable* attributes into the cohort code
spaces that condition EPC Prediction (ADR-0054).
The historic dump carries display text ("Semi-Detached", "Cavity wall, as
built, no insulation (assumed)"); the Comparable Properties cohort carries
gov-EPC codes. This module is the anti-corruption layer between the two: one
resolver per whitelisted stable attribute, each returning None on an
unresolvable value so its cohort filter is simply inactive. Volatile
attributes (heating system, hot water, glazing, PV, insulation states,
lighting) are deliberately absent a 14+-year-old observation of them is not
evidence about today.
"""
from __future__ import annotations
from dataclasses import dataclass, replace
from typing import Optional
from datatypes.epc.domain.historic_epc import HistoricEpc
from domain.epc.property_overrides.override_code_mapping import (
built_form_to_code,
property_type_to_code,
)
from domain.epc_prediction.prediction_target import (
PredictionTarget,
PredictionTargetAttributes,
)
# RdSAP `wall_construction` codes by material prefix — the same table
# `wall_type_overlay.py` pins (source: domain/sap10_ml/rdsap_uvalues.py).
# The historic description's material is everything before the first comma.
_WALL_MATERIAL_CONSTRUCTION: dict[str, int] = {
"Granite or whin": 1,
"Sandstone": 2,
"Solid brick": 3,
"Cavity wall": 4,
"Timber frame": 5,
"System built": 6,
"Cob": 7,
"Park home wall": 8,
}
# RdSAP Table S1 band letters — the code space cohort building parts carry
# (`sap_building_parts[].construction_age_band` is "A".."M" from the API).
# The old register lodged the full England-and-Wales display strings; a
# pre-2012 cert cannot carry the post-2012 bands, but they cost nothing.
_AGE_BAND_LETTERS: dict[str, str] = {
"England and Wales: before 1900": "A",
"England and Wales: 1900-1929": "B",
"England and Wales: 1930-1949": "C",
"England and Wales: 1950-1966": "D",
"England and Wales: 1967-1975": "E",
"England and Wales: 1976-1982": "F",
"England and Wales: 1983-1990": "G",
"England and Wales: 1991-1995": "H",
"England and Wales: 1996-2002": "I",
"England and Wales: 2003-2006": "J",
"England and Wales: 2007 onwards": "K",
"England and Wales: 2007-2011": "K",
"England and Wales: 2012-2022": "L",
"England and Wales: 2023 onwards": "M",
}
# Modern RdSAP-20/21 `main_fuel` codes (epc_codes.csv), keyed by the base fuel
# description — the same family `main_fuel_overlay.py` pins. The old register
# suffixes private fuels with " (not community)", stripped before lookup.
_FUEL_CODES: dict[str, int] = {
"mains gas": 26,
"mains gas (community)": 20,
"LPG": 27,
"bottled LPG": 3,
"LPG special condition": 17,
"oil": 28,
"electricity": 29,
"electricity (community)": 25,
"house coal": 33,
"smokeless coal": 15,
"dual fuel (mineral and wood)": 10,
"wood logs": 6,
"bulk wood pellets": 7,
"wood chips": 8,
"biomass (community)": 31,
}
# Roof FORM families by the description's pre-comma half. Only forms whose
# API code grouping is pinned (see comparable_properties.ROOF_FORM_BY_CONSTRUCTION
# for the empirical evidence); roof rooms and thatch stay None — never guess.
_ROOF_FORM_BY_PREFIX: dict[str, str] = {
"Pitched": "pitched",
"Flat": "flat",
"(another dwelling above)": "dwelling_above",
"(another premises above)": "premises_above",
}
_NOT_COMMUNITY_SUFFIX = " (not community)"
# Pre-RdSAP-17 lodgements carry a deprecation rider after the fuel name; the
# fuel named is the same physical fuel. Dominant in the pre-2012 dump (a
# 65-shard scan: 636/663 otherwise-unresolved values were these variants).
_LEGACY_SUFFIX = " - this is for backwards compatibility only and should not be used"
# SAP-style lodgements prefix the category; map the known forms to the base
# description the code table keys on.
_LEGACY_ALIASES: dict[str, str] = {
"Gas: mains gas": "mains gas",
"Electricity: electricity, unspecified tariff": "electricity",
"dual fuel - mineral + wood": "dual fuel (mineral and wood)",
}
@dataclass(frozen=True)
class HistoricConditioning:
"""The expired cert's stable attributes, in cohort code space; None means
"unresolved — do not condition on this"."""
property_type: Optional[str]
built_form: Optional[str]
wall_construction: Optional[int]
construction_age_band: Optional[str]
main_fuel: Optional[int]
total_floor_area_m2: Optional[float]
roof_form: Optional[str] = None
def _wall_construction(description: str) -> Optional[int]:
material = description.split(",", 1)[0].strip()
return _WALL_MATERIAL_CONSTRUCTION.get(material)
def _roof_form(description: str) -> Optional[str]:
form = description.split(",", 1)[0].strip()
return _ROOF_FORM_BY_PREFIX.get(form)
def _main_fuel(description: str) -> Optional[int]:
base = description.strip().removesuffix(_NOT_COMMUNITY_SUFFIX)
base = base.removesuffix(_LEGACY_SUFFIX)
base = _LEGACY_ALIASES.get(base, base)
return _FUEL_CODES.get(base)
def _floor_area(raw: str) -> Optional[float]:
try:
area = float(raw)
except ValueError:
return None
return area if area > 0 else None
def conditioning_from_historic(record: HistoricEpc) -> HistoricConditioning:
return HistoricConditioning(
property_type=property_type_to_code(record.property_type),
built_form=built_form_to_code(record.built_form),
wall_construction=_wall_construction(record.walls_description),
construction_age_band=_AGE_BAND_LETTERS.get(record.construction_age_band),
main_fuel=_main_fuel(record.main_fuel),
total_floor_area_m2=_floor_area(record.total_floor_area),
roof_form=_roof_form(record.roof_description),
)
def attributes_with_historic_fallback(
attributes: Optional[PredictionTargetAttributes],
conditioning: HistoricConditioning,
) -> PredictionTargetAttributes:
"""Landlord Overrides speak to *current* state, so they always win; the
expired observation only fills the gaps they left (ADR-0054) including
supplying the hard `property_type` gate when no override resolved one."""
if attributes is None:
attributes = PredictionTargetAttributes(property_type=None)
return PredictionTargetAttributes(
property_type=attributes.property_type or conditioning.property_type,
built_form=attributes.built_form or conditioning.built_form,
wall_construction=(
attributes.wall_construction
if attributes.wall_construction is not None
else conditioning.wall_construction
),
)
def target_with_conditioning(
target: PredictionTarget, conditioning: HistoricConditioning
) -> PredictionTarget:
"""The target enriched with the attributes only the expired cert observes:
age band, main fuel, and the ±5% floor-area band (ADR-0054)."""
return replace(
target,
construction_age_band=conditioning.construction_age_band,
main_fuel=conditioning.main_fuel,
total_floor_area_m2=conditioning.total_floor_area_m2,
roof_form=conditioning.roof_form,
)

View file

@ -34,6 +34,16 @@ class PredictionTarget:
# The target Property's own coordinates (resolved from its UPRN), against # The target Property's own coordinates (resolved from its UPRN), against
# which neighbours are distance-weighted. None disables geo-weighting. # which neighbours are distance-weighted. None disables geo-weighting.
coordinates: Optional[Coordinates] = None coordinates: Optional[Coordinates] = None
# Stable attributes observed on an expired Historic EPC condition selection
# the same way (ADR-0054): RdSAP Table S1 band letter, modern main_fuel
# code, and a ±5% floor-area band. None leaves each filter inactive.
construction_age_band: Optional[str] = None
main_fuel: Optional[int] = None
total_floor_area_m2: Optional[float] = None
# The roof FORM family ("pitched" / "flat" / "dwelling_above" /
# "premises_above") — the API's roof_construction codes group by form,
# so the filter matches families, never exact codes (ADR-0054 amendment).
roof_form: Optional[str] = None
@dataclass(frozen=True) @dataclass(frozen=True)

View file

@ -1,7 +1,13 @@
from __future__ import annotations from __future__ import annotations
import re
from dataclasses import dataclass from dataclasses import dataclass
# UK postcode format, matched against the canonical (no-space, upper) value.
# A pure structural check — it does not confirm the postcode actually exists
# (that would need a network lookup, which a value object must not do).
_UK_POSTCODE_RE = re.compile(r"[A-Z]{1,2}\d[A-Z\d]?\d[A-Z]{2}")
@dataclass(frozen=True) @dataclass(frozen=True)
class Postcode: class Postcode:
@ -13,3 +19,8 @@ class Postcode:
def __str__(self) -> str: def __str__(self) -> str:
return self.value return self.value
def is_valid(self) -> bool:
"""Whether the canonical value is a well-formed UK postcode (format
only no existence check, so it is pure and offline)."""
return _UK_POSTCODE_RE.fullmatch(self.value) is not None

View file

@ -56,23 +56,34 @@ def _load_cohort(
if not path.exists(): if not path.exists():
continue continue
raw: dict[str, Any] = json.loads(path.read_text()) raw: dict[str, Any] = json.loads(path.read_text())
try: comparable = comparable_from_payload(cert, raw, coordinates)
epc = EpcPropertyDataMapper.from_api_response(raw) if comparable is not None:
except Exception: # noqa: BLE001 — a bad cert must not abort the sweep cohort.append(comparable)
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,
)
)
return cohort 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]: def load_coordinates(corpus_dir: Path) -> dict[int, Coordinates]:
"""The optional `_coordinates.json` sidecar (`{uprn: [lon, lat]}`), resolved """The optional `_coordinates.json` sidecar (`{uprn: [lon, lat]}`), resolved
from the OS Open-UPRN data by `fetch_corpus_coordinates.py`. Absent for a from the OS Open-UPRN data by `fetch_corpus_coordinates.py`. Absent for a

View file

@ -0,0 +1,26 @@
from __future__ import annotations
from io import BytesIO
import pandas as pd
from infrastructure.s3.s3_client import S3Client
class GzipCsvS3Client(S3Client):
"""Reads a gzipped-CSV S3 object into a pandas DataFrame.
The S3-facing half of the historic-EPC read: an :class:`S3Client` (injected
boto client + bucket) plus the gzip/CSV decode, so a Repo can depend on this
instead of the ``utils.s3`` free functions. ``low_memory=False`` so the wide,
mixed-type historic-EPC columns infer a dtype from the whole column rather
than per-chunk (which would otherwise split one column across object/float).
"""
def read_csv_gz(self, key: str) -> pd.DataFrame:
raw = self.get_object(key)
# pandas-stubs' read_csv overloads carry bare generics of their own, so
# strict mode flags the member access; the call and return are typed.
return pd.read_csv( # pyright: ignore[reportUnknownMemberType]
BytesIO(raw), compression="gzip", low_memory=False
)

View file

@ -5,11 +5,18 @@ from dataclasses import dataclass
from typing import Any, Optional, Protocol from typing import Any, Optional, Protocol
from datatypes.epc.domain.epc_property_data import EpcPropertyData from datatypes.epc.domain.epc_property_data import EpcPropertyData
from datatypes.epc.domain.historic_epc import HistoricEpc
from domain.epc_prediction.comparable_properties import ( from domain.epc_prediction.comparable_properties import (
ComparableProperty, ComparableProperty,
select_comparables, select_comparables,
) )
from domain.epc_prediction.epc_prediction import EpcPrediction from domain.epc_prediction.epc_prediction import EpcPrediction
from domain.epc_prediction.historic_conditioning import (
HistoricConditioning,
attributes_with_historic_fallback,
conditioning_from_historic,
target_with_conditioning,
)
from domain.epc_prediction.prediction_target import ( from domain.epc_prediction.prediction_target import (
PredictionTargetAttributes, PredictionTargetAttributes,
build_prediction_target, build_prediction_target,
@ -17,6 +24,7 @@ from domain.epc_prediction.prediction_target import (
from domain.geospatial.coordinates import Coordinates from domain.geospatial.coordinates import Coordinates
from domain.geospatial.spatial_reference import SpatialReference from domain.geospatial.spatial_reference import SpatialReference
from domain.property.property import PropertyIdentity from domain.property.property import PropertyIdentity
from repositories.epc.epc_repository import EpcSource
from repositories.geospatial.geospatial_repository import GeospatialRepository from repositories.geospatial.geospatial_repository import GeospatialRepository
from repositories.unit_of_work import UnitOfWork from repositories.unit_of_work import UnitOfWork
@ -40,6 +48,13 @@ class PredictionAttributesReader(Protocol):
def attributes_for(self, property_id: int) -> PredictionTargetAttributes: ... def attributes_for(self, property_id: int) -> PredictionTargetAttributes: ...
class HistoricEpcReader(Protocol):
"""The slice of the Historic-EPC resolver Ingestion needs: the exact-UPRN
lookup whose stable attributes condition prediction (ADR-0054)."""
def record_for_uprn(self, uprn: str, postcode: str) -> Optional[HistoricEpc]: ...
class SolarFetcher(Protocol): class SolarFetcher(Protocol):
"""The slice of the Google Solar client Ingestion needs (e.g. GoogleSolarApiClient).""" """The slice of the Google Solar client Ingestion needs (e.g. GoogleSolarApiClient)."""
@ -69,6 +84,9 @@ class _Fetched:
predicted_epc: Optional[EpcPropertyData] predicted_epc: Optional[EpcPropertyData]
solar_insights: Optional[dict[str, Any]] solar_insights: Optional[dict[str, Any]]
spatial: Optional[SpatialReference] spatial: Optional[SpatialReference]
# "expired" when the prediction was conditioned by an expired Historic EPC
# (ADR-0054); plain "predicted" otherwise.
predicted_source: EpcSource = "predicted"
class IngestionOrchestrator: class IngestionOrchestrator:
@ -98,6 +116,7 @@ class IngestionOrchestrator:
comparables_repo: Optional[ComparablesRepo] = None, comparables_repo: Optional[ComparablesRepo] = None,
prediction_attributes_reader: Optional[PredictionAttributesReader] = None, prediction_attributes_reader: Optional[PredictionAttributesReader] = None,
epc_prediction: Optional[EpcPrediction] = None, epc_prediction: Optional[EpcPrediction] = None,
historic_epc_reader: Optional[HistoricEpcReader] = None,
) -> None: ) -> None:
self._unit_of_work = unit_of_work self._unit_of_work = unit_of_work
self._epc_fetcher = epc_fetcher self._epc_fetcher = epc_fetcher
@ -110,6 +129,7 @@ class IngestionOrchestrator:
self._comparables_repo = comparables_repo self._comparables_repo = comparables_repo
self._prediction_attributes_reader = prediction_attributes_reader self._prediction_attributes_reader = prediction_attributes_reader
self._epc_prediction = epc_prediction self._epc_prediction = epc_prediction
self._historic_epc_reader = historic_epc_reader
def run(self, property_ids: list[int]) -> None: def run(self, property_ids: list[int]) -> None:
preps = self._prepare(property_ids) preps = self._prepare(property_ids)
@ -149,33 +169,64 @@ class IngestionOrchestrator:
solar_insights = self._solar_fetcher.get_building_insights( solar_insights = self._solar_fetcher.get_building_insights(
coordinates.longitude, coordinates.latitude coordinates.longitude, coordinates.latitude
) )
predicted_epc = ( predicted_epc: Optional[EpcPropertyData] = None
self._predict(prep.identity, coordinates, prep.attributes) conditioning: Optional[HistoricConditioning] = None
if epc is None if epc is None:
else None conditioning = self._historic_conditioning(uprn, prep.identity.postcode)
predicted_epc = self._predict(
prep.identity, coordinates, prep.attributes, conditioning
)
predicted_source = (
"expired"
if predicted_epc is not None and conditioning is not None
else "predicted"
) )
return _Fetched( return _Fetched(
prep.property_id, uprn, epc, predicted_epc, solar_insights, spatial prep.property_id,
uprn,
epc,
predicted_epc,
solar_insights,
spatial,
predicted_source,
) )
def _historic_conditioning(
self, uprn: int, postcode: str
) -> Optional[HistoricConditioning]:
"""The expired Historic EPC's stable attributes for this exact UPRN, or
None when the reader is unwired or the backup holds no row (ADR-0054)."""
if self._historic_epc_reader is None:
return None
record = self._historic_epc_reader.record_for_uprn(str(uprn), postcode)
return conditioning_from_historic(record) if record is not None else None
def _predict( def _predict(
self, self,
identity: PropertyIdentity, identity: PropertyIdentity,
coordinates: Optional[Coordinates], coordinates: Optional[Coordinates],
attributes: Optional[PredictionTargetAttributes], attributes: Optional[PredictionTargetAttributes],
conditioning: Optional[HistoricConditioning] = None,
) -> Optional[EpcPropertyData]: ) -> Optional[EpcPropertyData]:
"""Synthesise the EPC-less Property's picture from its postcode cohort, or """Synthesise the EPC-less Property's picture from its postcode cohort, or
None when the predictor is unwired, the Property is gated out (unknown None when the predictor is unwired, the Property is gated out (unknown
property type), or no comparables survive selection (ADR-0031).""" property type), or no comparables survive selection (ADR-0031). An
expired Historic EPC's stable attributes fill override gaps and condition
the cohort; Landlord Overrides always win where both speak (ADR-0054)."""
if ( if (
self._comparables_repo is None self._comparables_repo is None
or self._epc_prediction is None or self._epc_prediction is None
or attributes is None or (attributes is None and conditioning is None)
): ):
return None return None
if conditioning is not None:
attributes = attributes_with_historic_fallback(attributes, conditioning)
assert attributes is not None # one of the two branches above supplied it
target = build_prediction_target(identity, coordinates, attributes) target = build_prediction_target(identity, coordinates, attributes)
if target is None: if target is None:
return None return None
if conditioning is not None:
target = target_with_conditioning(target, conditioning)
candidates = self._comparables_repo.candidates_for(identity.postcode) candidates = self._comparables_repo.candidates_for(identity.postcode)
comparables = select_comparables(target, candidates) comparables = select_comparables(target, candidates)
if not comparables.members: if not comparables.members:
@ -191,7 +242,7 @@ class IngestionOrchestrator:
uow.epc.save( uow.epc.save(
item.predicted_epc, item.predicted_epc,
property_id=item.property_id, property_id=item.property_id,
source="predicted", source=item.predicted_source,
) )
# The live `solar` table is keyed by UPRN and needs the fetch's # The live `solar` table is keyed by UPRN and needs the fetch's
# coordinates; insights are only set when those coordinates were # coordinates; insights are only set when those coordinates were

View file

@ -50,9 +50,20 @@ from infrastructure.postgres.epc_property_table import (
EpcRenewableHeatIncentiveModel, EpcRenewableHeatIncentiveModel,
EpcWindowModel, EpcWindowModel,
) )
from repositories.epc.epc_repository import EpcRepository, EpcSource from repositories.epc.epc_repository import (
PREDICTED_SLOT_SOURCES,
EpcRepository,
EpcSource,
)
from utilities.private import private from utilities.private import private
def _slot_sources(source: EpcSource) -> tuple[EpcSource, ...]:
"""The source family sharing `source`'s slot: "predicted" and "expired" are
one slot whose flavour a re-ingestion may flip, so every slot read and
slot-clearing delete addresses the family (ADR-0054)."""
return PREDICTED_SLOT_SOURCES if source in PREDICTED_SLOT_SOURCES else (source,)
_T = TypeVar("_T") _T = TypeVar("_T")
@ -320,7 +331,7 @@ class EpcPostgresRepository(EpcRepository):
for i in self._session.exec( for i in self._session.exec(
select(EpcPropertyModel.id) select(EpcPropertyModel.id)
.where(col(EpcPropertyModel.property_id).in_(property_ids)) .where(col(EpcPropertyModel.property_id).in_(property_ids))
.where(EpcPropertyModel.source == source) .where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
).all() ).all()
if i is not None if i is not None
] ]
@ -367,7 +378,7 @@ class EpcPostgresRepository(EpcRepository):
for i in self._session.exec( for i in self._session.exec(
select(EpcPropertyModel.id) select(EpcPropertyModel.id)
.where(EpcPropertyModel.property_id == property_id) .where(EpcPropertyModel.property_id == property_id)
.where(EpcPropertyModel.source == source) .where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
).all() ).all()
if i is not None if i is not None
] ]
@ -417,7 +428,7 @@ class EpcPostgresRepository(EpcRepository):
row = self._session.exec( row = self._session.exec(
select(EpcPropertyModel) select(EpcPropertyModel)
.where(EpcPropertyModel.property_id == property_id) .where(EpcPropertyModel.property_id == property_id)
.where(EpcPropertyModel.source == source) .where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
.order_by(EpcPropertyModel.id) # type: ignore[arg-type] .order_by(EpcPropertyModel.id) # type: ignore[arg-type]
).first() ).first()
if row is None or row.id is None: if row is None or row.id is None:
@ -444,7 +455,7 @@ class EpcPostgresRepository(EpcRepository):
parents = self._session.exec( parents = self._session.exec(
select(EpcPropertyModel) select(EpcPropertyModel)
.where(col(EpcPropertyModel.property_id).in_(property_ids)) .where(col(EpcPropertyModel.property_id).in_(property_ids))
.where(EpcPropertyModel.source == source) .where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
.order_by(EpcPropertyModel.id) # type: ignore[arg-type] .order_by(EpcPropertyModel.id) # type: ignore[arg-type]
).all() ).all()
parent_by_property: dict[int, EpcPropertyModel] = {} parent_by_property: dict[int, EpcPropertyModel] = {}

View file

@ -9,8 +9,15 @@ if TYPE_CHECKING:
from repositories.epc.epc_postgres_repository import EpcSaveRequest from repositories.epc.epc_postgres_repository import EpcSaveRequest
# Provenance of a persisted EPC picture (ADR-0031): a real "lodged" EPC, or a # Provenance of a persisted EPC picture (ADR-0031): a real "lodged" EPC, or a
# "predicted" one synthesised by EPC Prediction. A property can hold one of each. # "predicted" one synthesised by EPC Prediction — "expired" is a prediction
EpcSource = Literal["lodged", "predicted"] # enhanced by an expired Historic EPC's stable attributes (ADR-0054). A
# property holds one lodged picture and one predicted-slot picture.
EpcSource = Literal["lodged", "predicted", "expired"]
# The predicted slot's source family: "predicted" and "expired" occupy the SAME
# slot (a re-ingestion may flip the flavour), so slot reads and slot-clearing
# deletes must always address the family, never one member (ADR-0054).
PREDICTED_SLOT_SOURCES: tuple[EpcSource, ...] = ("predicted", "expired")
class EpcRepository(ABC): class EpcRepository(ABC):

View file

View file

@ -0,0 +1,30 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from datatypes.epc.domain.historic_epc import HistoricEpc
from domain.postcode import Postcode
class PostcodeNotFound(Exception):
"""The postcode is empty, so it cannot key a historic-EPC lookup.
Distinct from a non-empty postcode that simply has no stored data that
case returns an empty list, because a miss is the normal, expected outcome
of a best-effort historic lookup.
"""
class HistoricEpcRepository(ABC):
"""Reads the 'old EPC' backup — one flat ``HistoricEpc`` row per certificate,
sharded by postcode in S3 (``historical_epc/{POSTCODE}/data.csv.gz``).
A Repo, not a Fetcher (ADR-0011): it reads stored data with no live EPC API
call. Takes a normalised :class:`Postcode` value object (so the boundary is
strictly typed and never re-sanitises a raw string). A non-empty postcode
with no stored object returns ``[]``; an empty one raises
:class:`PostcodeNotFound`.
"""
@abstractmethod
def get_for_postcode(self, postcode: Postcode) -> list[HistoricEpc]: ...

View file

@ -0,0 +1,62 @@
from __future__ import annotations
from typing import Optional
from datatypes.epc.domain.historic_epc import HistoricEpc
from datatypes.epc.domain.historic_epc_matching import (
HistoricEpcMatches,
ScoredHistoricEpc,
rank_historic_epc,
)
from domain.postcode import Postcode
from repositories.historic_epc.historic_epc_repository import HistoricEpcRepository
class HistoricEpcResolver:
"""Resolves an address to a historic-EPC match by composing the repository
(fetch a postcode's records) with the matcher (score them against the
address). This is where ``address2uprn`` plugs onto the old-EPC backup.
"""
def __init__(self, repo: HistoricEpcRepository) -> None:
self._repo = repo
def match(self, user_address: str, postcode: str) -> HistoricEpcMatches:
"""All of the postcode's historic EPCs scored against ``user_address``."""
if not user_address:
raise ValueError("user_address must be non-empty")
pc = Postcode(postcode)
records: list[HistoricEpc] = self._repo.get_for_postcode(pc)
matches: list[ScoredHistoricEpc] = rank_historic_epc(records, user_address)
return HistoricEpcMatches(
user_address=user_address,
postcode=str(pc),
matches=matches,
)
def record_for_uprn(self, uprn: str, postcode: str) -> Optional[HistoricEpc]:
"""The postcode shard's certificate for ``uprn``, or None on a miss.
Exact UPRN equality only the prediction path must never import a
fuzzy address match's attributes as if observed (ADR-0054)."""
if not uprn:
return None
records: list[HistoricEpc] = self._repo.get_for_postcode(Postcode(postcode))
certs: list[HistoricEpc] = [r for r in records if r.uprn == uprn]
if not certs:
return None
return max(certs, key=lambda r: r.lodgement_date)
def resolve_uprn(
self, user_address: str, postcode: str
) -> Optional[tuple[str, str, float]]:
"""``(uprn, matched_address, lexiscore)`` for an unambiguous rank-1
match, else None (no data / ambiguous tie / zero score)."""
matches: HistoricEpcMatches = self.match(user_address, postcode)
uprn: Optional[str] = matches.unambiguous_uprn()
if not uprn or uprn == "nan":
return None
top: Optional[ScoredHistoricEpc] = matches.top()
if top is None:
return None
return uprn, top.record.address, top.lexiscore

View file

@ -0,0 +1,74 @@
from __future__ import annotations
from collections.abc import Hashable
from typing import Any
import pandas as pd
from botocore.exceptions import ClientError
from datatypes.epc.domain.historic_epc import HistoricEpc
from datatypes.epc.domain.historic_epc_matching import (
map_historic_epc_row_to_domain,
)
from domain.postcode import Postcode
from infrastructure.s3.gzip_csv_s3_client import GzipCsvS3Client
from infrastructure.s3.s3_uri import parse_s3_uri
from repositories.historic_epc.historic_epc_repository import (
HistoricEpcRepository,
PostcodeNotFound,
)
DEFAULT_S3_ROOT = "s3://retrofit-data-dev/historical_epc"
class HistoricEpcS3Repository(HistoricEpcRepository):
"""Reads per-postcode ``data.csv.gz`` shards of the historic EPC backup.
The bucket and the raw read live in the injected :class:`GzipCsvS3Client`
(``infrastructure/s3``); the Repo only builds the per-postcode key and maps
rows to domain records, so it holds no S3/HTTP code of its own the same
shape as :class:`UnstandardisedAddressListCsvS3Repository`, and unlike the
old ``utils.s3`` free-function dependency.
"""
def __init__(self, client: GzipCsvS3Client, root_prefix: str) -> None:
self._client = client
self._root_prefix = root_prefix
@classmethod
def with_default_s3_client(
cls, s3_root: str = DEFAULT_S3_ROOT
) -> "HistoricEpcS3Repository":
"""Build a repository backed by a real S3 client for ``s3_root``
(``s3://bucket/prefix``).
The composition-root convenience constructor used by the lambda and the
``match_addresses_for_postcode`` seam; tests inject a ``GzipCsvS3Client``
over a moto-mocked boto client through ``__init__`` instead.
"""
import boto3
bucket, root_prefix = parse_s3_uri(s3_root)
# boto3-stubs types client("s3") via an overload set in which services
# without installed stubs return Unknown, so strict mode flags the
# member access; the "s3" overload itself resolves to a typed S3Client.
boto_s3 = boto3.client("s3") # pyright: ignore[reportUnknownMemberType]
return cls(GzipCsvS3Client(boto_s3, bucket), root_prefix)
def get_for_postcode(self, postcode: Postcode) -> list[HistoricEpc]:
if not postcode.is_valid():
raise PostcodeNotFound(f"{postcode.value!r} is not a valid UK postcode")
key = f"{self._root_prefix.rstrip('/')}/{postcode}/data.csv.gz"
try:
df: pd.DataFrame = self._client.read_csv_gz(key)
except ClientError as e:
if e.response.get("Error", {}).get("Code") in ("NoSuchKey", "404"):
return []
raise
# pandas-stubs' to_dict overloads carry bare generics of their own, so
# strict mode flags the member access; the rows annotation keeps the
# downstream mapping fully typed.
rows: list[dict[Hashable, Any]] = df.to_dict( # pyright: ignore[reportUnknownMemberType]
orient="records"
)
return [map_historic_epc_row_to_domain(row) for row in rows]

View file

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

81
scripts/epc_disk_cache.py Normal file
View file

@ -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

View file

@ -0,0 +1,597 @@
"""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, 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 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 --out report.md
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
FLOOR_AREA_TOLERANCE,
ComparableProperty,
age_bands_within_one,
roof_form_of_construction,
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
_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
(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 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:
"""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."""
classification: dict[str, tuple[int, int]]
residuals: dict[str, list[float]]
sap_residuals: list[float]
def aggregate(scores: list[PairScore]) -> ArmScores:
counts: dict[str, list[int]] = defaultdict(lambda: [0, 0])
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["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(
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:
"""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 = _rate_cell(plain.classification.get(component, (0, 0)))
c = _rate_cell(conditioned.classification.get(component, (0, 0)))
lines.append(f"| {component} | {p} | {c} |")
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)
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 _part0(epc: EpcPropertyData) -> Optional[object]:
parts = epc.sap_building_parts
return parts[0] if parts else None
def _actual_roof_form(epc: EpcPropertyData) -> Optional[str]:
part = _part0(epc)
return roof_form_of_construction(getattr(part, "roof_construction", None))
def _actual_age_band(epc: EpcPropertyData) -> Optional[object]:
part = _part0(epc)
return getattr(part, "construction_age_band", None)
def _actual_wall(epc: EpcPropertyData) -> Optional[object]:
part = _part0(epc)
return getattr(part, "wall_construction", None)
def _actual_fuel(epc: EpcPropertyData) -> Optional[object]:
details = epc.sap_heating.main_heating_details
return details[0].main_fuel_type if details else None
def _tfa_within_band(actual_tfa: float, hist_tfa: Optional[float]) -> Optional[bool]:
if hist_tfa is None:
return None
return abs(actual_tfa - hist_tfa) <= FLOOR_AREA_TOLERANCE * hist_tfa
@dataclass(frozen=True)
class LadderStep:
"""One conditioning filter's fate in the sequential relax ladder: how many
of the incoming cohort matched, and whether it engaged (matches >= k)."""
matches: int
cohort_before: int
engaged: bool
def simulate_conditioning_ladder(
base: list[ComparableProperty],
*,
age_band: Optional[str],
main_fuel: Optional[int],
total_floor_area_m2: Optional[float],
roof_form: Optional[str] = None,
minimum_cohort: int = 5,
) -> dict[str, Optional[LadderStep]]:
"""Replay select_comparables' age->fuel->TFA conditioning sequence over the
plain arm's selected cohort, recording per filter whether it ENGAGED
(>= minimum_cohort matches survive) or RELAXED. None = attribute unresolved,
filter never active."""
steps: dict[str, Optional[LadderStep]] = {}
cohort = list(base)
def apply(name: str, active: bool, matches: list[ComparableProperty]) -> None:
nonlocal cohort
if not active:
steps[name] = None
return
engaged = len(matches) >= minimum_cohort
steps[name] = LadderStep(len(matches), len(cohort), engaged)
if engaged:
cohort = matches
apply(
"roof_form",
roof_form is not None,
[c for c in cohort if _actual_roof_form(c.epc) == roof_form],
)
apply(
"construction_age_band",
age_band is not None,
[c for c in cohort if age_bands_within_one(_actual_age_band(c.epc), age_band)],
)
apply(
"main_fuel",
main_fuel is not None,
[c for c in cohort if _actual_fuel(c.epc) == main_fuel],
)
apply(
"total_floor_area",
total_floor_area_m2 is not None,
[
c
for c in cohort
if total_floor_area_m2 is not None
and abs(c.epc.total_floor_area_m2 - total_floor_area_m2)
<= FLOOR_AREA_TOLERANCE * total_floor_area_m2
],
)
return steps
def format_diagnosis(rows: list[dict[str, object]]) -> str:
"""Aggregate the per-pair telemetry into the two diagnosis tables: did each
conditioning filter ever ENGAGE, and does the historic value still AGREE
with the newly lodged one (the staleness measurement)."""
if not rows:
return ""
filters = ("roof_form", "construction_age_band", "main_fuel", "total_floor_area")
lines = [
"",
"## Diagnosis — filter engagement (conditioned arm)",
"",
"| filter | resolved | engaged | relaxed (too few matches) |",
"|---|---|---|---|",
]
for name in filters:
resolved = engaged = 0
for row in rows:
step = row.get(f"ladder_{name}")
if step is None:
continue
resolved += 1
if isinstance(step, LadderStep) and step.engaged:
engaged += 1
lines.append(
f"| {name} | {resolved}/{len(rows)} | {engaged}/{resolved or 1} "
f"| {resolved - engaged}/{resolved or 1} |"
)
attrs = (
"property_type",
"built_form",
"wall_construction",
"roof_form",
"construction_age_band",
"main_fuel",
"tfa_within_band",
)
lines += [
"",
"## Diagnosis — historic vs newly-lodged agreement (staleness)",
"",
"| attribute | historic resolved | agrees with new cert |",
"|---|---|---|",
]
for name in attrs:
resolved = agrees = 0
for row in rows:
value = row.get(f"agrees_{name}")
if value is None:
continue
resolved += 1
if value:
agrees += 1
pct = f" ({agrees / resolved:.0%})" if resolved else ""
lines.append(f"| {name} | {resolved}/{len(rows)} | {agrees}/{resolved or 1}{pct} |")
sizes: list[int] = [
size
for row in rows
if isinstance((size := row.get("plain_cohort_size")), int)
]
if sizes:
lines += [
"",
f"Mean plain-arm cohort size: {sum(sizes) / len(sizes):.1f} "
f"(min {min(sizes)}, max {max(sizes)}); relax threshold k=5.",
]
return "\n".join(lines)
def run( # pragma: no cover - live IO composition
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
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()
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()
predictor = EpcPrediction()
calculator = Sap10Calculator()
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] = []
telemetry: list[dict[str, object]] = []
pairs = 0
def emit_telemetry(row: dict[str, object]) -> None:
# Append incrementally so a late crash loses nothing.
telemetry.append(row)
if telemetry_path is None:
return
import dataclasses as _dc
import json as _json
serialisable = {
k: (_dc.asdict(v) if isinstance(v, LadderStep) else v)
for k, v in row.items()
}
with telemetry_path.open("a") as handle:
handle.write(_json.dumps(serialisable) + "\n")
if telemetry_path is not None and telemetry_path.exists():
telemetry_path.unlink()
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))
)
# Pair-check first: only a postcode with a SAP-10.2 relodgement of a
# pre-2012 UPRN pays for the cohort fetch. A cert the mapper can't yet
# map (strict-raise) can't be a validation target either — skip it and
# keep sweeping; one bad cert must not kill a multi-hour run.
paired: list[tuple[str, HistoricEpc, EpcPropertyData]] = []
for uprn, record in historic.items():
try:
actual = epc_client.get_by_uprn(int(uprn))
except Exception as error:
print(f" {uprn}: unmappable cert skipped: {error}", file=sys.stderr)
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
try:
cohort = comparables_repo.candidates_for(postcode)
except Exception as error:
print(f" {postcode}: cohort fetch failed: {error}", file=sys.stderr)
continue
for uprn, record, actual in paired:
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_target is not None:
base = list(select_comparables(plain_target, loo_cohort).members)
ladder = simulate_conditioning_ladder(
base,
age_band=conditioning.construction_age_band,
main_fuel=conditioning.main_fuel,
total_floor_area_m2=conditioning.total_floor_area_m2,
roof_form=conditioning.roof_form,
)
emit_telemetry(
{
"postcode": postcode,
"uprn": uprn,
"plain_cohort_size": len(base),
# Raw values alongside the booleans, so band-width and
# near-miss questions are answerable post hoc.
"historic_age_band": conditioning.construction_age_band,
"actual_age_band": _actual_age_band(actual),
"historic_tfa": conditioning.total_floor_area_m2,
"actual_tfa": actual.total_floor_area_m2,
"historic_fuel": conditioning.main_fuel,
"actual_fuel": _actual_fuel(actual),
"historic_fuel_text": record.main_fuel,
**{f"ladder_{k}": v for k, v in ladder.items()},
"agrees_property_type": (
None
if conditioning.property_type is None
else conditioning.property_type == actual.property_type
),
"agrees_built_form": (
None
if conditioning.built_form is None
else conditioning.built_form == actual.built_form
),
"agrees_roof_form": (
None
if conditioning.roof_form is None
else conditioning.roof_form == _actual_roof_form(actual)
),
"agrees_wall_construction": (
None
if conditioning.wall_construction is None
else conditioning.wall_construction == _actual_wall(actual)
),
"agrees_construction_age_band": (
None
if conditioning.construction_age_band is None
else conditioning.construction_age_band
== _actual_age_band(actual)
),
"agrees_main_fuel": (
None
if conditioning.main_fuel is None
else conditioning.main_fuel == _actual_fuel(actual)
),
"agrees_tfa_within_band": _tfa_within_band(
actual.total_floor_area_m2,
conditioning.total_floor_area_m2,
),
}
)
if plain is not None:
plain_scores.append(
PairScore(compare_prediction(plain, actual), sap_residual(plain, actual))
)
if conditioned is not None:
conditioned_scores.append(
PairScore(
compare_prediction(conditioned, actual),
sap_residual(conditioned, actual),
)
)
report = format_report(
aggregate(plain_scores), aggregate(conditioned_scores), pairs
)
return report + format_diagnosis(telemetry)
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")
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:
postcodes += [
line.strip()
for line in args.postcodes_file.read_text().splitlines()
if line.strip()
]
if not postcodes:
parser.error("no postcodes given")
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)
if __name__ == "__main__":
main()

View file

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

View file

@ -8,7 +8,12 @@ enough remain, weighted by recency × similarity. Pure domain logic.
from datetime import date from datetime import date
from typing import Optional, Union from typing import Optional, Union
from datatypes.epc.domain.epc_property_data import EpcPropertyData, SapBuildingPart from datatypes.epc.domain.epc_property_data import (
EpcPropertyData,
MainHeatingDetail,
SapBuildingPart,
SapHeating,
)
from domain.epc_prediction.comparable_properties import ( from domain.epc_prediction.comparable_properties import (
ComparableProperty, ComparableProperty,
ComparableProperties, ComparableProperties,
@ -25,6 +30,10 @@ def _comparable(
wall_construction: Optional[Union[int, str]] = None, wall_construction: Optional[Union[int, str]] = None,
address: Optional[str] = None, address: Optional[str] = None,
registration_date: Optional[date] = None, registration_date: Optional[date] = None,
construction_age_band: Optional[str] = None,
main_fuel: Optional[int] = None,
total_floor_area_m2: Optional[float] = None,
roof_construction: Optional[int] = None,
) -> ComparableProperty: ) -> ComparableProperty:
"""A ComparableProperty carrying only the fields under test (opaque EpcPropertyData """A ComparableProperty carrying only the fields under test (opaque EpcPropertyData
with property_type / built_form / main wall set the partial-instance idiom).""" with property_type / built_form / main wall set the partial-instance idiom)."""
@ -34,7 +43,19 @@ def _comparable(
main: SapBuildingPart = object.__new__(SapBuildingPart) main: SapBuildingPart = object.__new__(SapBuildingPart)
if wall_construction is not None: if wall_construction is not None:
main.wall_construction = wall_construction main.wall_construction = wall_construction
if construction_age_band is not None:
main.construction_age_band = construction_age_band
if roof_construction is not None:
main.roof_construction = roof_construction
epc.sap_building_parts = [main] epc.sap_building_parts = [main]
if main_fuel is not None:
detail: MainHeatingDetail = object.__new__(MainHeatingDetail)
detail.main_fuel_type = main_fuel
heating: SapHeating = object.__new__(SapHeating)
heating.main_heating_details = [detail]
epc.sap_heating = heating
if total_floor_area_m2 is not None:
epc.total_floor_area_m2 = total_floor_area_m2
return ComparableProperty( return ComparableProperty(
epc=epc, epc=epc,
certificate_number=certificate_number, certificate_number=certificate_number,
@ -171,3 +192,169 @@ def test_known_wall_override_relaxes_when_too_few_match() -> None:
# Assert — relaxed: all eight houses retained. # Assert — relaxed: all eight houses retained.
assert len(result.members) == 8 assert len(result.members) == 8
def test_historic_age_band_conditions_the_cohort() -> None:
# Arrange — the expired cert observed band C (1930-1949); 5 band-C houses +
# 2 band-G houses in the cohort (ADR-0054).
target = PredictionTarget(
postcode="LS6 1AA", property_type="2", construction_age_band="C"
)
candidates = [
_comparable(
property_type="2", construction_age_band="C", certificate_number=f"C{i}"
)
for i in range(5)
] + [
_comparable(
property_type="2", construction_age_band="G", certificate_number=f"G{i}"
)
for i in range(2)
]
# Act
result: ComparableProperties = select_comparables(
target, candidates, minimum_cohort=5
)
# Assert — only the same-age-band comparables remain.
assert {c.certificate_number for c in result.members} == {
"C0", "C1", "C2", "C3", "C4"
}
def test_historic_main_fuel_conditions_the_cohort() -> None:
# Arrange — the expired cert observed mains gas (26); 5 gas + 2 electric
# (29) houses in the cohort (ADR-0054).
target = PredictionTarget(postcode="LS6 1AA", property_type="2", main_fuel=26)
candidates = [
_comparable(property_type="2", main_fuel=26, certificate_number=f"G{i}")
for i in range(5)
] + [
_comparable(property_type="2", main_fuel=29, certificate_number=f"E{i}")
for i in range(2)
]
# Act
result: ComparableProperties = select_comparables(
target, candidates, minimum_cohort=5
)
# Assert
assert {c.certificate_number for c in result.members} == {
"G0", "G1", "G2", "G3", "G4"
}
def test_floor_area_band_keeps_comparables_within_20_percent() -> None:
# Arrange — the expired cert observed 100 m²; the band is ±20% (ADR-0054 as
# amended: the 439-pair harness put historic-vs-new TFA agreement at only
# 45% within ±5% but 82% within ±20% — a coarse dwelling-size filter, not a
# precision match): 96, 118 and 84 are in, 125 and 79 are out.
target = PredictionTarget(
postcode="LS6 1AA", property_type="2", total_floor_area_m2=100.0
)
candidates = [
_comparable(property_type="2", total_floor_area_m2=96.0, certificate_number="A"),
_comparable(
property_type="2", total_floor_area_m2=118.0, certificate_number="B"
),
_comparable(property_type="2", total_floor_area_m2=84.0, certificate_number="C"),
_comparable(
property_type="2", total_floor_area_m2=125.0, certificate_number="X"
),
_comparable(property_type="2", total_floor_area_m2=79.0, certificate_number="Y"),
]
# Act
result: ComparableProperties = select_comparables(
target, candidates, minimum_cohort=2
)
# Assert
assert {c.certificate_number for c in result.members} == {"A", "B", "C"}
def test_historic_age_band_conditions_within_one_band() -> None:
# Arrange — the expired cert observed band C, but assessors re-band ±1
# constantly (harness: 52% exact agreement, 90% within one band), so the
# filter keeps the band NEIGHBOURHOOD: B/C/D match, G does not (ADR-0054 as
# amended).
target = PredictionTarget(
postcode="LS6 1AA", property_type="2", construction_age_band="C"
)
near = ["B", "C", "D", "D", "B"]
candidates = [
_comparable(
property_type="2",
construction_age_band=band,
certificate_number=f"N{i}",
)
for i, band in enumerate(near)
] + [
_comparable(
property_type="2", construction_age_band="G", certificate_number=f"G{i}"
)
for i in range(2)
]
# Act
result: ComparableProperties = select_comparables(
target, candidates, minimum_cohort=5
)
# Assert — the five band-neighbourhood comparables survive; band G drops.
assert {c.certificate_number for c in result.members} == {
"N0", "N1", "N2", "N3", "N4"
}
def test_historic_roof_form_conditions_the_cohort_by_family() -> None:
# Arrange — the expired cert observed a pitched roof. The API's
# roof_construction codes group into FORM families (empirical sweep:
# 4/5/8 = pitched, 1 = flat), so all pitched-family comparables match and
# the flat one drops (ADR-0054 as amended).
target = PredictionTarget(postcode="LS6 1AA", property_type="2", roof_form="pitched")
candidates = [
_comparable(property_type="2", roof_construction=4, certificate_number="P4a"),
_comparable(property_type="2", roof_construction=4, certificate_number="P4b"),
_comparable(property_type="2", roof_construction=5, certificate_number="P5"),
_comparable(property_type="2", roof_construction=8, certificate_number="P8"),
_comparable(property_type="2", roof_construction=4, certificate_number="P4c"),
_comparable(property_type="2", roof_construction=1, certificate_number="F1"),
]
# Act
result: ComparableProperties = select_comparables(
target, candidates, minimum_cohort=5
)
# Assert — the whole pitched family survives; the flat roof drops.
assert {c.certificate_number for c in result.members} == {
"P4a", "P4b", "P4c", "P5", "P8"
}
def test_floor_area_band_relaxes_when_too_few_match() -> None:
# Arrange — only one comparable inside the ±5% band (< k=2): the band must
# relax rather than starve the cohort (graceful degradation, ADR-0029).
target = PredictionTarget(
postcode="LS6 1AA", property_type="2", total_floor_area_m2=100.0
)
candidates = [
_comparable(property_type="2", total_floor_area_m2=99.0, certificate_number="A"),
_comparable(
property_type="2", total_floor_area_m2=140.0, certificate_number="X"
),
_comparable(
property_type="2", total_floor_area_m2=150.0, certificate_number="Y"
),
]
# Act
result: ComparableProperties = select_comparables(
target, candidates, minimum_cohort=2
)
# Assert — relaxed: all three retained.
assert len(result.members) == 3

View file

@ -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

View file

@ -0,0 +1,117 @@
"""HistoricConditioning resolves an expired Historic EPC's stable attributes
into the cohort's code spaces (ADR-0054).
Only stable attributes are resolved volatile ones (heating system, glazing,
PV, insulation states) never condition prediction. Every resolver degrades to
None on an unresolvable value, which leaves its cohort filter inactive.
"""
from __future__ import annotations
import dataclasses
from datatypes.epc.domain.historic_epc import HistoricEpc
from domain.epc_prediction.historic_conditioning import (
HistoricConditioning,
conditioning_from_historic,
)
def _hist(**overrides: str) -> HistoricEpc:
fields = {f.name: "" for f in dataclasses.fields(HistoricEpc)}
fields.update(overrides)
return HistoricEpc(**fields)
def test_resolves_stable_attributes_into_cohort_code_spaces():
# Arrange — display-text values exactly as the old register lodged them.
record = _hist(
property_type="House",
built_form="Semi-Detached",
walls_description="Cavity wall, as built, no insulation (assumed)",
roof_description="Pitched, 100 mm loft insulation",
construction_age_band="England and Wales: 1930-1949",
main_fuel="mains gas (not community)",
total_floor_area="84",
)
# Act
conditioning = conditioning_from_historic(record)
# Assert — each attribute lands in the code space its cohort filter compares.
assert isinstance(conditioning, HistoricConditioning)
assert conditioning.property_type == "0"
assert conditioning.built_form == "2"
assert conditioning.wall_construction == 4
assert conditioning.roof_form == "pitched"
assert conditioning.construction_age_band == "C"
assert conditioning.main_fuel == 26
assert conditioning.total_floor_area_m2 == 84.0
def test_roof_descriptions_resolve_to_form_families():
# Arrange / Act / Assert — the FORM half of the description (before the
# comma) maps to a family, because the API's roof_construction codes group
# that way (7,974-cert co-occurrence sweep: 1=Flat 98%, 4/5/8=Pitched
# 88-99%, 3=another dwelling above 100%; 7/9=another premises above per
# the #1452 suppression fix). Insulation state is volatile and ignored.
cases = {
"Pitched, 250 mm loft insulation": "pitched",
"Pitched, no insulation (assumed)": "pitched",
"Flat, insulated (assumed)": "flat",
"(another dwelling above)": "dwelling_above",
"(another premises above)": "premises_above",
}
for text, family in cases.items():
record = _hist(roof_description=text)
assert conditioning_from_historic(record).roof_form == family, text
# Unpinned forms (roof rooms, thatched) must not guess.
assert conditioning_from_historic(
_hist(roof_description="Roof room(s), insulated")
).roof_form is None
assert conditioning_from_historic(
_hist(roof_description="Thatched, with additional insulation")
).roof_form is None
def test_legacy_register_fuel_descriptions_resolve():
# Arrange / Act / Assert — the old register lodged pre-RdSAP-17 fuels with
# a "backwards compatibility" rider or a SAP-style prefix; they name the
# same physical fuels (dominant in the pre-2012 dump: a 65-shard scan found
# 636/663 unresolved values were these variants).
cases = {
"mains gas - this is for backwards compatibility only and should not be used": 26,
"electricity - this is for backwards compatibility only and should not be used": 29,
"LPG - this is for backwards compatibility only and should not be used": 27,
"oil - this is for backwards compatibility only and should not be used": 28,
"Gas: mains gas": 26,
"Electricity: electricity, unspecified tariff": 29,
"dual fuel - mineral + wood": 10,
}
for text, code in cases.items():
record = _hist(main_fuel=text)
assert conditioning_from_historic(record).main_fuel == code, text
def test_unresolvable_values_degrade_to_none():
# Arrange — junk and blanks must never guess a code.
record = _hist(
property_type="Castle",
built_form="Not Recorded",
walls_description="Average thermal transmittance 0.3 W/m²K",
construction_age_band="INVALID!",
main_fuel="To be used only when there is no heating system",
total_floor_area="",
)
# Act
conditioning = conditioning_from_historic(record)
# Assert
assert conditioning.property_type is None
assert conditioning.built_form is None
assert conditioning.wall_construction is None
assert conditioning.roof_form is None
assert conditioning.construction_age_band is None
assert conditioning.main_fuel is None
assert conditioning.total_floor_area_m2 is None

View file

@ -57,3 +57,18 @@ def test_postcode_is_frozen() -> None:
# act / assert # act / assert
with pytest.raises(dataclasses.FrozenInstanceError): with pytest.raises(dataclasses.FrozenInstanceError):
postcode.value = "OTHER" # type: ignore[misc] postcode.value = "OTHER" # type: ignore[misc]
@pytest.mark.parametrize(
"raw",
["sw1a 1aa", "M1 1AA", "b33 8th", "cr2 6xh", "dn55 1pt", "ec1a 1bb", "w1a 0ax"],
)
def test_valid_uk_postcode_formats_are_valid(raw: str) -> None:
# act / assert — a pure format check on the canonical value, no network.
assert Postcode(raw).is_valid() is True
@pytest.mark.parametrize("raw", ["", " ", "NONSENSE", "12345", "SW1A 1A", "ZZ"])
def test_malformed_or_empty_postcodes_are_invalid(raw: str) -> None:
# act / assert
assert Postcode(raw).is_valid() is False

364626
tests/fixtures/expired_prediction_pairs.json vendored Normal file

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,45 @@
import gzip
from collections.abc import Iterator
import pytest
from botocore.exceptions import ClientError
from moto import mock_aws
from infrastructure.s3.gzip_csv_s3_client import GzipCsvS3Client
from tests.infrastructure import make_boto_client
BUCKET = "gzip-csv-bucket"
@pytest.fixture
def gzip_client() -> Iterator[GzipCsvS3Client]:
with mock_aws():
boto_client = make_boto_client("s3")
boto_client.create_bucket(Bucket=BUCKET)
yield GzipCsvS3Client(boto_client, BUCKET)
def test_read_csv_gz_decodes_gzipped_csv_into_dataframe(
gzip_client: GzipCsvS3Client,
) -> None:
# arrange
csv = "ADDRESS,UPRN\n47 GORDON ROAD,100\n48 GORDON ROAD,200\n"
gzip_client.put_object("historical_epc/AB338AL/data.csv.gz", gzip.compress(csv.encode()))
# act
df = gzip_client.read_csv_gz("historical_epc/AB338AL/data.csv.gz")
# assert
assert list(df.columns) == ["ADDRESS", "UPRN"]
assert df.shape == (2, 2)
assert df["ADDRESS"].tolist() == ["47 GORDON ROAD", "48 GORDON ROAD"]
def test_read_csv_gz_raises_no_such_key_when_object_missing(
gzip_client: GzipCsvS3Client,
) -> None:
# act / assert — a missing object surfaces as a ClientError; translating that
# into a domain miss is the Repo's job, not the client's.
with pytest.raises(ClientError) as exc_info:
gzip_client.read_csv_gz("historical_epc/ZZ999ZZ/data.csv.gz")
assert exc_info.value.response["Error"]["Code"] == "NoSuchKey"

View file

@ -21,7 +21,11 @@ from repositories.epc.epc_postgres_repository import EpcSaveRequest
from repositories.plan.plan_repository import PlanRepository, PlanSaveRequest from repositories.plan.plan_repository import PlanRepository, PlanSaveRequest
from repositories.product.product_repository import ProductRepository from repositories.product.product_repository import ProductRepository
from repositories.property_baseline.property_baseline_repository import PropertyBaselineRepository from repositories.property_baseline.property_baseline_repository import PropertyBaselineRepository
from repositories.epc.epc_repository import EpcRepository, EpcSource from repositories.epc.epc_repository import (
PREDICTED_SLOT_SOURCES,
EpcRepository,
EpcSource,
)
from repositories.property.property_repository import ( from repositories.property.property_repository import (
PropertyIdentityInsert, PropertyIdentityInsert,
PropertyRepository, PropertyRepository,
@ -81,6 +85,7 @@ class FakePropertyRepo(PropertyRepository):
class FakeEpcRepo(EpcRepository): class FakeEpcRepo(EpcRepository):
def __init__(self, by_property: Optional[dict[int, EpcPropertyData]] = None) -> None: def __init__(self, by_property: Optional[dict[int, EpcPropertyData]] = None) -> None:
self.saved: list[tuple[EpcPropertyData, Optional[int]]] = [] self.saved: list[tuple[EpcPropertyData, Optional[int]]] = []
self.sources: list[EpcSource] = []
self._by_property = by_property or {} self._by_property = by_property or {}
# Predicted EPCs live in their own slot, coexisting with lodged (ADR-0031). # Predicted EPCs live in their own slot, coexisting with lodged (ADR-0031).
self._predicted_by_property: dict[int, EpcPropertyData] = {} self._predicted_by_property: dict[int, EpcPropertyData] = {}
@ -93,10 +98,11 @@ class FakeEpcRepo(EpcRepository):
source: EpcSource = "lodged", source: EpcSource = "lodged",
) -> int: ) -> int:
self.saved.append((data, property_id)) self.saved.append((data, property_id))
self.sources.append(source)
if property_id is not None: if property_id is not None:
slot = ( slot = (
self._predicted_by_property self._predicted_by_property
if source == "predicted" if source in PREDICTED_SLOT_SOURCES
else self._by_property else self._by_property
) )
slot[property_id] = data slot[property_id] = data

View file

@ -202,3 +202,119 @@ def test_a_lodged_epc_is_not_predicted_over() -> None:
assert comparables_repo.searched == [] assert comparables_repo.searched == []
assert epc_repo.get_for_property(10) == lodged assert epc_repo.get_for_property(10) == lodged
assert epc_repo.get_predicted_for_property(10) is None assert epc_repo.get_predicted_for_property(10) is None
def _historic(**overrides: str):
import dataclasses
from datatypes.epc.domain.historic_epc import HistoricEpc
fields = {f.name: "" for f in dataclasses.fields(HistoricEpc)}
fields.update(overrides)
return HistoricEpc(**fields)
class _FakeHistoricReader:
def __init__(self, record: Optional[Any]) -> None:
self._record = record
self.lookups: list[tuple[str, str]] = []
def record_for_uprn(self, uprn: str, postcode: str) -> Optional[Any]:
self.lookups.append((uprn, postcode))
return self._record
def test_expired_historic_epc_conditions_prediction_and_labels_the_source() -> None:
# Arrange — EPC-less, NO landlord overrides (property type unresolved), but
# the historic backup holds this UPRN's expired cert: its stable attributes
# supply the cohort gate and the persisted source is "expired" (ADR-0054).
epc_repo = FakeEpcRepo()
uow = FakeUnitOfWork(
property=FakePropertyRepo({10: _property(uprn=12345)}),
epc=epc_repo,
solar=FakeSolarRepo(),
)
comparables_repo = _FakeComparablesRepo(_cohort())
historic_reader = _FakeHistoricReader(_historic(property_type="House"))
orchestrator = IngestionOrchestrator(
unit_of_work=lambda: uow,
epc_fetcher=_FakeEpcFetcher(None),
geospatial_repo=_FakeGeospatialRepo(Coordinates(longitude=-0.1, latitude=51.5)),
solar_fetcher=_FakeSolarFetcher(),
comparables_repo=comparables_repo,
prediction_attributes_reader=_FakeAttributesReader(
PredictionTargetAttributes(property_type=None)
),
epc_prediction=EpcPrediction(),
historic_epc_reader=historic_reader,
)
# Act
orchestrator.run([10])
# Assert — looked up by exact UPRN + postcode; predicted via the historic
# gate; persisted to the predicted slot labelled "expired".
assert historic_reader.lookups == [("12345", "A0 0AA")]
assert comparables_repo.searched == ["A0 0AA"]
assert epc_repo.get_predicted_for_property(10) is not None
assert epc_repo.sources == ["expired"]
assert epc_repo.get_for_property(10) is None
def test_landlord_overrides_win_over_the_expired_historic_cert() -> None:
# Arrange — the landlord says Flat ("2") today; the 2009 cert said House.
# Overrides speak to current state so the target stays a flat — and the
# all-house cohort therefore yields no comparables and no prediction.
epc_repo = FakeEpcRepo()
uow = FakeUnitOfWork(
property=FakePropertyRepo({10: _property(uprn=12345)}),
epc=epc_repo,
solar=FakeSolarRepo(),
)
orchestrator = IngestionOrchestrator(
unit_of_work=lambda: uow,
epc_fetcher=_FakeEpcFetcher(None),
geospatial_repo=_FakeGeospatialRepo(Coordinates(longitude=-0.1, latitude=51.5)),
solar_fetcher=_FakeSolarFetcher(),
comparables_repo=_FakeComparablesRepo(_cohort()),
prediction_attributes_reader=_FakeAttributesReader(
PredictionTargetAttributes(property_type="2")
),
epc_prediction=EpcPrediction(),
historic_epc_reader=_FakeHistoricReader(_historic(property_type="House")),
)
# Act
orchestrator.run([10])
# Assert — the historic "House" did not overwrite the landlord's flat.
assert epc_repo.get_predicted_for_property(10) is None
def test_no_historic_record_keeps_the_plain_predicted_source() -> None:
# Arrange — historic reader wired but the shard has no row for this UPRN.
epc_repo = FakeEpcRepo()
uow = FakeUnitOfWork(
property=FakePropertyRepo({10: _property(uprn=12345)}),
epc=epc_repo,
solar=FakeSolarRepo(),
)
orchestrator = IngestionOrchestrator(
unit_of_work=lambda: uow,
epc_fetcher=_FakeEpcFetcher(None),
geospatial_repo=_FakeGeospatialRepo(Coordinates(longitude=-0.1, latitude=51.5)),
solar_fetcher=_FakeSolarFetcher(),
comparables_repo=_FakeComparablesRepo(_cohort()),
prediction_attributes_reader=_FakeAttributesReader(
PredictionTargetAttributes(property_type="0")
),
epc_prediction=EpcPrediction(),
historic_epc_reader=_FakeHistoricReader(None),
)
# Act
orchestrator.run([10])
# Assert — an ordinary prediction, labelled "predicted".
assert epc_repo.get_predicted_for_property(10) is not None
assert epc_repo.sources == ["predicted"]

View file

@ -80,3 +80,47 @@ def test_a_lodged_only_property_has_no_predicted_epc(db_engine: Engine) -> None:
repo = EpcPostgresRepository(session) repo = EpcPostgresRepository(session)
assert repo.get_predicted_for_property(9) is None assert repo.get_predicted_for_property(9) is None
assert repo.get_for_property(9) == epc assert repo.get_for_property(9) == epc
def test_an_expired_source_epc_lives_in_the_predicted_slot(db_engine: Engine) -> None:
# Arrange — an Expired-Enhanced Prediction is persisted with source="expired"
# (ADR-0054): enhanced by an expired historic observation, not totally
# predicted — but it occupies the same predicted slot.
epc = _epc()
with Session(db_engine) as session:
repo = EpcPostgresRepository(session)
repo.save(epc, property_id=11, source="expired")
session.commit()
# Act / Assert — readable through the predicted slot; the lodged slot is empty.
with Session(db_engine) as session:
repo = EpcPostgresRepository(session)
assert repo.get_predicted_for_property(11) == epc
assert repo.get_predicted_for_properties([11]) == {11: epc}
assert repo.get_for_property(11) is None
def test_saving_predicted_over_expired_replaces_the_slot_without_stranding(
db_engine: Engine,
) -> None:
# Arrange — a re-ingestion can flip the slot's flavour (expired -> predicted
# or back); the slot holds ONE row, never a stranded stale sibling.
from sqlmodel import select
from infrastructure.postgres.epc_property_table import EpcPropertyModel
epc = _epc()
with Session(db_engine) as session:
repo = EpcPostgresRepository(session)
repo.save(epc, property_id=13, source="expired")
repo.save(epc, property_id=13, source="predicted")
session.commit()
# Act
with Session(db_engine) as session:
rows = session.exec(
select(EpcPropertyModel).where(EpcPropertyModel.property_id == 13)
).all()
# Assert — exactly one slot row remains, the latest flavour.
assert [r.source for r in rows] == ["predicted"]

View file

@ -0,0 +1,164 @@
"""HistoricEpcResolver composes the repository with the address matcher.
Exercised against a fake in-memory repository (a dict of postcode -> records),
so the resolver's composition is tested with no S3 and no network — the matcher
and the repo each have their own tests.
"""
from __future__ import annotations
import dataclasses
from datatypes.epc.domain.historic_epc import HistoricEpc
from datatypes.epc.domain.historic_epc_matching import HistoricEpcMatches
from domain.postcode import Postcode
from repositories.historic_epc.historic_epc_repository import HistoricEpcRepository
from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver
def _hist(address: str, uprn: str, lodgement_date: str = "") -> HistoricEpc:
fields = {f.name: "" for f in dataclasses.fields(HistoricEpc)}
fields["address"] = address
fields["uprn"] = uprn
fields["lodgement_date"] = lodgement_date
return HistoricEpc(**fields)
class _FakeRepo(HistoricEpcRepository):
def __init__(self, by_postcode: dict[str, list[HistoricEpc]]) -> None:
self._by_postcode = by_postcode
def get_for_postcode(self, postcode: Postcode) -> list[HistoricEpc]:
return self._by_postcode.get(str(postcode), [])
def test_match_composes_repo_and_matcher_into_scored_matches():
# Arrange
repo = _FakeRepo(
{
"AB338AL": [
_hist("48 GORDON ROAD", "200"),
_hist("47 GORDON ROAD", "100"),
]
}
)
resolver = HistoricEpcResolver(repo)
# Act
result = resolver.match("47 Gordon Road", "AB33 8AL")
# Assert
assert isinstance(result, HistoricEpcMatches)
assert result.postcode == "AB338AL"
assert len(result.matches) == 2
top = result.top()
assert top is not None
assert top.record.address == "47 GORDON ROAD"
def test_resolve_uprn_returns_unambiguous_match():
# Arrange
repo = _FakeRepo(
{
"AB338AL": [
_hist("47 GORDON ROAD", "100"),
_hist("48 GORDON ROAD", "200"),
]
}
)
resolver = HistoricEpcResolver(repo)
# Act
result = resolver.resolve_uprn("47 Gordon Road", "AB33 8AL")
# Assert
assert result is not None
uprn, address, score = result
assert uprn == "100"
assert address == "47 GORDON ROAD"
assert score > 0
def test_resolve_uprn_is_none_when_postcode_has_no_data():
# Arrange — valid postcode, but the backup has no shard for it.
resolver = HistoricEpcResolver(_FakeRepo({}))
# Act / Assert
assert resolver.resolve_uprn("47 Gordon Road", "AB33 8AL") is None
def test_resolve_uprn_is_none_on_ambiguous_tie():
# Arrange — two identical addresses with different UPRNs share rank-1.
repo = _FakeRepo(
{
"AB338AL": [
_hist("47 GORDON ROAD", "100"),
_hist("47 GORDON ROAD", "200"),
]
}
)
# Act / Assert
assert HistoricEpcResolver(repo).resolve_uprn("47 Gordon Road", "AB33 8AL") is None
def test_record_for_uprn_returns_the_exact_uprn_match():
# Arrange — the prediction path knows the target's UPRN; the lookup must be
# exact equality, never the fuzzy address matcher (ADR-0054).
repo = _FakeRepo(
{
"AB338AL": [
_hist("47 GORDON ROAD", "100"),
_hist("48 GORDON ROAD", "200"),
]
}
)
# Act
record = HistoricEpcResolver(repo).record_for_uprn("200", "AB33 8AL")
# Assert
assert record is not None
assert record.address == "48 GORDON ROAD"
def test_record_for_uprn_is_none_when_uprn_not_in_shard():
# Arrange
repo = _FakeRepo({"AB338AL": [_hist("47 GORDON ROAD", "100")]})
# Act / Assert — a miss is the normal outcome of a best-effort lookup.
assert HistoricEpcResolver(repo).record_for_uprn("999", "AB33 8AL") is None
def test_record_for_uprn_picks_latest_lodgement_when_relodged():
# Arrange — the register re-lodged this UPRN; the newer observation wins.
repo = _FakeRepo(
{
"AB338AL": [
_hist("47 GORDON ROAD", "100", lodgement_date="2008-01-15"),
_hist("47 GORDON ROAD, DORRIDGE", "100", lodgement_date="2010-06-02"),
]
}
)
# Act
record = HistoricEpcResolver(repo).record_for_uprn("100", "AB33 8AL")
# Assert
assert record is not None
assert record.lodgement_date == "2010-06-02"
def test_resolve_uprn_is_none_when_all_scores_zero():
# Arrange — no candidate shares the user's building number => all hard-zero.
repo = _FakeRepo(
{
"AB338AL": [
_hist("999 ELSEWHERE", "100"),
_hist("888 ELSEWHERE", "200"),
]
}
)
# Act / Assert
assert HistoricEpcResolver(repo).resolve_uprn("47 Gordon Road", "AB33 8AL") is None

View file

@ -0,0 +1,195 @@
"""HistoricEpcS3Repository reads per-postcode shards of the old-EPC backup.
A reference-data lookup, not a Fetcher (ADR-0011): no live EPC API call. The
adapter takes a normalised ``Postcode`` and reads
``historical_epc/{POSTCODE}/data.csv.gz`` through an injected
``GzipCsvS3Client`` (infrastructure/s3) never the ``utils.s3`` free functions.
The client is wrapped over a tiny fake boto client (rather than moto) so the
tests can assert the *exact* S3 key the repo builds and inject a non-missing
``ClientError``; the real gzip/CSV decode in ``GzipCsvS3Client`` still runs.
"""
from __future__ import annotations
import csv
import dataclasses
import gzip
from io import BytesIO, StringIO
from typing import Any, Optional
from unittest.mock import patch
import pytest
from botocore.exceptions import ClientError
from datatypes.epc.domain.historic_epc import HistoricEpc
from domain.postcode import Postcode
from infrastructure.s3.gzip_csv_s3_client import GzipCsvS3Client
from repositories.historic_epc.historic_epc_repository import PostcodeNotFound
from repositories.historic_epc.historic_epc_s3_repository import (
HistoricEpcS3Repository,
)
# HistoricEpc requires every CSV column; derive the (upper-cased) column list
# straight from the dataclass so it can never drift from the domain type.
_COLS = [f.name.upper() for f in dataclasses.fields(HistoricEpc)]
_ROOT_PREFIX = "historical_epc"
def _gzip_csv(rows: list[tuple[str, str]]) -> bytes:
"""A gzipped CSV carrying every HistoricEpc column; only ADDRESS/UPRN set."""
buffer = StringIO()
writer = csv.DictWriter(buffer, fieldnames=_COLS)
writer.writeheader()
for address, uprn in rows:
row = {col: "" for col in _COLS}
row["ADDRESS"] = address
row["UPRN"] = uprn
writer.writerow(row)
return gzip.compress(buffer.getvalue().encode())
class _FakeBoto:
"""A minimal stand-in for a boto3 S3 client: serves one canned object (or
raises a chosen ``ClientError``) and records the ``(Bucket, Key)`` of every
request, so the repo tests can assert the exact S3 location without a live
bucket."""
def __init__(
self, *, body: Optional[bytes] = None, error_code: Optional[str] = None
) -> None:
self._body = body
self._error_code = error_code
self.calls: list[tuple[str, str]] = []
def get_object(self, *, Bucket: str, Key: str) -> dict[str, Any]:
self.calls.append((Bucket, Key))
if self._error_code is not None:
raise ClientError(
{"Error": {"Code": self._error_code, "Message": "x"}}, "GetObject"
)
return {"Body": BytesIO(self._body or b"")}
@property
def requested_keys(self) -> list[str]:
return [key for _bucket, key in self.calls]
def _repo(boto: _FakeBoto) -> HistoricEpcS3Repository:
client = GzipCsvS3Client(boto, "retrofit-data-dev")
return HistoricEpcS3Repository(client, _ROOT_PREFIX)
def test_get_for_postcode_maps_shard_rows_to_historic_epc_records():
# Arrange
repo = _repo(_FakeBoto(body=_gzip_csv([("47 GORDON ROAD", "100")])))
# Act
records = repo.get_for_postcode(Postcode("AB33 8AL"))
# Assert
assert len(records) == 1
assert isinstance(records[0], HistoricEpc)
assert records[0].address == "47 GORDON ROAD"
assert records[0].uprn == "100"
def test_builds_s3_key_from_postcode_and_root_prefix():
# Arrange
boto = _FakeBoto(body=_gzip_csv([("47 GORDON ROAD", "100")]))
repo = _repo(boto)
# Act — the Postcode value object has already normalised casing/spacing.
repo.get_for_postcode(Postcode("ab33 8al"))
# Assert
assert boto.requested_keys == ["historical_epc/AB338AL/data.csv.gz"]
def test_non_empty_postcode_with_no_stored_object_returns_empty_list():
# Arrange — a postcode whose shard does not exist in S3.
repo = _repo(_FakeBoto(error_code="NoSuchKey"))
# Act
records = repo.get_for_postcode(Postcode("AB33 8AL"))
# Assert — a miss is the normal, expected outcome, not an exception.
assert records == []
def test_non_missing_read_error_propagates():
# Arrange — an error that is NOT a missing object must not be swallowed.
repo = _repo(_FakeBoto(error_code="AccessDenied"))
# Act / Assert
with pytest.raises(ClientError):
repo.get_for_postcode(Postcode("AB33 8AL"))
def test_empty_postcode_raises_postcode_not_found():
# Arrange — Postcode normalises whitespace away, leaving an empty key.
repo = _repo(_FakeBoto())
# Act / Assert — an unusable key, distinct from a non-empty absent postcode.
with pytest.raises(PostcodeNotFound):
repo.get_for_postcode(Postcode(" "))
def test_malformed_postcode_raises_postcode_not_found():
# Arrange — a non-empty but malformed postcode can't key a real shard.
repo = _repo(_FakeBoto())
# Act / Assert
with pytest.raises(PostcodeNotFound):
repo.get_for_postcode(Postcode("NONSENSE"))
def test_shard_with_unexpected_extra_column_still_maps():
# Arrange — the upstream dump can gain columns the domain type doesn't
# know; mapping must ignore them rather than explode at construction.
buffer = StringIO()
writer = csv.DictWriter(buffer, fieldnames=_COLS + ["SOMETHING_NEW"])
writer.writeheader()
row = {col: "" for col in _COLS}
row["ADDRESS"] = "47 GORDON ROAD"
row["UPRN"] = "100"
row["SOMETHING_NEW"] = "surprise"
writer.writerow(row)
body = gzip.compress(buffer.getvalue().encode())
repo = _repo(_FakeBoto(body=body))
# Act
records = repo.get_for_postcode(Postcode("AB33 8AL"))
# Assert
assert records[0].address == "47 GORDON ROAD"
assert records[0].uprn == "100"
def test_uprn_trailing_dot_zero_is_stripped():
# Arrange — pandas reads an integer UPRN column written with a decimal as
# float, so the cell stringifies to "151020766.0"; the domain UPRN must be
# the bare integer string.
repo = _repo(_FakeBoto(body=_gzip_csv([("47 GORDON ROAD", "151020766.0")])))
# Act
records = repo.get_for_postcode(Postcode("AB33 8AL"))
# Assert
assert records[0].uprn == "151020766"
def test_with_default_s3_client_threads_bucket_and_key_from_s3_root():
# Arrange — the factory parses ``s3://bucket/prefix`` into the client's
# bucket and the repo's root prefix, so a read lands at the right location.
boto = _FakeBoto(error_code="NoSuchKey")
with patch("boto3.client", return_value=boto):
repo = HistoricEpcS3Repository.with_default_s3_client(
"s3://my-bucket/some/prefix/"
)
# Act
repo.get_for_postcode(Postcode("ab33 8al"))
# Assert — bucket from the URI authority, key from the URI path + postcode.
assert boto.calls == [("my-bucket", "some/prefix/AB338AL/data.csv.gz")]

View file

@ -0,0 +1,149 @@
"""Pure pair-selection and aggregation logic of the ADR-0054 pairs harness."""
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,
)
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 _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,
)
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_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
arm = aggregate(scores)
# 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([_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 — 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
def test_ladder_simulation_engages_only_with_enough_matches():
# Arrange — a 6-strong base cohort: 5 band-C (engages at k=5), then within
# the band-C survivors only 2 on fuel 26 (relaxes), and 5 within ±5% of
# 100 m² (engages on the un-shrunk cohort).
from scripts.expired_prediction_pairs_harness import simulate_conditioning_ladder
from tests.domain.epc_prediction.test_comparable_properties import _comparable
base = [
_comparable(
property_type="0",
certificate_number=f"C{i}",
construction_age_band="C",
main_fuel=26 if i < 2 else 29,
total_floor_area_m2=100.0 + i,
)
for i in range(5)
] + [
_comparable(
property_type="0",
certificate_number="G0",
construction_age_band="G",
main_fuel=26,
total_floor_area_m2=100.0,
)
]
# Act
steps = simulate_conditioning_ladder(
base, age_band="C", main_fuel=26, total_floor_area_m2=100.0
)
# Assert — age engaged (5 matches), fuel relaxed (2 < 5 within band-C
# survivors), TFA engaged (all 5 survivors within the band).
age = steps["construction_age_band"]
fuel = steps["main_fuel"]
tfa = steps["total_floor_area"]
assert age is not None and age.engaged and age.matches == 5
assert fuel is not None and not fuel.engaged and fuel.matches == 2
assert tfa is not None and tfa.engaged and tfa.matches == 5
def test_ladder_simulation_skips_unresolved_attributes():
from scripts.expired_prediction_pairs_harness import simulate_conditioning_ladder
steps = simulate_conditioning_ladder(
[], age_band=None, main_fuel=None, total_floor_area_m2=None
)
assert steps == {
"roof_form": None,
"construction_age_band": None,
"main_fuel": None,
"total_floor_area": None,
}