Model/scripts/audit/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.

This registry is meant to **compound**: each audit that confirms a new
systematic problem should leave behind a check (see the `audit-ara-portfolio`
skill's self-improve phase). So every check's docstring records its
**provenance** — the motivating cause and example properties — so a future reader
can re-verify it and judge whether it still earns its place. A threshold should
be justified against the real distribution, not guessed.

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 Connection, 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]
    scenario_budget: Optional[float]  # None == unlimited budget
    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]
    # Recommendation-level rollups for the default plan.
    solar_sap_points: Optional[float]  # max SAP a single solar_pv measure earns
    solar_bill_savings: Optional[float]  # the solar_pv measure's £/yr bill saving
    n_measures: int


@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("plan-stops-short-of-goal", Severity.HIGH)
def _plan_stops_short_of_goal(a: PropertyAudit) -> Optional[str]:
    """The default plan ends BELOW the scenario's goal band on a scenario that is
    NOT budget-capped — with an unlimited budget a plan should reach the goal
    unless it is physically impossible for that dwelling, so a shortfall is either
    a measure wrongly withheld (a bug) or a "sensible reason" the engine should be
    able to explain.

    This is the deterministic worklist for Khalim's "why didn't this get
    recommended <measure>" class (portfolio 814, pid 742121 — a dwelling left
    short of C with no HHRSH bundle). It only flags candidates; the deep-dive
    (``run_modelling_e2e`` + the candidates CSV) decides withheld-measure vs
    physically-unreachable. Budget-capped scenarios are excluded because there a
    shortfall is expected — the money simply ran out."""
    goal, post = _band_rank(a.scenario_goal_band), _band_rank(a.post_band)
    if goal is None or post is None or post <= goal:
        return None
    if a.scenario_budget is not None:
        # Budget-capped: falling short is expected once the money runs out.
        return None
    return (
        f"post {a.post_band} ({a.post_sap}) short of goal {a.scenario_goal_band} "
        f"on an unlimited-budget scenario (cost_of_works £{a.cost_of_works or 0:.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.

    Lodged scores below 13 are handed off to ``implausible-lodged-score`` — they
    indicate corrupt upstream EPC register data, not a modelling divergence."""
    if a.effective_sap is None or a.lodged_sap is None:
        return None
    if a.lodged_sap < 13:
        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("implausible-lodged-score", Severity.LOW)
def _implausible_lodged_score(a: PropertyAudit) -> Optional[str]:
    """Lodged SAP score is below 13 with a large effective-vs-lodged gap — the
    lodged figure is implausible upstream EPC register data, not a modelling bug.

    Floor of 13 is justified against the portfolio-796 distribution: every lodged
    score below 13 either matches a known Class C exemplar (SAP 1: pids 727752,
    727008, 731205) or shares the same bad-data fingerprint — implausibly low
    lodged paired with a reasonable effective in band D/E (pids 710449 SAP 5,
    722562 SAP 7, 725377 SAP 9, 727457 SAP 10, 723364 SAP 12). SAP 13 and above
    are left in scope as ambiguous.

    The gap guard (≥15) mirrors ``effective-lodged-divergence`` so the two checks
    partition the same population without overlap."""
    if a.lodged_sap is None or a.effective_sap is None:
        return None
    if a.lodged_sap >= 13:
        return None
    gap = a.effective_sap - a.lodged_sap
    if abs(gap) < 15:
        return None
    return (
        f"lodged SAP {a.lodged_sap:.0f} is implausible (< 13) — upstream EPC data,"
        f" not a modelling bug (effective {a.effective_sap:.0f}, Δ{gap:+.0f})"
    )


@check("impossible-sap-over-100", Severity.HIGH)
def _impossible_sap_over_100(a: PropertyAudit) -> Optional[str]:
    """A SAP score above 100 is impossible (SAP caps at 100) — a calculator /
    aggregation bug, or an oversized solar array pushing the score past the cap."""
    offenders = [
        f"{label} {value:.1f}"
        for label, value in (("post", a.post_sap), ("effective", a.effective_sap))
        if value is not None and value > 100.0
    ]
    if not offenders:
        return None
    return "SAP > 100: " + ", ".join(offenders)


@check("excessive-solar-sap", Severity.MEDIUM)
def _excessive_solar_sap(a: PropertyAudit) -> Optional[str]:
    """A single solar PV measure earns an implausibly large slice of SAP (> 25
    points; cohort avg ≈ 12.5). Usually an oversized array — Google footprint
    conflation borrowing a neighbour's / the whole building's roof (ADR-0038)."""
    if a.solar_sap_points is None or a.solar_sap_points <= 25.0:
        return None
    return f"solar PV alone earns {a.solar_sap_points:.1f} SAP points (likely oversized array)"


@check("unusually-high-post-sap", Severity.LOW)
def _unusually_high_post_sap(a: PropertyAudit) -> Optional[str]:
    """Post-works SAP at the very top of the scale (>= 95, near band A) — rare for
    a retrofit of existing stock; worth confirming it isn't an over-credit."""
    if a.post_sap is None or a.post_sap < 95.0 or a.post_sap > 100.0:
        return None
    return f"post SAP {a.post_sap:.1f} (near band A) — confirm not over-credited"


@check("low-solar-bill-savings", Severity.MEDIUM)
def _low_solar_bill_savings(a: PropertyAudit) -> Optional[str]:
    """A solar PV measure that barely cuts the bill (< £50/yr, or negative). Solar
    reliably saves on electricity, so a near-zero / negative figure points at a
    pricing bug — e.g. self-consumption or SEG export not credited (the Saltmead
    case: solar, D→C, but only ≈ −£62/yr)."""
    if a.solar_bill_savings is None or a.solar_bill_savings >= 50.0:
        return None
    return f"solar PV bill saving only £{a.solar_bill_savings:.0f}/yr — check self-consumption / SEG export"


@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, s.budget AS scenario_budget,
           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
      AND (:scenario_id IS NULL OR pl.scenario_id = :scenario_id)
    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
    """
)


# Bound every audit query so a bad plan aborts instead of saturating the shared
# DB. The `recommendation` table (~26m rows) has no index on `plan_id`, so any
# query reaching it via `plan_id` seq-scans the whole table — exactly what
# blocked the DB during the portfolio-796 audit. The timeout is the hard ceiling;
# `_guard_recommendation_scan` is the smarter gate that refuses such a plan up
# front rather than running it for the full timeout window.
_STATEMENT_TIMEOUT_MS = 120_000


# The rollup is kept as a string (not a pre-built ``text()``) so we can prepend
# ``EXPLAIN`` and inspect the plan before executing it. It reaches
# ``recommendation`` via the indexed ``property_id`` (never ``plan_id``); on a
# large portfolio the planner may still pick a seq-scan, which the EXPLAIN gate
# catches. Only the ``solar_pv`` rollups + ``n_measures`` come from here, feeding
# the two opt-in recommendation checks (excessive-solar-sap, low-solar-bill).
_ROLLUP_SQL = """
    SELECT r.property_id,
           MAX(r.sap_points) FILTER (WHERE r.type = 'solar_pv') AS solar_sap,
           MAX(r.energy_cost_savings) FILTER (WHERE r.type = 'solar_pv') AS solar_bill,
           COUNT(*) AS n_measures
    FROM recommendation r
    JOIN plan pl ON pl.id = r.plan_id AND pl.is_default = TRUE
      AND (:scenario_id IS NULL OR pl.scenario_id = :scenario_id)
    JOIN property p ON p.id = r.property_id
    WHERE (:portfolio_id IS NULL OR p.portfolio_id = :portfolio_id)
      AND (:property_id IS NULL OR p.id = :property_id)
    GROUP BY r.property_id
    """
_ROLLUP_QUERY = text(_ROLLUP_SQL)


class RecommendationScanError(RuntimeError):
    """Raised when a recommendation query would seq-scan the 26m-row table.

    The fix is operational, not a retry: add ``idx_recommendation_plan_id`` (see
    the audit skill's query-safety notes) or run without ``--with-recommendations``
    so the two solar checks are simply skipped.
    """


def _guard_recommendation_scan(
    conn: Connection, params: dict[str, Optional[int]]
) -> None:
    """Refuse to run the rollup if its plan seq-scans ``recommendation``.

    EXPLAIN (no ANALYZE) executes nothing, so this is free. A full seq-scan of
    the 26m-row table is the pattern that blocked the DB; aborting here is far
    cheaper than discovering it via the statement timeout.
    """
    plan = "\n".join(
        str(row[0]) for row in conn.execute(text("EXPLAIN " + _ROLLUP_SQL), params)
    )
    if "Seq Scan on recommendation" in plan:
        raise RecommendationScanError(
            "rollup would seq-scan the 26m-row `recommendation` table "
            f"(no index on plan_id; portfolio likely too large). Plan:\n{plan}"
        )


def _load(
    portfolio_id: Optional[int],
    property_id: Optional[int],
    scenario_id: Optional[int],
    with_recommendations: bool = False,
) -> list[PropertyAudit]:
    """Join one row per Property for the checks to run over.

    ``with_recommendations`` is OFF by default: the ``recommendation`` rollup is
    the only query that touches the 26m-row table and is the one that blocked the
    DB, so it is opt-in and EXPLAIN-gated. With it off, the two solar checks see
    ``None`` and are inert — every other check is bounded by the portfolio.
    """
    engine = build_engine()
    out: list[PropertyAudit] = []
    params: dict[str, Optional[int]] = {
        "portfolio_id": portfolio_id,
        "property_id": property_id,
        "scenario_id": scenario_id,
    }
    with engine.connect() as conn:
        # Hard ceiling: a runaway plan aborts instead of hammering the shared DB.
        conn.execute(text(f"SET statement_timeout = {_STATEMENT_TIMEOUT_MS}"))
        rollups: dict[int, tuple[Optional[float], Optional[float], int]] = {}
        if with_recommendations:
            _guard_recommendation_scan(conn, params)
            rollups = {
                m["property_id"]: (m["solar_sap"], m["solar_bill"], m["n_measures"])
                for m in (
                    row._mapping for row in conn.execute(_ROLLUP_QUERY, params)
                )
            }
        for r in conn.execute(_QUERY, params):
            m = r._mapping
            solar_sap, solar_bill, n_measures = rollups.get(m["id"], (None, None, 0))
            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"],
                    scenario_budget=m["scenario_budget"],
                    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"],
                    solar_sap_points=solar_sap,
                    solar_bill_savings=solar_bill,
                    n_measures=n_measures,
                )
            )
    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(
        "--scenario", type=int, default=None, help="restrict to one scenario_id"
    )
    parser.add_argument(
        "--severity",
        choices=[s.name.lower() for s in Severity],
        default="low",
        help="minimum severity to report (default: low — all)",
    )
    parser.add_argument(
        "--with-recommendations",
        action="store_true",
        help="run the recommendation rollup (the solar checks). OFF by default: "
        "it scans the 26m-row `recommendation` table and needs "
        "idx_recommendation_plan_id to be bounded — EXPLAIN-gated when on.",
    )
    args = parser.parse_args()
    min_severity = Severity[args.severity.upper()]

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

    if not args.with_recommendations:
        print(
            "NOTE: recommendation rollup skipped (--with-recommendations off) — "
            "the solar checks (excessive-solar-sap, low-solar-bill-savings) are inert."
        )
    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()