junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1527 from Hestia-Homes/feature/goal-aligned-objectives

Browse source 
Goal-aligned optimiser objectives: CO2 and bill £ (ADR-0062)
This commit is contained in:
KhalimCK 2026-07-10 12:35:07 +01:00 • committed by GitHub
commit 8b01d2518c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 525 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

62
docs/adr/0062-goal-aligned-optimiser-objectives.md Normal file
View file

@ -0,0 +1,62 @@
---
status: accepted (extends ADR-0016; composes with ADR-0061)
---
# Goal-aligned Optimiser objectives: each goal maximises its own metric
Every Scenario goal used to optimise SAP. The legacy engine returned no
target for Energy Savings / Reducing CO2 (`optimiser_functions.calculate_gain`
→ `None`) and maximised SAP gain within budget regardless of the goal, and the
new engine inherited that: the goal label changed nothing but the words on the
brief. The scorer already computes each package's carbon and (via SapResult →
EnergyBreakdown → BillDerivation) its annual bill, so aligning the objective
is a selection change, not a calculator change.
Decided in a grilling session with Khalim, 2026-07-09.
## Decision
**The Optimiser maximises the Scenario goal's own metric, as a pluggable
`objective: Callable[[Score], float]` (higher is better), with no target:
goal-aligned briefs are "reduce as much as possible within this budget".**
- **Reducing CO2 emissions** maximises annual kg CO2 saved
(`-score.co2_kg_per_yr`).
- **Energy Savings** maximises the annual bill £ saved, priced at the **live
Fuel Rates snapshot** (ADR-0014), not SAP's internal tariff book — that
difference is the point of the goal. SAP is itself a cost-shaped rating, so
the two frequently agree; they diverge exactly when current tariffs disagree
with SAP's assumptions (e.g. the gas/electricity price ratio).
- **Increasing EPC** keeps its SAP objective and band-target semantics
(least-cost-to-target, repair, max-gain fallback) unchanged.
- **Valuation Improvement / None** stay max-SAP-within-budget — SAP is a
defensible valuation proxy and `None` has no semantics to encode.
- **`goal_value` is ignored for the goal-aligned goals** — no percentage or
absolute target exists yet. If targets arrive later they slot into the
existing target machinery on the objective's scale.
- **A budget is mandatory** for the goal-aligned goals: unconstrained
"as much as possible" would recommend every beneficial measure. A
budget-less Energy/CO2 Scenario raises a `ValueError` naming the scenario
and goal — a loud misconfiguration, not a maximal plan.
- **One currency everywhere**: the role-1 group signals
(`independent_option_signals`), the forced Measure Dependency pricing, the
greedy-repair marginals, and Fabric First's phase-2 re-scoring
(ADR-0061) are all measured by the same objective, so a ventilation that
costs SAP but is carbon-neutral cannot sink a carbon-improving wall, and a
fabric-first phase 2 picks its heating on post-fabric carbon, not
post-fabric SAP.
## Consequences
- Selection changes; truth-telling does not. The Plan's persisted Scores,
Bills, and role-3 SAP attribution are computed exactly as before — only
*which* package is chosen responds to the goal.
- At a £16,000 budget on the uninsulated solid-brick corpus dwelling
(001431), the SAP objective buys wall + floor + gas boiler (SAP 72.9,
2,069 kg CO2/yr, £2,088/yr) while the carbon objective buys wall + floor +
storage heaters (SAP 69.2, 1,098 kg CO2/yr, £2,635/yr) — goals now trade
SAP, carbon and bills against each other visibly.
- The Energy Savings objective inherits the Fuel Rates snapshot's staleness
characteristics (quarterly Ofgem-cap cadence, ADR-0014).
- `independent_option_impacts` (role-1 SAP/CO2/kWh triple) is removed —
superseded by `independent_option_signals` in the objective's currency.

103
domain/modelling/optimisation/optimiser.py
View file

@ -21,7 +21,7 @@ from __future__ import annotations
import itertools
from dataclasses import dataclass
from typing import Optional, Protocol, Sequence
from typing import Callable, Optional, Protocol, Sequence
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.measure_type import FABRIC_MEASURE_TYPES, MeasureType
@ -33,8 +33,9 @@ from domain.modelling.simulation import EpcSimulation
@dataclass(frozen=True)
class ScoredOption:
"""A candidate Measure Option paired with its role-1 (independent-vs-
baseline) SAP gain — the optimiser's input signal. Cost is read from the
Option; the gain is supplied by scoring."""
baseline) gain in the goal objective's currency — SAP points by default,
kg CO2 / £ saved for the goal-aligned Scenarios (ADR-0062). Cost is read
from the Option; the gain is supplied by scoring."""
option: MeasureOption
sap_gain: float
@ -171,6 +172,12 @@ class OptimisedPackage:
score: Score
def sap_rating(score: Score) -> float:
"""The default Optimiser objective: the un-rounded SAP rating (higher is
better) — what every goal optimised before goal-aligned objectives."""
return score.sap_continuous
def optimise_package(
*,
groups: list[list[ScoredOption]],
@ -179,6 +186,7 @@ def optimise_package(
budget: Optional[float],
target_sap: Optional[float],
dependencies: Sequence[MeasureDependency] = (),
objective: Callable[[Score], float] = sap_rating,
) -> OptimisedPackage:
"""Select the Optimised Package for one Property + Scenario (ADR-0016 +
its amendment).
@ -197,26 +205,32 @@ def optimise_package(
Without a ``target_sap`` (other goals) it is max-gain-within-budget. Either
way forced dependencies are injected on every path and their cost counts
toward the spend; the returned `selected` includes them. ``budget`` of None
means unconstrained."""
baseline_sap: float = _score(scorer, baseline_epc, []).sap_continuous
means unconstrained.
``objective`` is the currency every internally-computed figure is measured
in (ADR-0062): the goal's metric, higher is better — SAP by default, CO2
reduction / bill saving for the goal-aligned Scenarios. The caller must
supply the group signals in the same currency; ``target_sap`` (when given)
is a value on the same scale."""
baseline_value: float = objective(_score(scorer, baseline_epc, []))
# Score each forced dependency's independent (role-1) impact so the selection
# can price the ventilation a wall drags in — negative for ventilation.
deps: list[MeasureDependency] = _with_role1_signals(
dependencies, scorer, baseline_epc, baseline_sap
dependencies, scorer, baseline_epc, baseline_value, objective
)
if target_sap is None:
return _max_gain_package(groups, scorer, baseline_epc, budget, deps)
target_gain: float = target_sap - baseline_sap
target_gain: float = target_sap - baseline_value
chosen: Optional[list[ScoredOption]] = optimise_min_cost(
groups, budget, target_gain, deps
)
if chosen is not None:
package: OptimisedPackage = _repair_to_target(
chosen, groups, deps, scorer, baseline_epc, budget, target_sap
chosen, groups, deps, scorer, baseline_epc, budget, target_sap, objective
)
if package.score.sap_continuous >= target_sap:
if objective(package.score) >= target_sap:
return package
# Target unreachable within budget (warm-start infeasible, or the repaired
# package still falls short) → best effort: the most improvement budget buys.
@ -231,6 +245,7 @@ def optimise_package_fabric_first(
budget: Optional[float],
target_sap: Optional[float],
dependencies: Sequence[MeasureDependency] = (),
objective: Callable[[Score], float] = sap_rating,
) -> OptimisedPackage:
"""Select the Optimised Package under the Fabric First constraint: optimise
the fabric measures (``FABRIC_MEASURE_TYPES``) first with the full budget;
@ -258,10 +273,11 @@ def optimise_package_fabric_first(
budget=budget,
target_sap=target_sap,
dependencies=dependencies,
objective=objective,
)
if (
target_sap is not None
and fabric_package.score.sap_continuous >= target_sap
and objective(fabric_package.score) >= target_sap
):
return fabric_package
if not fabric_package.selected:
@ -311,13 +327,15 @@ def optimise_package_fabric_first(
remaining_groups,
post_fabric_scorer,
baseline_epc,
start_sap=fabric_package.score.sap_continuous,
objective=objective,
start_value=objective(fabric_package.score),
),
scorer=post_fabric_scorer,
baseline_epc=baseline_epc,
budget=leftover_budget,
target_sap=target_sap,
dependencies=outstanding_dependencies,
objective=objective,
)
return OptimisedPackage(
selected=[*fabric_package.selected, *top_up.selected],
@ -338,23 +356,25 @@ def _rescored_groups(
scorer: Scorer,
baseline_epc: EpcPropertyData,
*,
start_sap: float,
objective: Callable[[Score], float],
start_value: float,
) -> list[list[ScoredOption]]:
"""The groups with every Option's role-1 warm-start signal re-scored
through ``scorer`` — for phase 2, its independent gain on the post-fabric
dwelling rather than the raw baseline, so options whose worth changes once
the envelope is treated (a boiler on an insulated home) are re-ranked.
``start_sap`` is the score of ``baseline_epc`` through ``scorer`` with no
candidate applied — the caller already has it (the phase-1 package score),
so it is threaded in rather than re-computed."""
through ``scorer`` in the ``objective``'s currency — for phase 2, its
independent gain on the post-fabric dwelling rather than the raw baseline,
so options whose worth changes once the envelope is treated (a boiler on
an insulated home) are re-ranked. ``start_value`` is the objective value of
``baseline_epc`` through ``scorer`` with no candidate applied — the caller
already has it (the phase-1 package score in the objective's currency), so
it is threaded in rather than re-computed."""
return [
[
ScoredOption(
option=scored.option,
sap_gain=scorer.score(
baseline_epc, [scored.option.overlay]
).sap_continuous
- start_sap,
sap_gain=objective(
scorer.score(baseline_epc, [scored.option.overlay])
)
- start_value,
)
for scored in group
]
@ -383,18 +403,20 @@ def _with_role1_signals(
dependencies: Sequence[MeasureDependency],
scorer: Scorer,
baseline_epc: EpcPropertyData,
baseline_sap: float,
baseline_value: float,
objective: Callable[[Score], float],
) -> list[MeasureDependency]:
"""Replace each dependency's placeholder role-1 signal with its true
independent-vs-baseline SAP impact, so the selectors price what the
dependency really does to the package (ADR-0016 amendment)."""
independent-vs-baseline impact **in the objective's currency**, so the
selectors price what the dependency really does to the package (ADR-0016
amendment; ADR-0062 for the currency)."""
scored: list[MeasureDependency] = []
for dependency in dependencies:
signal: float = (
scorer.score(
baseline_epc, [dependency.required.option.overlay]
).sap_continuous
- baseline_sap
objective(
scorer.score(baseline_epc, [dependency.required.option.overlay])
)
- baseline_value
)
scored.append(
MeasureDependency(
@ -432,16 +454,17 @@ def _repair_to_target(
baseline_epc: EpcPropertyData,
budget: Optional[float],
target_sap: float,
objective: Callable[[Score], float],
) -> OptimisedPackage:
"""Inject dependencies onto the warm-start, re-score for the truth, then
greedy-add the untreated-group Option with the best marginal SAP-per-£ (its
own dependency folded in) until the true SAP clears ``target_sap`` or no
affordable improving Option remains."""
greedy-add the untreated-group Option with the best marginal objective-per-£
(its own dependency folded in) until the true objective value clears
``target_sap`` or no affordable improving Option remains."""
selected: list[ScoredOption] = _inject(chosen, dependencies)
score: Score = _score(scorer, baseline_epc, selected)
while score.sap_continuous < target_sap:
while objective(score) < target_sap:
candidate = _best_repair_candidate(
groups, chosen, dependencies, scorer, baseline_epc, score, budget
groups, chosen, dependencies, scorer, baseline_epc, score, budget, objective
)
if candidate is None:
break
@ -499,14 +522,16 @@ def _best_repair_candidate(
baseline_epc: EpcPropertyData,
current: Score,
budget: Optional[float],
objective: Callable[[Score], float],
) -> Optional[ScoredOption]:
"""The untreated-group Option giving the best **marginal** SAP-per-£ when
added to the current package — re-scored (not the role-1 signal) with any
ventilation dependency it newly triggers folded in, so both its SAP and its
incremental cost are truthful. Affordable when the resulting whole-package
"""The untreated-group Option giving the best **marginal** objective-per-£
when added to the current package — re-scored (not the role-1 signal) with
any ventilation dependency it newly triggers folded in, so both its gain and
its incremental cost are truthful. Affordable when the resulting whole-package
cost is within ``budget`` and strictly improving. None if there is none."""
used: set[int] = _used_group_indices(groups, chosen)
base_cost: float = _package_cost(_inject(chosen, dependencies))
current_value: float = objective(current)
best: Optional[ScoredOption] = None
best_ratio: float = 0.0
for index, group in enumerate(groups):
@ -520,7 +545,7 @@ def _best_repair_candidate(
if budget is not None and package_cost > budget:
continue
trial: Score = _score(scorer, baseline_epc, trial_selected)
marginal: float = trial.sap_continuous - current.sap_continuous
marginal: float = objective(trial) - current_value
if marginal <= 0.0:
continue
incremental: float = package_cost - base_cost

43
domain/modelling/scoring/scoring.py
View file

@ -15,7 +15,7 @@ truthful. The whole-package re-score (role 2) is `PackageScorer.score` directly.
"""
from dataclasses import dataclass
from typing import Sequence
from typing import Callable, Optional, Sequence
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.modelling.scoring.package_scorer import PackageScorer, Score
@ -83,33 +83,28 @@ def marginal_impacts(
return marginals_from_scores(cascade_scores(scorer, baseline, overlays))
def independent_option_impacts(
def independent_option_signals(
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] = []
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 = next(
(impact for overlay, impact in scored if overlay == option.overlay), None
cached: Optional[float] = next(
(signal for overlay, signal 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
),
)
cached = objective(scorer.score(baseline, [option.overlay])) - base_value
scored.append((option.overlay, cached))
impacts.append(cached)
return impacts
signals.append(cached)
return signals

4
harness/console.py
View file

@ -39,6 +39,7 @@ from orchestration.modelling_orchestrator import (
_candidate_recommendations, # pyright: ignore[reportPrivateUsage]
)
from orchestration.property_baseline_orchestrator import PropertyBaselineOrchestrator
from repositories.fuel_rates.fuel_rates_repository import FuelRatesRepository
from repositories.fuel_rates.fuel_rates_static_file_repository import (
FuelRatesStaticFileRepository,
)
@ -182,6 +183,7 @@ def run_modelling(
considered_measures: Optional[frozenset[MeasureType]] = None,
products: Optional[ProductRepository] = None,
scenario: Optional[Scenario] = None,
fuel_rates: Optional[FuelRatesRepository] = None,
print_table: bool = True,
) -> Plan:
"""Run ONLY the Modelling stage over ``epc`` with no database — skipping
@ -240,7 +242,7 @@ def run_modelling(
ModellingOrchestrator(
unit_of_work=lambda: unit,
calculator=Sap10Calculator(),
fuel_rates=FuelRatesStaticFileRepository(),
fuel_rates=fuel_rates or FuelRatesStaticFileRepository(),
).run(
property_ids=[_PROPERTY_ID],
scenario_ids=[scenario_id],

89
orchestration/modelling_orchestrator.py
View file

@ -20,16 +20,18 @@ from domain.modelling.optimisation.optimiser import (
ScoredOption,
optimise_package,
optimise_package_fabric_first,
sap_rating,
)
from domain.modelling.scoring.package_scorer import PackageScorer, Score
from domain.modelling.plan import Plan, PlanMeasure
from domain.modelling.recommendation import MeasureOption, Recommendation
from domain.modelling.generators.roof_recommendation import recommend_roof_insulation
from domain.modelling.portfolio_goal import PortfolioGoal
from domain.modelling.scenario import Scenario
from domain.modelling.scoring.scoring import (
MeasureImpact,
cascade_scores,
independent_option_impacts,
independent_option_signals,
marginals_from_scores,
)
from domain.modelling.generators.wall_recommendation import recommend_cavity_wall
@ -49,12 +51,6 @@ from repositories.product.product_repository import ProductRepository
from repositories.solar.solar_repository import SolarRepository
from repositories.unit_of_work import UnitOfWork
# The PortfolioGoal value that targets a SAP band (cf.
# backend.app.db.models.portfolio.PortfolioGoal.INCREASING_EPC). Other goals
# (Energy Savings, Reducing CO2 emissions) don't yet set a SAP repair target —
# the optimiser just maximises SAP gain within budget for them (later slice).
_INCREASING_EPC_GOAL: Final[str] = "Increasing EPC"
# Best-practice install sequence for the role-3 attribution cascade (ADR-0016):
# walls → roof → ventilation → floor, per the legacy `Recommendations` class.
# Ventilation sits after the fabric that triggers it so its (negative) marginal
@ -176,6 +172,11 @@ class ModellingOrchestrator:
considered: Optional[frozenset[MeasureType]] = combine_considered_measures(
scenario.considered_measures(), considered_measures
)
# The Optimiser speaks the goal's currency (ADR-0062): group signals,
# dependency pricing and repair marginals are all measured by this
# objective — SAP by default, carbon reduction for a Reducing-CO2 goal.
_require_budget_for_goal_aligned(scenario)
objective: Callable[[Score], float] = _objective_for(scenario, bill_derivation)
groups: list[list[ScoredOption]] = _scored_candidate_groups(
scorer,
effective_epc,
@ -183,6 +184,7 @@ class ModellingOrchestrator:
planning_restrictions,
solar_potential,
considered,
objective,
)
# Forced Measure Dependencies (ventilation) are excluded from the pool
# but injected into the package before the re-score (ADR-0016).
@ -202,6 +204,7 @@ class ModellingOrchestrator:
budget=scenario.budget,
target_sap=_target_sap(scenario),
dependencies=dependencies,
objective=objective,
)
# Role-3 attribution: re-apply the *selected* set in best-practice order
@ -395,9 +398,11 @@ def _scored_candidate_groups(
planning_restrictions: PlanningRestrictions,
solar_potential: Optional[SolarPotential],
considered_measures: Optional[frozenset[MeasureType]],
objective: Callable[[Score], float],
) -> list[list[ScoredOption]]:
"""One group per Recommendation: each Option scored independently against
the baseline (role-1 warm-start signal, ADR-0016)."""
the baseline (role-1 warm-start signal, ADR-0016), in the goal objective's
currency (ADR-0062)."""
# The SAP design heat loss sizes the ASHP to the dwelling (ADR-0049); read it
# off a baseline score, which the group scoring computes anyway.
baseline_result = scorer.score(effective_epc, []).sap_result
@ -414,22 +419,80 @@ def _scored_candidate_groups(
design_heat_loss_kw,
):
options = list(recommendation.options)
impacts: list[MeasureImpact] = independent_option_impacts(
scorer, effective_epc, options
signals: list[float] = independent_option_signals(
scorer, effective_epc, options, objective
)
groups.append(
[
ScoredOption(option=option, sap_gain=impact.sap_points)
for option, impact in zip(options, impacts, strict=True)
ScoredOption(option=option, sap_gain=signal)
for option, signal in zip(options, signals, strict=True)
]
)
return groups
def _carbon_reduction(score: Score) -> float:
"""The Reducing-CO2 objective: annual kg CO2 below zero-point, negated so
higher is better (a saved kg scores +1)."""
return -score.co2_kg_per_yr
def _bill_saving(bill_derivation: BillDerivation) -> Callable[[Score], float]:
"""The Energy-Savings objective: the annual Bill at the current Fuel Rates
snapshot, negated so higher is better (a saved £ scores +1). Priced at the
live snapshot, not SAP's internal tariff book — that difference is the
point of the goal (ADR-0062)."""
def objective(score: Score) -> float:
return -_bill_for(bill_derivation, score).total_gbp
return objective
# The goal-aligned goals (ADR-0062): each maximises its own metric within the
# Scenario budget and sets no SAP target. One table is the single source of
# "which goals are goal-aligned" — both the objective dispatch and the
# budget-required guard read it, so a new goal-aligned goal cannot be added to
# one without the other. Each entry builds its objective from the plan's
# BillDerivation (the carbon objective ignores it; the bill objective needs it).
# A goal absent from the table optimises SAP, as every goal did before.
_GOAL_ALIGNED_OBJECTIVES: Final[
dict[str, Callable[[BillDerivation], Callable[[Score], float]]]
] = {
PortfolioGoal.REDUCING_CO2_EMISSIONS.value: lambda _bill_derivation: (
_carbon_reduction
),
PortfolioGoal.ENERGY_SAVINGS.value: _bill_saving,
}
def _require_budget_for_goal_aligned(scenario: Scenario) -> None:
"""A goal-aligned Scenario is 'reduce as much as possible within this
budget' — undefined without one (unconstrained, it would recommend every
beneficial measure). Fail the misconfiguration loudly (ADR-0062)."""
if scenario.budget is None and scenario.goal in _GOAL_ALIGNED_OBJECTIVES:
raise ValueError(
f"scenario {scenario.id} has goal {scenario.goal!r} but no budget; "
"goal-aligned scenarios require a budget"
)
def _objective_for(
scenario: Scenario, bill_derivation: BillDerivation
) -> Callable[[Score], float]:
"""The metric the Scenario's goal maximises (ADR-0062), as an Optimiser
objective (higher is better). Goals without an aligned metric optimise
SAP, as every goal did before."""
build_objective = _GOAL_ALIGNED_OBJECTIVES.get(scenario.goal)
if build_objective is None:
return sap_rating
return build_objective(bill_derivation)
def _target_sap(scenario: Scenario) -> Optional[float]:
"""The SAP rating the Optimiser repairs toward — the floor of the goal
band for an INCREASING_EPC goal, else None (no SAP target)."""
if scenario.goal != _INCREASING_EPC_GOAL:
if scenario.goal != PortfolioGoal.INCREASING_EPC.value:
return None
return float(Epc(scenario.goal_value).sap_lower_bound())

162
tests/domain/modelling/test_optimiser_goal_objective.py Normal file
View file

@ -0,0 +1,162 @@
"""Behaviour of the Optimiser under a goal-aligned objective (ADR-0062): a
Scenario whose goal is Reducing CO2 emissions / Energy Savings optimises its
own metric, not SAP. The caller supplies group signals already measured in the
objective's currency; the optimiser must price everything *it* computes — the
forced Measure Dependency signals — in the same currency, so a ventilation
that costs SAP but is carbon-neutral cannot sink a carbon-improving wall.
"""
from __future__ import annotations
from typing import Sequence
from datatypes.epc.domain.epc_property_data import (
BuildingPartIdentifier,
EpcPropertyData,
)
from domain.modelling.measure_type import MeasureType
from domain.modelling.optimisation.optimiser import (
OptimisedPackage,
ScoredOption,
optimise_package,
optimise_package_fabric_first,
)
from domain.modelling.scoring.package_scorer import Score
from domain.modelling.simulation import BuildingPartOverlay, EpcSimulation
from tests.domain.modelling._optimiser_fixtures import (
ASHP_OVERLAY,
BOILER_OVERLAY,
WALL_OVERLAY,
scored_option,
selected_types,
ventilation_dependency,
)
from tests.domain.sap10_calculator.worksheet._elmhurst_worksheet_000490 import (
build_epc,
)
class _CarbonScorer:
"""A stub where the wall is a small carbon win (−20 kg/yr) and a large SAP
win (+6), while its forced ventilation is carbon-neutral but SAP-ruinous
(−30): SAP-priced dependency signals sink the wall; carbon-priced ones
keep it."""
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
sap, co2 = 60.0, 500.0
for sim in simulations:
if sim.ventilation is not None:
sap -= 30.0
for part in sim.building_parts.values():
if part.wall_insulation_type is not None:
sap += 6.0
co2 -= 20.0
return Score(
sap_continuous=sap, co2_kg_per_yr=co2, primary_energy_kwh_per_yr=0.0
)
def _carbon_reduction(score: Score) -> float:
return -score.co2_kg_per_yr
def test_dependency_signals_are_priced_in_the_objective_currency() -> None:
# Arrange — the wall's signal (supplied by the caller, +20 kg CO2 saved)
# and the ventilation it forces in (carbon-neutral). Under legacy SAP
# pricing the ventilation's −30 SAP would outweigh the wall's +20 signal
# and the package would collapse to nothing.
groups: list[list[ScoredOption]] = [
[scored_option("cavity_wall_insulation", gain=20.0, cost=1000.0, overlay=WALL_OVERLAY)],
]
dependency = ventilation_dependency(
cost=300.0, triggers=frozenset({MeasureType.CAVITY_WALL_INSULATION})
)
# Act — a Reducing-CO2 brief: maximise carbon reduction within budget.
package: OptimisedPackage = optimise_package(
groups=groups,
scorer=_CarbonScorer(),
baseline_epc=build_epc(),
budget=5000.0,
target_sap=None,
dependencies=[dependency],
objective=_carbon_reduction,
)
# Assert — the wall survives with its ventilation: the dependency is worth
# 0 kg CO2, not −30 SAP, so the package is a net +20 kg saving.
assert selected_types(package.selected) == {
"cavity_wall_insulation",
"mechanical_ventilation",
}
assert abs(package.score.co2_kg_per_yr - 480.0) <= 1e-9
# Internal wall insulation — a distinct fabric overlay so the fabric-first
# phase-1 pick is unambiguous. No shared fixture (the shared WALL_OVERLAY is a
# cavity fill, type 2); this is a solid-wall internal treatment, type 3.
_IWI_OVERLAY = EpcSimulation(
building_parts={
BuildingPartIdentifier.MAIN: BuildingPartOverlay(wall_insulation_type=3)
}
)
class _CarbonHeatingScorer:
"""A stub where the boiler wins on SAP (+10 vs +2) but the heat pump wins
on carbon (−50 vs −5 kg/yr): a fabric-first phase 2 that re-scores its
candidates in SAP picks the wrong heating for a Reducing-CO2 brief."""
def score(
self, baseline: EpcPropertyData, simulations: Sequence[EpcSimulation]
) -> Score:
sap, co2 = 60.0, 500.0
for sim in simulations:
for part in sim.building_parts.values():
if part.wall_insulation_type is not None:
sap += 5.0
co2 -= 10.0
if sim.heating is None:
continue
if sim.heating.sap_main_heating_code is not None:
sap += 10.0
co2 -= 5.0
if sim.heating.main_heating_index_number is not None:
sap += 2.0
co2 -= 50.0
return Score(
sap_continuous=sap, co2_kg_per_yr=co2, primary_energy_kwh_per_yr=0.0
)
def test_fabric_first_phase_two_rescores_in_the_objective_currency() -> None:
# Arrange — a fabric-first Reducing-CO2 brief. Phase 1 commits the wall;
# phase 2 must choose the heating on its post-fabric *carbon* worth, not
# its SAP worth. Signals are supplied in kg CO2 saved (the caller's job).
groups: list[list[ScoredOption]] = [
[scored_option("internal_wall_insulation", gain=10.0, cost=1000.0, overlay=_IWI_OVERLAY)],
[
scored_option("gas_boiler_upgrade", gain=5.0, cost=2000.0, overlay=BOILER_OVERLAY),
scored_option("air_source_heat_pump", gain=50.0, cost=6000.0, overlay=ASHP_OVERLAY),
],
]
# Act — no target (goal-aligned briefs have none), generous budget.
package: OptimisedPackage = optimise_package_fabric_first(
groups=groups,
scorer=_CarbonHeatingScorer(),
baseline_epc=build_epc(),
budget=10000.0,
target_sap=None,
objective=_carbon_reduction,
)
# Assert — the wall plus the heat pump (−50 kg), not the SAP-favoured
# boiler; the truthful package carbon is 500 − 10 − 50 = 440.
assert selected_types(package.selected) == {
"internal_wall_insulation",
"air_source_heat_pump",
}
assert abs(package.score.co2_kg_per_yr - 440.0) <= 1e-9

12
tests/domain/modelling/test_scoring.py
View file

@ -16,7 +16,7 @@ from domain.modelling.recommendation import MeasureOption
from domain.modelling.scoring.scoring import (
MeasureImpact,
cascade_scores,
independent_option_impacts,
independent_option_signals,
marginal_impacts,
marginals_from_scores,
)
@ -64,7 +64,7 @@ def _option(overlay: EpcSimulation) -> MeasureOption:
)
def test_independent_option_impacts_score_each_distinct_overlay_once() -> None:
def test_independent_option_signals_score_each_distinct_overlay_once() -> None:
# Arrange
baseline: EpcPropertyData = build_epc()
scorer = _CountingScorer()
@ -86,15 +86,15 @@ def test_independent_option_impacts_score_each_distinct_overlay_once() -> None:
options = [_option(overlay_a), _option(overlay_a_dup), _option(overlay_b)]
# Act
impacts: list[MeasureImpact] = independent_option_impacts(
scorer, baseline, options
signals: list[float] = independent_option_signals(
scorer, baseline, options, lambda score: score.sap_continuous
)
# 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
assert signals[0] == signals[1] == 2.0
assert signals[2] == 3.0
def test_single_overlay_marginal_is_its_improvement_over_baseline() -> None:

133
tests/orchestration/test_modelling_goal_objectives.py Normal file
View file

@ -0,0 +1,133 @@
"""The ModellingOrchestrator aligns the Optimiser's objective with the
Scenario's goal (ADR-0062): Reducing CO2 emissions maximises the carbon
reduction the budget buys, Energy Savings maximises the annual bill saving,
and Increasing EPC keeps its SAP target semantics. End-to-end through
``run_modelling`` (no database) with the real calculator, against the
uninsulated solid-brick 001431 dwelling where the SAP-optimal and
carbon-optimal packages diverge at a £16,000 budget.
"""
from __future__ import annotations
import dataclasses
import pytest
from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.fuel_rates.fuel import Fuel
from domain.fuel_rates.fuel_rates import FuelRate, FuelRates
from domain.modelling.measure_type import MeasureType
from domain.modelling.plan import Plan
from domain.modelling.scenario import Scenario
from harness.console import run_modelling
from repositories.fuel_rates.fuel_rates_repository import FuelRatesRepository
from repositories.fuel_rates.fuel_rates_static_file_repository import (
FuelRatesStaticFileRepository,
)
from tests.domain.modelling._elmhurst_recommendation import (
parse_recommendation_summary,
)
def _solid_brick_dwelling() -> EpcPropertyData:
return parse_recommendation_summary("solid_brick_ewi_001431_before.pdf")
def _scenario(goal: str, *, budget: float) -> Scenario:
return Scenario(
id=999, goal=goal, goal_value="", budget=budget, is_default=True
)
def test_reducing_co2_scenario_buys_carbon_not_sap() -> None:
# Arrange — at £16,000 the SAP objective buys the wall + floor + £3,200
# gas boiler package (~2,069 kg CO2/yr, SAP 72.9). The carbon objective
# swaps the boiler for electric storage heaters (~1,098 kg/yr) — a lower
# SAP, but ~970 kg/yr less carbon on the low-carbon grid.
epc = _solid_brick_dwelling()
# Act — the same dwelling and budget under each goal.
sap_led: Plan = run_modelling(
epc,
scenario=_scenario("Valuation Improvement", budget=16000.0),
print_table=False,
)
carbon_led: Plan = run_modelling(
epc,
scenario=_scenario("Reducing CO2 emissions", budget=16000.0),
print_table=False,
)
# Assert — the goal changes the outcome in the goal's favour: the carbon
# plan cuts materially more CO2 than the SAP plan buys with the same
# money, and the gas boiler that wins on SAP-per-£ is rejected.
assert (
carbon_led.post_retrofit.co2_kg_per_yr
< sap_led.post_retrofit.co2_kg_per_yr - 500.0
)
selected = {measure.measure_type for measure in carbon_led.measures}
assert MeasureType.GAS_BOILER_UPGRADE not in selected
def test_a_goal_aligned_scenario_without_a_budget_fails_loudly() -> None:
# Arrange — 'reduce as much as possible within this budget' is undefined
# without a budget: unconstrained it would recommend every beneficial
# measure. A budget-less goal-aligned Scenario is a misconfiguration and
# must fail visibly, not produce a maximal plan.
epc = _solid_brick_dwelling()
budgetless = Scenario(
id=999,
goal="Reducing CO2 emissions",
goal_value="",
budget=None,
is_default=True,
)
# Act / Assert
with pytest.raises(ValueError, match="budget"):
run_modelling(epc, scenario=budgetless, print_table=False)
class _FixedFuelRates(FuelRatesRepository):
def __init__(self, rates: FuelRates) -> None:
self._rates = rates
def get_current(self) -> FuelRates:
return self._rates
def _cheap_electricity_snapshot() -> FuelRates:
"""The committed snapshot with electricity at 1p/kWh — a world where any
electric heating out-bills gas, while SAP's internal price book (which the
calculator rates against) is unmoved."""
base = FuelRatesStaticFileRepository().get_current()
rates = dict(base.rates)
rates[Fuel.ELECTRICITY] = FuelRate(
unit_rate_p_per_kwh=1.0,
standing_charge_p_per_day=rates[Fuel.ELECTRICITY].standing_charge_p_per_day,
)
return dataclasses.replace(base, rates=rates)
def test_energy_savings_scenario_prices_packages_at_the_live_fuel_rates() -> None:
# Arrange — SAP is itself a cost metric, but it prices energy from its
# internal tariff book. The Energy Savings goal must price at the *live*
# Fuel Rates snapshot: with 1p/kWh electricity, electric heating slashes
# the bill even though SAP still scores the gas boiler package higher.
epc = _solid_brick_dwelling()
# Act
plan: Plan = run_modelling(
epc,
scenario=_scenario("Energy Savings", budget=16000.0),
fuel_rates=_FixedFuelRates(_cheap_electricity_snapshot()),
print_table=False,
)
# Assert — the bill objective abandons the boiler for electric heating.
selected = {measure.measure_type for measure in plan.measures}
assert MeasureType.GAS_BOILER_UPGRADE not in selected
assert selected & {
MeasureType.AIR_SOURCE_HEAT_PUMP,
MeasureType.HIGH_HEAT_RETENTION_STORAGE_HEATERS,
}