Goal-aligned dispatch reads one enum-keyed table 🟪

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>
This commit is contained in:
Khalim Conn-Kowlessar 2026-07-10 11:17:27 +00:00
parent daf4449d0d
commit 98d2f7aa16
2 changed files with 37 additions and 31 deletions

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 Callable, 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
@ -99,7 +99,7 @@ def independent_option_signals(
scored: list[tuple[EpcSimulation, float]] = []
signals: list[float] = []
for option in options:
cached: float | None = next(
cached: Optional[float] = next(
(signal for overlay, signal in scored if overlay == option.overlay),
None,
)

View file

@ -26,6 +26,7 @@ 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,
@ -50,14 +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). The
# goal-aligned goals (ADR-0062) set no target: they maximise their own metric
# within the Scenario budget.
_INCREASING_EPC_GOAL: Final[str] = "Increasing EPC"
_REDUCING_CO2_GOAL: Final[str] = "Reducing CO2 emissions"
_ENERGY_SAVINGS_GOAL: Final[str] = "Energy Savings"
# 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
@ -405,7 +398,7 @@ def _scored_candidate_groups(
planning_restrictions: PlanningRestrictions,
solar_potential: Optional[SolarPotential],
considered_measures: Optional[frozenset[MeasureType]],
objective: Callable[[Score], float] = sap_rating,
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), in the goal objective's
@ -438,20 +431,6 @@ def _scored_candidate_groups(
return groups
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 (
_REDUCING_CO2_GOAL,
_ENERGY_SAVINGS_GOAL,
):
raise ValueError(
f"scenario {scenario.id} has goal {scenario.goal!r} but no budget; "
"goal-aligned scenarios require a budget"
)
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)."""
@ -470,23 +449,50 @@ def _bill_saving(bill_derivation: BillDerivation) -> Callable[[Score], float]:
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."""
if scenario.goal == _REDUCING_CO2_GOAL:
return _carbon_reduction
if scenario.goal == _ENERGY_SAVINGS_GOAL:
return _bill_saving(bill_derivation)
return sap_rating
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())