Model/domain/modelling/scoring.py
Khalim Conn-Kowlessar 13dd5fe81a 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>
2026-06-03 08:50:49 +00:00

91 lines
3.8 KiB
Python

"""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