From 9204228366c8b78949b05b09d105633bd74a68cc Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Mon, 6 Jul 2026 08:38:07 +0000 Subject: [PATCH] =?UTF-8?q?An=20expired=20historic=20cert=20conditions=20i?= =?UTF-8?q?ngestion's=20prediction=20and=20labels=20it=20=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../epc_prediction/historic_conditioning.py | 39 ++++++++++++- orchestration/ingestion_orchestrator.py | 57 ++++++++++++++++--- 2 files changed, 87 insertions(+), 9 deletions(-) diff --git a/domain/epc_prediction/historic_conditioning.py b/domain/epc_prediction/historic_conditioning.py index c52f94f91..a0c884e0b 100644 --- a/domain/epc_prediction/historic_conditioning.py +++ b/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, + ) diff --git a/orchestration/ingestion_orchestrator.py b/orchestration/ingestion_orchestrator.py index 7411b9a36..f9f162950 100644 --- a/orchestration/ingestion_orchestrator.py +++ b/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