An expired historic cert conditions ingestion's prediction and labels it 🟥

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

orchestration/ingestion_orchestrator.py

@@ -5,6 +5,7 @@ from dataclasses import dataclass
 from typing import Any, Optional, Protocol
 
 from datatypes.epc.domain.epc_property_data import EpcPropertyData
+from datatypes.epc.domain.historic_epc import HistoricEpc
 from domain.epc_prediction.comparable_properties import (
     ComparableProperty,
     select_comparables,
@@ -40,6 +41,13 @@ class PredictionAttributesReader(Protocol):
     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):
     """The slice of the Google Solar client Ingestion needs (e.g. GoogleSolarApiClient)."""
 
@@ -98,6 +106,7 @@ class IngestionOrchestrator:
         comparables_repo: Optional[ComparablesRepo] = None,
         prediction_attributes_reader: Optional[PredictionAttributesReader] = None,
         epc_prediction: Optional[EpcPrediction] = None,
+        historic_epc_reader: Optional[HistoricEpcReader] = None,
     ) -> None:
         self._unit_of_work = unit_of_work
         self._epc_fetcher = epc_fetcher
@@ -110,6 +119,7 @@ class IngestionOrchestrator:
         self._comparables_repo = comparables_repo
         self._prediction_attributes_reader = prediction_attributes_reader
         self._epc_prediction = epc_prediction
+        self._historic_epc_reader = historic_epc_reader
 
     def run(self, property_ids: list[int]) -> None:
         preps = self._prepare(property_ids)

tests/orchestration/fakes.py

@@ -85,6 +85,7 @@ class FakePropertyRepo(PropertyRepository):
 class FakeEpcRepo(EpcRepository):
     def __init__(self, by_property: Optional[dict[int, EpcPropertyData]] = None) -> None:
         self.saved: list[tuple[EpcPropertyData, Optional[int]]] = []
+        self.sources: list[EpcSource] = []
         self._by_property = by_property or {}
         # Predicted EPCs live in their own slot, coexisting with lodged (ADR-0031).
         self._predicted_by_property: dict[int, EpcPropertyData] = {}
@@ -97,6 +98,7 @@ class FakeEpcRepo(EpcRepository):
         source: EpcSource = "lodged",
     ) -> int:
         self.saved.append((data, property_id))
+        self.sources.append(source)
         if property_id is not None:
             slot = (
                 self._predicted_by_property

tests/orchestration/test_ingestion_prediction.py

@@ -202,3 +202,119 @@ def test_a_lodged_epc_is_not_predicted_over() -> None:
     assert comparables_repo.searched == []
     assert epc_repo.get_for_property(10) == lodged
     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"]