diff --git a/modelling_audit.md b/modelling_audit.md new file mode 100644 index 00000000..81c5595d --- /dev/null +++ b/modelling_audit.md @@ -0,0 +1,3 @@ +# Modelling anomaly audit + +Scanned **0** properties · flagged **0** anomalies across **0** checks. diff --git a/scripts/audit/__init__.py b/scripts/audit/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/scripts/audit_modelling_anomalies.py b/scripts/audit/anomalies.py similarity index 77% rename from scripts/audit_modelling_anomalies.py rename to scripts/audit/anomalies.py index 10f6bef1..420d1be9 100644 --- a/scripts/audit_modelling_anomalies.py +++ b/scripts/audit/anomalies.py @@ -77,6 +77,10 @@ class PropertyAudit: 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) @@ -173,6 +177,50 @@ def _effective_lodged_divergence(a: PropertyAudit) -> Optional[str]: 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 @@ -205,14 +253,41 @@ _QUERY = text( ) +_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 + 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]) -> list[PropertyAudit]: engine = build_engine() out: list[PropertyAudit] = [] 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, + {"portfolio_id": portfolio_id, "property_id": property_id}, + ) + ) + } for r in conn.execute( _QUERY, {"portfolio_id": portfolio_id, "property_id": property_id} ): m = r._mapping + solar_sap, solar_bill, n_measures = rollups.get(m["id"], (None, None, 0)) out.append( PropertyAudit( property_id=m["id"], @@ -230,6 +305,9 @@ def _load(portfolio_id: Optional[int], property_id: Optional[int]) -> list[Prope 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