refactor(modelling): load ASHP rates from a committed costs file

Slice 10 of ADR-0025 costing. The Southern Housing rate table moves from code
constants into ashp_rates.json (structured rows the flat scalar catalogue can't
hold), loaded via AshpRates.from_json. Products takes an injected AshpRates
(default: the committed sheet), so rates are now data -- tunable (e.g.
reuse_distribution_fraction) without a code change, and ready for ETL/DB-supplied
rates later. Behaviour-preserving: the 6 pinned cost tests still hold against the
default, plus a new test proving injected rates drive the total.

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

domain/modelling/ashp_rates.json

@@ -0,0 +1,26 @@
+{
+  "_source": "Southern Housing Group ASHP rates (HEAT PUMPS tab, ECOHT01-68); see ADR-0025. Fully-loaded supply+install rates in GBP.",
+  "decommission": {
+    "electric_storage_small": 570.0,
+    "electric_storage_large": 840.0,
+    "gas": 720.0,
+    "oil": 720.0,
+    "lpg": 960.0
+  },
+  "heat_pump_bands": [[5.0, 9720.0], [8.0, 9840.0], [11.0, 10200.0], [15.0, 10680.0]],
+  "heat_pump_top_price": 11400.0,
+  "cylinder": 2382.60,
+  "distribution_by_radiators": {
+    "4": 2220.0,
+    "5": 2550.0,
+    "6": 3084.0,
+    "7": 3618.0,
+    "8": 4152.0,
+    "9": 4680.0,
+    "10": 5220.0,
+    "11": 5754.0,
+    "12": 6288.0
+  },
+  "distribution_flush": 168.0,
+  "reuse_distribution_fraction": 0.5
+}

domain/modelling/products.py

@@ -15,61 +15,83 @@ lives in the modelling layer (ADR-0025).
 
 from __future__ import annotations
 
+import json
 from dataclasses import dataclass
 from enum import Enum
+from pathlib import Path
+from typing import Any
 
 from domain.modelling.contingencies import contingency_rate
 from domain.modelling.recommendation import Cost
 
 _ASHP_MEASURE_TYPE = "air_source_heat_pump"
 
-# --- Southern Housing Group ASHP rates (committed constants; moved to the
-# costs file in a later slice). Each is a fully-loaded supply+install rate. ---
+# The committed ASHP rate sheet (ADR-0025) — structured rate rows the flat
+# scalar catalogue cannot hold; loaded into `AshpRates`.
+_ASHP_RATES_PATH = Path(__file__).resolve().parent / "ashp_rates.json"
 
-# Decommission an existing electric-storage system, by property size band.
-_DECOMMISSION_ELECTRIC_STORAGE_SMALL = 570.0
-_DECOMMISSION_ELECTRIC_STORAGE_LARGE = 840.0
-# Decommission an existing wet (boiler) system — flat across property size for
-# gas and oil; LPG carries the extra tank/fuel removal (ECOHT06-08, 03-04).
-_DECOMMISSION_GAS = 720.0
-_DECOMMISSION_OIL = 720.0
-_DECOMMISSION_LPG = 960.0
-
-# Heat-pump install (MONOBLOC, brand-neutral), by kW size band — design heat
-# loss is rounded up to the next band (ECOHT09-13).
-_PUMP_BANDS: tuple[tuple[float, float], ...] = (
-    (5.0, 9720.0),
-    (8.0, 9840.0),
-    (11.0, 10200.0),
-    (15.0, 10680.0),
-)
-_PUMP_TOP_PRICE = 11400.0
-
-# Fixed unvented hot-water cylinder (200 L) — one per install; the cylinder-size
-# spread on the sheet is £188, treated as noise (ADR-0025).
-_CYLINDER = 2382.60
-
-# Full new wet central-heating distribution, by radiator count (ECOHT40-48).
-_DISTRIBUTION_BY_RADIATORS: dict[int, float] = {
-    4: 2220.0,
-    5: 2550.0,
-    6: 3084.0,
-    7: 3618.0,
-    8: 4152.0,
-    9: 4680.0,
-    10: 5220.0,
-    11: 5754.0,
-    12: 6288.0,
-}
 _MIN_RADIATORS = 4
 _MAX_RADIATORS = 12
-# Power-flush + inhibitor when reusing an existing wet system (ECOHT67).
-_DISTRIBUTION_FLUSH = 168.0
-# Fraction of a full new distribution charged when reusing an existing wet
-# system — a stand-in for partial radiator upsizing at low ASHP flow temps.
-# The headline uncertainty in the model; recalibrate against real reuse-job
-# costs / survey data (ADR-0025).
-_REUSE_DISTRIBUTION_FRACTION = 0.5
+
+
+@dataclass(frozen=True)
+class AshpRates:
+    """The Southern Housing Group ASHP rate table (ADR-0025) — fully-loaded
+    supply+install rates, one row per priced line item. Data, not code: the
+    committed default loads from `ashp_rates.json`, and a caller can inject a
+    variant (e.g. to recalibrate `reuse_distribution_fraction`)."""
+
+    decommission_electric_storage_small: float
+    decommission_electric_storage_large: float
+    decommission_gas: float
+    decommission_oil: float
+    decommission_lpg: float
+    # Heat-pump install bands (max_kw, price), ascending; design heat loss rounds
+    # up to the first covering band, else `heat_pump_top_price`.
+    heat_pump_bands: tuple[tuple[float, float], ...]
+    heat_pump_top_price: float
+    # Fixed unvented cylinder — one per install (size spread on the sheet is £188).
+    cylinder: float
+    # Full new wet distribution, by radiator count.
+    distribution_by_radiators: dict[int, float]
+    # Power-flush + inhibitor when reusing an existing wet system.
+    distribution_flush: float
+    # Fraction of a full distribution charged on reuse — a stand-in for partial
+    # radiator upsizing at low ASHP flow temps; the headline uncertainty.
+    reuse_distribution_fraction: float
+
+    @classmethod
+    def default(cls) -> "AshpRates":
+        """Load the committed Southern Housing rate sheet."""
+        return cls.from_json(_ASHP_RATES_PATH)
+
+    @classmethod
+    def from_json(cls, path: Path) -> "AshpRates":
+        with path.open(encoding="utf-8") as handle:
+            raw: dict[str, Any] = json.load(handle)
+        decommission: dict[str, Any] = raw["decommission"]
+        return cls(
+            decommission_electric_storage_small=float(
+                decommission["electric_storage_small"]
+            ),
+            decommission_electric_storage_large=float(
+                decommission["electric_storage_large"]
+            ),
+            decommission_gas=float(decommission["gas"]),
+            decommission_oil=float(decommission["oil"]),
+            decommission_lpg=float(decommission["lpg"]),
+            heat_pump_bands=tuple(
+                (float(kw), float(price)) for kw, price in raw["heat_pump_bands"]
+            ),
+            heat_pump_top_price=float(raw["heat_pump_top_price"]),
+            cylinder=float(raw["cylinder"]),
+            distribution_by_radiators={
+                int(rads): float(price)
+                for rads, price in raw["distribution_by_radiators"].items()
+            },
+            distribution_flush=float(raw["distribution_flush"]),
+            reuse_distribution_fraction=float(raw["reuse_distribution_fraction"]),
+        )
 
 
 class AshpExistingSystem(Enum):
@@ -100,7 +122,11 @@ class AshpCostInputs:
 
 class Products:
     """The catalogue collection. Owns cost composition for measures whose price
-    is not a single catalogue scalar (the ASHP bundle — ADR-0025)."""
+    is not a single catalogue scalar (the ASHP bundle — ADR-0025). The ASHP rate
+    table is data, injected as `AshpRates` (default: the committed rate sheet)."""
+
+    def __init__(self, rates: AshpRates | None = None) -> None:
+        self._rates: AshpRates = rates if rates is not None else AshpRates.default()
 
     def ashp_bundle_cost(self, inputs: AshpCostInputs) -> Cost:
         """Compose the fully-loaded ASHP bundle total for a dwelling and pair it
@@ -108,7 +134,7 @@ class Products:
         total: float = (
             self._decommission(inputs)
             + self._heat_pump(inputs.design_heat_loss_kw)
-            + _CYLINDER
+            + self._rates.cylinder
             + self._distribution(inputs)
         )
         return Cost(
@@ -118,24 +144,26 @@ class Products:
     def _heat_pump(self, design_heat_loss_kw: float) -> float:
         """Price the install at the smallest band that covers the design heat
         loss (round up); above the largest band, the top rate applies."""
-        for max_kw, price in _PUMP_BANDS:
+        for max_kw, price in self._rates.heat_pump_bands:
             if design_heat_loss_kw <= max_kw:
                 return price
-        return _PUMP_TOP_PRICE
+        return self._rates.heat_pump_top_price
 
     def _decommission(self, inputs: AshpCostInputs) -> float:
+        rates = self._rates
+        electric_storage: float = (
+            rates.decommission_electric_storage_small
+            if inputs.is_small_property
+            else rates.decommission_electric_storage_large
+        )
         if inputs.existing_system is AshpExistingSystem.ELECTRIC_STORAGE:
-            return (
-                _DECOMMISSION_ELECTRIC_STORAGE_SMALL
-                if inputs.is_small_property
-                else _DECOMMISSION_ELECTRIC_STORAGE_LARGE
-            )
+            return electric_storage
         if inputs.existing_system is AshpExistingSystem.GAS:
-            return _DECOMMISSION_GAS
+            return rates.decommission_gas
         if inputs.existing_system is AshpExistingSystem.OIL:
-            return _DECOMMISSION_OIL
+            return rates.decommission_oil
         if inputs.existing_system is AshpExistingSystem.LPG:
-            return _DECOMMISSION_LPG
+            return rates.decommission_lpg
         # Systems off the rate sheet: ASHP is still offered (ADR-0025), so price
         # a fallback rather than raise. Nothing to remove for no system; electric
         # room/panel heaters are comparable work to storage heaters; anything
@@ -143,14 +171,17 @@ class Products:
         if inputs.existing_system is AshpExistingSystem.NONE:
             return 0.0
         if inputs.existing_system is AshpExistingSystem.ELECTRIC_OTHER:
-            return _DECOMMISSION_ELECTRIC_STORAGE_SMALL if inputs.is_small_property else _DECOMMISSION_ELECTRIC_STORAGE_LARGE
-        return _DECOMMISSION_GAS
+            return electric_storage
+        return rates.decommission_gas
 
     def _distribution(self, inputs: AshpCostInputs) -> float:
         radiators: int = max(_MIN_RADIATORS, min(_MAX_RADIATORS, inputs.radiator_count))
-        full: float = _DISTRIBUTION_BY_RADIATORS[radiators]
+        full: float = self._rates.distribution_by_radiators[radiators]
         # An existing wet system is reused, not rebuilt: a flush plus a fraction
         # of the full distribution to cover partial radiator upsizing.
         if inputs.has_reusable_wet_system:
-            return _DISTRIBUTION_FLUSH + _REUSE_DISTRIBUTION_FRACTION * full
+            return (
+                self._rates.distribution_flush
+                + self._rates.reuse_distribution_fraction * full
+            )
         return full

tests/domain/modelling/test_products.py

@@ -9,9 +9,12 @@ Costs are pinned against the real Southern Housing Group rate sheet, so the
 totals are exact (delta <= 1e-9), mirroring the cascade-pin philosophy.
 """
 
+from dataclasses import replace
+
 from domain.modelling.products import (
     AshpCostInputs,
     AshpExistingSystem,
+    AshpRates,
     Products,
 )
 from domain.modelling.recommendation import Cost
@@ -39,6 +42,27 @@ def test_ashp_bundle_cost_composes_an_electric_storage_full_distribution_dwellin
     assert abs(cost.contingency_rate - 0.25) <= 1e-9
 
 
+def test_ashp_bundle_cost_uses_injected_rates() -> None:
+    # Arrange — the rate table is data (ADR-0025): a Products built with a tweaked
+    # cylinder rate prices that cylinder, not the committed default.
+    rates: AshpRates = replace(AshpRates.default(), cylinder=1000.0)
+    products = Products(rates=rates)
+    inputs = AshpCostInputs(
+        existing_system=AshpExistingSystem.ELECTRIC_STORAGE,
+        is_small_property=True,
+        design_heat_loss_kw=4.0,
+        radiator_count=7,
+        has_reusable_wet_system=False,
+    )
+
+    # Act
+    cost: Cost = products.ashp_bundle_cost(inputs)
+
+    # Assert — decommission 570 + pump 9720 + injected cylinder 1000 +
+    # distribution 3618 = 14908.0.
+    assert abs(cost.total - 14908.0) <= 1e-9
+
+
 def _large_no_reuse(system: AshpExistingSystem) -> AshpCostInputs:
     """A large dwelling, 8 kW band, 8 radiators, no reusable wet system — so the
     only thing varying with ``system`` is the decommission line."""