|
"""Aggregate-only measurement of UC caseload by element/family type, council tax by |
|
nation, and owned_land by nation on an enhanced FRS artifact. Prints NO individual |
|
records (UKDS data-protection rule). Usage: python measure_uc_ct.py <h5> <tag> [year]""" |
|
import sys, json, hashlib, subprocess, numpy as np, pandas as pd |
|
from policyengine_uk import Microsimulation |
|
import policyengine_uk, policyengine_core |
|
|
|
path, tag = sys.argv[1], sys.argv[2] |
|
year = int(sys.argv[3]) if len(sys.argv) > 3 else 2025 |
|
sha = hashlib.sha256(open(path, "rb").read()).hexdigest() |
|
sim = Microsimulation(dataset=path) |
|
|
|
def arr(v, y=year, map_to=None): |
|
x = sim.calculate(v, y, map_to=map_to) if map_to else sim.calculate(v, y) |
|
vals = x.values if hasattr(x, "values") else x |
|
if hasattr(vals, "decode_to_str"): |
|
vals = vals.decode_to_str() |
|
return np.asarray(vals) |
|
|
|
out = {"artifact": path.split("/")[-1], "sha256": sha, "tag": tag, "year": year, |
|
"policyengine_uk_version": __import__("importlib.metadata").metadata.version("policyengine-uk"), |
|
"policyengine_core_version": __import__("importlib.metadata").metadata.version("policyengine-core"), |
|
"model_commit": subprocess.run(["git", "-C", "/Users/maxghenis/PolicyEngine/_worktrees/ukdef-uk-model", "rev-parse", "--short", "HEAD"], capture_output=True, text=True).stdout.strip()} |
|
|
|
# ---- benunit level ---- |
|
w = arr("benunit_weight").astype(float) |
|
uc = arr("universal_credit").astype(float) |
|
on_uc = uc > 0 |
|
out["uc_benunits_total_m"] = round(float(w[on_uc].sum()) / 1e6, 4) |
|
out["uc_spend_bn"] = round(float((w * uc).sum()) / 1e9, 3) |
|
out["benunits_unweighted_on_uc"] = int(on_uc.sum()) |
|
out["benunits_unweighted_total"] = int(len(w)) |
|
for v in ["would_claim_uc", "is_uc_entitled", "is_uc_eligible"]: |
|
try: |
|
b = arr(v).astype(bool) |
|
out[f"{v}_m"] = round(float(w[b].sum()) / 1e6, 4) |
|
out[f"{v}_unweighted"] = int(b.sum()) |
|
except Exception as e: |
|
out[f"{v}_error"] = str(e)[:120] |
|
try: |
|
rep = arr("universal_credit_reported").astype(float) > 0 |
|
out["uc_reported_benunits_m"] = round(float(w[rep].sum()) / 1e6, 4) |
|
out["uc_reported_and_on_uc_m"] = round(float(w[rep & on_uc].sum()) / 1e6, 4) |
|
out["uc_on_uc_not_reported_m"] = round(float(w[on_uc & ~rep].sum()) / 1e6, 4) |
|
out["uc_reported_unweighted"] = int(rep.sum()) |
|
except Exception as e: |
|
out["uc_reported_error"] = str(e)[:120] |
|
try: |
|
ent = arr("is_uc_entitled").astype(bool); wc = arr("would_claim_uc").astype(bool) |
|
out["entitled_not_claiming_m"] = round(float(w[ent & ~wc].sum()) / 1e6, 4) |
|
out["entitled_and_claiming_m"] = round(float(w[ent & wc].sum()) / 1e6, 4) |
|
out["entitled_claiming_but_uc_zero_m"] = round(float(w[ent & wc & ~on_uc].sum()) / 1e6, 4) |
|
out["max_support_entitled_m"] = round(float(w[ent].sum()) / 1e6, 4) |
|
except Exception as e: |
|
out["entitlement_error"] = str(e)[:120] |
|
|
|
elements = {"childcare": "uc_childcare_element", "housing": "uc_housing_costs_element", |
|
"lcwra": "uc_LCWRA_element", "carer": "uc_carer_element", "child": "uc_child_element", |
|
"disability_any": "uc_disability_elements"} |
|
out["elements"] = {} |
|
for k, v in elements.items(): |
|
try: |
|
e = arr(v).astype(float) |
|
out["elements"][k] = {"on_uc_with_element_m": round(float(w[on_uc & (e > 0)].sum()) / 1e6, 4), |
|
"any_with_element_m": round(float(w[e > 0].sum()) / 1e6, 4), |
|
"element_spend_bn_on_uc": round(float((w * e)[on_uc].sum()) / 1e9, 3), |
|
"unweighted_on_uc_with_element": int((on_uc & (e > 0)).sum())} |
|
except Exception as e: |
|
out["elements"][k] = {"error": str(e)[:160]} |
|
|
|
ft = arr("family_type"); nk = arr("benunit_count_children").astype(int) |
|
out["family_type_on_uc_m"] = {str(f): round(float(w[on_uc & (ft == f)].sum()) / 1e6, 4) for f in sorted(set(ft.tolist()))} |
|
nkb = np.minimum(nk, 5) |
|
out["children_on_uc_m"] = {("5_or_more" if n == 5 else str(n)): round(float(w[on_uc & (nkb == n)].sum()) / 1e6, 4) for n in range(6)} |
|
has_kids = nk > 0 |
|
out["with_children_on_uc_m"] = round(float(w[on_uc & has_kids].sum()) / 1e6, 4) |
|
out["no_children_on_uc_m"] = round(float(w[on_uc & ~has_kids].sum()) / 1e6, 4) |
|
|
|
# ---- GB-only (admin UC facts are Great Britain) ---- |
|
reg = arr("benunit_region"); gb = reg != "NORTHERN_IRELAND" |
|
out["gb"] = {"uc_benunits_total_m": round(float(w[on_uc & gb].sum())/1e6,4), "uc_spend_bn": round(float((w*uc)[on_uc & gb].sum())/1e9,3), "records_on_uc": int((on_uc & gb).sum()), |
|
"ni_on_uc_m": round(float(w[on_uc & ~gb].sum())/1e6,4)} |
|
out["gb"]["elements"] = {} |
|
for k, v in elements.items(): |
|
e = arr(v).astype(float) |
|
out["gb"]["elements"][k] = {"on_uc_with_element_m": round(float(w[on_uc & gb & (e > 0)].sum())/1e6,4), "records": int((on_uc & gb & (e>0)).sum())} |
|
out["gb"]["family_type_on_uc_m"] = {str(f): round(float(w[on_uc & gb & (ft == f)].sum())/1e6,4) for f in sorted(set(ft.tolist()))} |
|
out["gb"]["children_on_uc_m"] = {("5_or_more" if n == 5 else str(n)): round(float(w[on_uc & gb & (nkb == n)].sum())/1e6,4) for n in range(6)} |
|
out["family_type_all_benunits_m"] = {str(f): round(float(w[ft == f].sum()) / 1e6, 4) for f in sorted(set(ft.tolist()))} |
|
|
|
# ---- household level ---- |
|
hw = arr("household_weight").astype(float) |
|
try: |
|
nation = arr("country") |
|
except Exception: |
|
reg = arr("region"); nation = np.where(reg == "SCOTLAND", "SCOTLAND", np.where(reg == "WALES", "WALES", np.where(reg == "NORTHERN_IRELAND", "NORTHERN_IRELAND", "ENGLAND"))) |
|
out["households_m"] = round(float(hw.sum()) / 1e6, 4) |
|
ct = arr("council_tax").astype(float) |
|
try: |
|
ctb = arr("council_tax_benefit", map_to="household").astype(float) |
|
except Exception as e: |
|
ctb = np.zeros_like(ct); out["ctb_error"] = str(e)[:120] |
|
out["council_tax"] = {} |
|
for n in sorted(set(nation.tolist())): |
|
m = nation == n |
|
hh = float(hw[m].sum()) |
|
out["council_tax"][n] = {"households_m": round(hh / 1e6, 4), |
|
"gross_bn": round(float((hw * ct)[m].sum()) / 1e9, 3), |
|
"ctr_bn": round(float((hw * ctb)[m].sum()) / 1e9, 3), |
|
"net_bn": round(float((hw * (ct - ctb))[m].sum()) / 1e9, 3), |
|
"avg_gross_gbp": round(float((hw * ct)[m].sum() / hh), 0), |
|
"avg_net_gbp": round(float((hw * (ct - ctb))[m].sum() / hh), 0), |
|
"share_ct_zero": round(float(hw[m & (ct <= 0)].sum() / hh), 4)} |
|
try: |
|
band = arr("council_tax_band") |
|
out["council_tax_band_shares"] = {} |
|
for n in sorted(set(nation.tolist())): |
|
m = nation == n; hh = float(hw[m].sum()) |
|
out["council_tax_band_shares"][n] = {str(b): round(float(hw[m & (band == b)].sum() / hh), 4) for b in sorted(set(band[m].tolist()))} |
|
except Exception as e: |
|
out["band_error"] = str(e)[:120] |
|
try: |
|
ol = arr("owned_land").astype(float) |
|
out["owned_land"] = {} |
|
for n in sorted(set(nation.tolist())): |
|
m = nation == n; hh = float(hw[m].sum()) |
|
tot = float((hw * ol)[m].sum()) |
|
out["owned_land"][n] = {"total_bn": round(tot / 1e9, 3), "avg_per_hh_gbp": round(tot / hh, 0), |
|
"share_hh_positive": round(float(hw[m & (ol > 0)].sum() / hh), 4), |
|
"unweighted_positive": int((m & (ol > 0)).sum()), |
|
"top10_records_share_of_nation_total": round(float(np.sort((hw * ol)[m])[-10:].sum() / tot), 3) if tot > 0 else None} |
|
out["owned_land"]["UK_total_bn"] = round(float((hw * ol).sum()) / 1e9, 3) |
|
try: |
|
hlv = arr("household_land_value").astype(float) |
|
out["owned_land"]["UK_share_of_household_land_value"] = round(float((hw * ol).sum() / (hw * hlv).sum()), 4) |
|
for n in sorted(set(nation.tolist())): |
|
m = nation == n |
|
out["owned_land"][n]["share_of_nation_household_land_value"] = round(float((hw * ol)[m].sum() / (hw * hlv)[m].sum()), 4) |
|
except Exception as e: |
|
out["owned_land"]["hlv_error"] = str(e)[:120] |
|
except Exception as e: |
|
out["owned_land_error"] = str(e)[:120] |
|
|
|
json.dump(out, open(f"/Users/maxghenis/PolicyEngine/_uk-defects-452-448/measure/{tag}_{year}.json", "w"), indent=1) |
|
print(json.dumps({k: v for k, v in out.items() if not isinstance(v, dict)}, indent=1)) |
