Model/domain/epc_prediction/validation.py

"""Component Accuracy aggregation for EPC Prediction (ADR-0030).

The leave-one-out scorer, calculator-FREE on purpose: it holds out each SAP 10.2
target, predicts it from its (all-vintage) Comparable Properties, and aggregates
the per-component classification hits + geometry residuals from
`compare_prediction`. This is the *primary*, calculator-independent signal — the
end-to-end SAP / carbon / PE check (which needs the calculator) is layered on top
by the runner. The same function backs both the committed ratcheting gate and the
offline national battle-test (one scorer, two harnesses).

Pure given the loaded cohorts: corpus IO (reading + mapping cert payloads) is the
caller's job, so this is directly unit-testable.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import date
from typing import Iterable, Iterator, Optional, Sequence

from datatypes.epc.domain.epc_property_data import EpcPropertyData
from domain.epc_prediction.comparable_properties import (
    Comparable,
    PredictionTarget,
    select_comparables,
)
from domain.epc_prediction.epc_prediction import EpcPrediction
from domain.epc_prediction.prediction_comparison import compare_prediction

# Only SAP 10.2 certs are valid held-out targets (ADR-0030) — the only vintage
# with full-fidelity lodged components. The source cohort keeps all vintages.
_SAP_10_2: float = 10.2


def _empty_classification() -> dict[str, list[int]]:
    return {}


def _empty_residuals() -> dict[str, list[float]]:
    return {}


@dataclass
class ComponentAccuracy:
    """Aggregated leave-one-out Component Accuracy over a corpus.

    `classification` maps a component name to [hits, applicable-total] (a
    not-applicable `None` hit is excluded from the total); `residuals` maps a
    numeric component to its signed (predicted − actual) values. `targets` counts
    the held-out SAP 10.2 properties scored.
    """

    classification: dict[str, list[int]] = field(
        default_factory=_empty_classification
    )
    residuals: dict[str, list[float]] = field(default_factory=_empty_residuals)
    targets: int = 0

    def rate(self, component: str) -> Optional[float]:
        """The classification hit-rate for a component, or None when nothing was
        applicable."""
        hits, total = self.classification.get(component, [0, 0])
        return hits / total if total else None

    def mean_abs_residual(self, component: str) -> Optional[float]:
        """Mean absolute residual for a numeric component, or None when empty."""
        values = self.residuals.get(component, [])
        return sum(abs(v) for v in values) / len(values) if values else None


def _recency_key(comparable: Comparable) -> tuple[date, str]:
    return (
        comparable.registration_date or date.min,
        comparable.certificate_number,
    )


def _latest_per_address(cohort: Sequence[Comparable]) -> list[Comparable]:
    """One held-out property per address — the latest cert, the best ground
    truth. Comparables with no address each stand alone."""
    latest: dict[str, Comparable] = {}
    standalone: list[Comparable] = []
    for c in cohort:
        if c.address is None:
            standalone.append(c)
        elif c.address not in latest or _recency_key(c) > _recency_key(
            latest[c.address]
        ):
            latest[c.address] = c
    return list(latest.values()) + standalone


def iter_predictions(
    cohorts: Iterable[Sequence[Comparable]],
    *,
    target_sap_version: float = _SAP_10_2,
) -> Iterator[tuple[EpcPropertyData, EpcPropertyData]]:
    """Yield `(predicted, actual)` for every SAP-`target_sap_version` held-out
    target across the cohorts — the single leave-one-out orchestration the
    Component Accuracy scorer and the runner's calculator end-to-end both consume
    (ADR-0030: one scorer, two harnesses). A target is held out by whole address
    (so a re-lodgement can't leak) and predicted from its all-vintage cohort."""
    predictor = EpcPrediction()
    for cohort in cohorts:
        for held_out in _latest_per_address(cohort):
            if held_out.epc.sap_version != target_sap_version:
                continue
            others = [
                c
                for c in cohort
                if c.address is None or c.address != held_out.address
            ]
            actual = held_out.epc
            target = PredictionTarget(
                postcode=actual.postcode,
                property_type=actual.property_type or "",
                built_form=actual.built_form,
            )
            comparables = select_comparables(target, others)
            if not comparables.members:
                continue
            yield predictor.predict(target, comparables), actual


def evaluate_component_accuracy(
    cohorts: Iterable[Sequence[Comparable]],
    *,
    target_sap_version: float = _SAP_10_2,
) -> ComponentAccuracy:
    """Score Component Accuracy by leave-one-out over each postcode cohort —
    aggregating the `compare_prediction` hits + residuals across every held-out
    SAP-`target_sap_version` target. Calculator-free (the primary signal)."""
    accuracy = ComponentAccuracy()
    for predicted, actual in iter_predictions(
        cohorts, target_sap_version=target_sap_version
    ):
        comparison = compare_prediction(predicted, actual)
        accuracy.targets += 1
        for name, hit in comparison.categorical_hits.items():
            counter = accuracy.classification.setdefault(name, [0, 0])
            if hit is not None:
                counter[1] += 1
                counter[0] += int(hit)
        accuracy.residuals.setdefault("floor_area", []).append(
            comparison.floor_area_residual
        )
        accuracy.residuals.setdefault("window_count", []).append(
            float(comparison.window_count_residual)
        )
        accuracy.residuals.setdefault("total_window_area", []).append(
            comparison.total_window_area_residual
        )
        accuracy.residuals.setdefault("building_parts", []).append(
            float(comparison.building_parts_residual)
        )
        accuracy.residuals.setdefault("door_count", []).append(
            float(comparison.door_count_residual)
        )
    return accuracy