Merge pull request #1549 from Hestia-Homes/feature/expired-predicted-source

feat(modelling_e2e): condition prediction on the expired Historic EPC (ADR-0054)

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"]

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

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-<stage>`
+  # 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
   }
 }
 

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

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"