mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
2d6b078bd8
Add Phase 6 (self-improve) to audit-ara-portfolio: when a run confirms a novel systematic problem, codify it as a check — gated on systematic (>=5 props, root-caused), not-already-covered, and /grill-me-pressure-tested. Each check records provenance (motivating cause + example properties) so the registry stays sharp and compounds every run. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
404 lines
16 KiB
Python
404 lines
16 KiB
Python
"""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 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]
|
||
# 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("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("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,
|
||
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
|
||
"""
|
||
)
|
||
|
||
|
||
_ROLLUP_QUERY = text(
|
||
"""
|
||
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
|
||
"""
|
||
)
|
||
|
||
|
||
def _load(
|
||
portfolio_id: Optional[int],
|
||
property_id: Optional[int],
|
||
scenario_id: Optional[int],
|
||
) -> list[PropertyAudit]:
|
||
engine = build_engine()
|
||
out: list[PropertyAudit] = []
|
||
params = {
|
||
"portfolio_id": portfolio_id,
|
||
"property_id": property_id,
|
||
"scenario_id": scenario_id,
|
||
}
|
||
with engine.connect() as conn:
|
||
rollups: dict[int, tuple[Optional[float], Optional[float], int]] = {
|
||
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"],
|
||
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)",
|
||
)
|
||
args = parser.parse_args()
|
||
min_severity = Severity[args.severity.upper()]
|
||
|
||
audits = _load(args.portfolio, args.property, args.scenario)
|
||
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()
|