feat(modelling): attribute per-measure bill savings via a telescoping cascade

`_plan_for` now scores the baseline + every cumulative prefix once
(`cascade_scores`, best-practice order) and reuses those Scores for both the
role-3 marginal attribution and a per-measure bill cascade: bill each prefix at
one Fuel Rates snapshot and take consecutive Bill deltas as each measure's
marginal delivered-kWh and £ saving. Saving is signed (ventilation is
negative) and telescopes exactly to the Plan headline savings, because the
Plan's baseline/post Bills are now the same cascade endpoints (`bills[0]` /
`bills[-1]`) — which also drops the redundant standalone baseline `calculate`.

`recommendation.kwh_savings` / `energy_cost_savings` are filled from these.
Adds `Bill.total_consumption_kwh` (shared by Plan + the orchestrator). Pinned
end-to-end on the real calculator: Σ per-measure savings == the Plan totals
(ADR-0014 amendment).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

domain/billing/bill.py

@@ -127,3 +127,9 @@ class Bill:
     standing_charges_gbp: float
     seg_credit_gbp: float
     total_gbp: float
+
+    @property
+    def total_consumption_kwh(self) -> float:
+        """Total delivered energy (kWh) across the billed sections. Standing
+        charges and the SEG credit are £, not energy, so they don't count."""
+        return sum((section.kwh for section in self.sections.values()), 0.0)

domain/modelling/plan.py

@@ -20,12 +20,6 @@ from domain.modelling.recommendation import Cost
 from domain.modelling.scoring.scoring import MeasureImpact
 
 
-def _total_consumption_kwh(bill: Bill) -> float:
-    """A Bill's total delivered energy (kWh) — the sum of its section kWh
-    (standing charges and the SEG credit are £, not energy)."""
-    return sum((section.kwh for section in bill.sections.values()), 0.0)
-
-
 @dataclass(frozen=True)
 class PlanMeasure:
     """One selected Measure Option as it lands in a Plan: the measure, its
@@ -112,7 +106,7 @@ class Plan:
     @property
     def post_energy_consumption(self) -> Optional[float]:
         """The post-package total delivered energy (kWh), or None if not billed."""
-        return None if self.post_bill is None else _total_consumption_kwh(self.post_bill)
+        return None if self.post_bill is None else self.post_bill.total_consumption_kwh
 
     @property
     def energy_consumption_savings(self) -> Optional[float]:
@@ -120,6 +114,7 @@ class Plan:
         both bills were derived."""
         if self.baseline_bill is None or self.post_bill is None:
             return None
-        return _total_consumption_kwh(self.baseline_bill) - _total_consumption_kwh(
-            self.post_bill
+        return (
+            self.baseline_bill.total_consumption_kwh
+            - self.post_bill.total_consumption_kwh
         )

orchestration/modelling_orchestrator.py

@@ -22,8 +22,9 @@ from domain.modelling.generators.roof_recommendation import recommend_loft_insul
 from domain.modelling.scenario import Scenario
 from domain.modelling.scoring.scoring import (
     MeasureImpact,
+    cascade_scores,
     independent_option_impacts,
-    marginal_impacts,
+    marginals_from_scores,
 )
 from domain.modelling.generators.wall_recommendation import recommend_cavity_wall
 from domain.sap10_calculator.calculator import SapCalculator
@@ -139,22 +140,31 @@ class ModellingOrchestrator:
         ordered: list[MeasureOption] = sorted(
             (scored.option for scored in package.selected), key=_best_practice_key
         )
-        impacts: list[MeasureImpact] = marginal_impacts(
+        # Score the baseline + every cumulative prefix once (cascade[0] is the
+        # baseline, cascade[-1] the whole package), then reuse those Scores for
+        # both the marginal attribution and the per-measure bill cascade.
+        cascade: list[Score] = cascade_scores(
             scorer, effective_epc, [option.overlay for option in ordered]
         )
-        baseline: Score = scorer.score(effective_epc, [])
+        impacts: list[MeasureImpact] = marginals_from_scores(cascade)
+        # Bill every prefix at one Fuel Rates snapshot; consecutive Bill deltas
+        # are each measure's marginal energy/cost saving — negative for
+        # ventilation — telescoping exactly to the Plan totals (ADR-0014). The
+        # Plan's baseline/post Bills are the cascade endpoints, so the
+        # per-measure savings and the headline savings share one source.
+        bills: list[Bill] = [_bill_for(bill_derivation, score) for score in cascade]
         measures: tuple[PlanMeasure, ...] = tuple(
-            _plan_measure(option, impact)
-            for option, impact in zip(ordered, impacts, strict=True)
+            _plan_measure(option, impact, before, after)
+            for option, impact, before, after in zip(
+                ordered, impacts, bills[:-1], bills[1:], strict=True
+            )
         )
-        # Price the unmodified and post-package end-states at the same Fuel
-        # Rates, reusing SapResults already scored — no extra calculate.
         return Plan(
             measures=measures,
-            baseline=baseline,
+            baseline=cascade[0],
             post_retrofit=package.score,
-            baseline_bill=_bill_for(bill_derivation, baseline),
-            post_bill=_bill_for(bill_derivation, package.score),
+            baseline_bill=bills[0],
+            post_bill=bills[-1],
         )
 
 
@@ -232,7 +242,12 @@ def _best_practice_key(option: MeasureOption) -> int:
         return len(_BEST_PRACTICE_ORDER)
 
 
-def _plan_measure(option: MeasureOption, impact: MeasureImpact) -> PlanMeasure:
+def _plan_measure(
+    option: MeasureOption, impact: MeasureImpact, before: Bill, after: Bill
+) -> PlanMeasure:
+    """Assemble a Plan Measure, attributing this measure's marginal bill saving
+    as the delta between the running package Bill before and after it (delivered
+    kWh and £). Signed so positive is a saving; ventilation is negative."""
     if option.cost is None:
         raise ValueError(
             f"measure option {option.measure_type!r} has no cost; cannot persist"
@@ -242,4 +257,6 @@ def _plan_measure(option: MeasureOption, impact: MeasureImpact) -> PlanMeasure:
         description=option.description,
         cost=option.cost,
         impact=impact,
+        kwh_savings=before.total_consumption_kwh - after.total_consumption_kwh,
+        energy_cost_savings=before.total_gbp - after.total_gbp,
     )

tests/orchestration/test_ara_first_run_pipeline_integration.py

@@ -318,6 +318,21 @@ def test_modelling_optimises_and_persists_a_multi_measure_plan(
     assert wall_sap is not None and vent_sap is not None
     assert wall_sap > 0.0
     assert vent_sap <= 0.0
+    # Per-measure bill savings (telescoping cascade, ADR-0014 amendment): each
+    # measure carries its delivered-kWh and £ saving, and they telescope exactly
+    # to the Plan's headline savings. Ventilation increases energy, so its
+    # savings are negative — and the telescoping still holds.
+    for rec in rec_rows:
+        assert rec.kwh_savings is not None
+        assert rec.energy_cost_savings is not None
+    vent_kwh: float | None = by_type["mechanical_ventilation"].kwh_savings
+    assert vent_kwh is not None and vent_kwh < 0.0
+    kwh_total: float = sum(rec.kwh_savings or 0.0 for rec in rec_rows)
+    cost_total: float = sum(rec.energy_cost_savings or 0.0 for rec in rec_rows)
+    assert plan.energy_consumption_savings is not None
+    assert plan.energy_bill_savings is not None
+    assert abs(kwh_total - plan.energy_consumption_savings) <= 1e-6
+    assert abs(cost_total - plan.energy_bill_savings) <= 1e-6
 
 
 def test_modelling_recommends_nothing_when_already_at_the_target_band(