junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Pairs harness reports the full Component Accuracy suite per arm 🟩

Browse source 
Classification hit-rates for every compare_prediction component, all five
numeric residuals, and the secondary calculator-floored SAP residual
(calc(predicted) − lodged), plain vs conditioned side by side — the same
metric shape as the prediction-corpus gate (ADR-0030). Pair-check now
precedes the cohort fetch so a national postcode sweep only pays the
expensive search-by-postcode for shards that actually hold a pair.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-07-06 09:05:59 +00:00
parent d9576db429
commit 1d83637afa
2 changed files with 178 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

187
scripts/expired_prediction_pairs_harness.py
View file

@ -1,22 +1,29 @@
"""Pre-2012 x post-June-2025 pairs harness for Expired-Enhanced Prediction (ADR-0054).
"""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). Each pair is predicted twice from its leave-one-out postcode cohort:
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 components with
`compare_prediction`; the report prints per-attribute hit rates side by side,
so any whitelist member (main fuel is the judgement call) can be promoted or
demoted with evidence. A report, not a CI gate.
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
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.
"""
@ -59,6 +66,14 @@ 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
@ -76,54 +91,102 @@ def latest_pre_2012_by_uprn(records: list[HistoricEpc]) -> dict[str, HistoricEpc
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:
"""Aggregated categorical hit counts for one arm: component -> (hits, scored)."""
"""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."""
hits: dict[str, tuple[int, int]]
floor_area_abs_residuals: list[float]
classification: dict[str, tuple[int, int]]
residuals: dict[str, list[float]]
sap_residuals: list[float]
def aggregate(comparisons: list[PredictionComparison]) -> ArmScores:
def aggregate(scores: list[PairScore]) -> ArmScores:
counts: dict[str, list[int]] = defaultdict(lambda: [0, 0])
residuals: list[float] = []
for comparison in comparisons:
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.append(abs(comparison.floor_area_residual))
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(
hits={k: (v[0], v[1]) for k, v in counts.items()},
floor_area_abs_residuals=residuals,
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:
components = sorted(set(plain.hits) | set(conditioned.hits))
"""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_hit, p_all = plain.hits.get(component, (0, 0))
c_hit, c_all = conditioned.hits.get(component, (0, 0))
p = f"{p_hit}/{p_all}" if p_all else "n/a"
c = f"{c_hit}/{c_all}" if c_all else "n/a"
p = _rate_cell(plain.classification.get(component, (0, 0)))
c = _rate_cell(conditioned.classification.get(component, (0, 0)))
lines.append(f"| {component} | {p} | {c} |")
def _mean(values: list[float]) -> str:
return f"{sum(values) / len(values):.1f}" if values else "n/a"
lines.append(
f"| floor_area mean abs residual (m²) | "
f"{_mean(plain.floor_area_abs_residuals)} | "
f"{_mean(conditioned.floor_area_abs_residuals)} |"
)
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)
@ -141,6 +204,7 @@ def _predict_arm(
def run(postcodes: list[str]) -> str: # pragma: no cover - live IO composition
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,
@ -154,29 +218,50 @@ def run(postcodes: list[str]) -> str: # pragma: no cover - live IO composition
from scripts.e2e_common import load_env, s3_parquet_reader
load_env()
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()
plain_comparisons: list[PredictionComparison] = []
conditioned_comparisons: list[PredictionComparison] = []
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] = []
pairs = 0
for raw_postcode in postcodes:
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))
)
if not historic:
print(f"{postcode}: no pre-2012 historic certs", file=sys.stderr)
continue
cohort = comparables_repo.candidates_for(postcode)
# Pair-check first: only a postcode with a SAP-10.2 relodgement of a
# pre-2012 UPRN pays for the cohort fetch.
paired: list[tuple[str, HistoricEpc, EpcPropertyData]] = []
for uprn, record in historic.items():
actual = epc_client.get_by_uprn(int(uprn))
if actual is None or actual.sap_version != _VALIDATION_SAP_VERSION:
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
cohort = comparables_repo.candidates_for(postcode)
for uprn, record, actual in paired:
pairs += 1
loo_cohort = [c for c in cohort if c.epc.uprn != int(uprn)]
identity = PropertyIdentity(
@ -193,20 +278,25 @@ def run(postcodes: list[str]) -> str: # pragma: no cover - live IO composition
plain = _predict_arm(plain_target, loo_cohort, predictor)
conditioned = _predict_arm(conditioned_target, loo_cohort, predictor)
if plain is not None:
plain_comparisons.append(compare_prediction(plain, actual))
plain_scores.append(
PairScore(compare_prediction(plain, actual), sap_residual(plain, actual))
)
if conditioned is not None:
conditioned_comparisons.append(compare_prediction(conditioned, actual))
print(f"{postcode} {uprn}: pair scored", file=sys.stderr)
conditioned_scores.append(
PairScore(
compare_prediction(conditioned, actual),
sap_residual(conditioned, actual),
)
)
return format_report(
aggregate(plain_comparisons), aggregate(conditioned_comparisons), pairs
)
return format_report(aggregate(plain_scores), aggregate(conditioned_scores), pairs)
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")
args = parser.parse_args()
postcodes: list[str] = list(args.postcodes)
if args.postcodes_file is not None:
@ -217,7 +307,10 @@ def main() -> None: # pragma: no cover - CLI entry
]
if not postcodes:
parser.error("no postcodes given")
print(run(postcodes))
report = run(postcodes)
if args.out is not None:
args.out.write_text(report + "\n")
print(report)
if __name__ == "__main__":

61
tests/scripts/test_expired_prediction_pairs_harness.py
View file

@ -3,10 +3,12 @@
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,
@ -20,14 +22,19 @@ def _hist(uprn: str, lodgement_date: str) -> HistoricEpc:
return HistoricEpc(**fields)
def _comparison(hits: dict[str, bool | None]) -> PredictionComparison:
return PredictionComparison(
categorical_hits=hits,
floor_area_residual=-4.0,
building_parts_residual=0,
window_count_residual=0,
total_window_area_residual=0.0,
door_count_residual=0,
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,
)
@ -48,31 +55,39 @@ def test_pairs_keep_only_the_latest_pre_2012_cert_per_uprn():
assert by_uprn["100"].lodgement_date == "2010-11-30"
def test_aggregate_counts_hits_and_skips_not_applicable():
# Arrange — two comparisons; wall scored twice (1 hit), roof scored once
# (None means the actual lodged no value — out of the denominator).
comparisons = [
_comparison({"wall_construction": True, "roof_construction": None}),
_comparison({"wall_construction": False, "roof_construction": True}),
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
scores = aggregate(comparisons)
arm = aggregate(scores)
# Assert
assert scores.hits["wall_construction"] == (1, 2)
assert scores.hits["roof_construction"] == (1, 1)
assert scores.floor_area_abs_residuals == [4.0, 4.0]
# 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([_comparison({"wall_construction": False})])
conditioned = aggregate([_comparison({"wall_construction": True})])
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
assert "| wall_construction | 0/1 | 1/1 |" in report
# 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