Model/domain/epc_prediction/epc_prediction.py

"""EPC Prediction synthesis (ADR-0029).

`EpcPrediction.predict` turns the selected `ComparableProperties` into a
predicted `EpcPropertyData`: copy a coherent representative template's structure
(building parts, windows, geometry), set the homogeneous categoricals to the
recency-weighted cohort mode, then apply Landlord Overrides on top. Pure domain
logic — deterministic neighbour synthesis, not ML.
"""

from __future__ import annotations

import copy
import math
import statistics
from collections import Counter, defaultdict
from datetime import date
from typing import Iterable, Optional, Union

from datatypes.epc.domain.epc_property_data import (
    EpcPropertyData,
    SapBuildingPart,
)
from domain.epc_prediction.comparable_properties import (
    Comparable,
    ComparableProperties,
    PredictionTarget,
)


class EpcPrediction:
    """Synthesises a predicted `EpcPropertyData` from Comparable Properties."""

    def predict(
        self, target: PredictionTarget, comparables: ComparableProperties
    ) -> EpcPropertyData:
        """Predict the target's EPC picture: copy a representative template's
        structure (coherent for the calculator), then set the homogeneous
        categoricals to the cohort mode."""
        template: Comparable = self._template(comparables)
        predicted: EpcPropertyData = copy.deepcopy(template.epc)
        self._apply_categorical_modes(predicted, comparables)
        self._apply_overrides(predicted, target)
        return predicted

    @staticmethod
    def _template(comparables: ComparableProperties) -> Comparable:
        """The representative comparable whose structure seeds the prediction:
        the member whose floor area is closest to the cohort median. A single
        neighbour's geometry is copied wholesale, so a size-representative
        template keeps the prediction off the cohort's size outliers (ADR-0029
        decision 4: closest on size)."""
        members: tuple[Comparable, ...] = comparables.members
        median_area: float = statistics.median(
            c.epc.total_floor_area_m2 for c in members
        )
        return min(
            members,
            key=lambda c: abs(c.epc.total_floor_area_m2 - median_area),
        )

    @staticmethod
    def _apply_categorical_modes(
        predicted: EpcPropertyData, comparables: ComparableProperties
    ) -> None:
        """Override the predicted picture's homogeneous categoricals — wall /
        roof / floor construction + insulation, age band — with the cohort mode
        (robust to an atypical template, per ADR-0029 decision 4). The template
        still supplies the geometry; only the categorical codes move to the mode.
        (Glazing type is deliberately left on the template — moding it is
        marginal and noisy; revisit with a larger corpus.)"""
        if not predicted.sap_building_parts:
            return
        main: SapBuildingPart = predicted.sap_building_parts[0]
        members = comparables.members
        for attr in _MAIN_PART_CATEGORICALS:
            if attr in _RECENCY_WEIGHTED_CATEGORICALS:
                mode = _recency_weighted_mode(members, attr)
            else:
                mode = _mode(_main_part_attr(c, attr) for c in members)
            if mode is not None:
                setattr(main, attr, mode)
        floor_dims = main.sap_floor_dimensions
        if floor_dims:
            for attr in _FLOOR_DIM_CATEGORICALS:
                floor_mode = _int_mode(_main_floor_attr(c, attr) for c in members)
                if floor_mode is not None:
                    setattr(floor_dims[0], attr, floor_mode)

    @staticmethod
    def _apply_overrides(
        predicted: EpcPropertyData, target: PredictionTarget
    ) -> None:
        """Apply the known Landlord Overrides on top of the estimate — a known
        value always wins over the cohort mode (ADR-0029)."""
        if not predicted.sap_building_parts:
            return
        if target.wall_construction is not None:
            predicted.sap_building_parts[0].wall_construction = (
                target.wall_construction
            )


# The homogeneous categoricals carried directly on the main building part. Floor
# categoricals live on the main floor dimension and glazing on the windows; both
# are handled separately.
_MAIN_PART_CATEGORICALS: tuple[str, ...] = (
    "wall_construction",
    "wall_insulation_type",
    "construction_age_band",
    "roof_construction",
    "roof_insulation_thickness",
)

# Integer-coded categoricals on the main building part's ground-floor dimension.
_FLOOR_DIM_CATEGORICALS: tuple[str, ...] = (
    "floor_construction",
    "floor_insulation",
)

# Categoricals whose physical value CHANGES over time (e.g. loft top-ups), so a
# recent neighbour reflects the current state better than an old one — these take
# a recency-WEIGHTED mode. Permanent categoricals (wall / age) take the plain
# mode: recency-weighting them was net-negative on the validation corpus (it
# discards data that is still valid). `_RECENCY_TAU_YEARS` is the exponential
# decay constant (≈2.8-year half-life), chosen on the corpus (roof insulation
# +4pp / +12pp on the fixture).
_RECENCY_WEIGHTED_CATEGORICALS: frozenset[str] = frozenset(
    {"roof_insulation_thickness"}
)
_RECENCY_TAU_YEARS: float = 4.0
_DAYS_PER_YEAR: float = 365.0


def _main_part_attr(
    comparable: Comparable, attr: str
) -> Optional[Union[int, str]]:
    parts: list[SapBuildingPart] = comparable.epc.sap_building_parts
    return getattr(parts[0], attr) if parts else None


def _main_floor_attr(comparable: Comparable, attr: str) -> Optional[int]:
    parts: list[SapBuildingPart] = comparable.epc.sap_building_parts
    if not parts:
        return None
    dims = parts[0].sap_floor_dimensions
    value: Optional[int] = getattr(dims[0], attr) if dims else None
    return value


def _mode(
    values: Iterable[Optional[Union[int, str]]],
) -> Optional[Union[int, str]]:
    """The most common non-None value, or None when there are none."""
    present = [v for v in values if v is not None]
    if not present:
        return None
    return Counter(present).most_common(1)[0][0]


def _recency_weighted_mode(
    members: tuple[Comparable, ...], attr: str
) -> Optional[Union[int, str]]:
    """The cohort mode of a main-part attribute, weighting each comparable's vote
    by recency — an exponential decay in the cert's age relative to the newest in
    the cohort. Newer neighbours dominate, so a stale majority can't outvote the
    current state. Falls back to a plain mode when no registration dates are
    lodged (all ages 0 ⇒ equal weight)."""
    newest: date = max(
        (c.registration_date or date.min for c in members), default=date.min
    )
    weights: dict[Union[int, str], float] = defaultdict(float)
    for comparable in members:
        value = _main_part_attr(comparable, attr)
        if value is None:
            continue
        lodged: date = comparable.registration_date or date.min
        age_years: float = (newest - lodged).days / _DAYS_PER_YEAR
        weights[value] += math.exp(-age_years / _RECENCY_TAU_YEARS)
    if not weights:
        return None
    return max(weights, key=lambda value: weights[value])


def _int_mode(values: Iterable[Optional[int]]) -> Optional[int]:
    """`_mode` narrowed to int-coded fields (keeps pyright strict happy when the
    target attribute is typed `Optional[int]`)."""
    present = [v for v in values if v is not None]
    if not present:
        return None
    return Counter(present).most_common(1)[0][0]