diff --git a/applications/modelling_e2e/Dockerfile b/applications/modelling_e2e/Dockerfile index fa13f8efa..1b7ee0804 100644 --- a/applications/modelling_e2e/Dockerfile +++ b/applications/modelling_e2e/Dockerfile @@ -19,6 +19,19 @@ COPY tests/__init__.py tests/__init__.py COPY tests/orchestration/__init__.py tests/orchestration/__init__.py COPY tests/orchestration/fakes.py tests/orchestration/fakes.py +# The Historic EPC S3 repo (ADR-0054) maps shard rows through +# datatypes/epc/domain/historic_epc_matching.py, which reaches back into the +# legacy address matcher. Copied file-by-file rather than `COPY backend/ backend/` +# — backend/ is the whole legacy engine, and the closure needs only these. Their +# third-party deps (pandas, requests) are already in requirements.txt. +COPY utils/__init__.py utils/__init__.py +COPY utils/pandas_utils.py utils/pandas_utils.py +COPY backend/__init__.py backend/__init__.py +COPY backend/utils/__init__.py backend/utils/__init__.py +COPY backend/utils/addressMatch.py backend/utils/addressMatch.py +COPY backend/address2UPRN/__init__.py backend/address2UPRN/__init__.py +COPY backend/address2UPRN/scoring.py backend/address2UPRN/scoring.py + COPY applications/ applications/ CMD ["applications.modelling_e2e.handler.handler"] diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index 37161dee8..3442cb362 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -54,6 +54,11 @@ from domain.epc_prediction.comparable_properties import ( 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_target import ( PredictionTarget, build_prediction_target, @@ -68,6 +73,7 @@ from domain.sap10_calculator.calculator import Sap10Calculator from domain.tasks.subtasks import SubTask, SubTaskFailure from domain.tasks.tasks import Source from harness.console import run_modelling +from orchestration.ingestion_orchestrator import HistoricEpcReader from orchestration.task_orchestrator import TaskOrchestrator from orchestration.property_baseline_orchestrator import ( PropertyBaselineOrchestrator, @@ -93,6 +99,11 @@ from repositories.comparable_properties.epc_comparable_properties_repository imp SkippedCohortCert, ) from repositories.epc.epc_postgres_repository import EpcPostgresRepository, EpcSaveRequest +from repositories.epc.epc_repository import EpcSource +from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver +from repositories.historic_epc.historic_epc_s3_repository import ( + HistoricEpcS3Repository, +) from repositories.plan.plan_repository import PlanSaveRequest from repositories.geospatial.geospatial_s3_repository import ( GeospatialS3Repository, @@ -177,6 +188,12 @@ class _PropertyWrite: solar: Optional[_SolarWrite] plan: Plan has_recommendations: bool + # Which flavour of the predicted slot this prediction belongs in: "expired" + # when an expired Historic EPC conditioned it, "predicted" otherwise + # (ADR-0054). Only read when `predicted_epc_is_new` — a prediction reused + # from the DB is never re-written, so it keeps whatever source it was + # stored under. + predicted_source: EpcSource = "predicted" def _flush_writes(engine: Engine, writes: list[_PropertyWrite]) -> None: @@ -196,7 +213,12 @@ def _flush_writes(engine: Engine, writes: list[_PropertyWrite]) -> None: if w.lodged_epc is not None and w.lodged_epc_is_new ] predicted_requests = [ - EpcSaveRequest(w.predicted_epc, property_id=w.property_id, portfolio_id=w.portfolio_id, source="predicted") + EpcSaveRequest( + w.predicted_epc, + property_id=w.property_id, + portfolio_id=w.portfolio_id, + source=w.predicted_source, + ) for w in writes if w.predicted_epc is not None and w.predicted_epc_is_new ] @@ -323,6 +345,21 @@ def _solar_min_request_interval_seconds() -> float: return _DEFAULT_SOLAR_MIN_REQUEST_INTERVAL_SECONDS +def _historic_epc_reader() -> Optional[HistoricEpcReader]: + """The expired-Historic-EPC backup (ADR-0054), read from the S3 root in + ``HISTORIC_EPC_S3_ROOT`` — or None, which leaves Expired-Enhanced Prediction + off and every prediction plain "predicted", exactly as before. + + Gated on the env var rather than defaulting to ``DEFAULT_S3_ROOT`` because + that constant names the *dev* bucket: a prod lambda silently reading dev data + is worse than the feature staying off. Terraform sets it per environment. + """ + root = os.environ.get("HISTORIC_EPC_S3_ROOT") + if not root: + return None + return HistoricEpcResolver(HistoricEpcS3Repository.with_default_s3_client(root)) + + def _solar_insights_for( solar_client: GoogleSolarApiClient, spatial: Optional[SpatialReference] ) -> Optional[dict[str, Any]]: @@ -357,18 +394,28 @@ def _predict_epc( postcode: str, portfolio_id: int, attributes_reader: OverrideBackedPredictionAttributesReader, + historic_epc_reader: Optional[HistoricEpcReader], coordinates: Optional[Coordinates], cohort_for: Callable[[str], list[ComparableProperty]], broaden: Callable[[PredictionTarget], list[ComparableProperty]], predictor: EpcPrediction, -) -> EpcPropertyData: +) -> tuple[EpcPropertyData, EpcSource]: """Synthesise an EpcPropertyData for an EPC-less property from its postcode - cohort (EPC Prediction Path 3, ADR-0031). + cohort (EPC Prediction Path 3, ADR-0031), paired with the source flavour to + persist it under. When the property's own postcode holds no same-type comparables (a sparse postcode — e.g. the only flat among houses), the cohort is broadened to the real unit postcodes physically nearest it (``broaden``) before giving up. + An expired Historic EPC — the pre-2012 certificate the current gov API no + longer serves — conditions the prediction where one exists for this UPRN: its + stable attributes fill the gaps Landlord Overrides left, and the observed age + band, fuel and floor area narrow the cohort. Landlord Overrides speak to + *current* state, so they always win where both speak (ADR-0054). A prediction + so conditioned is persisted as "expired" rather than "predicted", which is + what lets reporting tell "no EPC at all" apart from "only an expired one". + Raises a specific ``PropertyNotModellableError`` subclass — naming the cause and carrying the property's identity — when it cannot predict: property_type unresolved, an empty same-type cohort, or a degenerate (no MAIN part) @@ -379,6 +426,14 @@ def _predict_epc( identity = PropertyIdentity( portfolio_id=portfolio_id, postcode=postcode, address="", uprn=uprn ) + record = ( + historic_epc_reader.record_for_uprn(str(uprn), postcode) + if historic_epc_reader is not None + else None + ) + conditioning = conditioning_from_historic(record) if record is not None else None + if conditioning is not None: + attributes = attributes_with_historic_fallback(attributes, conditioning) target = build_prediction_target(identity, coordinates, attributes) if target is None: raise UnresolvedPropertyTypeError( @@ -389,6 +444,8 @@ def _predict_epc( property_type=attributes.property_type, built_form=attributes.built_form, ) + if conditioning is not None: + target = target_with_conditioning(target, conditioning) comparables = select_comparables(target, cohort_for(target.postcode)) broadened = False if not comparables.members: @@ -416,7 +473,7 @@ def _predict_epc( property_type=target.property_type, cohort_size=len(comparables.members), ) - return predicted + return predicted, ("expired" if conditioning is not None else "predicted") @task_handler( @@ -480,6 +537,7 @@ def handler( widen_nearby_postcodes=PostcodesIoClient(radius_m=3000, limit=60), ) predictor = EpcPrediction() + historic_epc_reader = _historic_epc_reader() def _get_cohort(postcode: str) -> list[ComparableProperty]: if postcode not in _cohort_cache: @@ -588,6 +646,7 @@ def handler( overrides = overlays_from(resolved_overrides) predicted_epc: Optional[EpcPropertyData] = None predicted_epc_is_new = False + predicted_source: EpcSource = "predicted" if epc is not None: logger.info(f"property={pid} lodged EPC found") @@ -610,12 +669,13 @@ def handler( ) predicted_epc = stored_predicted else: - predicted_epc = _predict_epc( + predicted_epc, predicted_source = _predict_epc( property_id=pid, uprn=uprn, postcode=postcode, portfolio_id=portfolio_id, attributes_reader=prediction_attrs_reader, + historic_epc_reader=historic_epc_reader, coordinates=coordinates, cohort_for=_get_cohort, broaden=_broaden, @@ -701,6 +761,7 @@ def handler( solar=solar_write, plan=plan, has_recommendations=bool(plan.measures), + predicted_source=predicted_source, ) ) logger.info(f"property={pid} queued for write") diff --git a/deployment/terraform/lambda/modelling_e2e/main.tf b/deployment/terraform/lambda/modelling_e2e/main.tf index 4be3bd2a8..9995f7fe4 100644 --- a/deployment/terraform/lambda/modelling_e2e/main.tf +++ b/deployment/terraform/lambda/modelling_e2e/main.tf @@ -24,6 +24,20 @@ locals { # truth. At N=32 → 4.0s. Consumed by the handler's throttle (see # infrastructure/solar/google_solar_api_client.py). solar_min_request_interval_seconds = var.maximum_concurrency / (0.8 * 600 / 60) + + # Root of the expired-EPC backup — the old register's final dump, sharded one + # `{POSTCODE}/data.csv.gz` per postcode under `historical_epc/` in the data bucket. + # Turns on Expired-Enhanced Prediction (ADR-0054): a prediction conditioned by the + # dwelling's expired pre-2012 certificate is persisted as source="expired" rather + # than "predicted", so reporting can tell "no EPC at all" apart from "only an + # expired one". + # + # `retrofit_sap_data_bucket_name` is — despite the name — the `retrofit-data-` + # bucket (shared/main.tf); engine and fast-api already take it from this output rather + # than rebuilding the string, so the bucket name has one source of truth. Reads need no + # IAM change: modelling_e2e_s3_read already grants GetObject + ListBucket across that + # whole bucket, which covers historical_epc/. + historic_epc_s3_root = "s3://${data.terraform_remote_state.shared.outputs.retrofit_sap_data_bucket_name}/historical_epc" } module "lambda" { @@ -55,6 +69,10 @@ module "lambda" { DATA_BUCKET = "retrofit-data-${var.stage}" SOLAR_MIN_REQUEST_INTERVAL_SECONDS = tostring(local.solar_min_request_interval_seconds) + + # Expired-Enhanced Prediction (ADR-0054). Empty ⇒ the handler never builds the + # reader and every prediction stays plain "predicted". + HISTORIC_EPC_S3_ROOT = local.historic_epc_s3_root } } diff --git a/tests/applications/modelling_e2e/test_predict_epc_errors.py b/tests/applications/modelling_e2e/test_predict_epc_errors.py index 715da6212..2eb902b3a 100644 --- a/tests/applications/modelling_e2e/test_predict_epc_errors.py +++ b/tests/applications/modelling_e2e/test_predict_epc_errors.py @@ -79,6 +79,7 @@ def test_unresolved_property_type_raises_with_property_identity() -> None: postcode="SE4 1DX", portfolio_id=796, attributes_reader=reader, + historic_epc_reader=None, coordinates=None, cohort_for=_no_cohort(), broaden=_no_broaden(), @@ -105,6 +106,7 @@ def test_resolved_type_but_empty_cohort_raises_no_same_type_comparables() -> Non postcode="SE13 5RT", portfolio_id=796, attributes_reader=reader, + historic_epc_reader=None, coordinates=None, cohort_for=_no_cohort(), broaden=_no_broaden(), @@ -137,6 +139,7 @@ def test_cohort_found_but_prediction_has_no_main_part_raises_degenerate() -> Non postcode="LS6 1AA", portfolio_id=796, attributes_reader=reader, + historic_epc_reader=None, coordinates=None, cohort_for=lambda _postcode: cohort, broaden=_no_broaden(), diff --git a/tests/applications/modelling_e2e/test_predict_epc_historic.py b/tests/applications/modelling_e2e/test_predict_epc_historic.py new file mode 100644 index 000000000..cd817e88f --- /dev/null +++ b/tests/applications/modelling_e2e/test_predict_epc_historic.py @@ -0,0 +1,188 @@ +"""``_predict_epc`` conditions its prediction on an expired Historic EPC and +labels the result "expired" (ADR-0054). + +The gov EPC API only serves certificates registered since 2012, so a dwelling +whose only certificate predates that is EPC-less to Ingestion and gets a +predicted picture. Persisting that picture as plain "predicted" loses the fact +that a real — if stale — certificate exists, which is exactly the distinction +portfolio reporting draws between "no EPC at all" and "only an expired one". + +`IngestionOrchestrator` already does this; `modelling_e2e` is the pipeline that +actually runs, and predicts through its own `_predict_epc`. These tests pin the +source label there. +""" + +from __future__ import annotations + +import dataclasses +import json +from datetime import date +from pathlib import Path +from typing import Any, Callable, Optional, cast + +from applications.modelling_e2e.handler import ( + _predict_epc, # pyright: ignore[reportPrivateUsage] +) +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 +from domain.epc_prediction.epc_prediction import EpcPrediction +from domain.epc_prediction.prediction_target import ( + PredictionTarget, + PredictionTargetAttributes, +) +from repositories.property.override_backed_prediction_attributes_reader import ( + OverrideBackedPredictionAttributesReader, +) + +_JSON_SAMPLES = Path(__file__).resolve().parents[3] / "backend/epc_api/json_samples" + + +def _epc() -> EpcPropertyData: + raw: dict[str, Any] = json.loads( + (_JSON_SAMPLES / "RdSAP-Schema-21.0.0" / "epc.json").read_text() + ) + return EpcPropertyDataMapper.from_api_response(raw) + + +def _cohort() -> list[ComparableProperty]: + # Three same-type neighbours (property_type "0"), distinct addresses so the + # dedupe keeps all three. + return [ + ComparableProperty( + epc=_epc(), + certificate_number=f"CERT-{i}", + address=f"{i} Some Street", + registration_date=date(2023, 1, i + 1), + ) + for i in range(3) + ] + + +def _historic(**overrides: str) -> HistoricEpc: + """A pre-2012 certificate. Every field defaults to "" — unset fields resolve + to None in `conditioning_from_historic`, so a test conditions on exactly the + attributes it names and nothing else.""" + fields = {f.name: "" for f in dataclasses.fields(HistoricEpc)} + fields.update(overrides) + return HistoricEpc(**fields) + + +class _HistoricReaderReturning: + """The historic-EPC backup, holding one record — or none, on a miss.""" + + def __init__(self, record: Optional[HistoricEpc]) -> None: + self._record = record + self.lookups: list[tuple[str, str]] = [] + + def record_for_uprn(self, uprn: str, postcode: str) -> Optional[HistoricEpc]: + self.lookups.append((uprn, postcode)) + return self._record + + +class _FixedAttributesReader: + def __init__(self, attributes: PredictionTargetAttributes) -> None: + self._attributes = attributes + + def attributes_for(self, property_id: int) -> PredictionTargetAttributes: + return self._attributes + + +def _reader( + attributes: PredictionTargetAttributes, +) -> OverrideBackedPredictionAttributesReader: + return cast( + OverrideBackedPredictionAttributesReader, _FixedAttributesReader(attributes) + ) + + +def _no_broaden() -> Callable[[PredictionTarget], list[ComparableProperty]]: + return lambda _target: [] + + +def _predict( + *, + attributes: PredictionTargetAttributes, + historic: Optional[_HistoricReaderReturning], +) -> tuple[EpcPropertyData, str]: + return _predict_epc( + property_id=733290, + uprn=100021987320, + postcode="SE4 1DX", + portfolio_id=796, + attributes_reader=_reader(attributes), + historic_epc_reader=historic, + coordinates=None, + cohort_for=lambda _postcode: _cohort(), + broaden=_no_broaden(), + predictor=EpcPrediction(), + ) + + +def test_prediction_conditioned_by_expired_historic_epc_is_sourced_expired() -> None: + # Arrange — an override resolves the type, and the backup also holds this + # UPRN's expired pre-2012 certificate. + historic = _HistoricReaderReturning(_historic(property_type="House")) + + # Act + _predicted, source = _predict( + attributes=PredictionTargetAttributes(property_type="0"), historic=historic + ) + + # Assert — the expired certificate is recorded in the source, not discarded, + # and it was looked up by this Property's UPRN and postcode. + assert source == "expired" + assert historic.lookups == [("100021987320", "SE4 1DX")] + + +def test_prediction_without_a_historic_record_is_sourced_predicted() -> None: + # Arrange — the backup is wired, but holds nothing for this UPRN. + # Act + _predicted, source = _predict( + attributes=PredictionTargetAttributes(property_type="0"), + historic=_HistoricReaderReturning(None), + ) + + # Assert — a blind gap-fill stays plain "predicted". + assert source == "predicted" + + +def test_unwired_historic_reader_leaves_every_prediction_predicted() -> None: + # Arrange — the feature is off (no HISTORIC_EPC_S3_ROOT in the environment). + # Act + _predicted, source = _predict( + attributes=PredictionTargetAttributes(property_type="0"), historic=None + ) + + # Assert — behaviour is exactly as before the ADR-0054 wiring. + assert source == "predicted" + + +def test_historic_epc_supplies_the_property_type_when_no_override_resolved() -> None: + # Arrange — no Landlord Override resolves a property_type, which on its own + # gates the Property out of prediction entirely (UnresolvedPropertyTypeError). + # The expired certificate observed one, so the Property becomes predictable. + # Act + _predicted, source = _predict( + attributes=PredictionTargetAttributes(property_type=None), + historic=_HistoricReaderReturning(_historic(property_type="House")), + ) + + # Assert — rescued by the historic observation rather than failing to model. + assert source == "expired" + + +def test_landlord_override_wins_over_the_expired_observation() -> None: + # Arrange — the override and the expired cert disagree on property type. The + # override speaks to *current* state, so it wins; the historic cert only + # fills gaps (ADR-0054). "0" (House) is the cohort's type — were the historic + # "Flat" to win, no same-type comparable would survive and this would raise. + # Act + _predicted, source = _predict( + attributes=PredictionTargetAttributes(property_type="0"), + historic=_HistoricReaderReturning(_historic(property_type="Flat")), + ) + + # Assert — predicted against the override's cohort, still labelled expired. + assert source == "expired"