feat(modelling): per-measure scoring — marginal cascade + per-Option signal (#1156)

scoring.py adds the telescoping marginal cascade that serves two of the three
ADR-0016 scoring roles:
- marginal_impacts(scorer, baseline, overlays): applies overlays cumulatively
  in order and reports each measure's marginal MeasureImpact (sap_points +
  carbon/energy savings). Role 3 (final-package attribution) — the marginals
  telescope EXACTLY to the whole-package total.
- independent_option_impacts(scorer, baseline, options): role 1 — scores each
  Option's overlay independently vs baseline, scoring each DISTINCT overlay
  once (Options sharing an overlay reuse the result). Approximate signal for
  the optimiser; never surfaced as a measure's true impact.

Role 2 (whole-package re-score) is PackageScorer.score directly. Three
behaviour tests on the real Sap10Calculator / a counting stand-in (hand-built
EPD): single-overlay marginal == improvement-over-baseline; two-overlay
marginals telescope to the package total; per-Option dedup scores each
distinct overlay once. Closes #1156. pyright strict clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-06-03 08:50:49 +00:00
parent 7a478cff6e
commit 13dd5fe81a
2 changed files with 233 additions and 0 deletions

View file

@ -0,0 +1,91 @@
"""Per-measure scoring — the telescoping marginal cascade (ADR-0016).
`marginal_impacts` applies overlays one at a time in the given order and
reports each measure's marginal contribution. It serves two of the three
scoring roles:
- role 1 (per-Option optimiser signal): call per Option as a 1-element
sequence -> its independent-vs-baseline impact;
- role 3 (final-package display attribution): call once with the selected
overlays in best-practice order -> per-measure impacts that telescope
exactly to the whole-package total.
Per-Option (role 1) figures are an approximate signal and must not be surfaced
as a measure's true impact — only the final-package cascade (role 3) is
truthful. The whole-package re-score (role 2) is `PackageScorer.score` directly.
"""
from dataclasses import dataclass
from typing import Sequence
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.package_scorer import PackageScorer, Score
from domain.modelling.recommendation import MeasureOption
from domain.modelling.simulation import EpcSimulation
@dataclass(frozen=True)
class MeasureImpact:
"""One measure's marginal contribution, signed so positive is always an
improvement: `sap_points` is the SAP gain; the savings are reductions
(baseline-at-this-step minus the new value)."""
sap_points: float
co2_savings_kg_per_yr: float
energy_savings_kwh_per_yr: float
def marginal_impacts(
scorer: PackageScorer,
baseline: EpcPropertyData,
overlays: Sequence[EpcSimulation],
) -> list[MeasureImpact]:
"""Apply overlays cumulatively in order; return each one's marginal impact
over the running state. The marginals telescope to the whole-package total."""
impacts: list[MeasureImpact] = []
previous: Score = scorer.score(baseline, [])
for index in range(len(overlays)):
current: Score = scorer.score(baseline, list(overlays[: index + 1]))
impacts.append(
MeasureImpact(
sap_points=current.sap_continuous - previous.sap_continuous,
co2_savings_kg_per_yr=previous.co2_kg_per_yr - current.co2_kg_per_yr,
energy_savings_kwh_per_yr=(
previous.primary_energy_kwh_per_yr
- current.primary_energy_kwh_per_yr
),
)
)
previous = current
return impacts
def independent_option_impacts(
scorer: PackageScorer,
baseline: EpcPropertyData,
options: Sequence[MeasureOption],
) -> list[MeasureImpact]:
"""Score each Option's overlay independently against the baseline (role 1 —
the optimiser's approximate input signal). Each *distinct* Simulation Overlay
is scored once (Options sharing an overlay reuse the result), so the baseline
is scored once plus one score per distinct overlay. Results follow the input
order. These figures are an approximate signal never surface them as a
measure's true impact."""
base: Score = scorer.score(baseline, [])
scored: list[tuple[EpcSimulation, MeasureImpact]] = []
impacts: list[MeasureImpact] = []
for option in options:
cached = next(
(impact for overlay, impact in scored if overlay == option.overlay), None
)
if cached is None:
current: Score = scorer.score(baseline, [option.overlay])
cached = MeasureImpact(
sap_points=current.sap_continuous - base.sap_continuous,
co2_savings_kg_per_yr=base.co2_kg_per_yr - current.co2_kg_per_yr,
energy_savings_kwh_per_yr=(
base.primary_energy_kwh_per_yr - current.primary_energy_kwh_per_yr
),
)
scored.append((option.overlay, cached))
impacts.append(cached)
return impacts

View file

@ -0,0 +1,142 @@
"""Behaviour of per-measure scoring: the telescoping marginal cascade that
serves both the per-Option optimiser signal (role 1) and the final-package
display attribution (role 3) ADR-0016. Exercises the real calculator on a
hand-built EPD; no PDF/parser involved.
"""
from typing import Sequence
from datatypes.epc.domain.epc_property_data import (
BuildingPartIdentifier,
EpcPropertyData,
)
from domain.modelling.package_scorer import PackageScorer, Score
from domain.modelling.recommendation import MeasureOption
from domain.modelling.scoring import (
MeasureImpact,
independent_option_impacts,
marginal_impacts,
)
from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation
from domain.sap10_calculator.calculator import Sap10Calculator
from tests.domain.sap10_calculator.worksheet._elmhurst_worksheet_000490 import (
build_epc,
)
_MAIN_CAVITY = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=2)
}
)
_EXT1_CAVITY = EpcSimulation(
building_parts={
BuildingPartIdentifier.EXTENSION_1: BuildingPartOverlay(wall_insulation_type=2)
}
)
class _CountingScorer(PackageScorer):
"""A PackageScorer stand-in that counts score() calls; the score is a
deterministic function of the overlays so distinct overlays differ."""
def __init__(self) -> None:
self.calls = 0
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
self.calls += 1
total = 0.0
for sim in simulations:
for overlay in sim.building_parts.values():
total += overlay.wall_insulation_type or 0
return Score(
sap_continuous=total, co2_kg_per_yr=0.0, primary_energy_kwh_per_yr=0.0
)
def _option(overlay: EpcSimulation) -> MeasureOption:
return MeasureOption(
measure_type="cavity_wall_insulation", description="opt", overlay=overlay
)
def test_independent_option_impacts_score_each_distinct_overlay_once() -> None:
# Arrange
baseline: EpcPropertyData = build_epc()
scorer = _CountingScorer()
overlay_a = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=2)
}
)
overlay_a_dup = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=2)
}
)
overlay_b = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=3)
}
)
options = [_option(overlay_a), _option(overlay_a_dup), _option(overlay_b)]
# Act
impacts: list[MeasureImpact] = independent_option_impacts(
scorer, baseline, options
)
# Assert
# baseline scored once + one score per DISTINCT overlay (a, b) = 3, not 4
assert scorer.calls == 3
assert impacts[0].sap_points == impacts[1].sap_points == 2.0
assert impacts[2].sap_points == 3.0
def test_single_overlay_marginal_is_its_improvement_over_baseline() -> None:
# Arrange
baseline: EpcPropertyData = build_epc()
scorer = PackageScorer(Sap10Calculator())
base: Score = scorer.score(baseline, [])
filled: Score = scorer.score(baseline, [_MAIN_CAVITY])
# Act
impacts: list[MeasureImpact] = marginal_impacts(scorer, baseline, [_MAIN_CAVITY])
# Assert
assert len(impacts) == 1
assert impacts[0].sap_points > 0 # cavity fill improves SAP
assert (
abs(impacts[0].sap_points - (filled.sap_continuous - base.sap_continuous))
<= 1e-9
)
def test_marginals_telescope_to_the_whole_package_total() -> None:
# Arrange
baseline: EpcPropertyData = build_epc()
scorer = PackageScorer(Sap10Calculator())
overlays = [_MAIN_CAVITY, _EXT1_CAVITY]
base: Score = scorer.score(baseline, [])
full: Score = scorer.score(baseline, overlays)
# Act
impacts: list[MeasureImpact] = marginal_impacts(scorer, baseline, overlays)
# Assert
assert len(impacts) == 2
assert (
abs(
sum(i.sap_points for i in impacts)
- (full.sap_continuous - base.sap_continuous)
)
<= 1e-9
)
assert (
abs(
sum(i.energy_savings_kwh_per_yr for i in impacts)
- (base.primary_energy_kwh_per_yr - full.primary_energy_kwh_per_yr)
)
<= 1e-6
)