mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
6c47951d60
Electric-tariff cluster investigation. The 112-cert electric-main cohort is mixed-sign within every category/meter - no systematic tariff bias. Root cause of the biggest non-xfail divergences (10012334488 +13.2, 10091578598 +7.81): SAP code 195 = 'Electric water storage boiler' (Table 4a p.170), which SAP 10.2 §12 Rule 2 correctly bills mostly at the 7-hour off-peak rate. Both are pure cost gaps (PE/CO2 match lodged -> demand right); the lodged software over-billed the storage boilers at peak rate. The engine is SPEC-FAITHFUL - matching lodged would tune against §12/Table-4a. Pinned to the spec-faithful engine values. Adds scripts/corpus_1000/elec_cohort.py (electric-main cohort analyzer). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
61 lines
2.8 KiB
Python
61 lines
2.8 KiB
Python
"""Electric-main cohort analysis: for every corpus cert whose main heating is
|
|
electric, dump heating SAP code / tariff (meter_type) / lodged vs engine SAP /
|
|
cost-vs-demand signature. Used to steer a spec-grounded Table-12a tariff fix
|
|
without regressing in-band certs."""
|
|
from __future__ import annotations
|
|
import json
|
|
from pathlib import Path
|
|
from datatypes.epc.domain.mapper import EpcPropertyDataMapper
|
|
from domain.sap10_calculator.calculator import calculate_sap_from_inputs
|
|
from domain.sap10_calculator.rdsap.cert_to_inputs import (
|
|
SAP_10_2_SPEC_PRICES, cert_to_demand_inputs, cert_to_inputs, _is_electric_main,
|
|
_first_main_heating,
|
|
)
|
|
|
|
CORPUS = Path("backend/epc_api/json_samples/RdSAP-Schema-21.0.1/corpus.jsonl")
|
|
|
|
def hcode(doc):
|
|
sh = doc.get("sap_heating") or {}
|
|
ds = sh.get("main_heating_details") or [{}]
|
|
return ds[0].get("main_heating_category"), ds[0].get("main_fuel_type")
|
|
|
|
rows = []
|
|
for line in CORPUS.read_text().splitlines():
|
|
doc = json.loads(line)
|
|
try:
|
|
epc = EpcPropertyDataMapper.from_api_response(doc)
|
|
except Exception:
|
|
continue
|
|
main = _first_main_heating(epc)
|
|
if not _is_electric_main(main):
|
|
continue
|
|
try:
|
|
r = calculate_sap_from_inputs(cert_to_inputs(epc, prices=SAP_10_2_SPEC_PRICES))
|
|
d = calculate_sap_from_inputs(cert_to_demand_inputs(epc, prices=SAP_10_2_SPEC_PRICES))
|
|
except Exception as e:
|
|
continue
|
|
L = doc.get("energy_rating_current"); pe_l = doc.get("energy_consumption_current")
|
|
if L is None: continue
|
|
dsap = r.sap_score_continuous - L
|
|
dpe = (d.pe_per_m2 - pe_l) if (pe_l is not None and hasattr(d,'pe_per_m2')) else None
|
|
cat, fuel = hcode(doc)
|
|
mt = (doc.get("sap_energy_source") or {}).get("meter_type")
|
|
sh = doc.get("sap_heating") or {}
|
|
code = (sh.get("main_heating_details") or [{}])[0].get("main_heating_code") \
|
|
or (sh.get("main_heating_details") or [{}])[0].get("main_heating_index_number")
|
|
rows.append((doc.get("uprn"), code, cat, mt, sh.get("water_heating_code"),
|
|
L, round(r.sap_score_continuous,1), round(dsap,2)))
|
|
|
|
rows.sort(key=lambda x:-abs(x[7]))
|
|
print(f"electric-main certs: {len(rows)}")
|
|
print(f"{'uprn':14s} {'hcode':>6} {'cat':>4} {'mtr':>4} {'whc':>5} {'L':>3} {'E':>6} {'dSAP':>6}")
|
|
for r in rows:
|
|
print(f"{str(r[0]):14s} {str(r[1]):>6} {str(r[2]):>4} {str(r[3]):>4} {str(r[4]):>5} {r[5]:3d} {r[6]:6.1f} {r[7]:+6.2f}")
|
|
# summary by (hcode, meter)
|
|
from collections import defaultdict
|
|
agg = defaultdict(lambda:[0,0.0,0,0])
|
|
for r in rows:
|
|
k=(r[1], r[3]); a=agg[k]; a[0]+=1; a[1]+=r[7]; a[2]+=(abs(r[7])>0.5); a[3]+=(abs(r[7])<=0.5)
|
|
print("\n=== by (hcode, meter): n, meanΔ, #div>0.5, #inband ===")
|
|
for k,a in sorted(agg.items(), key=lambda x:-abs(x[1][1])):
|
|
print(f" code {str(k[0]):>6} meter {str(k[1]):>4}: n={a[0]:3d} meanΔ={a[1]/a[0]:+6.2f} div={a[2]:3d} inband={a[3]:3d}")
|