mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
98d2f7aa16
Review findings on PR #1527: - The 'which goals are goal-aligned' set lived in two adjacent if-chains (_objective_for and _require_budget_for_goal_aligned) that had to stay in sync — a new goal-aligned goal added to one but not the other would slip the budget guard. Both now read a single _GOAL_ALIGNED_OBJECTIVES table. - The goal strings are the canonical PortfolioGoal enum values, not re-declared string constants, so goal-value drift can't silently degrade a goal to max-SAP; _target_sap reads the enum too. - _scored_candidate_groups takes objective without a default (its only caller passes it). - scoring.py: 'cached: float | None' -> Optional[float] per the CLAUDE.md 'Use Optional over | None' rule. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
110 lines
4.5 KiB
Python
110 lines
4.5 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 Callable, Optional, Sequence
|
|
|
|
from datatypes.epc.domain.epc_property_data import EpcPropertyData
|
|
from domain.modelling.scoring.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 cascade_scores(
|
|
scorer: PackageScorer,
|
|
baseline: EpcPropertyData,
|
|
overlays: Sequence[EpcSimulation],
|
|
) -> list[Score]:
|
|
"""Score the cumulative prefixes of `overlays` in order: index 0 is the
|
|
baseline (empty prefix), index i the state after the first i overlays. The
|
|
list has `len(overlays) + 1` entries — one calculator run each.
|
|
|
|
Each Score carries its `SapResult`, so the same cascade powers both the
|
|
role-3 marginal attribution (`marginals_from_scores`) and the telescoping
|
|
per-measure bill cascade — neither needs to re-score (ADR-0014 / ADR-0016)."""
|
|
return [
|
|
scorer.score(baseline, list(overlays[:prefix_length]))
|
|
for prefix_length in range(len(overlays) + 1)
|
|
]
|
|
|
|
|
|
def marginals_from_scores(scores: Sequence[Score]) -> list[MeasureImpact]:
|
|
"""Each measure's marginal impact from a precomputed cumulative-prefix
|
|
cascade (`scores[0]` is the baseline). Signed so positive is an improvement;
|
|
the marginals telescope to `scores[-1]` vs `scores[0]`."""
|
|
impacts: list[MeasureImpact] = []
|
|
for index in range(1, len(scores)):
|
|
previous: Score = scores[index - 1]
|
|
current: Score = scores[index]
|
|
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
|
|
),
|
|
)
|
|
)
|
|
return impacts
|
|
|
|
|
|
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."""
|
|
return marginals_from_scores(cascade_scores(scorer, baseline, overlays))
|
|
|
|
|
|
def independent_option_signals(
|
|
scorer: PackageScorer,
|
|
baseline: EpcPropertyData,
|
|
options: Sequence[MeasureOption],
|
|
objective: Callable[[Score], float],
|
|
) -> list[float]:
|
|
"""Each Option's independent-vs-baseline gain **in the objective's
|
|
currency** (role 1 — the optimiser's approximate input signal, ADR-0062):
|
|
SAP points for an Increasing-EPC goal, kg CO2 saved for Reducing CO2, £
|
|
saved for Energy Savings. Each distinct Simulation Overlay is scored once
|
|
(Options sharing an overlay reuse the result); results follow the input
|
|
order."""
|
|
base_value: float = objective(scorer.score(baseline, []))
|
|
scored: list[tuple[EpcSimulation, float]] = []
|
|
signals: list[float] = []
|
|
for option in options:
|
|
cached: Optional[float] = next(
|
|
(signal for overlay, signal in scored if overlay == option.overlay),
|
|
None,
|
|
)
|
|
if cached is None:
|
|
cached = objective(scorer.score(baseline, [option.overlay])) - base_value
|
|
scored.append((option.overlay, cached))
|
|
signals.append(cached)
|
|
return signals
|