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 dataclasses import dataclass
from datetime import date
from typing import Callable, Iterable, Optional, Union

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


@dataclass(frozen=True)
class PredictionConfidence:
    """A compute-only confidence signal for a prediction (ADR-0029 open item).

    `cohort_size` is the number of Comparable Properties the prediction drew on;
    `component_agreement` maps a homogeneous component to the cohort's *agreement*
    — the modal value's share (0..1) of the neighbours that lodge one. A small or
    split cohort flags a component downstream may want to treat cautiously (e.g.
    the per-dwelling fields with a low accuracy ceiling). Surfacing / persisting
    this is a separate HITL follow-up; here it is computed only.
    """

    cohort_size: int
    component_agreement: dict[str, float]

    def agreement(self, component: str) -> Optional[float]:
        """The cohort's modal-value share for a component, or None when no
        neighbour lodges one (it was not applicable)."""
        return self.component_agreement.get(component)


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), set the predicted floor area to
        the cohort median (the best point estimate of the target's size, decoupled
        from the one template's own area), then set the homogeneous categoricals
        to the cohort mode."""
        template: Comparable = self._template(comparables)
        predicted: EpcPropertyData = copy.deepcopy(template.epc)
        predicted.total_floor_area_m2 = _median_floor_area(comparables.members)
        self._apply_categorical_modes(predicted, comparables)
        self._apply_glazing_mode(predicted, comparables)
        self._apply_heating_donor(predicted, comparables)
        self._apply_overrides(predicted, target)
        return predicted

    @staticmethod
    def _apply_heating_donor(
        predicted: EpcPropertyData, comparables: ComparableProperties
    ) -> None:
        """Replace the structural template's heating with a coherent donor's whole
        `SapHeating` cluster (ADR-0029; issue #1225). Heating sub-fields can't be
        field-moded without breaking system coherence (e.g. a fuel that doesn't
        match the emitter), so the cluster is copied as a unit from a single
        neighbour: the one matching the cohort's modal heating *signature* (main
        fuel + category + cylinder presence), the most recent among those matches
        (a recent cert reflects the current system). This makes the predicted
        heating both representative and internally coherent, rather than whatever
        the size-representative template happened to carry. No donor (no neighbour
        lodges a main heating system) leaves the template's heating in place."""
        donor = _heating_donor(comparables.members)
        if donor is None:
            return
        predicted.sap_heating = copy.deepcopy(donor.epc.sap_heating)
        predicted.has_hot_water_cylinder = donor.epc.has_hot_water_cylinder
        predicted.solar_water_heating = donor.epc.solar_water_heating

    @staticmethod
    def _apply_glazing_mode(
        predicted: EpcPropertyData, comparables: ComparableProperties
    ) -> None:
        """Set every window's glazing type to the recency-weighted cohort mode.
        Glazing is retrofitted over a dwelling's life (single → double), so a
        recent neighbour reflects the current state — its correct method is the
        recency-weighted mode (like roof insulation), NOT the plain mode (which
        regressed) or the template copy. The window geometry (size, count) is
        left on the template; only the glazing categorical moves."""
        glazing = _recency_weighted_choice(
            comparables.members, _comparable_modal_glazing
        )
        if glazing is None:
            return
        for window in predicted.sap_windows:
            window.glazing_type = glazing

    def confidence(
        self, comparables: ComparableProperties
    ) -> PredictionConfidence:
        """Compute the per-prediction confidence from the cohort: its size plus,
        for each homogeneous categorical, the modal value's share among the
        neighbours that lodge one (ADR-0029). Compute-only — it never alters the
        prediction, only annotates how much the cohort agreed."""
        members: tuple[Comparable, ...] = comparables.members
        agreement: dict[str, float] = {}
        for attr in _MAIN_PART_CATEGORICALS:
            share: Optional[float] = _modal_share(
                _main_part_attr(c, attr) for c in members
            )
            if share is not None:
                agreement[attr] = share
        for attr in _FLOOR_DIM_CATEGORICALS:
            floor_share: Optional[float] = _modal_share(
                _main_floor_attr(c, attr) for c in members
            )
            if floor_share is not None:
                agreement[attr] = floor_share
        return PredictionConfidence(
            cohort_size=len(members), component_agreement=agreement
        )

    @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 mode is
        physically-similarity-weighted (decision 5): each neighbour's vote decays
        with its distance from the cohort's physical centre, so the mode leans on
        the most representative neighbours rather than treating every survivor
        equally. 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
        weights: list[float] = _similarity_weights(members)
        for attr in _MAIN_PART_CATEGORICALS:
            if attr in _RECENCY_WEIGHTED_CATEGORICALS:
                mode = _recency_weighted_mode(members, attr)
            else:
                mode = _weighted_mode(
                    (_main_part_attr(c, attr) for c in members), weights
                )
            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 = _weighted_int_mode(
                    (_main_floor_attr(c, attr) for c in members), weights
                )
                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

# Physical-similarity weighting of the categorical mode (ADR-0029 decision 5): a
# comparable's vote decays exponentially with how far it sits from the cohort's
# physical centre — floor area from the median, construction age from the modal
# band — so an outlier-sized or outlier-era neighbour can't sway the mode. Scales
# chosen on the validation corpus (wall-insulation +2.8pp / roof +1.1pp /
# floor-construction +2.4pp / floor-insulation +1.2pp; gate-safe, no regression).
_SIMILARITY_SIZE_SCALE_M2: float = 20.0
_SIMILARITY_AGE_WEIGHT: float = 0.5
_AGE_BAND_ORDER: str = "ABCDEFGHIJKL"


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 _median_floor_area(members: tuple[Comparable, ...]) -> float:
    """The cohort's median floor area — the point estimate of the target's size.
    The median minimises mean absolute deviation, so it is the best single guess
    for an unknown neighbour's area; it is set independently of the structural
    template (the calculator derives heat loss from the building-part geometry,
    not this scalar, so the two need not agree)."""
    return statistics.median(c.epc.total_floor_area_m2 for c in members)


def _age_band_index(comparable: Comparable) -> Optional[int]:
    """The main building part's construction-age-band position (A=0 … L=11), or
    None when no recognisable band is lodged."""
    band = _main_part_attr(comparable, "construction_age_band")
    if isinstance(band, str) and band in _AGE_BAND_ORDER:
        return _AGE_BAND_ORDER.index(band)
    return None


def _similarity_weights(members: tuple[Comparable, ...]) -> list[float]:
    """A physical-similarity weight per comparable (ADR-0029 decision 5): the
    product of an exponential decay in its floor-area distance from the cohort
    median and in its age-band distance from the cohort's modal band. A neighbour
    missing a size or age contributes a neutral weight on that axis, so it is
    never penalised for absent data. Aligned with `members` index-for-index."""
    if not members:
        return []
    median_area: float = statistics.median(
        c.epc.total_floor_area_m2 for c in members
    )
    age_indices: list[Optional[int]] = [_age_band_index(c) for c in members]
    present_ages: list[int] = [i for i in age_indices if i is not None]
    modal_age: Optional[float] = (
        statistics.median(present_ages) if present_ages else None
    )
    weights: list[float] = []
    for comparable, age_index in zip(members, age_indices):
        size_term: float = math.exp(
            -abs(comparable.epc.total_floor_area_m2 - median_area)
            / _SIMILARITY_SIZE_SCALE_M2
        )
        age_term: float = (
            math.exp(-_SIMILARITY_AGE_WEIGHT * abs(age_index - modal_age))
            if modal_age is not None and age_index is not None
            else 1.0
        )
        weights.append(size_term * age_term)
    return weights


def _weighted_mode(
    values: Iterable[Optional[Union[int, str]]], weights: list[float]
) -> Optional[Union[int, str]]:
    """The value with the greatest total similarity weight (ties broken by first
    appearance, matching `_mode`), or None when no non-None value is present."""
    totals: dict[Union[int, str], float] = defaultdict(float)
    for value, weight in zip(values, weights):
        if value is not None:
            totals[value] += weight
    if not totals:
        return None
    return max(totals, key=lambda value: totals[value])


def _weighted_int_mode(
    values: Iterable[Optional[int]], weights: list[float]
) -> Optional[int]:
    """`_weighted_mode` narrowed to int-coded fields (keeps pyright strict happy
    when the target attribute is typed `Optional[int]`)."""
    totals: dict[int, float] = defaultdict(float)
    for value, weight in zip(values, weights):
        if value is not None:
            totals[value] += weight
    if not totals:
        return None
    return max(totals, key=lambda value: totals[value])


def _modal_share(
    values: Iterable[Optional[Union[int, str]]],
) -> Optional[float]:
    """The most common value's share of the present (non-None) values — a 0..1
    measure of how much the cohort agrees — or None when none are present."""
    present = [v for v in values if v is not None]
    if not present:
        return None
    modal_count: int = Counter(present).most_common(1)[0][1]
    return modal_count / len(present)


def _recency_weighted_choice(
    members: tuple[Comparable, ...],
    value_of: Callable[[Comparable], Optional[Union[int, str]]],
) -> Optional[Union[int, str]]:
    """The recency-weighted cohort mode of a per-comparable value: each
    neighbour's vote decays exponentially with the cert's age relative to the
    newest in the cohort, so newer neighbours dominate and 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). Returns None when no comparable
    supplies a value. Used for the time-varying components — those upgraded over a
    dwelling's life (loft top-ups, glazing retrofits)."""
    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 = value_of(comparable)
        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 _recency_weighted_mode(
    members: tuple[Comparable, ...], attr: str
) -> Optional[Union[int, str]]:
    """`_recency_weighted_choice` over a main building-part attribute."""
    return _recency_weighted_choice(
        members, lambda comparable: _main_part_attr(comparable, attr)
    )


def _comparable_modal_glazing(
    comparable: Comparable,
) -> Optional[Union[int, str]]:
    """A comparable's modal glazing type — the most common across its windows, or
    None when it lodges none. One glazing signal per neighbour, robust to a single
    odd window, matching how the harness scores `modal_glazing_type`."""
    types = [window.glazing_type for window in comparable.epc.sap_windows]
    return Counter(types).most_common(1)[0][0] if types else None


def _main_heating_detail(comparable: Comparable) -> Optional[MainHeatingDetail]:
    """The primary heating system's detail row, or None when none is lodged."""
    details = comparable.epc.sap_heating.main_heating_details
    return details[0] if details else None


def _heating_signature(
    comparable: Comparable,
) -> Optional[tuple[Union[int, str], Optional[int], bool]]:
    """The donor-matching signature — main fuel + heating category + cylinder
    presence: the coarse identity of the heating system. None when no main heating
    system is lodged, so the comparable is not a donor candidate."""
    detail = _main_heating_detail(comparable)
    if detail is None:
        return None
    return (
        detail.main_fuel_type,
        detail.main_heating_category,
        comparable.epc.has_hot_water_cylinder,
    )


def _heating_donor(members: tuple[Comparable, ...]) -> Optional[Comparable]:
    """The coherent heating donor: the comparable whose heating signature is the
    cohort mode, breaking ties toward the most recent cert (then certificate
    number, for determinism). None when no neighbour lodges a heating system."""
    signed = [(c, _heating_signature(c)) for c in members]
    signatures = [sig for _, sig in signed if sig is not None]
    if not signatures:
        return None
    modal = Counter(signatures).most_common(1)[0][0]
    matches = [c for c, sig in signed if sig == modal]
    return max(
        matches,
        key=lambda c: (c.registration_date or date.min, c.certificate_number),
    )