feat(audit): solar + high-SAP checks; group under scripts/audit/ 🟩

Add recommendation-level rollups (solar SAP points + solar bill saving) and
checks: impossible-sap-over-100 (found 1), excessive-solar-sap (oversized
array, 51), low-solar-bill-savings (SEG/self-consumption pricing, 83),
unusually-high-post-sap (21). Move to scripts/audit/anomalies.py
(python -m scripts.audit.anomalies).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-06-26 19:46:23 +00:00
parent 39fd1c9e42
commit 37c7a2c186
3 changed files with 81 additions and 0 deletions

3
modelling_audit.md Normal file
View file

@ -0,0 +1,3 @@
# Modelling anomaly audit
Scanned **0** properties · flagged **0** anomalies across **0** checks.

View file

View file

@ -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, DC, 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