Model/scripts/audit_modelling_anomalies.py

"""Audit modelled Properties for *odd results* — a growing, pluggable set of
checks that read the DB and flag plans / baselines / recommendations that look
wrong, so the team can triage them instead of hunting by hand in the FE.

Run:
    python -m scripts.audit_modelling_anomalies --portfolio 796
    python -m scripts.audit_modelling_anomalies --portfolio 796 --severity high
    python -m scripts.audit_modelling_anomalies --property 725634

Writes ``modelling_audit.md`` + ``modelling_audit.csv`` and prints a summary.

ADDING A CHECK: write a function ``(a: PropertyAudit) -> Optional[str]`` that
returns a one-line reason when the Property looks wrong (else None), and decorate
it with ``@check("kebab-name", Severity.HIGH)``. That is the whole contract — the
runner discovers it, runs it over every Property, and reports the reasons. Keep
each check small and single-purpose; lean on the shared `PropertyAudit` bundle
rather than re-querying.

Read-only: this script never writes to the DB.
"""

from __future__ import annotations

import argparse
import csv
from dataclasses import dataclass
from enum import IntEnum
from typing import Callable, Optional

from sqlalchemy import text

from datatypes.epc.domain.epc import Epc
from scripts.e2e_common import build_engine, load_env

# A..G, A best — index is the rank (lower = better) for band comparisons.
_BANDS = "ABCDEFG"


def _band_rank(band: Optional[str]) -> Optional[int]:
    if band is None or band not in _BANDS:
        return None
    return _BANDS.index(band)


def _band_of(score: Optional[float]) -> Optional[str]:
    if score is None:
        return None
    return Epc.from_sap_score(round(score)).value


class Severity(IntEnum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3


@dataclass(frozen=True)
class PropertyAudit:
    """Everything a check needs about one modelled Property, joined once.

    The *default* plan is the one shown in the FE; ``None`` when the Property has
    no plan for the scenario. All performance figures are the persisted ones.
    """

    property_id: int
    uprn: Optional[int]
    portfolio_id: int
    scenario_id: Optional[int]
    scenario_goal_band: Optional[str]
    lodged_sap: Optional[float]
    lodged_band: Optional[str]
    effective_sap: Optional[float]
    effective_band: Optional[str]
    rebaseline_reason: Optional[str]
    post_sap: Optional[float]
    post_band: Optional[str]
    cost_of_works: Optional[float]
    energy_bill_savings: Optional[float]
    energy_consumption_savings: Optional[float]


@dataclass(frozen=True)
class Anomaly:
    property_id: int
    uprn: Optional[int]
    check: str
    severity: Severity
    detail: str


Check = Callable[[PropertyAudit], Optional[str]]
_REGISTRY: list[tuple[str, Severity, Check]] = []


def check(name: str, severity: Severity) -> Callable[[Check], Check]:
    def register(fn: Check) -> Check:
        _REGISTRY.append((name, severity, fn))
        return fn

    return register


# ───────────────────────── checks ─────────────────────────
# Each returns a reason string when the Property looks wrong, else None.


@check("plan-below-baseline-band", Severity.HIGH)
def _plan_below_baseline_band(a: PropertyAudit) -> Optional[str]:
    """The default plan's post-works band is WORSE than the baseline band — a
    retrofit plan should never end below where the property started."""
    base, post = _band_rank(a.effective_band), _band_rank(a.post_band)
    if base is None or post is None or post <= base:
        return None
    return f"post {a.post_band} ({a.post_sap}) worse than effective baseline {a.effective_band} ({a.effective_sap})"


@check("plan-score-below-baseline", Severity.HIGH)
def _plan_score_below_baseline(a: PropertyAudit) -> Optional[str]:
    """Post-works SAP is materially BELOW the baseline SAP — works that lower the
    score, or a plan/baseline computed from different pictures."""
    if a.effective_sap is None or a.post_sap is None:
        return None
    if a.post_sap >= a.effective_sap - 0.5:
        return None
    return f"post SAP {a.post_sap:.1f} below effective baseline {a.effective_sap:.1f} (Δ{a.post_sap - a.effective_sap:.1f})"


@check("already-meets-goal-with-works", Severity.MEDIUM)
def _already_meets_goal_with_works(a: PropertyAudit) -> Optional[str]:
    """The property already meets/exceeds the scenario's goal band, yet the plan
    spends money on measures — nothing should be recommended."""
    goal, base = _band_rank(a.scenario_goal_band), _band_rank(a.effective_band)
    if goal is None or base is None or base > goal:
        return None
    if (a.cost_of_works or 0.0) <= 0.0:
        return None
    return f"already {a.effective_band} >= goal {a.scenario_goal_band} but cost_of_works £{a.cost_of_works:.0f}"


@check("post-band-score-mismatch", Severity.MEDIUM)
def _post_band_score_mismatch(a: PropertyAudit) -> Optional[str]:
    """The persisted post band disagrees with the band the post SAP implies — a
    rounding/derivation bug between score and rating."""
    implied = _band_of(a.post_sap)
    if implied is None or a.post_band is None or implied == a.post_band:
        return None
    return f"post_epc_rating {a.post_band} but post_sap_points {a.post_sap:.1f} implies {implied}"


@check("zero-works-post-differs", Severity.MEDIUM)
def _zero_works_post_differs(a: PropertyAudit) -> Optional[str]:
    """A no-op plan (£0 of works) whose post SAP differs from the baseline — the
    baseline and the plan's starting point disagree (stale or inconsistent)."""
    if a.effective_sap is None or a.post_sap is None:
        return None
    if (a.cost_of_works or 0.0) > 0.0:
        return None
    if abs(a.post_sap - a.effective_sap) <= 0.5:
        return None
    return f"£0 works but post SAP {a.post_sap:.1f} != effective {a.effective_sap:.1f}"


@check("effective-lodged-divergence", Severity.LOW)
def _effective_lodged_divergence(a: PropertyAudit) -> Optional[str]:
    """The Effective baseline is far from the lodged accredited figure (≥15 SAP).
    Often legitimate (overrides / pre-SAP10 rebaseline), but worth a look — a big
    gap can also mean a bad override or a calculator divergence."""
    if a.effective_sap is None or a.lodged_sap is None:
        return None
    gap = a.effective_sap - a.lodged_sap
    if abs(gap) < 15:
        return None
    return f"effective {a.effective_sap:.0f} vs lodged {a.lodged_sap:.0f} (Δ{gap:+.0f}, reason={a.rebaseline_reason})"


@check("negative-bill-savings", Severity.LOW)
def _negative_bill_savings(a: PropertyAudit) -> Optional[str]:
    """The plan INCREASES the annual bill — can be legitimate on a fuel-switch
    (gas→ASHP), but a recommended plan that costs more to run is worth review."""
    if a.energy_bill_savings is None or a.energy_bill_savings >= 0:
        return None
    if (a.cost_of_works or 0.0) <= 0.0:
        return None
    return f"energy_bill_savings £{a.energy_bill_savings:.0f}/yr on £{a.cost_of_works:.0f} of works"


# ─────────────────────── runner ───────────────────────

_QUERY = text(
    """
    SELECT p.id, p.uprn, p.portfolio_id,
           pl.scenario_id, s.goal_value AS goal_band,
           pbp.lodged_sap_score, pbp.lodged_epc_band,
           pbp.effective_sap_score, pbp.effective_epc_band, pbp.rebaseline_reason,
           pl.post_sap_points, pl.post_epc_rating, pl.cost_of_works,
           pl.energy_bill_savings, pl.energy_consumption_savings
    FROM property p
    LEFT JOIN property_baseline_performance pbp ON pbp.property_id = p.id
    LEFT JOIN plan pl ON pl.property_id = p.id AND pl.is_default = TRUE
    LEFT JOIN scenario s ON s.id = pl.scenario_id
    WHERE (:portfolio_id IS NULL OR p.portfolio_id = :portfolio_id)
      AND (:property_id IS NULL OR p.id = :property_id)
    ORDER BY p.id
    """
)


def _load(portfolio_id: Optional[int], property_id: Optional[int]) -> list[PropertyAudit]:
    engine = build_engine()
    out: list[PropertyAudit] = []
    with engine.connect() as conn:
        for r in conn.execute(
            _QUERY, {"portfolio_id": portfolio_id, "property_id": property_id}
        ):
            m = r._mapping
            out.append(
                PropertyAudit(
                    property_id=m["id"],
                    uprn=m["uprn"],
                    portfolio_id=m["portfolio_id"],
                    scenario_id=m["scenario_id"],
                    scenario_goal_band=m["goal_band"],
                    lodged_sap=m["lodged_sap_score"],
                    lodged_band=m["lodged_epc_band"],
                    effective_sap=m["effective_sap_score"],
                    effective_band=m["effective_epc_band"],
                    rebaseline_reason=m["rebaseline_reason"],
                    post_sap=m["post_sap_points"],
                    post_band=m["post_epc_rating"],
                    cost_of_works=m["cost_of_works"],
                    energy_bill_savings=m["energy_bill_savings"],
                    energy_consumption_savings=m["energy_consumption_savings"],
                )
            )
    return out


def run(audits: list[PropertyAudit], min_severity: Severity) -> list[Anomaly]:
    found: list[Anomaly] = []
    for a in audits:
        for name, severity, fn in _REGISTRY:
            if severity < min_severity:
                continue
            detail = fn(a)
            if detail is not None:
                found.append(Anomaly(a.property_id, a.uprn, name, severity, detail))
    found.sort(key=lambda x: (-x.severity, x.check, x.property_id))
    return found


def _write_reports(anomalies: list[Anomaly], scanned: int) -> None:
    with open("modelling_audit.csv", "w", newline="") as f:
        w = csv.writer(f)
        w.writerow(["property_id", "uprn", "severity", "check", "detail"])
        for a in anomalies:
            w.writerow([a.property_id, a.uprn, a.severity.name, a.check, a.detail])

    by_check: dict[str, list[Anomaly]] = {}
    for a in anomalies:
        by_check.setdefault(a.check, []).append(a)
    lines = [
        "# Modelling anomaly audit",
        "",
        f"Scanned **{scanned}** properties · flagged **{len(anomalies)}** anomalies "
        f"across **{len(by_check)}** checks.",
        "",
    ]
    for name in sorted(by_check, key=lambda n: (-by_check[n][0].severity, n)):
        rows = by_check[name]
        lines.append(f"## {name} ({rows[0].severity.name}) — {len(rows)}")
        lines.append("")
        for a in rows[:50]:
            lines.append(f"- property **{a.property_id}** (uprn {a.uprn}): {a.detail}")
        if len(rows) > 50:
            lines.append(f"- … and {len(rows) - 50} more (see CSV)")
        lines.append("")
    with open("modelling_audit.md", "w") as f:
        f.write("\n".join(lines))


def main() -> None:
    load_env()
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--portfolio", type=int, default=None, help="portfolio_id to scan")
    parser.add_argument("--property", type=int, default=None, help="a single property_id")
    parser.add_argument(
        "--severity",
        choices=[s.name.lower() for s in Severity],
        default="low",
        help="minimum severity to report (default: low — all)",
    )
    args = parser.parse_args()
    min_severity = Severity[args.severity.upper()]

    audits = _load(args.portfolio, args.property)
    anomalies = run(audits, min_severity)
    _write_reports(anomalies, len(audits))

    print(f"scanned {len(audits)} properties · {len(anomalies)} anomalies "
          f"(>= {min_severity.name})")
    counts: dict[str, int] = {}
    for a in anomalies:
        counts[a.check] = counts.get(a.check, 0) + 1
    for name, severity, _ in _REGISTRY:
        if severity >= min_severity:
            print(f"  [{severity.name:>6}] {name}: {counts.get(name, 0)}")
    print("wrote modelling_audit.md / modelling_audit.csv")


if __name__ == "__main__":
    main()