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

Ingestion's lookup chain becomes: new EPC API -> historic backup by exact
UPRN -> plain prediction. A found record's stable attributes fill the
gaps Landlord Overrides left (overrides always win) and enrich the target
with age band / fuel / TFA band; the persisted predicted-slot row carries
source="expired". Reader unwired or shard miss -> behaviour unchanged.

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

domain/epc_prediction/historic_conditioning.py

@@ -13,7 +13,7 @@ evidence about today.
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from typing import Optional
 
 from datatypes.epc.domain.historic_epc import HistoricEpc
@@ -21,6 +21,10 @@ 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).
@@ -121,3 +125,36 @@ def conditioning_from_historic(record: HistoricEpc) -> HistoricConditioning:
         main_fuel=_main_fuel(record.main_fuel),
         total_floor_area_m2=_floor_area(record.total_floor_area),
     )
+
+
+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,
+    )

orchestration/ingestion_orchestrator.py

@@ -11,6 +11,12 @@ from domain.epc_prediction.comparable_properties import (
     select_comparables,
 )
 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 (
     PredictionTargetAttributes,
     build_prediction_target,
@@ -18,6 +24,7 @@ from domain.epc_prediction.prediction_target import (
 from domain.geospatial.coordinates import Coordinates
 from domain.geospatial.spatial_reference import SpatialReference
 from domain.property.property import PropertyIdentity
+from repositories.epc.epc_repository import EpcSource
 from repositories.geospatial.geospatial_repository import GeospatialRepository
 from repositories.unit_of_work import UnitOfWork
 
@@ -77,6 +84,9 @@ class _Fetched:
     predicted_epc: Optional[EpcPropertyData]
     solar_insights: Optional[dict[str, Any]]
     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:
@@ -159,33 +169,64 @@ class IngestionOrchestrator:
             solar_insights = self._solar_fetcher.get_building_insights(
                 coordinates.longitude, coordinates.latitude
             )
-        predicted_epc = (
-            self._predict(prep.identity, coordinates, prep.attributes)
-            if epc is None
-            else None
+        predicted_epc: Optional[EpcPropertyData] = None
+        conditioning: Optional[HistoricConditioning] = None
+        if epc is 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(
-            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(
         self,
         identity: PropertyIdentity,
         coordinates: Optional[Coordinates],
         attributes: Optional[PredictionTargetAttributes],
+        conditioning: Optional[HistoricConditioning] = None,
     ) -> Optional[EpcPropertyData]:
         """Synthesise the EPC-less Property's picture from its postcode cohort, or
         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 (
             self._comparables_repo is None
             or self._epc_prediction is None
-            or attributes is None
+            or (attributes is None and conditioning is 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)
         if target is None:
             return None
+        if conditioning is not None:
+            target = target_with_conditioning(target, conditioning)
         candidates = self._comparables_repo.candidates_for(identity.postcode)
         comparables = select_comparables(target, candidates)
         if not comparables.members:
@@ -201,7 +242,7 @@ class IngestionOrchestrator:
                     uow.epc.save(
                         item.predicted_epc,
                         property_id=item.property_id,
-                        source="predicted",
+                        source=item.predicted_source,
                     )
                 # The live `solar` table is keyed by UPRN and needs the fetch's
                 # coordinates; insights are only set when those coordinates were