"""exercise_2b.py Exercise II-B. The audit program for the delivered cooling center analysis. The delivered script, cooling_centers_analysis.py, stays exactly as the consulting firm shipped it. This program audits it: it measures the error the invalid median aggregation introduces, corrects it with the published tract medians, compares the two recommendations, and then searches the neighboring designs for the point at which the defect starts to change the answer. python exercise_2b.py """ from __future__ import annotations import sys from pathlib import Path import numpy as np sys.path.insert(0, str(Path(__file__).resolve().parent)) import common as C # noqa: E402 import cooling_centers_analysis as CC # noqa: E402 TOPN = 10 WEIGHTS = [1, 2, 3, 5, 8] SIZES = [5, 10, 15, 20] def joined(): bad = CC.build() m = C.load_acs() tr = m[(m.level == "tract") & (m.code.str.startswith(CC.COUNTY))] tr = tr[(tr["pop"] > 0) & (tr.hh > 0) & (tr.pov_univ > 0)] return bad.merge(tr[["code", "medinc"]], left_on="tract", right_on="code") def ranking(j, income_col, weight=1): s = (j.pct65.rank(pct=True) + j.pct_pov.rank(pct=True) + j.pct_nv.rank(pct=True) + weight * (-j[income_col]).rank(pct=True)) return j.assign(score=s).sort_values("score", ascending=False) def top(j, income_col, weight=1, k=TOPN): return set(ranking(j, income_col, weight).head(k)["tract"]) def main(): C.ensure_dirs() j = joined() C.banner("exercise II-B: auditing the delivered analysis", "ACS 2023 5-year, Hinds County tracts", "acs2023_mississippi_asi.csv", {"tracts": len(j)}) out = {"tracts": int(len(j))} print("\n[1] THE DEFECT, IN DOLLARS") err = (j["median_income"] - j["medinc"]).abs() out["error"] = {"mean_usd": int(round(err.mean())), "median_usd": int(round(err.median())), "max_usd": int(round(err.max())), "pct_over_5000": round(100 * float((err > 5000).mean()), 1)} print(f" mean absolute error ${err.mean():,.0f}") print(f" median absolute error ${err.median():,.0f}") print(f" largest single error ${err.max():,.0f}") print(f" tracts off by over $5,000 " f"{out['error']['pct_over_5000']}%") moved = int(((-j["median_income"]).rank() - (-j["medinc"]).rank()) .abs().gt(5).sum()) print(f" tracts moving more than five places in income rank " f"{moved} of {len(j)}") out["income_rank_moves_over_5"] = moved print("\n[2] THE DEFECT, IN THE RECOMMENDATION") a, b = top(j, "median_income"), top(j, "medinc") out["recommendation"] = {"overlap": len(a & b), "jaccard": round(C.jaccard(a, b), 3)} print(f" recommended ten, delivered against corrected: " f"{len(a & b)} of {TOPN} shared, Jaccard {C.jaccard(a, b):.3f}") rd = ranking(j, "median_income").reset_index(drop=True) rc = ranking(j, "medinc").reset_index(drop=True) pos = {t: i for i, t in enumerate(rd.tract)} shift = {t: pos[t] - i for i, t in enumerate(rc.tract)} gain = max(shift, key=shift.get) lose = min(shift, key=shift.get) print(f" largest gain after correction tract {gain}, up " f"{shift[gain]} places") print(f" largest loss after correction tract {lose}, down " f"{-shift[lose]} places") out["largest_gain"] = {"tract": gain, "places": int(shift[gain])} out["largest_loss"] = {"tract": lose, "places": int(-shift[lose])} print("\n[3] WHERE IT STARTS TO MATTER") print(f" {'income weight':>14}{'shared of 10':>14}{'jaccard':>9}") out["weight_sweep"] = [] for w in WEIGHTS: a, b = top(j, "median_income", w), top(j, "medinc", w) print(f" {w:>14}{len(a & b):>14}{C.jaccard(a, b):>9.3f}") out["weight_sweep"].append({"weight": w, "shared": len(a & b), "jaccard": round(C.jaccard(a, b), 3)}) print(f" {'places funded':>14}{'shared':>14}{'jaccard':>9}") out["size_sweep"] = [] for k in SIZES: a, b = top(j, "median_income", 1, k), top(j, "medinc", 1, k) print(f" {k:>14}{len(a & b):>14}{C.jaccard(a, b):>9.3f}") out["size_sweep"].append({"places": k, "shared": len(a & b), "jaccard": round(C.jaccard(a, b), 3)}) def _words(v): v = [str(x) for x in v] return (" and ".join(v) if len(v) < 3 else ", ".join(v[:-1]) + " and " + v[-1]) or "none" hit = [r["places"] for r in out["size_sweep"] if r["shared"] < r["places"]] quiet = [r["places"] for r in out["size_sweep"] if r["shared"] == r["places"]] print(f" At equal weights the defect changes the funded list at " f"{_words(hit)} places and leaves it unchanged at {_words(quiet)}.") try: import make_figures as MF except ImportError as exc: print(f"\n audit figure skipped: {exc.name} is not installed") else: MF.fig_before_after() tbl = j[["tract", "median_income", "medinc", "pct65", "pct_pov", "pct_nv"]].rename(columns={"median_income": "delivered_median", "medinc": "published_median"}) tbl["absolute_error"] = err.values tbl["delivered_rank"] = tbl.tract.map(pos).add(1) tbl["corrected_rank"] = tbl.tract.map( {t: i + 1 for i, t in enumerate(rc.tract)}) C.write_handoff("exercise_2b", tbl, { "correction": "published tract median replaces the mean of block " "group medians", "index": "four equally weighted percentile ranks, as delivered", "sweeps": {"income_weight": WEIGHTS, "places_funded": SIZES}, }, [ "The defect changes nothing at the delivered design. Is the analysis " "therefore acceptable, and would you sign off on it?", "The method is invalid regardless of its consequence here. What would " "you require the firm to change even though the list is the same?", "Which neighboring design would a county plausibly adopt next year, " "and would the defect bite there?", ]) C.save_checkvalues("exercise_2b", out) if __name__ == "__main__": main()