Model/scripts/hyde/case_a_population_sweep.py

"""Local population sanity sweep for the full-SAP baseline-downgrade fix (ADR-0037).

For a sample of Case A properties (portfolio 796, full-SAP path: epc_property
assessment_type+sap_version NULL), re-map the real cert with the NEW mapper, run
the SAP-10.2 calculator, and compute the post-fix Effective SAP/band. Compare to
the lodged figures to (a) confirm the fix flips Effective off the stale lodged
value and (b) flag survivors — drops too large to be a 2012→10.2 methodology
shift (>15 SAP or >=2 bands below lodged), candidate deeper-calc/cert bugs.

Read-only. No DB writes.
"""
from __future__ import annotations

import os
import random  # noqa: F401 — only used with a fixed seed below

from scripts.e2e_common import load_env, build_engine
from sqlalchemy import text

load_env()

from infrastructure.epc_client.epc_client_service import EpcClientService
from datatypes.epc.domain.mapper import EpcPropertyDataMapper
from domain.sap10_calculator.calculator import Sap10Calculator
from datatypes.epc.domain.epc import Epc

_BAND_IDX = {b: i for i, b in enumerate(["G", "F", "E", "D", "C", "B", "A"])}


def _band_letter(epc_band: object) -> str:
    s = str(epc_band)
    return s.split(".")[-1] if "." in s else s


def main() -> None:
    eng = build_engine()
    with eng.connect() as c:
        # 30 worst old-drops + 30 spread across the cohort (deterministic order).
        worst = c.execute(text("""
            SELECT p.uprn, pbp.lodged_sap_score, pbp.lodged_epc_band,
                   pbp.effective_sap_score AS old_eff, pl.post_sap_points AS old_post
            FROM property p
            JOIN epc_property ep ON ep.uprn=p.uprn
            JOIN property_baseline_performance pbp ON pbp.property_id=p.id
            JOIN plan pl ON pl.property_id=p.id
            WHERE p.portfolio_id=796 AND ep.assessment_type IS NULL
              AND ep.sap_version IS NULL
            ORDER BY (pbp.effective_sap_score - pl.post_sap_points) DESC
            LIMIT 30
        """)).fetchall()
        spread = c.execute(text("""
            SELECT p.uprn, pbp.lodged_sap_score, pbp.lodged_epc_band,
                   pbp.effective_sap_score AS old_eff, pl.post_sap_points AS old_post
            FROM property p
            JOIN epc_property ep ON ep.uprn=p.uprn
            JOIN property_baseline_performance pbp ON pbp.property_id=p.id
            JOIN plan pl ON pl.property_id=p.id
            WHERE p.portfolio_id=796 AND ep.assessment_type IS NULL
              AND ep.sap_version IS NULL AND (p.uprn % 7) = 0
            ORDER BY p.uprn LIMIT 30
        """)).fetchall()

    rows = {r._mapping["uprn"]: dict(r._mapping) for r in worst}
    for r in spread:
        rows.setdefault(r._mapping["uprn"], dict(r._mapping))

    svc = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"])
    calc = Sap10Calculator()

    n = 0
    flipped = 0
    calc_errors = 0
    survivors = []
    drops = []
    for uprn, row in rows.items():
        n += 1
        try:
            results = svc._search(uprn=int(uprn))
            if not results:
                print(f"  uprn={uprn} NO_CERT")
                continue
            latest = max(results, key=lambda x: x.registration_date)
            raw = svc._fetch_certificate(latest.certificate_number)
            epc = EpcPropertyDataMapper.from_api_response(raw)
            new_sap = calc.calculate(epc).sap_score
        except Exception as e:  # noqa: BLE001 — sweep tolerates per-cert failure
            calc_errors += 1
            print(f"  uprn={uprn} CALC_ERROR {type(e).__name__}: {str(e)[:90]}")
            continue
        new_band = _band_letter(Epc.from_sap_score(new_sap))
        lodged = row["lodged_sap_score"]
        lodged_band = row["lodged_epc_band"]
        drop = lodged - new_sap
        drops.append(drop)
        if epc.sap_version is not None and epc.sap_version < 10.2:
            flipped += 1
        band_drop = _BAND_IDX.get(lodged_band, 0) - _BAND_IDX.get(new_band, 0)
        survivor = drop > 15 or band_drop >= 2
        tag = "  *** SURVIVOR" if survivor else ""
        if survivor:
            survivors.append((uprn, lodged, lodged_band, new_sap, new_band, drop))
        print(
            f"  uprn={uprn} lodged={lodged}/{lodged_band} "
            f"new_eff={new_sap}/{new_band} drop={drop} "
            f"(old_eff={row['old_eff']} old_post={round(float(row['old_post']),1)}){tag}"
        )

    computed = len(drops)
    print("\n==== SUMMARY ====")
    print(f"sampled={n}  computed={computed}  calc_errors={calc_errors}")
    print(f"rebaseline flipped (sap_version<10.2 now set) = {flipped}/{computed}")
    if drops:
        drops_sorted = sorted(drops)
        print(
            f"lodged-new_eff drop: min={min(drops)} median="
            f"{drops_sorted[len(drops_sorted)//2]} max={max(drops)} "
            f"mean={round(sum(drops)/len(drops),1)}"
        )
    print(f"survivors (>15 SAP or >=2 bands below lodged) = {len(survivors)}")
    for s in survivors:
        print(f"   SURVIVOR uprn={s[0]} lodged={s[1]}/{s[2]} new={s[3]}/{s[4]} drop={s[5]}")


if __name__ == "__main__":
    main()