"""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 " 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()