mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
Merge pull request #1466 from Hestia-Homes/feature/historic-epc-enhanced-prediction
Expired-enhanced prediction: historic EPC conditions the cohort (ADR-0054)
This commit is contained in:
Jun-te Kim
GitHub
commit
0507b6b0f5
24 changed files with 366931 additions and 33 deletions
|
|
@ -81,6 +81,14 @@ _Avoid_: neighbours, similar properties, peer set
|
|||
Producing a Property's `EpcPropertyData` picture from its **Comparable Properties** when it has no EPC (~30% of UK homes, typically long-tenure). **Deterministic** neighbour synthesis (k-NN-style — *not* ML; no trained model): take the cohort **mode** for the homogeneous categoricals (wall / roof / floor construction + insulation, construction age band), copy a single representative comparable's **structure** wholesale (building parts, per-window dimensions + orientations, floor dimensions) so the picture stays internally consistent for the calculator, then apply **Landlord Overrides** and the known inputs on top. The result is scored through **SAP10 Calculation** like any other **Effective EPC**, so a predicted Property flows through Rebaselining, Bill Derivation, and Modelling unchanged — held in a **distinct predicted-EPC slot** that coexists with any lodged EPC (so provenance is structural and the UI can flag it; see ADR-0031). A **known property type is required** — the hard cohort filter (a flat is never sized from houses) — supplied by a **Landlord Override** (or, later, an Ordnance Survey lookup); a Property whose property type is genuinely unknown is **gated out**, never predicted from a mixed-type cohort and never given a national default. The same cohort machinery also produces **EPC Anomaly Flags** for Properties that *do* have an EPC. A future learned-weighting refinement is possible but separate, as with the calculator's ML residual head.
|
||||
_Avoid_: EpcPredictionService (no "service" suffix — name the operation), ML prediction (it is deterministic), EPC estimation
|
||||
|
||||
**Historic EPC**:
|
||||
One certificate row from the final data dump of the shut-down old EPC register, held at `s3://retrofit-data-dev/historical_epc/{POSTCODE}/data.csv.gz` and read through the `HistoricEpcRepository` port. Partial, tabular, display-text data (the old API never exposed full SAP inputs), covering certificates the new gov API (registered ≥ 1 Jan 2012) cannot see. Consumed by `address2UPRN` (fuzzy address→UPRN resolution) and by **Expired-Enhanced Prediction** (exact-UPRN lookup only).
|
||||
_Avoid_: old EPC (ambiguous with a pre-SAP10 cert from the new API), historical EPC API (the API is gone; only the backup exists)
|
||||
|
||||
**Expired-Enhanced Prediction**:
|
||||
**EPC Prediction** for a Property whose only certificate predates 2012: the expired **Historic EPC**, found by exact UPRN in its postcode shard, **conditions** cohort selection with its *stable* attributes (property type, built form, wall material, roof construction, the age band's ±1-band neighbourhood, main fuel, and a ±20% floor-area band) exactly as a Landlord Override would. Band widths are harness-evidenced, not guessed (ADR-0054 amendment: age band agrees with a relodged cert 52% exactly / 90% within one band; TFA 45% within ±5% / 82% within ±20%). Volatile attributes (heating, hot water, glazing, PV, insulation, lighting) are excluded — 14+ years stale — and stay neighbour-predicted; the historic cert is **never copied into the Effective EPC as current state**. Persisted to the predicted slot with `source="expired"`. Scoped to the historic backup only; post-2012 expired certs from the new API keep their existing treatment (ADR-0054).
|
||||
_Avoid_: historic override (it is conditioning, not an override the effective picture trusts), expired EPC path (names the input, not the operation)
|
||||
|
||||
### Survey documents
|
||||
|
||||
**Ventilation Audit**:
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ from orchestration.ara_first_run_pipeline import AraFirstRunPipeline
|
|||
from orchestration.ingestion_orchestrator import (
|
||||
ComparablesRepo,
|
||||
EpcFetcher,
|
||||
HistoricEpcReader,
|
||||
IngestionOrchestrator,
|
||||
PredictionAttributesReader,
|
||||
SolarFetcher,
|
||||
|
|
@ -70,6 +71,7 @@ def build_first_run_pipeline(
|
|||
solar_fetcher: SolarFetcher,
|
||||
comparables_repo: Optional[ComparablesRepo] = None,
|
||||
prediction_attributes_reader: Optional[PredictionAttributesReader] = None,
|
||||
historic_epc_reader: Optional[HistoricEpcReader] = None,
|
||||
) -> AraFirstRunPipeline:
|
||||
"""Compose the real three-stage pipeline on a Unit-of-Work factory.
|
||||
|
||||
|
|
@ -95,6 +97,9 @@ def build_first_run_pipeline(
|
|||
comparables_repo=comparables_repo,
|
||||
prediction_attributes_reader=prediction_attributes_reader,
|
||||
epc_prediction=EpcPrediction(),
|
||||
# Expired-Enhanced Prediction (ADR-0054): off until a resolver over
|
||||
# the historic S3 backup is supplied, like the two readers above.
|
||||
historic_epc_reader=historic_epc_reader,
|
||||
),
|
||||
baseline=PropertyBaselineOrchestrator(
|
||||
unit_of_work=unit_of_work,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,111 @@
|
|||
# The expired historic EPC conditions prediction with its stable attributes; it is never trusted as current state
|
||||
|
||||
## Status
|
||||
|
||||
accepted
|
||||
|
||||
## Context
|
||||
|
||||
The old EPC register API has been shut down. Before it went, we captured its
|
||||
final data dump to `s3://retrofit-data-dev/historical_epc/{POSTCODE}/data.csv.gz`
|
||||
— one flat `HistoricEpc` row per certificate, partial tabular data (the old API
|
||||
never exposed full SAP inputs). PR #1356 lifted that backup into a DDD stack:
|
||||
`HistoricEpcRepository` (port) / `HistoricEpcS3Repository` (adapter) /
|
||||
`HistoricEpcResolver`, so far consumed only by `address2UPRN`.
|
||||
|
||||
The new gov EPC API (get-energy-performance-data.communities.gov.uk) only
|
||||
covers certificates registered **since 1 January 2012**. A Property whose only
|
||||
certificate predates 2012 is EPC-less to Ingestion today, so **EPC Prediction**
|
||||
synthesises its picture blind from `property_type` + `built_form` (+ any
|
||||
Landlord Overrides). But the historic dump holds *observed* attributes for that
|
||||
exact dwelling — wall, roof, floor area, fuel, age band, heating, glazing, PV.
|
||||
|
||||
The tension: a pre-2012 certificate is 14+ years stale. Construction and
|
||||
geometry do not change (wall material, built form, age band, roughly the floor
|
||||
area); the rest very plausibly has (heating system, hot water, glazing, PV,
|
||||
insulation levels, lighting). Trusting a 2009 "back boiler, single glazed"
|
||||
observation as override-grade current truth could make the prediction *worse*
|
||||
than the neighbour-based default.
|
||||
|
||||
## Decision
|
||||
|
||||
An expired historic EPC **conditions** EPC Prediction the way a Landlord
|
||||
Override does — it narrows and enriches the cohort — and is **never copied
|
||||
into the Effective EPC as current state**.
|
||||
|
||||
1. **Stable attributes only.** The historic certificate contributes:
|
||||
`property_type` (the hard cohort filter, as today), `built_form`, wall
|
||||
construction **material** (the RdSAP `wall_construction` code resolved from
|
||||
the description's material prefix, per `wall_type_overlay.py`), roof
|
||||
construction, `construction_age_band`, and `main_fuel`. Volatile attributes
|
||||
— heating system, hot water, glazing, PV, insulation states, lighting — are
|
||||
excluded and stay neighbour-predicted.
|
||||
2. **Floor area is a tolerance band, not an override.** Comparables are
|
||||
soft-filtered to within **±20%** of the historic certificate's total floor
|
||||
area — a coarse dwelling-size filter, not a precision match. The predicted
|
||||
floor area remains the cohort's geo-weighted median. *(Amended from ±5% —
|
||||
see Amendment below.)*
|
||||
3. **Every historic filter rides the existing filter-then-relax ladder**
|
||||
(ADR-0029): an attribute that cannot be resolved into the cohort's code
|
||||
space maps to `None` and its filter is simply inactive; a filter that would
|
||||
starve the cohort below the minimum is relaxed. Degradation is graceful by
|
||||
construction.
|
||||
4. **Exact-UPRN lookup.** The prediction path fetches the Property's postcode
|
||||
shard and matches on UPRN equality (multiple rows for one UPRN → latest
|
||||
lodgement date). Fuzzy address matching stays quarantined in `address2UPRN`;
|
||||
a fuzzy hit will never import a neighbour's attributes as if observed.
|
||||
5. **Provenance: `EpcSource` gains `"expired"`.** The enhanced prediction is
|
||||
persisted to the predicted slot with `source="expired"` — enhanced by an
|
||||
expired observation of this dwelling, rather than totally predicted.
|
||||
`Property.source_path` / Effective-EPC precedence are unchanged.
|
||||
6. **Scoped to the historic source.** Post-2012 certificates from the new API
|
||||
— including expired or replaced ones — keep today's treatment. The
|
||||
epistemic inconsistency (an expired 2014 cert is trusted as current; an
|
||||
expired 2009 one only conditions) is acknowledged and deliberate: changing
|
||||
the post-2012 path is a separate, larger-blast-radius decision.
|
||||
7. **Validated by a pairs harness, not reasoning alone.** A repeatable script
|
||||
finds properties holding a pre-2012 historic certificate *and* a
|
||||
post-June-2025 (RdSAP 10 / SAP 10.2) lodged certificate, predicts from the
|
||||
historic attributes, and reports Component Accuracy against the lodged
|
||||
truth **per attribute** — so any whitelist member (main fuel is the
|
||||
judgement call) can be promoted or demoted with evidence. A report, not a
|
||||
CI gate.
|
||||
|
||||
## Consequences
|
||||
|
||||
- The lookup chain at Ingestion becomes: new EPC API → historic backup by UPRN
|
||||
→ plain prediction. A historic miss costs one S3 GET.
|
||||
- `PredictionTarget` grows optional stable-attribute fields and
|
||||
`select_comparables` gains their soft filters (including the first
|
||||
numeric-tolerance filter); all default to `None`/inactive, so existing
|
||||
callers and behaviour are untouched.
|
||||
- `EpcSource` widens to `Literal["lodged", "predicted", "expired"]`; the
|
||||
`epc_property.source` column is already TEXT, so no migration.
|
||||
- Downstream consumers that branch on `source == "predicted"` treat
|
||||
`"expired"` the same unless they opt into the distinction (it lives in the
|
||||
predicted slot; `source_path` is unchanged).
|
||||
- The whitelist is a hypothesis until the pairs harness reports; attribute
|
||||
membership changes are cheap (one resolver each).
|
||||
|
||||
## Amendment (2026-07-06): band widths set by the pairs harness
|
||||
|
||||
The harness ran at scale (2,000 stride-sampled postcodes → **439 pairs**, 419
|
||||
scored per arm; tables on PR #1466) and two parameters changed with evidence:
|
||||
|
||||
- **Age band conditions on the ±1-band neighbourhood, not equality.** The
|
||||
historic band agrees with the newly lodged one only **52%** exactly but
|
||||
**90%** within one band — assessors re-band constantly (skewing newer: 31%
|
||||
of new certs band later). Equality-conditioning steered engaged cohorts
|
||||
toward a coin flip.
|
||||
- **The floor-area band widens ±5% → ±20%.** Historic-vs-new TFA agreement:
|
||||
45% within ±5%, 63% within ±10%, 72% within ±15%, **82% within ±20%**
|
||||
(remeasurement + extensions). At ±5% the filter engaged on only 19% of
|
||||
pairs and worsened the floor-area residual.
|
||||
|
||||
Also established by the same evidence: main fuel is **95%** stable (after
|
||||
resolving the register's legacy "backwards compatibility" fuel descriptions),
|
||||
property type **91%**, wall construction **79%** — the whitelist's core holds.
|
||||
Known collateral to watch: `roof_insulation_thickness` transfers slightly
|
||||
worse from age-conditioned cohorts (negative in all three harness rounds) —
|
||||
candidate follow-up: exclude the roof-insulation mode when age conditioning
|
||||
engaged.
|
||||
|
|
@ -21,6 +21,47 @@ from domain.geospatial.coordinates import Coordinates
|
|||
# else it is relaxed (ADR-0029 filter-then-relax ladder).
|
||||
_DEFAULT_MINIMUM_COHORT = 5
|
||||
|
||||
# Half-width of the floor-area conditioning band: an expired Historic EPC's
|
||||
# observed floor area keeps comparables within ±20% of it — a coarse
|
||||
# dwelling-size filter, not a precision match. Evidence (ADR-0054 amendment,
|
||||
# 439-pair harness): the historic TFA agrees with the newly lodged one within
|
||||
# ±5% only 45% of the time (remeasurement + extensions) but within ±20% 82%.
|
||||
FLOOR_AREA_TOLERANCE = 0.20
|
||||
|
||||
# RdSAP Table S1 band letters in chronological order, for band-distance checks.
|
||||
_AGE_BAND_ORDER = "ABCDEFGHIJKLM"
|
||||
|
||||
# roof_construction codes by FORM family. Pinned empirically — a 7,974-cert
|
||||
# co-occurrence sweep of code x roofs[0].description over single-building-part
|
||||
# certs (scripts/roof_construction_code_sweep.py): 1=Flat 98%, 4=Pitched 99%,
|
||||
# 5/8=Pitched 88%, 3="(another dwelling above)" 100% — plus 7/9 = "(another
|
||||
# premises above)", established by the #1452 heat-loss suppression fix.
|
||||
ROOF_FORM_BY_CONSTRUCTION: dict[int, str] = {
|
||||
1: "flat",
|
||||
3: "dwelling_above",
|
||||
4: "pitched",
|
||||
5: "pitched",
|
||||
7: "premises_above",
|
||||
8: "pitched",
|
||||
9: "premises_above",
|
||||
}
|
||||
|
||||
|
||||
def roof_form_of_construction(code: object) -> Optional[str]:
|
||||
"""The form family of a roof_construction code, or None when unknown."""
|
||||
return ROOF_FORM_BY_CONSTRUCTION.get(code) if isinstance(code, int) else None
|
||||
|
||||
|
||||
def age_bands_within_one(candidate: object, target_band: object) -> bool:
|
||||
"""Whether two Table S1 band letters are at most one band apart. Assessors
|
||||
re-band constantly (harness: 52% exact agreement historic-vs-new, 90%
|
||||
within one band), so the age filter keeps the band NEIGHBOURHOOD."""
|
||||
if not (isinstance(candidate, str) and isinstance(target_band, str)):
|
||||
return False
|
||||
if candidate not in _AGE_BAND_ORDER or target_band not in _AGE_BAND_ORDER:
|
||||
return False
|
||||
return abs(_AGE_BAND_ORDER.index(candidate) - _AGE_BAND_ORDER.index(target_band)) <= 1
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ComparableProperty:
|
||||
|
|
@ -88,6 +129,35 @@ def select_comparables(
|
|||
active=target.wall_construction is not None,
|
||||
minimum_cohort=minimum_cohort,
|
||||
)
|
||||
cohort = _maybe_filter(
|
||||
cohort,
|
||||
lambda c: _main_roof_form(c) == target.roof_form,
|
||||
active=target.roof_form is not None,
|
||||
minimum_cohort=minimum_cohort,
|
||||
)
|
||||
cohort = _maybe_filter(
|
||||
cohort,
|
||||
lambda c: age_bands_within_one(
|
||||
_main_construction_age_band(c), target.construction_age_band
|
||||
),
|
||||
active=target.construction_age_band is not None,
|
||||
minimum_cohort=minimum_cohort,
|
||||
)
|
||||
cohort = _maybe_filter(
|
||||
cohort,
|
||||
lambda c: _main_fuel_type(c) == target.main_fuel,
|
||||
active=target.main_fuel is not None,
|
||||
minimum_cohort=minimum_cohort,
|
||||
)
|
||||
target_area = target.total_floor_area_m2
|
||||
cohort = _maybe_filter(
|
||||
cohort,
|
||||
lambda c: target_area is not None
|
||||
and abs(c.epc.total_floor_area_m2 - target_area)
|
||||
<= FLOOR_AREA_TOLERANCE * target_area,
|
||||
active=target_area is not None,
|
||||
minimum_cohort=minimum_cohort,
|
||||
)
|
||||
return ComparableProperties(members=tuple(cohort))
|
||||
|
||||
|
||||
|
|
@ -124,3 +194,21 @@ def _main_wall_construction(comparable: ComparableProperty) -> object:
|
|||
"""The main building part's wall construction, or None when no part lodged."""
|
||||
parts = comparable.epc.sap_building_parts
|
||||
return parts[0].wall_construction if parts else None
|
||||
|
||||
|
||||
def _main_roof_form(comparable: ComparableProperty) -> Optional[str]:
|
||||
"""The main building part's roof-form family, or None when unresolvable."""
|
||||
parts = comparable.epc.sap_building_parts
|
||||
return roof_form_of_construction(parts[0].roof_construction) if parts else None
|
||||
|
||||
|
||||
def _main_construction_age_band(comparable: ComparableProperty) -> object:
|
||||
"""The main building part's Table S1 band letter, or None when no part lodged."""
|
||||
parts = comparable.epc.sap_building_parts
|
||||
return parts[0].construction_age_band if parts else None
|
||||
|
||||
|
||||
def _main_fuel_type(comparable: ComparableProperty) -> object:
|
||||
"""The primary heating fuel code, or None when no main heating lodged."""
|
||||
details = comparable.epc.sap_heating.main_heating_details
|
||||
return details[0].main_fuel_type if details else None
|
||||
|
|
|
|||
193
domain/epc_prediction/historic_conditioning.py
Normal file
193
domain/epc_prediction/historic_conditioning.py
Normal file
|
|
@ -0,0 +1,193 @@
|
|||
"""Resolve an expired Historic EPC's *stable* attributes into the cohort code
|
||||
spaces that condition EPC Prediction (ADR-0054).
|
||||
|
||||
The historic dump carries display text ("Semi-Detached", "Cavity wall, as
|
||||
built, no insulation (assumed)"); the Comparable Properties cohort carries
|
||||
gov-EPC codes. This module is the anti-corruption layer between the two: one
|
||||
resolver per whitelisted stable attribute, each returning None on an
|
||||
unresolvable value so its cohort filter is simply inactive. Volatile
|
||||
attributes (heating system, hot water, glazing, PV, insulation states,
|
||||
lighting) are deliberately absent — a 14+-year-old observation of them is not
|
||||
evidence about today.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, replace
|
||||
from typing import Optional
|
||||
|
||||
from datatypes.epc.domain.historic_epc import HistoricEpc
|
||||
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).
|
||||
# The historic description's material is everything before the first comma.
|
||||
_WALL_MATERIAL_CONSTRUCTION: dict[str, int] = {
|
||||
"Granite or whin": 1,
|
||||
"Sandstone": 2,
|
||||
"Solid brick": 3,
|
||||
"Cavity wall": 4,
|
||||
"Timber frame": 5,
|
||||
"System built": 6,
|
||||
"Cob": 7,
|
||||
"Park home wall": 8,
|
||||
}
|
||||
|
||||
# RdSAP Table S1 band letters — the code space cohort building parts carry
|
||||
# (`sap_building_parts[].construction_age_band` is "A".."M" from the API).
|
||||
# The old register lodged the full England-and-Wales display strings; a
|
||||
# pre-2012 cert cannot carry the post-2012 bands, but they cost nothing.
|
||||
_AGE_BAND_LETTERS: dict[str, str] = {
|
||||
"England and Wales: before 1900": "A",
|
||||
"England and Wales: 1900-1929": "B",
|
||||
"England and Wales: 1930-1949": "C",
|
||||
"England and Wales: 1950-1966": "D",
|
||||
"England and Wales: 1967-1975": "E",
|
||||
"England and Wales: 1976-1982": "F",
|
||||
"England and Wales: 1983-1990": "G",
|
||||
"England and Wales: 1991-1995": "H",
|
||||
"England and Wales: 1996-2002": "I",
|
||||
"England and Wales: 2003-2006": "J",
|
||||
"England and Wales: 2007 onwards": "K",
|
||||
"England and Wales: 2007-2011": "K",
|
||||
"England and Wales: 2012-2022": "L",
|
||||
"England and Wales: 2023 onwards": "M",
|
||||
}
|
||||
|
||||
# Modern RdSAP-20/21 `main_fuel` codes (epc_codes.csv), keyed by the base fuel
|
||||
# description — the same family `main_fuel_overlay.py` pins. The old register
|
||||
# suffixes private fuels with " (not community)", stripped before lookup.
|
||||
_FUEL_CODES: dict[str, int] = {
|
||||
"mains gas": 26,
|
||||
"mains gas (community)": 20,
|
||||
"LPG": 27,
|
||||
"bottled LPG": 3,
|
||||
"LPG special condition": 17,
|
||||
"oil": 28,
|
||||
"electricity": 29,
|
||||
"electricity (community)": 25,
|
||||
"house coal": 33,
|
||||
"smokeless coal": 15,
|
||||
"dual fuel (mineral and wood)": 10,
|
||||
"wood logs": 6,
|
||||
"bulk wood pellets": 7,
|
||||
"wood chips": 8,
|
||||
"biomass (community)": 31,
|
||||
}
|
||||
|
||||
# Roof FORM families by the description's pre-comma half. Only forms whose
|
||||
# API code grouping is pinned (see comparable_properties.ROOF_FORM_BY_CONSTRUCTION
|
||||
# for the empirical evidence); roof rooms and thatch stay None — never guess.
|
||||
_ROOF_FORM_BY_PREFIX: dict[str, str] = {
|
||||
"Pitched": "pitched",
|
||||
"Flat": "flat",
|
||||
"(another dwelling above)": "dwelling_above",
|
||||
"(another premises above)": "premises_above",
|
||||
}
|
||||
|
||||
_NOT_COMMUNITY_SUFFIX = " (not community)"
|
||||
|
||||
# Pre-RdSAP-17 lodgements carry a deprecation rider after the fuel name; the
|
||||
# fuel named is the same physical fuel. Dominant in the pre-2012 dump (a
|
||||
# 65-shard scan: 636/663 otherwise-unresolved values were these variants).
|
||||
_LEGACY_SUFFIX = " - this is for backwards compatibility only and should not be used"
|
||||
|
||||
# SAP-style lodgements prefix the category; map the known forms to the base
|
||||
# description the code table keys on.
|
||||
_LEGACY_ALIASES: dict[str, str] = {
|
||||
"Gas: mains gas": "mains gas",
|
||||
"Electricity: electricity, unspecified tariff": "electricity",
|
||||
"dual fuel - mineral + wood": "dual fuel (mineral and wood)",
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HistoricConditioning:
|
||||
"""The expired cert's stable attributes, in cohort code space; None means
|
||||
"unresolved — do not condition on this"."""
|
||||
|
||||
property_type: Optional[str]
|
||||
built_form: Optional[str]
|
||||
wall_construction: Optional[int]
|
||||
construction_age_band: Optional[str]
|
||||
main_fuel: Optional[int]
|
||||
total_floor_area_m2: Optional[float]
|
||||
roof_form: Optional[str] = None
|
||||
|
||||
|
||||
def _wall_construction(description: str) -> Optional[int]:
|
||||
material = description.split(",", 1)[0].strip()
|
||||
return _WALL_MATERIAL_CONSTRUCTION.get(material)
|
||||
|
||||
|
||||
def _roof_form(description: str) -> Optional[str]:
|
||||
form = description.split(",", 1)[0].strip()
|
||||
return _ROOF_FORM_BY_PREFIX.get(form)
|
||||
|
||||
|
||||
def _main_fuel(description: str) -> Optional[int]:
|
||||
base = description.strip().removesuffix(_NOT_COMMUNITY_SUFFIX)
|
||||
base = base.removesuffix(_LEGACY_SUFFIX)
|
||||
base = _LEGACY_ALIASES.get(base, base)
|
||||
return _FUEL_CODES.get(base)
|
||||
|
||||
|
||||
def _floor_area(raw: str) -> Optional[float]:
|
||||
try:
|
||||
area = float(raw)
|
||||
except ValueError:
|
||||
return None
|
||||
return area if area > 0 else None
|
||||
|
||||
|
||||
def conditioning_from_historic(record: HistoricEpc) -> HistoricConditioning:
|
||||
return HistoricConditioning(
|
||||
property_type=property_type_to_code(record.property_type),
|
||||
built_form=built_form_to_code(record.built_form),
|
||||
wall_construction=_wall_construction(record.walls_description),
|
||||
construction_age_band=_AGE_BAND_LETTERS.get(record.construction_age_band),
|
||||
main_fuel=_main_fuel(record.main_fuel),
|
||||
total_floor_area_m2=_floor_area(record.total_floor_area),
|
||||
roof_form=_roof_form(record.roof_description),
|
||||
)
|
||||
|
||||
|
||||
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,
|
||||
roof_form=conditioning.roof_form,
|
||||
)
|
||||
|
|
@ -34,6 +34,16 @@ class PredictionTarget:
|
|||
# The target Property's own coordinates (resolved from its UPRN), against
|
||||
# which neighbours are distance-weighted. None disables geo-weighting.
|
||||
coordinates: Optional[Coordinates] = None
|
||||
# Stable attributes observed on an expired Historic EPC condition selection
|
||||
# the same way (ADR-0054): RdSAP Table S1 band letter, modern main_fuel
|
||||
# code, and a ±5% floor-area band. None leaves each filter inactive.
|
||||
construction_age_band: Optional[str] = None
|
||||
main_fuel: Optional[int] = None
|
||||
total_floor_area_m2: Optional[float] = None
|
||||
# The roof FORM family ("pitched" / "flat" / "dwelling_above" /
|
||||
# "premises_above") — the API's roof_construction codes group by form,
|
||||
# so the filter matches families, never exact codes (ADR-0054 amendment).
|
||||
roof_form: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
|
|
|||
|
|
@ -56,23 +56,34 @@ def _load_cohort(
|
|||
if not path.exists():
|
||||
continue
|
||||
raw: dict[str, Any] = json.loads(path.read_text())
|
||||
try:
|
||||
epc = EpcPropertyDataMapper.from_api_response(raw)
|
||||
except Exception: # noqa: BLE001 — a bad cert must not abort the sweep
|
||||
continue
|
||||
uprn = _uprn(raw)
|
||||
cohort.append(
|
||||
ComparableProperty(
|
||||
epc=epc,
|
||||
certificate_number=cert,
|
||||
address=_address(raw),
|
||||
registration_date=_registration_date(raw),
|
||||
coordinates=coordinates.get(uprn) if uprn is not None else None,
|
||||
)
|
||||
)
|
||||
comparable = comparable_from_payload(cert, raw, coordinates)
|
||||
if comparable is not None:
|
||||
cohort.append(comparable)
|
||||
return cohort
|
||||
|
||||
|
||||
def comparable_from_payload(
|
||||
cert: str,
|
||||
raw: dict[str, Any],
|
||||
coordinates: dict[int, Coordinates],
|
||||
) -> Optional[ComparableProperty]:
|
||||
"""One frozen cert payload -> a ComparableProperty, or None when the mapper
|
||||
can't take it (a bad cert must never abort a corpus load). Shared by the
|
||||
per-file corpus above and the single-file expired-pairs corpus."""
|
||||
try:
|
||||
epc = EpcPropertyDataMapper.from_api_response(raw)
|
||||
except Exception: # noqa: BLE001
|
||||
return None
|
||||
uprn = _uprn(raw)
|
||||
return ComparableProperty(
|
||||
epc=epc,
|
||||
certificate_number=cert,
|
||||
address=_address(raw),
|
||||
registration_date=_registration_date(raw),
|
||||
coordinates=coordinates.get(uprn) if uprn is not None else None,
|
||||
)
|
||||
|
||||
|
||||
def load_coordinates(corpus_dir: Path) -> dict[int, Coordinates]:
|
||||
"""The optional `_coordinates.json` sidecar (`{uprn: [lon, lat]}`), resolved
|
||||
from the OS Open-UPRN data by `fetch_corpus_coordinates.py`. Absent for a
|
||||
|
|
|
|||
|
|
@ -5,11 +5,18 @@ 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,
|
||||
)
|
||||
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,
|
||||
|
|
@ -17,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
|
||||
|
||||
|
|
@ -40,6 +48,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)."""
|
||||
|
||||
|
|
@ -69,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:
|
||||
|
|
@ -98,6 +116,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 +129,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)
|
||||
|
|
@ -149,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:
|
||||
|
|
@ -191,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
|
||||
|
|
|
|||
|
|
@ -50,9 +50,20 @@ from infrastructure.postgres.epc_property_table import (
|
|||
EpcRenewableHeatIncentiveModel,
|
||||
EpcWindowModel,
|
||||
)
|
||||
from repositories.epc.epc_repository import EpcRepository, EpcSource
|
||||
from repositories.epc.epc_repository import (
|
||||
PREDICTED_SLOT_SOURCES,
|
||||
EpcRepository,
|
||||
EpcSource,
|
||||
)
|
||||
from utilities.private import private
|
||||
|
||||
|
||||
def _slot_sources(source: EpcSource) -> tuple[EpcSource, ...]:
|
||||
"""The source family sharing `source`'s slot: "predicted" and "expired" are
|
||||
one slot whose flavour a re-ingestion may flip, so every slot read and
|
||||
slot-clearing delete addresses the family (ADR-0054)."""
|
||||
return PREDICTED_SLOT_SOURCES if source in PREDICTED_SLOT_SOURCES else (source,)
|
||||
|
||||
_T = TypeVar("_T")
|
||||
|
||||
|
||||
|
|
@ -320,7 +331,7 @@ class EpcPostgresRepository(EpcRepository):
|
|||
for i in self._session.exec(
|
||||
select(EpcPropertyModel.id)
|
||||
.where(col(EpcPropertyModel.property_id).in_(property_ids))
|
||||
.where(EpcPropertyModel.source == source)
|
||||
.where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
|
||||
).all()
|
||||
if i is not None
|
||||
]
|
||||
|
|
@ -367,7 +378,7 @@ class EpcPostgresRepository(EpcRepository):
|
|||
for i in self._session.exec(
|
||||
select(EpcPropertyModel.id)
|
||||
.where(EpcPropertyModel.property_id == property_id)
|
||||
.where(EpcPropertyModel.source == source)
|
||||
.where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
|
||||
).all()
|
||||
if i is not None
|
||||
]
|
||||
|
|
@ -417,7 +428,7 @@ class EpcPostgresRepository(EpcRepository):
|
|||
row = self._session.exec(
|
||||
select(EpcPropertyModel)
|
||||
.where(EpcPropertyModel.property_id == property_id)
|
||||
.where(EpcPropertyModel.source == source)
|
||||
.where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
|
||||
.order_by(EpcPropertyModel.id) # type: ignore[arg-type]
|
||||
).first()
|
||||
if row is None or row.id is None:
|
||||
|
|
@ -444,7 +455,7 @@ class EpcPostgresRepository(EpcRepository):
|
|||
parents = self._session.exec(
|
||||
select(EpcPropertyModel)
|
||||
.where(col(EpcPropertyModel.property_id).in_(property_ids))
|
||||
.where(EpcPropertyModel.source == source)
|
||||
.where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
|
||||
.order_by(EpcPropertyModel.id) # type: ignore[arg-type]
|
||||
).all()
|
||||
parent_by_property: dict[int, EpcPropertyModel] = {}
|
||||
|
|
|
|||
|
|
@ -9,8 +9,15 @@ if TYPE_CHECKING:
|
|||
from repositories.epc.epc_postgres_repository import EpcSaveRequest
|
||||
|
||||
# Provenance of a persisted EPC picture (ADR-0031): a real "lodged" EPC, or a
|
||||
# "predicted" one synthesised by EPC Prediction. A property can hold one of each.
|
||||
EpcSource = Literal["lodged", "predicted"]
|
||||
# "predicted" one synthesised by EPC Prediction — "expired" is a prediction
|
||||
# enhanced by an expired Historic EPC's stable attributes (ADR-0054). A
|
||||
# property holds one lodged picture and one predicted-slot picture.
|
||||
EpcSource = Literal["lodged", "predicted", "expired"]
|
||||
|
||||
# The predicted slot's source family: "predicted" and "expired" occupy the SAME
|
||||
# slot (a re-ingestion may flip the flavour), so slot reads and slot-clearing
|
||||
# deletes must always address the family, never one member (ADR-0054).
|
||||
PREDICTED_SLOT_SOURCES: tuple[EpcSource, ...] = ("predicted", "expired")
|
||||
|
||||
|
||||
class EpcRepository(ABC):
|
||||
|
|
|
|||
|
|
@ -34,6 +34,19 @@ class HistoricEpcResolver:
|
|||
matches=matches,
|
||||
)
|
||||
|
||||
def record_for_uprn(self, uprn: str, postcode: str) -> Optional[HistoricEpc]:
|
||||
"""The postcode shard's certificate for ``uprn``, or None on a miss.
|
||||
|
||||
Exact UPRN equality only — the prediction path must never import a
|
||||
fuzzy address match's attributes as if observed (ADR-0054)."""
|
||||
if not uprn:
|
||||
return None
|
||||
records: list[HistoricEpc] = self._repo.get_for_postcode(Postcode(postcode))
|
||||
certs: list[HistoricEpc] = [r for r in records if r.uprn == uprn]
|
||||
if not certs:
|
||||
return None
|
||||
return max(certs, key=lambda r: r.lodgement_date)
|
||||
|
||||
def resolve_uprn(
|
||||
self, user_address: str, postcode: str
|
||||
) -> Optional[tuple[str, str, float]]:
|
||||
|
|
|
|||
140
scripts/build_expired_pairs_corpus.py
Normal file
140
scripts/build_expired_pairs_corpus.py
Normal file
|
|
@ -0,0 +1,140 @@
|
|||
"""Freeze a subsample of harness pairs into the committed integration fixture.
|
||||
|
||||
Turns the pairs harness's live evidence into a deterministic, offline
|
||||
regression gate (the ADR-0030 corpus pattern): anonymised RAW API payloads
|
||||
loaded through ``EpcPropertyDataMapper``, so the gate keeps exercising the
|
||||
mapper and survives domain-dataclass changes. Layout, extending the
|
||||
``tests/fixtures/epc_prediction`` conventions:
|
||||
|
||||
ONE json file (a thousand per-cert files would drown a PR diff):
|
||||
pairs: [{postcode, uprn, actual, historic: {...}}]
|
||||
cohorts: {postcode: {token: anonymised payload}}
|
||||
actuals: {token: anonymised payload} (the lodged SAP-10.2 certs)
|
||||
|
||||
Reads the raw-JSON disk cache (scripts/epc_disk_cache.py) for everything the
|
||||
API served, and the historic S3 backup for the pre-2012 records. Pairs are
|
||||
subsampled deterministically (sorted, strided) from the harness telemetry.
|
||||
|
||||
Usage:
|
||||
python scripts/build_expired_pairs_corpus.py \
|
||||
--telemetry pairs_telemetry.jsonl --cache-dir .epc_cache \
|
||||
--out tests/fixtures/expired_prediction_pairs.json --sample 30
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import dataclasses
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from datatypes.epc.domain.historic_epc import HistoricEpc # noqa: E402
|
||||
from domain.postcode import Postcode # noqa: E402
|
||||
from harness.epc_prediction_corpus import anonymise_payload, stable_hash # noqa: E402
|
||||
|
||||
# Free-text fields blanked on the frozen historic record; the postcode is kept
|
||||
# (coarse open data, the shard key) and the address becomes a stable token so
|
||||
# nothing joins back to a household.
|
||||
_HISTORIC_PII_BLANK = ("address1", "address2", "address3", "posttown")
|
||||
|
||||
|
||||
def anonymise_historic(record: HistoricEpc) -> dict[str, str]:
|
||||
row = dataclasses.asdict(record)
|
||||
for field in _HISTORIC_PII_BLANK:
|
||||
row[field] = ""
|
||||
row["address"] = stable_hash("addr", record.address) if record.address else ""
|
||||
row["lmk_key"] = stable_hash("lmk", record.lmk_key) if record.lmk_key else ""
|
||||
return row
|
||||
|
||||
|
||||
def _read_cache(cache_dir: Path, key: str) -> Optional[Any]:
|
||||
path = cache_dir / f"{key}.json"
|
||||
return json.loads(path.read_text()) if path.exists() else None
|
||||
|
||||
|
||||
def subsample(rows: list[dict[str, Any]], count: int) -> list[dict[str, Any]]:
|
||||
"""A deterministic spread across the telemetry: sort, stride."""
|
||||
ordered = sorted(rows, key=lambda r: (str(r["postcode"]), str(r["uprn"])))
|
||||
if len(ordered) <= count:
|
||||
return ordered
|
||||
stride = len(ordered) // count
|
||||
return ordered[::stride][:count]
|
||||
|
||||
|
||||
def main() -> None: # pragma: no cover - IO composition
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--telemetry", type=Path, required=True)
|
||||
parser.add_argument("--cache-dir", type=Path, required=True)
|
||||
parser.add_argument("--out", type=Path, required=True)
|
||||
parser.add_argument("--sample", type=int, default=30)
|
||||
args = parser.parse_args()
|
||||
|
||||
from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver
|
||||
from repositories.historic_epc.historic_epc_s3_repository import (
|
||||
HistoricEpcS3Repository,
|
||||
)
|
||||
|
||||
resolver = HistoricEpcResolver(HistoricEpcS3Repository.with_default_s3_client())
|
||||
|
||||
telemetry = [json.loads(line) for line in args.telemetry.read_text().splitlines()]
|
||||
chosen = subsample(telemetry, args.sample)
|
||||
|
||||
cohorts: dict[str, dict[str, Any]] = {}
|
||||
actuals: dict[str, Any] = {}
|
||||
pairs: list[dict[str, Any]] = []
|
||||
for row in chosen:
|
||||
postcode, uprn = str(row["postcode"]), str(row["uprn"])
|
||||
historic = resolver.record_for_uprn(uprn, postcode)
|
||||
if historic is None:
|
||||
print(f"{postcode} {uprn}: historic record gone — skipped", file=sys.stderr)
|
||||
continue
|
||||
search = _read_cache(args.cache_dir, f"search_uprn_{uprn}")
|
||||
if not search:
|
||||
print(f"{postcode} {uprn}: uprn search not cached — skipped", file=sys.stderr)
|
||||
continue
|
||||
latest = max(search, key=lambda r: str(r["registration_date"]))
|
||||
actual_raw = _read_cache(args.cache_dir, f"cert_{latest['certificate_number']}")
|
||||
cohort_search = _read_cache(args.cache_dir, f"search_pc_{postcode}")
|
||||
if actual_raw is None or cohort_search is None:
|
||||
print(f"{postcode} {uprn}: cert/cohort not cached — skipped", file=sys.stderr)
|
||||
continue
|
||||
|
||||
if postcode not in cohorts:
|
||||
payloads: dict[str, Any] = {}
|
||||
for result in cohort_search:
|
||||
raw = _read_cache(
|
||||
args.cache_dir, f"cert_{result['certificate_number']}"
|
||||
)
|
||||
if raw is None:
|
||||
continue
|
||||
token = stable_hash("cert", str(result["certificate_number"]))
|
||||
payloads[token] = anonymise_payload(raw)
|
||||
cohorts[postcode] = payloads
|
||||
|
||||
actual_token = stable_hash("cert", str(latest["certificate_number"]))
|
||||
actuals[actual_token] = anonymise_payload(actual_raw)
|
||||
pairs.append(
|
||||
{
|
||||
"postcode": str(Postcode(postcode)),
|
||||
"uprn": uprn,
|
||||
"actual": actual_token,
|
||||
"historic": anonymise_historic(historic),
|
||||
}
|
||||
)
|
||||
print(f"{postcode} {uprn}: frozen", file=sys.stderr)
|
||||
|
||||
args.out.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.out.write_text(
|
||||
json.dumps(
|
||||
{"pairs": pairs, "cohorts": cohorts, "actuals": actuals}, indent=1
|
||||
)
|
||||
)
|
||||
print(f"{len(pairs)} pairs frozen across {len(cohorts)} postcodes -> {args.out}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
81
scripts/epc_disk_cache.py
Normal file
81
scripts/epc_disk_cache.py
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
"""A raw-JSON disk cache around the EPC-API client, for harness scripts.
|
||||
|
||||
Caches at the RAW layer — certificate payloads exactly as the API returned
|
||||
them, search results as plain JSON rows — so that:
|
||||
|
||||
1. **Iteration is fast**: the pairs harness re-reads the same certs on every
|
||||
run (pairs and cohorts don't change), so each re-run was re-paying 10-20k
|
||||
sequential HTTP calls; a warm cache replays in minutes.
|
||||
2. **The cache is fixture material**: a committed integration-test corpus
|
||||
should hold anonymised raw API JSON loaded through
|
||||
``EpcPropertyDataMapper.from_api_response`` (the ADR-0030 corpus pattern),
|
||||
so replays keep exercising the mapper and survive domain-dataclass changes.
|
||||
Pickles of internal dataclasses would do neither.
|
||||
|
||||
Successful responses only — errors propagate uncached. Script-support code,
|
||||
not a repository: production ingestion must keep reading live (a re-lodgement
|
||||
must be seen), so this stays out of the DDD layers.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional, cast
|
||||
|
||||
from datatypes.epc.search.epc_search_result import EpcSearchResult
|
||||
from infrastructure.epc_client.epc_client_service import EpcClientService
|
||||
|
||||
|
||||
def _safe(key: str) -> str:
|
||||
return re.sub(r"[^A-Za-z0-9_-]", "_", key)
|
||||
|
||||
|
||||
class JsonCachingEpcClient(EpcClientService):
|
||||
"""An ``EpcClientService`` whose raw HTTP layer is disk-cached as JSON.
|
||||
|
||||
Overrides the two private fetchers, so retry, mapping and search parsing
|
||||
all still run on every call — a warm replay exercises the exact production
|
||||
code path minus the network.
|
||||
"""
|
||||
|
||||
def __init__(self, auth_token: str, cache_dir: Path) -> None:
|
||||
super().__init__(auth_token)
|
||||
self._cache_dir = cache_dir
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def _path(self, key: str) -> Path:
|
||||
return self._cache_dir / f"{_safe(key)}.json"
|
||||
|
||||
def _read(self, key: str) -> Optional[Any]:
|
||||
path = self._path(key)
|
||||
if not path.exists():
|
||||
return None
|
||||
return json.loads(path.read_text())
|
||||
|
||||
def _write(self, key: str, value: Any) -> None:
|
||||
self._path(key).write_text(json.dumps(value))
|
||||
|
||||
def _fetch_certificate(self, cert_num: str) -> dict[str, Any]:
|
||||
cached = self._read(f"cert_{cert_num}")
|
||||
if cached is not None:
|
||||
return cast(dict[str, Any], cached)
|
||||
raw = super()._fetch_certificate(cert_num)
|
||||
self._write(f"cert_{cert_num}", raw)
|
||||
return raw
|
||||
|
||||
def _search(
|
||||
self,
|
||||
postcode: Optional[str] = None,
|
||||
uprn: Optional[int] = None,
|
||||
) -> list[EpcSearchResult]:
|
||||
key = f"search_pc_{postcode}" if postcode else f"search_uprn_{uprn}"
|
||||
cached = self._read(key)
|
||||
if cached is not None:
|
||||
rows = cast(list[dict[str, Any]], cached)
|
||||
return [EpcSearchResult(**row) for row in rows]
|
||||
results = super()._search(postcode=postcode, uprn=uprn)
|
||||
self._write(key, [dataclasses.asdict(r) for r in results])
|
||||
return results
|
||||
597
scripts/expired_prediction_pairs_harness.py
Normal file
597
scripts/expired_prediction_pairs_harness.py
Normal file
|
|
@ -0,0 +1,597 @@
|
|||
"""Pre-2012 x SAP-10.2 pairs harness for Expired-Enhanced Prediction (ADR-0054).
|
||||
|
||||
For each postcode, find properties holding BOTH a pre-2012 cert in the historic
|
||||
S3 backup AND a SAP-10.2 cert on the new gov API (RdSAP 10 went live June 2025;
|
||||
only a same-spec lodged figure is a valid validation target — see Component
|
||||
Accuracy, ADR-0030). Each pair is predicted twice from its leave-one-out
|
||||
postcode cohort:
|
||||
|
||||
- PLAIN arm: property type + built form only (what a blind prediction sees);
|
||||
- CONDITIONED arm: the historic cert's stable attributes conditioning cohort
|
||||
selection (ADR-0054).
|
||||
|
||||
Both arms are scored against the lodged SAP-10.2 cert with the SAME metric
|
||||
suite as the prediction corpus (ADR-0030 Component Accuracy): per-component
|
||||
classification hit-rates, mean-absolute numeric residuals, plus the secondary
|
||||
calculator-floored SAP residual (calc(predicted) vs the lodged score). The
|
||||
per-attribute breakdown is the whitelist evidence: an attribute whose
|
||||
conditioned hit-rate is WORSE than plain is stale and gets demoted.
|
||||
|
||||
The expensive cohort fetch (search-by-postcode + per-cert fetch) happens only
|
||||
for postcodes where a pair actually exists, so the script can sweep hundreds
|
||||
of postcodes cheaply.
|
||||
|
||||
Usage:
|
||||
python scripts/expired_prediction_pairs_harness.py "B93 8SY" "LS6 1AA"
|
||||
python scripts/expired_prediction_pairs_harness.py --postcodes-file pcs.txt --out report.md
|
||||
|
||||
Env: OPEN_EPC_API_TOKEN, DATA_BUCKET; ambient AWS credentials for S3.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from datatypes.epc.domain.epc_property_data import EpcPropertyData # noqa: E402
|
||||
from datatypes.epc.domain.historic_epc import HistoricEpc # noqa: E402
|
||||
from domain.epc_prediction.comparable_properties import ( # noqa: E402
|
||||
FLOOR_AREA_TOLERANCE,
|
||||
ComparableProperty,
|
||||
age_bands_within_one,
|
||||
roof_form_of_construction,
|
||||
select_comparables,
|
||||
)
|
||||
from domain.epc_prediction.epc_prediction import EpcPrediction # noqa: E402
|
||||
from domain.epc_prediction.historic_conditioning import ( # noqa: E402
|
||||
attributes_with_historic_fallback,
|
||||
conditioning_from_historic,
|
||||
target_with_conditioning,
|
||||
)
|
||||
from domain.epc_prediction.prediction_comparison import ( # noqa: E402
|
||||
PredictionComparison,
|
||||
compare_prediction,
|
||||
)
|
||||
from domain.epc_prediction.prediction_target import ( # noqa: E402
|
||||
PredictionTarget,
|
||||
build_prediction_target,
|
||||
)
|
||||
from domain.postcode import Postcode # noqa: E402
|
||||
from domain.property.property import PropertyIdentity # noqa: E402
|
||||
|
||||
_PRE_2012 = "2012-01-01"
|
||||
_VALIDATION_SAP_VERSION = 10.2
|
||||
|
||||
_RESIDUAL_COMPONENTS = (
|
||||
"floor_area_m2",
|
||||
"building_parts",
|
||||
"window_count",
|
||||
"total_window_area_m2",
|
||||
"door_count",
|
||||
)
|
||||
|
||||
|
||||
def latest_pre_2012_by_uprn(records: list[HistoricEpc]) -> dict[str, HistoricEpc]:
|
||||
"""One historic cert per UPRN: the latest lodgement strictly before 2012
|
||||
(ISO date strings compare lexicographically). UPRN-less rows are dropped —
|
||||
without a UPRN there is nothing to pair."""
|
||||
by_uprn: dict[str, HistoricEpc] = {}
|
||||
for record in records:
|
||||
if not record.uprn or not record.lodgement_date:
|
||||
continue
|
||||
if record.lodgement_date >= _PRE_2012:
|
||||
continue
|
||||
current = by_uprn.get(record.uprn)
|
||||
if current is None or record.lodgement_date > current.lodgement_date:
|
||||
by_uprn[record.uprn] = record
|
||||
return by_uprn
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PairScore:
|
||||
"""One arm's score for one pair: the component comparison plus the
|
||||
calculator-floored SAP residual (calc(predicted) − lodged), None when the
|
||||
calculator could not score the predicted picture."""
|
||||
|
||||
comparison: PredictionComparison
|
||||
sap_residual: Optional[float]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ArmScores:
|
||||
"""One arm aggregated in the ComponentAccuracy shape (ADR-0030):
|
||||
classification maps component -> (hits, applicable-total); residuals maps a
|
||||
numeric component -> signed values; sap_residuals are calc − lodged."""
|
||||
|
||||
classification: dict[str, tuple[int, int]]
|
||||
residuals: dict[str, list[float]]
|
||||
sap_residuals: list[float]
|
||||
|
||||
|
||||
def aggregate(scores: list[PairScore]) -> ArmScores:
|
||||
counts: dict[str, list[int]] = defaultdict(lambda: [0, 0])
|
||||
residuals: dict[str, list[float]] = defaultdict(list)
|
||||
sap_residuals: list[float] = []
|
||||
for score in scores:
|
||||
comparison = score.comparison
|
||||
for component, hit in comparison.categorical_hits.items():
|
||||
if hit is None:
|
||||
continue
|
||||
counts[component][1] += 1
|
||||
if hit:
|
||||
counts[component][0] += 1
|
||||
residuals["floor_area_m2"].append(comparison.floor_area_residual)
|
||||
residuals["building_parts"].append(float(comparison.building_parts_residual))
|
||||
residuals["window_count"].append(float(comparison.window_count_residual))
|
||||
residuals["total_window_area_m2"].append(
|
||||
comparison.total_window_area_residual
|
||||
)
|
||||
residuals["door_count"].append(float(comparison.door_count_residual))
|
||||
if score.sap_residual is not None:
|
||||
sap_residuals.append(score.sap_residual)
|
||||
return ArmScores(
|
||||
classification={k: (v[0], v[1]) for k, v in counts.items()},
|
||||
residuals=dict(residuals),
|
||||
sap_residuals=sap_residuals,
|
||||
)
|
||||
|
||||
|
||||
def _mean_abs(values: list[float]) -> str:
|
||||
return f"{sum(abs(v) for v in values) / len(values):.1f}" if values else "n/a"
|
||||
|
||||
|
||||
def _rate_cell(hits: tuple[int, int]) -> str:
|
||||
hit, total = hits
|
||||
return f"{hit}/{total} ({hit / total:.0%})" if total else "n/a"
|
||||
|
||||
|
||||
def format_report(plain: ArmScores, conditioned: ArmScores, pairs: int) -> str:
|
||||
"""The two arms side by side, in the corpus gate's shape: classification
|
||||
hit-rates per component, then mean-abs residuals, then the secondary SAP
|
||||
residual."""
|
||||
components = sorted(set(plain.classification) | set(conditioned.classification))
|
||||
lines = [
|
||||
f"# Expired-Enhanced Prediction pairs report ({pairs} pairs)",
|
||||
"",
|
||||
"## Classification hit-rates (hits/applicable)",
|
||||
"",
|
||||
"| component | plain | conditioned |",
|
||||
"|---|---|---|",
|
||||
]
|
||||
for component in components:
|
||||
p = _rate_cell(plain.classification.get(component, (0, 0)))
|
||||
c = _rate_cell(conditioned.classification.get(component, (0, 0)))
|
||||
lines.append(f"| {component} | {p} | {c} |")
|
||||
lines += [
|
||||
"",
|
||||
"## Mean absolute residuals (predicted − actual)",
|
||||
"",
|
||||
"| component | plain | conditioned |",
|
||||
"|---|---|---|",
|
||||
]
|
||||
for component in _RESIDUAL_COMPONENTS:
|
||||
p = _mean_abs(plain.residuals.get(component, []))
|
||||
c = _mean_abs(conditioned.residuals.get(component, []))
|
||||
lines.append(f"| {component} | {p} | {c} |")
|
||||
lines += [
|
||||
"",
|
||||
"## SAP residual — secondary, calculator-floored (calc(predicted) − lodged)",
|
||||
"",
|
||||
"| metric | plain | conditioned |",
|
||||
"|---|---|---|",
|
||||
f"| mean abs | {_mean_abs(plain.sap_residuals)} "
|
||||
f"| {_mean_abs(conditioned.sap_residuals)} |",
|
||||
f"| scored | {len(plain.sap_residuals)} | {len(conditioned.sap_residuals)} |",
|
||||
]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _predict_arm(
|
||||
target: Optional[PredictionTarget],
|
||||
cohort: list[ComparableProperty],
|
||||
predictor: EpcPrediction,
|
||||
) -> Optional[EpcPropertyData]:
|
||||
if target is None:
|
||||
return None
|
||||
comparables = select_comparables(target, cohort)
|
||||
if not comparables.members:
|
||||
return None
|
||||
return predictor.predict(target, comparables)
|
||||
|
||||
|
||||
def _part0(epc: EpcPropertyData) -> Optional[object]:
|
||||
parts = epc.sap_building_parts
|
||||
return parts[0] if parts else None
|
||||
|
||||
|
||||
def _actual_roof_form(epc: EpcPropertyData) -> Optional[str]:
|
||||
part = _part0(epc)
|
||||
return roof_form_of_construction(getattr(part, "roof_construction", None))
|
||||
|
||||
|
||||
def _actual_age_band(epc: EpcPropertyData) -> Optional[object]:
|
||||
part = _part0(epc)
|
||||
return getattr(part, "construction_age_band", None)
|
||||
|
||||
|
||||
def _actual_wall(epc: EpcPropertyData) -> Optional[object]:
|
||||
part = _part0(epc)
|
||||
return getattr(part, "wall_construction", None)
|
||||
|
||||
|
||||
def _actual_fuel(epc: EpcPropertyData) -> Optional[object]:
|
||||
details = epc.sap_heating.main_heating_details
|
||||
return details[0].main_fuel_type if details else None
|
||||
|
||||
|
||||
def _tfa_within_band(actual_tfa: float, hist_tfa: Optional[float]) -> Optional[bool]:
|
||||
if hist_tfa is None:
|
||||
return None
|
||||
return abs(actual_tfa - hist_tfa) <= FLOOR_AREA_TOLERANCE * hist_tfa
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LadderStep:
|
||||
"""One conditioning filter's fate in the sequential relax ladder: how many
|
||||
of the incoming cohort matched, and whether it engaged (matches >= k)."""
|
||||
|
||||
matches: int
|
||||
cohort_before: int
|
||||
engaged: bool
|
||||
|
||||
|
||||
def simulate_conditioning_ladder(
|
||||
base: list[ComparableProperty],
|
||||
*,
|
||||
age_band: Optional[str],
|
||||
main_fuel: Optional[int],
|
||||
total_floor_area_m2: Optional[float],
|
||||
roof_form: Optional[str] = None,
|
||||
minimum_cohort: int = 5,
|
||||
) -> dict[str, Optional[LadderStep]]:
|
||||
"""Replay select_comparables' age->fuel->TFA conditioning sequence over the
|
||||
plain arm's selected cohort, recording per filter whether it ENGAGED
|
||||
(>= minimum_cohort matches survive) or RELAXED. None = attribute unresolved,
|
||||
filter never active."""
|
||||
steps: dict[str, Optional[LadderStep]] = {}
|
||||
cohort = list(base)
|
||||
|
||||
def apply(name: str, active: bool, matches: list[ComparableProperty]) -> None:
|
||||
nonlocal cohort
|
||||
if not active:
|
||||
steps[name] = None
|
||||
return
|
||||
engaged = len(matches) >= minimum_cohort
|
||||
steps[name] = LadderStep(len(matches), len(cohort), engaged)
|
||||
if engaged:
|
||||
cohort = matches
|
||||
|
||||
apply(
|
||||
"roof_form",
|
||||
roof_form is not None,
|
||||
[c for c in cohort if _actual_roof_form(c.epc) == roof_form],
|
||||
)
|
||||
apply(
|
||||
"construction_age_band",
|
||||
age_band is not None,
|
||||
[c for c in cohort if age_bands_within_one(_actual_age_band(c.epc), age_band)],
|
||||
)
|
||||
apply(
|
||||
"main_fuel",
|
||||
main_fuel is not None,
|
||||
[c for c in cohort if _actual_fuel(c.epc) == main_fuel],
|
||||
)
|
||||
apply(
|
||||
"total_floor_area",
|
||||
total_floor_area_m2 is not None,
|
||||
[
|
||||
c
|
||||
for c in cohort
|
||||
if total_floor_area_m2 is not None
|
||||
and abs(c.epc.total_floor_area_m2 - total_floor_area_m2)
|
||||
<= FLOOR_AREA_TOLERANCE * total_floor_area_m2
|
||||
],
|
||||
)
|
||||
return steps
|
||||
|
||||
|
||||
def format_diagnosis(rows: list[dict[str, object]]) -> str:
|
||||
"""Aggregate the per-pair telemetry into the two diagnosis tables: did each
|
||||
conditioning filter ever ENGAGE, and does the historic value still AGREE
|
||||
with the newly lodged one (the staleness measurement)."""
|
||||
if not rows:
|
||||
return ""
|
||||
filters = ("roof_form", "construction_age_band", "main_fuel", "total_floor_area")
|
||||
lines = [
|
||||
"",
|
||||
"## Diagnosis — filter engagement (conditioned arm)",
|
||||
"",
|
||||
"| filter | resolved | engaged | relaxed (too few matches) |",
|
||||
"|---|---|---|---|",
|
||||
]
|
||||
for name in filters:
|
||||
resolved = engaged = 0
|
||||
for row in rows:
|
||||
step = row.get(f"ladder_{name}")
|
||||
if step is None:
|
||||
continue
|
||||
resolved += 1
|
||||
if isinstance(step, LadderStep) and step.engaged:
|
||||
engaged += 1
|
||||
lines.append(
|
||||
f"| {name} | {resolved}/{len(rows)} | {engaged}/{resolved or 1} "
|
||||
f"| {resolved - engaged}/{resolved or 1} |"
|
||||
)
|
||||
attrs = (
|
||||
"property_type",
|
||||
"built_form",
|
||||
"wall_construction",
|
||||
"roof_form",
|
||||
"construction_age_band",
|
||||
"main_fuel",
|
||||
"tfa_within_band",
|
||||
)
|
||||
lines += [
|
||||
"",
|
||||
"## Diagnosis — historic vs newly-lodged agreement (staleness)",
|
||||
"",
|
||||
"| attribute | historic resolved | agrees with new cert |",
|
||||
"|---|---|---|",
|
||||
]
|
||||
for name in attrs:
|
||||
resolved = agrees = 0
|
||||
for row in rows:
|
||||
value = row.get(f"agrees_{name}")
|
||||
if value is None:
|
||||
continue
|
||||
resolved += 1
|
||||
if value:
|
||||
agrees += 1
|
||||
pct = f" ({agrees / resolved:.0%})" if resolved else ""
|
||||
lines.append(f"| {name} | {resolved}/{len(rows)} | {agrees}/{resolved or 1}{pct} |")
|
||||
sizes: list[int] = [
|
||||
size
|
||||
for row in rows
|
||||
if isinstance((size := row.get("plain_cohort_size")), int)
|
||||
]
|
||||
if sizes:
|
||||
lines += [
|
||||
"",
|
||||
f"Mean plain-arm cohort size: {sum(sizes) / len(sizes):.1f} "
|
||||
f"(min {min(sizes)}, max {max(sizes)}); relax threshold k=5.",
|
||||
]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def run( # pragma: no cover - live IO composition
|
||||
postcodes: list[str],
|
||||
telemetry_path: Optional[Path] = None,
|
||||
cache_dir: Optional[Path] = None,
|
||||
) -> str:
|
||||
from domain.sap10_calculator.calculator import Sap10Calculator
|
||||
from infrastructure.epc_client.epc_client_service import EpcClientService
|
||||
from repositories.comparable_properties.epc_comparable_properties_repository import (
|
||||
EpcComparablePropertiesRepository,
|
||||
)
|
||||
from repositories.geospatial.geospatial_s3_repository import (
|
||||
GeospatialS3Repository,
|
||||
)
|
||||
from repositories.historic_epc.historic_epc_s3_repository import (
|
||||
HistoricEpcS3Repository,
|
||||
)
|
||||
from scripts.e2e_common import load_env, s3_parquet_reader
|
||||
|
||||
load_env()
|
||||
if cache_dir is not None:
|
||||
from scripts.epc_disk_cache import JsonCachingEpcClient
|
||||
|
||||
epc_client: EpcClientService = JsonCachingEpcClient(
|
||||
os.environ["OPEN_EPC_API_TOKEN"], cache_dir
|
||||
)
|
||||
else:
|
||||
epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"])
|
||||
geospatial = GeospatialS3Repository(s3_parquet_reader(os.environ["DATA_BUCKET"]))
|
||||
comparables_repo = EpcComparablePropertiesRepository(epc_client, geospatial)
|
||||
historic_repo = HistoricEpcS3Repository.with_default_s3_client()
|
||||
predictor = EpcPrediction()
|
||||
calculator = Sap10Calculator()
|
||||
|
||||
def sap_residual(
|
||||
predicted: Optional[EpcPropertyData], actual: EpcPropertyData
|
||||
) -> Optional[float]:
|
||||
if predicted is None or actual.energy_rating_current is None:
|
||||
return None
|
||||
try:
|
||||
calculated: float = calculator.calculate(predicted).sap_score_continuous
|
||||
except Exception as error: # the calculator strict-raises on gaps
|
||||
print(f" calculator raised: {error}", file=sys.stderr)
|
||||
return None
|
||||
return calculated - float(actual.energy_rating_current)
|
||||
|
||||
plain_scores: list[PairScore] = []
|
||||
conditioned_scores: list[PairScore] = []
|
||||
telemetry: list[dict[str, object]] = []
|
||||
pairs = 0
|
||||
|
||||
def emit_telemetry(row: dict[str, object]) -> None:
|
||||
# Append incrementally so a late crash loses nothing.
|
||||
telemetry.append(row)
|
||||
if telemetry_path is None:
|
||||
return
|
||||
import dataclasses as _dc
|
||||
import json as _json
|
||||
|
||||
serialisable = {
|
||||
k: (_dc.asdict(v) if isinstance(v, LadderStep) else v)
|
||||
for k, v in row.items()
|
||||
}
|
||||
with telemetry_path.open("a") as handle:
|
||||
handle.write(_json.dumps(serialisable) + "\n")
|
||||
|
||||
if telemetry_path is not None and telemetry_path.exists():
|
||||
telemetry_path.unlink()
|
||||
for index, raw_postcode in enumerate(postcodes):
|
||||
postcode = str(Postcode(raw_postcode))
|
||||
historic = latest_pre_2012_by_uprn(
|
||||
historic_repo.get_for_postcode(Postcode(raw_postcode))
|
||||
)
|
||||
# Pair-check first: only a postcode with a SAP-10.2 relodgement of a
|
||||
# pre-2012 UPRN pays for the cohort fetch. A cert the mapper can't yet
|
||||
# map (strict-raise) can't be a validation target either — skip it and
|
||||
# keep sweeping; one bad cert must not kill a multi-hour run.
|
||||
paired: list[tuple[str, HistoricEpc, EpcPropertyData]] = []
|
||||
for uprn, record in historic.items():
|
||||
try:
|
||||
actual = epc_client.get_by_uprn(int(uprn))
|
||||
except Exception as error:
|
||||
print(f" {uprn}: unmappable cert skipped: {error}", file=sys.stderr)
|
||||
continue
|
||||
if actual is not None and actual.sap_version == _VALIDATION_SAP_VERSION:
|
||||
paired.append((uprn, record, actual))
|
||||
print(
|
||||
f"[{index + 1}/{len(postcodes)}] {postcode}: "
|
||||
f"{len(historic)} pre-2012 UPRNs, {len(paired)} pairs",
|
||||
file=sys.stderr,
|
||||
flush=True,
|
||||
)
|
||||
if not paired:
|
||||
continue
|
||||
try:
|
||||
cohort = comparables_repo.candidates_for(postcode)
|
||||
except Exception as error:
|
||||
print(f" {postcode}: cohort fetch failed: {error}", file=sys.stderr)
|
||||
continue
|
||||
for uprn, record, actual in paired:
|
||||
pairs += 1
|
||||
loo_cohort = [c for c in cohort if c.epc.uprn != int(uprn)]
|
||||
identity = PropertyIdentity(
|
||||
portfolio_id=0, postcode=postcode, address=record.address, uprn=int(uprn)
|
||||
)
|
||||
conditioning = conditioning_from_historic(record)
|
||||
attributes = attributes_with_historic_fallback(None, conditioning)
|
||||
plain_target = build_prediction_target(identity, None, attributes)
|
||||
conditioned_target = (
|
||||
target_with_conditioning(plain_target, conditioning)
|
||||
if plain_target is not None
|
||||
else None
|
||||
)
|
||||
plain = _predict_arm(plain_target, loo_cohort, predictor)
|
||||
conditioned = _predict_arm(conditioned_target, loo_cohort, predictor)
|
||||
if plain_target is not None:
|
||||
base = list(select_comparables(plain_target, loo_cohort).members)
|
||||
ladder = simulate_conditioning_ladder(
|
||||
base,
|
||||
age_band=conditioning.construction_age_band,
|
||||
main_fuel=conditioning.main_fuel,
|
||||
total_floor_area_m2=conditioning.total_floor_area_m2,
|
||||
roof_form=conditioning.roof_form,
|
||||
)
|
||||
emit_telemetry(
|
||||
{
|
||||
"postcode": postcode,
|
||||
"uprn": uprn,
|
||||
"plain_cohort_size": len(base),
|
||||
# Raw values alongside the booleans, so band-width and
|
||||
# near-miss questions are answerable post hoc.
|
||||
"historic_age_band": conditioning.construction_age_band,
|
||||
"actual_age_band": _actual_age_band(actual),
|
||||
"historic_tfa": conditioning.total_floor_area_m2,
|
||||
"actual_tfa": actual.total_floor_area_m2,
|
||||
"historic_fuel": conditioning.main_fuel,
|
||||
"actual_fuel": _actual_fuel(actual),
|
||||
"historic_fuel_text": record.main_fuel,
|
||||
**{f"ladder_{k}": v for k, v in ladder.items()},
|
||||
"agrees_property_type": (
|
||||
None
|
||||
if conditioning.property_type is None
|
||||
else conditioning.property_type == actual.property_type
|
||||
),
|
||||
"agrees_built_form": (
|
||||
None
|
||||
if conditioning.built_form is None
|
||||
else conditioning.built_form == actual.built_form
|
||||
),
|
||||
"agrees_roof_form": (
|
||||
None
|
||||
if conditioning.roof_form is None
|
||||
else conditioning.roof_form == _actual_roof_form(actual)
|
||||
),
|
||||
"agrees_wall_construction": (
|
||||
None
|
||||
if conditioning.wall_construction is None
|
||||
else conditioning.wall_construction == _actual_wall(actual)
|
||||
),
|
||||
"agrees_construction_age_band": (
|
||||
None
|
||||
if conditioning.construction_age_band is None
|
||||
else conditioning.construction_age_band
|
||||
== _actual_age_band(actual)
|
||||
),
|
||||
"agrees_main_fuel": (
|
||||
None
|
||||
if conditioning.main_fuel is None
|
||||
else conditioning.main_fuel == _actual_fuel(actual)
|
||||
),
|
||||
"agrees_tfa_within_band": _tfa_within_band(
|
||||
actual.total_floor_area_m2,
|
||||
conditioning.total_floor_area_m2,
|
||||
),
|
||||
}
|
||||
)
|
||||
if plain is not None:
|
||||
plain_scores.append(
|
||||
PairScore(compare_prediction(plain, actual), sap_residual(plain, actual))
|
||||
)
|
||||
if conditioned is not None:
|
||||
conditioned_scores.append(
|
||||
PairScore(
|
||||
compare_prediction(conditioned, actual),
|
||||
sap_residual(conditioned, actual),
|
||||
)
|
||||
)
|
||||
|
||||
report = format_report(
|
||||
aggregate(plain_scores), aggregate(conditioned_scores), pairs
|
||||
)
|
||||
return report + format_diagnosis(telemetry)
|
||||
|
||||
|
||||
def main() -> None: # pragma: no cover - CLI entry
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("postcodes", nargs="*", help="postcodes to scan")
|
||||
parser.add_argument("--postcodes-file", type=Path, default=None)
|
||||
parser.add_argument("--out", type=Path, default=None, help="write the report here")
|
||||
parser.add_argument(
|
||||
"--telemetry", type=Path, default=None, help="write per-pair JSONL here"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--cache-dir",
|
||||
type=Path,
|
||||
default=None,
|
||||
help="raw-JSON disk cache for API responses (fast re-runs; fixture material)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
postcodes: list[str] = list(args.postcodes)
|
||||
if args.postcodes_file is not None:
|
||||
postcodes += [
|
||||
line.strip()
|
||||
for line in args.postcodes_file.read_text().splitlines()
|
||||
if line.strip()
|
||||
]
|
||||
if not postcodes:
|
||||
parser.error("no postcodes given")
|
||||
report = run(postcodes, telemetry_path=args.telemetry, cache_dir=args.cache_dir)
|
||||
if args.out is not None:
|
||||
args.out.write_text(report + "\n")
|
||||
print(report)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
114
scripts/roof_construction_code_sweep.py
Normal file
114
scripts/roof_construction_code_sweep.py
Normal file
|
|
@ -0,0 +1,114 @@
|
|||
"""Pin the RdSAP ``roof_construction`` code table empirically (ADR-0054 gap).
|
||||
|
||||
The gov-EPC API lodges ``sap_building_parts[].roof_construction`` as an int
|
||||
whose enumeration is published nowhere we hold; the historic dump lodges roofs
|
||||
as display text. The wall equivalent was pinned the same way this script works
|
||||
(the basement wall code 6: a bulk co-occurrence sweep). Method:
|
||||
|
||||
- fetch every cert in the given postcodes' cohorts;
|
||||
- keep only certs with EXACTLY one building part and one roofs element, so the
|
||||
(code, description) pairing is unambiguous;
|
||||
- cross-tabulate code x description-prefix (text before the first comma) and
|
||||
report each code's dominant prefix with its purity.
|
||||
|
||||
Usage:
|
||||
python scripts/roof_construction_code_sweep.py --postcodes-file pcs.txt \
|
||||
--cache-dir .epc_cache --out roof_codes.md
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
from collections import Counter, defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from datatypes.epc.domain.epc_property_data import EpcPropertyData # noqa: E402
|
||||
|
||||
|
||||
def prefix_of(description: str) -> str:
|
||||
"""The roof description's construction/form half — the text before the
|
||||
first comma ("Pitched, 100 mm loft insulation" -> "Pitched")."""
|
||||
return description.split(",", 1)[0].strip()
|
||||
|
||||
|
||||
def tabulate(certs: list[EpcPropertyData]) -> dict[int, Counter[str]]:
|
||||
"""code -> Counter(description prefix), over unambiguous certs only."""
|
||||
table: dict[int, Counter[str]] = defaultdict(Counter)
|
||||
for cert in certs:
|
||||
if len(cert.sap_building_parts) != 1 or len(cert.roofs) != 1:
|
||||
continue
|
||||
code = cert.sap_building_parts[0].roof_construction
|
||||
if code is None:
|
||||
continue
|
||||
table[code][prefix_of(cert.roofs[0].description)] += 1
|
||||
return dict(table)
|
||||
|
||||
|
||||
def format_table(table: dict[int, Counter[str]]) -> str:
|
||||
lines = [
|
||||
"# roof_construction code x description-prefix co-occurrence",
|
||||
"",
|
||||
"| code | n | dominant prefix | purity | runners-up |",
|
||||
"|---|---|---|---|---|",
|
||||
]
|
||||
for code in sorted(table):
|
||||
counts = table[code]
|
||||
total = sum(counts.values())
|
||||
(top, top_n), *rest = counts.most_common()
|
||||
runners = ", ".join(f"{p} ({n})" for p, n in rest[:3]) or "—"
|
||||
lines.append(
|
||||
f"| {code} | {total} | {top} | {top_n / total:.0%} | {runners} |"
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main() -> None: # pragma: no cover - CLI/IO composition
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--postcodes-file", type=Path, required=True)
|
||||
parser.add_argument("--cache-dir", type=Path, required=True)
|
||||
parser.add_argument("--out", type=Path, default=None)
|
||||
args = parser.parse_args()
|
||||
|
||||
from scripts.e2e_common import load_env
|
||||
from scripts.epc_disk_cache import JsonCachingEpcClient
|
||||
|
||||
load_env()
|
||||
client = JsonCachingEpcClient(os.environ["OPEN_EPC_API_TOKEN"], args.cache_dir)
|
||||
|
||||
certs: list[EpcPropertyData] = []
|
||||
postcodes = [
|
||||
line.strip()
|
||||
for line in args.postcodes_file.read_text().splitlines()
|
||||
if line.strip()
|
||||
]
|
||||
for index, postcode in enumerate(postcodes):
|
||||
try:
|
||||
results = client.search_by_postcode(postcode)
|
||||
except Exception as error:
|
||||
print(f"{postcode}: search failed: {error}", file=sys.stderr)
|
||||
continue
|
||||
for result in results:
|
||||
try:
|
||||
certs.append(
|
||||
client.get_by_certificate_number(result.certificate_number)
|
||||
)
|
||||
except Exception:
|
||||
continue # unmappable cert — not usable evidence
|
||||
print(
|
||||
f"[{index + 1}/{len(postcodes)}] {postcode}: {len(certs)} certs so far",
|
||||
file=sys.stderr,
|
||||
flush=True,
|
||||
)
|
||||
|
||||
report = format_table(tabulate(certs))
|
||||
if args.out is not None:
|
||||
args.out.write_text(report + "\n")
|
||||
print(report)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -8,7 +8,12 @@ enough remain, weighted by recency × similarity. Pure domain logic.
|
|||
from datetime import date
|
||||
from typing import Optional, Union
|
||||
|
||||
from datatypes.epc.domain.epc_property_data import EpcPropertyData, SapBuildingPart
|
||||
from datatypes.epc.domain.epc_property_data import (
|
||||
EpcPropertyData,
|
||||
MainHeatingDetail,
|
||||
SapBuildingPart,
|
||||
SapHeating,
|
||||
)
|
||||
from domain.epc_prediction.comparable_properties import (
|
||||
ComparableProperty,
|
||||
ComparableProperties,
|
||||
|
|
@ -25,6 +30,10 @@ def _comparable(
|
|||
wall_construction: Optional[Union[int, str]] = None,
|
||||
address: Optional[str] = None,
|
||||
registration_date: Optional[date] = None,
|
||||
construction_age_band: Optional[str] = None,
|
||||
main_fuel: Optional[int] = None,
|
||||
total_floor_area_m2: Optional[float] = None,
|
||||
roof_construction: Optional[int] = None,
|
||||
) -> ComparableProperty:
|
||||
"""A ComparableProperty carrying only the fields under test (opaque EpcPropertyData
|
||||
with property_type / built_form / main wall set — the partial-instance idiom)."""
|
||||
|
|
@ -34,7 +43,19 @@ def _comparable(
|
|||
main: SapBuildingPart = object.__new__(SapBuildingPart)
|
||||
if wall_construction is not None:
|
||||
main.wall_construction = wall_construction
|
||||
if construction_age_band is not None:
|
||||
main.construction_age_band = construction_age_band
|
||||
if roof_construction is not None:
|
||||
main.roof_construction = roof_construction
|
||||
epc.sap_building_parts = [main]
|
||||
if main_fuel is not None:
|
||||
detail: MainHeatingDetail = object.__new__(MainHeatingDetail)
|
||||
detail.main_fuel_type = main_fuel
|
||||
heating: SapHeating = object.__new__(SapHeating)
|
||||
heating.main_heating_details = [detail]
|
||||
epc.sap_heating = heating
|
||||
if total_floor_area_m2 is not None:
|
||||
epc.total_floor_area_m2 = total_floor_area_m2
|
||||
return ComparableProperty(
|
||||
epc=epc,
|
||||
certificate_number=certificate_number,
|
||||
|
|
@ -171,3 +192,169 @@ def test_known_wall_override_relaxes_when_too_few_match() -> None:
|
|||
|
||||
# Assert — relaxed: all eight houses retained.
|
||||
assert len(result.members) == 8
|
||||
|
||||
|
||||
def test_historic_age_band_conditions_the_cohort() -> None:
|
||||
# Arrange — the expired cert observed band C (1930-1949); 5 band-C houses +
|
||||
# 2 band-G houses in the cohort (ADR-0054).
|
||||
target = PredictionTarget(
|
||||
postcode="LS6 1AA", property_type="2", construction_age_band="C"
|
||||
)
|
||||
candidates = [
|
||||
_comparable(
|
||||
property_type="2", construction_age_band="C", certificate_number=f"C{i}"
|
||||
)
|
||||
for i in range(5)
|
||||
] + [
|
||||
_comparable(
|
||||
property_type="2", construction_age_band="G", certificate_number=f"G{i}"
|
||||
)
|
||||
for i in range(2)
|
||||
]
|
||||
|
||||
# Act
|
||||
result: ComparableProperties = select_comparables(
|
||||
target, candidates, minimum_cohort=5
|
||||
)
|
||||
|
||||
# Assert — only the same-age-band comparables remain.
|
||||
assert {c.certificate_number for c in result.members} == {
|
||||
"C0", "C1", "C2", "C3", "C4"
|
||||
}
|
||||
|
||||
|
||||
def test_historic_main_fuel_conditions_the_cohort() -> None:
|
||||
# Arrange — the expired cert observed mains gas (26); 5 gas + 2 electric
|
||||
# (29) houses in the cohort (ADR-0054).
|
||||
target = PredictionTarget(postcode="LS6 1AA", property_type="2", main_fuel=26)
|
||||
candidates = [
|
||||
_comparable(property_type="2", main_fuel=26, certificate_number=f"G{i}")
|
||||
for i in range(5)
|
||||
] + [
|
||||
_comparable(property_type="2", main_fuel=29, certificate_number=f"E{i}")
|
||||
for i in range(2)
|
||||
]
|
||||
|
||||
# Act
|
||||
result: ComparableProperties = select_comparables(
|
||||
target, candidates, minimum_cohort=5
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert {c.certificate_number for c in result.members} == {
|
||||
"G0", "G1", "G2", "G3", "G4"
|
||||
}
|
||||
|
||||
|
||||
def test_floor_area_band_keeps_comparables_within_20_percent() -> None:
|
||||
# Arrange — the expired cert observed 100 m²; the band is ±20% (ADR-0054 as
|
||||
# amended: the 439-pair harness put historic-vs-new TFA agreement at only
|
||||
# 45% within ±5% but 82% within ±20% — a coarse dwelling-size filter, not a
|
||||
# precision match): 96, 118 and 84 are in, 125 and 79 are out.
|
||||
target = PredictionTarget(
|
||||
postcode="LS6 1AA", property_type="2", total_floor_area_m2=100.0
|
||||
)
|
||||
candidates = [
|
||||
_comparable(property_type="2", total_floor_area_m2=96.0, certificate_number="A"),
|
||||
_comparable(
|
||||
property_type="2", total_floor_area_m2=118.0, certificate_number="B"
|
||||
),
|
||||
_comparable(property_type="2", total_floor_area_m2=84.0, certificate_number="C"),
|
||||
_comparable(
|
||||
property_type="2", total_floor_area_m2=125.0, certificate_number="X"
|
||||
),
|
||||
_comparable(property_type="2", total_floor_area_m2=79.0, certificate_number="Y"),
|
||||
]
|
||||
|
||||
# Act
|
||||
result: ComparableProperties = select_comparables(
|
||||
target, candidates, minimum_cohort=2
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert {c.certificate_number for c in result.members} == {"A", "B", "C"}
|
||||
|
||||
|
||||
def test_historic_age_band_conditions_within_one_band() -> None:
|
||||
# Arrange — the expired cert observed band C, but assessors re-band ±1
|
||||
# constantly (harness: 52% exact agreement, 90% within one band), so the
|
||||
# filter keeps the band NEIGHBOURHOOD: B/C/D match, G does not (ADR-0054 as
|
||||
# amended).
|
||||
target = PredictionTarget(
|
||||
postcode="LS6 1AA", property_type="2", construction_age_band="C"
|
||||
)
|
||||
near = ["B", "C", "D", "D", "B"]
|
||||
candidates = [
|
||||
_comparable(
|
||||
property_type="2",
|
||||
construction_age_band=band,
|
||||
certificate_number=f"N{i}",
|
||||
)
|
||||
for i, band in enumerate(near)
|
||||
] + [
|
||||
_comparable(
|
||||
property_type="2", construction_age_band="G", certificate_number=f"G{i}"
|
||||
)
|
||||
for i in range(2)
|
||||
]
|
||||
|
||||
# Act
|
||||
result: ComparableProperties = select_comparables(
|
||||
target, candidates, minimum_cohort=5
|
||||
)
|
||||
|
||||
# Assert — the five band-neighbourhood comparables survive; band G drops.
|
||||
assert {c.certificate_number for c in result.members} == {
|
||||
"N0", "N1", "N2", "N3", "N4"
|
||||
}
|
||||
|
||||
|
||||
def test_historic_roof_form_conditions_the_cohort_by_family() -> None:
|
||||
# Arrange — the expired cert observed a pitched roof. The API's
|
||||
# roof_construction codes group into FORM families (empirical sweep:
|
||||
# 4/5/8 = pitched, 1 = flat), so all pitched-family comparables match and
|
||||
# the flat one drops (ADR-0054 as amended).
|
||||
target = PredictionTarget(postcode="LS6 1AA", property_type="2", roof_form="pitched")
|
||||
candidates = [
|
||||
_comparable(property_type="2", roof_construction=4, certificate_number="P4a"),
|
||||
_comparable(property_type="2", roof_construction=4, certificate_number="P4b"),
|
||||
_comparable(property_type="2", roof_construction=5, certificate_number="P5"),
|
||||
_comparable(property_type="2", roof_construction=8, certificate_number="P8"),
|
||||
_comparable(property_type="2", roof_construction=4, certificate_number="P4c"),
|
||||
_comparable(property_type="2", roof_construction=1, certificate_number="F1"),
|
||||
]
|
||||
|
||||
# Act
|
||||
result: ComparableProperties = select_comparables(
|
||||
target, candidates, minimum_cohort=5
|
||||
)
|
||||
|
||||
# Assert — the whole pitched family survives; the flat roof drops.
|
||||
assert {c.certificate_number for c in result.members} == {
|
||||
"P4a", "P4b", "P4c", "P5", "P8"
|
||||
}
|
||||
|
||||
|
||||
def test_floor_area_band_relaxes_when_too_few_match() -> None:
|
||||
# Arrange — only one comparable inside the ±5% band (< k=2): the band must
|
||||
# relax rather than starve the cohort (graceful degradation, ADR-0029).
|
||||
target = PredictionTarget(
|
||||
postcode="LS6 1AA", property_type="2", total_floor_area_m2=100.0
|
||||
)
|
||||
candidates = [
|
||||
_comparable(property_type="2", total_floor_area_m2=99.0, certificate_number="A"),
|
||||
_comparable(
|
||||
property_type="2", total_floor_area_m2=140.0, certificate_number="X"
|
||||
),
|
||||
_comparable(
|
||||
property_type="2", total_floor_area_m2=150.0, certificate_number="Y"
|
||||
),
|
||||
]
|
||||
|
||||
# Act
|
||||
result: ComparableProperties = select_comparables(
|
||||
target, candidates, minimum_cohort=2
|
||||
)
|
||||
|
||||
# Assert — relaxed: all three retained.
|
||||
assert len(result.members) == 3
|
||||
|
|
|
|||
165
tests/domain/epc_prediction/test_expired_pairs_gate.py
Normal file
165
tests/domain/epc_prediction/test_expired_pairs_gate.py
Normal file
|
|
@ -0,0 +1,165 @@
|
|||
"""Tier-1 ratcheting gate for Expired-Enhanced Prediction (ADR-0054).
|
||||
|
||||
Replays the pairs harness OFFLINE over the committed, anonymised fixture
|
||||
(`tests/fixtures/expired_prediction_pairs` — pre-2012 historic records paired
|
||||
with their lodged SAP-10.2 certs and full postcode cohorts, frozen from the
|
||||
2,000-postcode national sweep). Both arms run the real production path minus
|
||||
the network: the raw payloads go through `EpcPropertyDataMapper`, conditioning
|
||||
through `conditioning_from_historic`, selection through `select_comparables`,
|
||||
synthesis through `EpcPrediction`. Deterministic, so every run reproduces the
|
||||
same numbers exactly — a failure is a real regression in the conditioning
|
||||
path, never sample noise.
|
||||
|
||||
Floors are the measured values over the frozen fixture and only ever
|
||||
**tighten** (the repo's no-tolerance-widening ethos), exactly like the
|
||||
Component Accuracy gate this extends.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import pytest
|
||||
|
||||
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,
|
||||
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_comparison import (
|
||||
PredictionComparison,
|
||||
compare_prediction,
|
||||
)
|
||||
from domain.epc_prediction.prediction_target import build_prediction_target
|
||||
from domain.property.property import PropertyIdentity
|
||||
from harness.epc_prediction_corpus import comparable_from_payload
|
||||
|
||||
_FIXTURE = (
|
||||
Path(__file__).parents[3]
|
||||
/ "tests"
|
||||
/ "fixtures"
|
||||
/ "expired_prediction_pairs.json"
|
||||
)
|
||||
|
||||
# Conditioned-arm classification floors (hit-rate over the frozen 30-pair /
|
||||
# 28-scoreable fixture) and residual ceilings — the measured values; tighten,
|
||||
# never loosen. The general (unconditioned) prediction floors live in
|
||||
# test_component_accuracy_gate.py; this gate guards the CONDITIONING path.
|
||||
_CLASSIFICATION_FLOORS: dict[str, float] = {
|
||||
"construction_age_band": 0.5357,
|
||||
"construction_age_band_pm1": 0.8214,
|
||||
"cylinder_insulation_type": 0.8000,
|
||||
"floor_construction": 0.8636,
|
||||
"floor_insulation": 1.0000,
|
||||
"has_hot_water_cylinder": 0.8214,
|
||||
"has_pv": 0.8929,
|
||||
"has_room_in_roof": 0.8571,
|
||||
"heating_main_category": 1.0000,
|
||||
"heating_main_control": 0.6071,
|
||||
"heating_main_fuel": 1.0000,
|
||||
"modal_glazing_type": 0.5000,
|
||||
"roof_construction": 0.7143,
|
||||
"roof_insulation_thickness": 0.3333,
|
||||
"roof_insulation_thickness_pm1": 0.4815,
|
||||
"secondary_heating_type": 0.1667,
|
||||
"solar_water_heating": 1.0000,
|
||||
"wall_construction": 0.8929,
|
||||
"wall_insulation_type": 0.7500,
|
||||
"water_heating_code": 1.0000,
|
||||
"water_heating_fuel": 1.0000,
|
||||
}
|
||||
_FLOOR_AREA_MAE_CEILING: Optional[float] = 21.121
|
||||
|
||||
|
||||
def _pairs() -> list[tuple[HistoricEpc, EpcPropertyData, list[ComparableProperty]]]:
|
||||
corpus = json.loads(_FIXTURE.read_text())
|
||||
cohorts: dict[str, list[ComparableProperty]] = {
|
||||
postcode: [
|
||||
comparable
|
||||
for token, payload in payloads.items()
|
||||
if (comparable := comparable_from_payload(token, payload, {})) is not None
|
||||
]
|
||||
for postcode, payloads in corpus["cohorts"].items()
|
||||
}
|
||||
return [
|
||||
(
|
||||
HistoricEpc(**pair["historic"]),
|
||||
EpcPropertyDataMapper.from_api_response(corpus["actuals"][pair["actual"]]),
|
||||
cohorts[pair["postcode"]],
|
||||
)
|
||||
for pair in corpus["pairs"]
|
||||
]
|
||||
|
||||
|
||||
def _conditioned_comparisons() -> list[PredictionComparison]:
|
||||
predictor = EpcPrediction()
|
||||
comparisons: list[PredictionComparison] = []
|
||||
for historic, actual, cohort in _pairs():
|
||||
conditioning = conditioning_from_historic(historic)
|
||||
attributes = attributes_with_historic_fallback(None, conditioning)
|
||||
identity = PropertyIdentity(
|
||||
portfolio_id=0,
|
||||
postcode=historic.postcode,
|
||||
address=historic.address,
|
||||
uprn=int(historic.uprn),
|
||||
)
|
||||
target = build_prediction_target(identity, None, attributes)
|
||||
if target is None:
|
||||
continue
|
||||
target = target_with_conditioning(target, conditioning)
|
||||
loo = [c for c in cohort if c.epc.uprn != int(historic.uprn)]
|
||||
comparables = select_comparables(target, loo)
|
||||
if not comparables.members:
|
||||
continue
|
||||
predicted = predictor.predict(target, comparables)
|
||||
comparisons.append(compare_prediction(predicted, actual))
|
||||
return comparisons
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def comparisons() -> list[PredictionComparison]:
|
||||
if not _FIXTURE.exists():
|
||||
pytest.skip("expired-pairs fixture not present")
|
||||
return _conditioned_comparisons()
|
||||
|
||||
|
||||
def test_fixture_yields_the_expected_pair_count(
|
||||
comparisons: list[PredictionComparison],
|
||||
) -> None:
|
||||
# The frozen fixture must keep producing its full set of scoreable pairs —
|
||||
# a drop means the fixture, the conditioning gate, or selection changed.
|
||||
# (30 frozen; 2 are gated out / find no comparables, deterministically.)
|
||||
assert len(comparisons) == 28
|
||||
|
||||
|
||||
@pytest.mark.parametrize("component,floor", sorted(_CLASSIFICATION_FLOORS.items()))
|
||||
def test_conditioned_classification_rate_does_not_regress(
|
||||
comparisons: list[PredictionComparison], component: str, floor: float
|
||||
) -> None:
|
||||
applicable = [
|
||||
hit
|
||||
for comparison in comparisons
|
||||
if (hit := comparison.categorical_hits.get(component)) is not None
|
||||
]
|
||||
assert applicable, component
|
||||
rate = sum(applicable) / len(applicable)
|
||||
assert rate >= floor - 1e-3, f"{component}: {rate:.4f} < floor {floor:.4f}"
|
||||
|
||||
|
||||
def test_conditioned_floor_area_mae_does_not_regress(
|
||||
comparisons: list[PredictionComparison],
|
||||
) -> None:
|
||||
if _FLOOR_AREA_MAE_CEILING is None:
|
||||
pytest.skip("ceiling not yet pinned")
|
||||
mae = sum(abs(c.floor_area_residual) for c in comparisons) / len(comparisons)
|
||||
assert mae <= _FLOOR_AREA_MAE_CEILING + 1e-3
|
||||
117
tests/domain/epc_prediction/test_historic_conditioning.py
Normal file
117
tests/domain/epc_prediction/test_historic_conditioning.py
Normal file
|
|
@ -0,0 +1,117 @@
|
|||
"""HistoricConditioning resolves an expired Historic EPC's stable attributes
|
||||
into the cohort's code spaces (ADR-0054).
|
||||
|
||||
Only stable attributes are resolved — volatile ones (heating system, glazing,
|
||||
PV, insulation states) never condition prediction. Every resolver degrades to
|
||||
None on an unresolvable value, which leaves its cohort filter inactive.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
|
||||
from datatypes.epc.domain.historic_epc import HistoricEpc
|
||||
from domain.epc_prediction.historic_conditioning import (
|
||||
HistoricConditioning,
|
||||
conditioning_from_historic,
|
||||
)
|
||||
|
||||
|
||||
def _hist(**overrides: str) -> HistoricEpc:
|
||||
fields = {f.name: "" for f in dataclasses.fields(HistoricEpc)}
|
||||
fields.update(overrides)
|
||||
return HistoricEpc(**fields)
|
||||
|
||||
|
||||
def test_resolves_stable_attributes_into_cohort_code_spaces():
|
||||
# Arrange — display-text values exactly as the old register lodged them.
|
||||
record = _hist(
|
||||
property_type="House",
|
||||
built_form="Semi-Detached",
|
||||
walls_description="Cavity wall, as built, no insulation (assumed)",
|
||||
roof_description="Pitched, 100 mm loft insulation",
|
||||
construction_age_band="England and Wales: 1930-1949",
|
||||
main_fuel="mains gas (not community)",
|
||||
total_floor_area="84",
|
||||
)
|
||||
|
||||
# Act
|
||||
conditioning = conditioning_from_historic(record)
|
||||
|
||||
# Assert — each attribute lands in the code space its cohort filter compares.
|
||||
assert isinstance(conditioning, HistoricConditioning)
|
||||
assert conditioning.property_type == "0"
|
||||
assert conditioning.built_form == "2"
|
||||
assert conditioning.wall_construction == 4
|
||||
assert conditioning.roof_form == "pitched"
|
||||
assert conditioning.construction_age_band == "C"
|
||||
assert conditioning.main_fuel == 26
|
||||
assert conditioning.total_floor_area_m2 == 84.0
|
||||
|
||||
|
||||
def test_roof_descriptions_resolve_to_form_families():
|
||||
# Arrange / Act / Assert — the FORM half of the description (before the
|
||||
# comma) maps to a family, because the API's roof_construction codes group
|
||||
# that way (7,974-cert co-occurrence sweep: 1=Flat 98%, 4/5/8=Pitched
|
||||
# 88-99%, 3=another dwelling above 100%; 7/9=another premises above per
|
||||
# the #1452 suppression fix). Insulation state is volatile and ignored.
|
||||
cases = {
|
||||
"Pitched, 250 mm loft insulation": "pitched",
|
||||
"Pitched, no insulation (assumed)": "pitched",
|
||||
"Flat, insulated (assumed)": "flat",
|
||||
"(another dwelling above)": "dwelling_above",
|
||||
"(another premises above)": "premises_above",
|
||||
}
|
||||
for text, family in cases.items():
|
||||
record = _hist(roof_description=text)
|
||||
assert conditioning_from_historic(record).roof_form == family, text
|
||||
# Unpinned forms (roof rooms, thatched) must not guess.
|
||||
assert conditioning_from_historic(
|
||||
_hist(roof_description="Roof room(s), insulated")
|
||||
).roof_form is None
|
||||
assert conditioning_from_historic(
|
||||
_hist(roof_description="Thatched, with additional insulation")
|
||||
).roof_form is None
|
||||
|
||||
|
||||
def test_legacy_register_fuel_descriptions_resolve():
|
||||
# Arrange / Act / Assert — the old register lodged pre-RdSAP-17 fuels with
|
||||
# a "backwards compatibility" rider or a SAP-style prefix; they name the
|
||||
# same physical fuels (dominant in the pre-2012 dump: a 65-shard scan found
|
||||
# 636/663 unresolved values were these variants).
|
||||
cases = {
|
||||
"mains gas - this is for backwards compatibility only and should not be used": 26,
|
||||
"electricity - this is for backwards compatibility only and should not be used": 29,
|
||||
"LPG - this is for backwards compatibility only and should not be used": 27,
|
||||
"oil - this is for backwards compatibility only and should not be used": 28,
|
||||
"Gas: mains gas": 26,
|
||||
"Electricity: electricity, unspecified tariff": 29,
|
||||
"dual fuel - mineral + wood": 10,
|
||||
}
|
||||
for text, code in cases.items():
|
||||
record = _hist(main_fuel=text)
|
||||
assert conditioning_from_historic(record).main_fuel == code, text
|
||||
|
||||
|
||||
def test_unresolvable_values_degrade_to_none():
|
||||
# Arrange — junk and blanks must never guess a code.
|
||||
record = _hist(
|
||||
property_type="Castle",
|
||||
built_form="Not Recorded",
|
||||
walls_description="Average thermal transmittance 0.3 W/m²K",
|
||||
construction_age_band="INVALID!",
|
||||
main_fuel="To be used only when there is no heating system",
|
||||
total_floor_area="",
|
||||
)
|
||||
|
||||
# Act
|
||||
conditioning = conditioning_from_historic(record)
|
||||
|
||||
# Assert
|
||||
assert conditioning.property_type is None
|
||||
assert conditioning.built_form is None
|
||||
assert conditioning.wall_construction is None
|
||||
assert conditioning.roof_form is None
|
||||
assert conditioning.construction_age_band is None
|
||||
assert conditioning.main_fuel is None
|
||||
assert conditioning.total_floor_area_m2 is None
|
||||
364626
tests/fixtures/expired_prediction_pairs.json
vendored
Normal file
364626
tests/fixtures/expired_prediction_pairs.json
vendored
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -21,7 +21,11 @@ from repositories.epc.epc_postgres_repository import EpcSaveRequest
|
|||
from repositories.plan.plan_repository import PlanRepository, PlanSaveRequest
|
||||
from repositories.product.product_repository import ProductRepository
|
||||
from repositories.property_baseline.property_baseline_repository import PropertyBaselineRepository
|
||||
from repositories.epc.epc_repository import EpcRepository, EpcSource
|
||||
from repositories.epc.epc_repository import (
|
||||
PREDICTED_SLOT_SOURCES,
|
||||
EpcRepository,
|
||||
EpcSource,
|
||||
)
|
||||
from repositories.property.property_repository import (
|
||||
PropertyIdentityInsert,
|
||||
PropertyRepository,
|
||||
|
|
@ -81,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] = {}
|
||||
|
|
@ -93,10 +98,11 @@ 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
|
||||
if source == "predicted"
|
||||
if source in PREDICTED_SLOT_SOURCES
|
||||
else self._by_property
|
||||
)
|
||||
slot[property_id] = data
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
|
|
|||
|
|
@ -80,3 +80,47 @@ def test_a_lodged_only_property_has_no_predicted_epc(db_engine: Engine) -> None:
|
|||
repo = EpcPostgresRepository(session)
|
||||
assert repo.get_predicted_for_property(9) is None
|
||||
assert repo.get_for_property(9) == epc
|
||||
|
||||
|
||||
def test_an_expired_source_epc_lives_in_the_predicted_slot(db_engine: Engine) -> None:
|
||||
# Arrange — an Expired-Enhanced Prediction is persisted with source="expired"
|
||||
# (ADR-0054): enhanced by an expired historic observation, not totally
|
||||
# predicted — but it occupies the same predicted slot.
|
||||
epc = _epc()
|
||||
with Session(db_engine) as session:
|
||||
repo = EpcPostgresRepository(session)
|
||||
repo.save(epc, property_id=11, source="expired")
|
||||
session.commit()
|
||||
|
||||
# Act / Assert — readable through the predicted slot; the lodged slot is empty.
|
||||
with Session(db_engine) as session:
|
||||
repo = EpcPostgresRepository(session)
|
||||
assert repo.get_predicted_for_property(11) == epc
|
||||
assert repo.get_predicted_for_properties([11]) == {11: epc}
|
||||
assert repo.get_for_property(11) is None
|
||||
|
||||
|
||||
def test_saving_predicted_over_expired_replaces_the_slot_without_stranding(
|
||||
db_engine: Engine,
|
||||
) -> None:
|
||||
# Arrange — a re-ingestion can flip the slot's flavour (expired -> predicted
|
||||
# or back); the slot holds ONE row, never a stranded stale sibling.
|
||||
from sqlmodel import select
|
||||
|
||||
from infrastructure.postgres.epc_property_table import EpcPropertyModel
|
||||
|
||||
epc = _epc()
|
||||
with Session(db_engine) as session:
|
||||
repo = EpcPostgresRepository(session)
|
||||
repo.save(epc, property_id=13, source="expired")
|
||||
repo.save(epc, property_id=13, source="predicted")
|
||||
session.commit()
|
||||
|
||||
# Act
|
||||
with Session(db_engine) as session:
|
||||
rows = session.exec(
|
||||
select(EpcPropertyModel).where(EpcPropertyModel.property_id == 13)
|
||||
).all()
|
||||
|
||||
# Assert — exactly one slot row remains, the latest flavour.
|
||||
assert [r.source for r in rows] == ["predicted"]
|
||||
|
|
|
|||
|
|
@ -16,10 +16,11 @@ from repositories.historic_epc.historic_epc_repository import HistoricEpcReposit
|
|||
from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver
|
||||
|
||||
|
||||
def _hist(address: str, uprn: str) -> HistoricEpc:
|
||||
def _hist(address: str, uprn: str, lodgement_date: str = "") -> HistoricEpc:
|
||||
fields = {f.name: "" for f in dataclasses.fields(HistoricEpc)}
|
||||
fields["address"] = address
|
||||
fields["uprn"] = uprn
|
||||
fields["lodgement_date"] = lodgement_date
|
||||
return HistoricEpc(**fields)
|
||||
|
||||
|
||||
|
|
@ -101,6 +102,53 @@ def test_resolve_uprn_is_none_on_ambiguous_tie():
|
|||
assert HistoricEpcResolver(repo).resolve_uprn("47 Gordon Road", "AB33 8AL") is None
|
||||
|
||||
|
||||
def test_record_for_uprn_returns_the_exact_uprn_match():
|
||||
# Arrange — the prediction path knows the target's UPRN; the lookup must be
|
||||
# exact equality, never the fuzzy address matcher (ADR-0054).
|
||||
repo = _FakeRepo(
|
||||
{
|
||||
"AB338AL": [
|
||||
_hist("47 GORDON ROAD", "100"),
|
||||
_hist("48 GORDON ROAD", "200"),
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
# Act
|
||||
record = HistoricEpcResolver(repo).record_for_uprn("200", "AB33 8AL")
|
||||
|
||||
# Assert
|
||||
assert record is not None
|
||||
assert record.address == "48 GORDON ROAD"
|
||||
|
||||
|
||||
def test_record_for_uprn_is_none_when_uprn_not_in_shard():
|
||||
# Arrange
|
||||
repo = _FakeRepo({"AB338AL": [_hist("47 GORDON ROAD", "100")]})
|
||||
|
||||
# Act / Assert — a miss is the normal outcome of a best-effort lookup.
|
||||
assert HistoricEpcResolver(repo).record_for_uprn("999", "AB33 8AL") is None
|
||||
|
||||
|
||||
def test_record_for_uprn_picks_latest_lodgement_when_relodged():
|
||||
# Arrange — the register re-lodged this UPRN; the newer observation wins.
|
||||
repo = _FakeRepo(
|
||||
{
|
||||
"AB338AL": [
|
||||
_hist("47 GORDON ROAD", "100", lodgement_date="2008-01-15"),
|
||||
_hist("47 GORDON ROAD, DORRIDGE", "100", lodgement_date="2010-06-02"),
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
# Act
|
||||
record = HistoricEpcResolver(repo).record_for_uprn("100", "AB33 8AL")
|
||||
|
||||
# Assert
|
||||
assert record is not None
|
||||
assert record.lodgement_date == "2010-06-02"
|
||||
|
||||
|
||||
def test_resolve_uprn_is_none_when_all_scores_zero():
|
||||
# Arrange — no candidate shares the user's building number => all hard-zero.
|
||||
repo = _FakeRepo(
|
||||
|
|
|
|||
149
tests/scripts/test_expired_prediction_pairs_harness.py
Normal file
149
tests/scripts/test_expired_prediction_pairs_harness.py
Normal file
|
|
@ -0,0 +1,149 @@
|
|||
"""Pure pair-selection and aggregation logic of the ADR-0054 pairs harness."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
from typing import Optional
|
||||
|
||||
from datatypes.epc.domain.historic_epc import HistoricEpc
|
||||
from domain.epc_prediction.prediction_comparison import PredictionComparison
|
||||
from scripts.expired_prediction_pairs_harness import (
|
||||
PairScore,
|
||||
aggregate,
|
||||
format_report,
|
||||
latest_pre_2012_by_uprn,
|
||||
)
|
||||
|
||||
|
||||
def _hist(uprn: str, lodgement_date: str) -> HistoricEpc:
|
||||
fields = {f.name: "" for f in dataclasses.fields(HistoricEpc)}
|
||||
fields["uprn"] = uprn
|
||||
fields["lodgement_date"] = lodgement_date
|
||||
return HistoricEpc(**fields)
|
||||
|
||||
|
||||
def _score(
|
||||
hits: dict[str, Optional[bool]], sap_residual: Optional[float] = None
|
||||
) -> PairScore:
|
||||
return PairScore(
|
||||
comparison=PredictionComparison(
|
||||
categorical_hits=hits,
|
||||
floor_area_residual=-4.0,
|
||||
building_parts_residual=1,
|
||||
window_count_residual=-2,
|
||||
total_window_area_residual=3.5,
|
||||
door_count_residual=0,
|
||||
),
|
||||
sap_residual=sap_residual,
|
||||
)
|
||||
|
||||
|
||||
def test_pairs_keep_only_the_latest_pre_2012_cert_per_uprn():
|
||||
# Arrange — one UPRN lodged twice pre-2012, once post-2012; one UPRN-less row.
|
||||
records = [
|
||||
_hist("100", "2008-05-01"),
|
||||
_hist("100", "2010-11-30"),
|
||||
_hist("100", "2013-01-01"),
|
||||
_hist("", "2009-01-01"),
|
||||
]
|
||||
|
||||
# Act
|
||||
by_uprn = latest_pre_2012_by_uprn(records)
|
||||
|
||||
# Assert — the 2010 cert wins; the 2013 one can never seed an expired pair.
|
||||
assert set(by_uprn) == {"100"}
|
||||
assert by_uprn["100"].lodgement_date == "2010-11-30"
|
||||
|
||||
|
||||
def test_aggregate_matches_the_component_accuracy_shape():
|
||||
# Arrange — two pairs; wall scored twice (1 hit), roof scored once (None
|
||||
# means the actual lodges no value — out of the denominator, per ADR-0030).
|
||||
scores = [
|
||||
_score({"wall_construction": True, "roof_construction": None}, sap_residual=6.0),
|
||||
_score({"wall_construction": False, "roof_construction": True}),
|
||||
]
|
||||
|
||||
# Act
|
||||
arm = aggregate(scores)
|
||||
|
||||
# Assert — classification (hits, applicable), all five residual components,
|
||||
# and the SAP residual list (only the scored pair contributes).
|
||||
assert arm.classification["wall_construction"] == (1, 2)
|
||||
assert arm.classification["roof_construction"] == (1, 1)
|
||||
assert arm.residuals["floor_area_m2"] == [-4.0, -4.0]
|
||||
assert arm.residuals["building_parts"] == [1.0, 1.0]
|
||||
assert arm.residuals["window_count"] == [-2.0, -2.0]
|
||||
assert arm.residuals["total_window_area_m2"] == [3.5, 3.5]
|
||||
assert arm.residuals["door_count"] == [0.0, 0.0]
|
||||
assert arm.sap_residuals == [6.0]
|
||||
|
||||
|
||||
def test_report_prints_both_arms_side_by_side():
|
||||
# Arrange
|
||||
plain = aggregate([_score({"wall_construction": False}, sap_residual=-8.0)])
|
||||
conditioned = aggregate([_score({"wall_construction": True}, sap_residual=2.0)])
|
||||
|
||||
# Act
|
||||
report = format_report(plain, conditioned, pairs=1)
|
||||
|
||||
# Assert — hit-rates, residuals and the SAP arm all present, side by side.
|
||||
assert "| wall_construction | 0/1 (0%) | 1/1 (100%) |" in report
|
||||
assert "| floor_area_m2 | 4.0 | 4.0 |" in report
|
||||
assert "| mean abs | 8.0 | 2.0 |" in report
|
||||
assert "1 pairs" in report
|
||||
|
||||
|
||||
def test_ladder_simulation_engages_only_with_enough_matches():
|
||||
# Arrange — a 6-strong base cohort: 5 band-C (engages at k=5), then within
|
||||
# the band-C survivors only 2 on fuel 26 (relaxes), and 5 within ±5% of
|
||||
# 100 m² (engages on the un-shrunk cohort).
|
||||
from scripts.expired_prediction_pairs_harness import simulate_conditioning_ladder
|
||||
from tests.domain.epc_prediction.test_comparable_properties import _comparable
|
||||
|
||||
base = [
|
||||
_comparable(
|
||||
property_type="0",
|
||||
certificate_number=f"C{i}",
|
||||
construction_age_band="C",
|
||||
main_fuel=26 if i < 2 else 29,
|
||||
total_floor_area_m2=100.0 + i,
|
||||
)
|
||||
for i in range(5)
|
||||
] + [
|
||||
_comparable(
|
||||
property_type="0",
|
||||
certificate_number="G0",
|
||||
construction_age_band="G",
|
||||
main_fuel=26,
|
||||
total_floor_area_m2=100.0,
|
||||
)
|
||||
]
|
||||
|
||||
# Act
|
||||
steps = simulate_conditioning_ladder(
|
||||
base, age_band="C", main_fuel=26, total_floor_area_m2=100.0
|
||||
)
|
||||
|
||||
# Assert — age engaged (5 matches), fuel relaxed (2 < 5 within band-C
|
||||
# survivors), TFA engaged (all 5 survivors within the band).
|
||||
age = steps["construction_age_band"]
|
||||
fuel = steps["main_fuel"]
|
||||
tfa = steps["total_floor_area"]
|
||||
assert age is not None and age.engaged and age.matches == 5
|
||||
assert fuel is not None and not fuel.engaged and fuel.matches == 2
|
||||
assert tfa is not None and tfa.engaged and tfa.matches == 5
|
||||
|
||||
|
||||
def test_ladder_simulation_skips_unresolved_attributes():
|
||||
from scripts.expired_prediction_pairs_harness import simulate_conditioning_ladder
|
||||
|
||||
steps = simulate_conditioning_ladder(
|
||||
[], age_band=None, main_fuel=None, total_floor_area_m2=None
|
||||
)
|
||||
|
||||
assert steps == {
|
||||
"roof_form": None,
|
||||
"construction_age_band": None,
|
||||
"main_fuel": None,
|
||||
"total_floor_area": None,
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue