From 39fd1c9e42b63c81d7e40ef132624b5ec1e8432e Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Fri, 26 Jun 2026 19:38:28 +0000 Subject: [PATCH] =?UTF-8?q?feat(audit):=20pluggable=20modelling-anomaly=20?= =?UTF-8?q?audit=20over=20the=20DB=20=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A check-registry that reads property/baseline/plan/scenario and flags odd results (plan-below-baseline, already-meets-goal-with-works, band/score mismatch, zero-works-post-differs, effective-lodged divergence, negative bill savings). Writes modelling_audit.md/.csv. Adding a check = one decorated function. Co-Authored-By: Claude Opus 4.8 (1M context) --- scripts/audit_modelling_anomalies.py | 311 +++++++++++++++++++++++++++ 1 file changed, 311 insertions(+) create mode 100644 scripts/audit_modelling_anomalies.py diff --git a/scripts/audit_modelling_anomalies.py b/scripts/audit_modelling_anomalies.py new file mode 100644 index 00000000..10f6bef1 --- /dev/null +++ b/scripts/audit_modelling_anomalies.py @@ -0,0 +1,311 @@ +"""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()