#!/usr/bin/env python3 """ Estimate how Warner (Democratic) and Mizusawa (Republican) voters split on Virginia's three 2026 constitutional amendments, from precinct results. Reads data/ (see bin/fetch), writes out/analysis.json, prints a summary. """ import csv import collections import json import random import re import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent DATA = ROOT / "data" OUT = ROOT / "out" SENATE = "Member, United States Senate" AMENDMENTS = {"q1": "abortion", "q2": "marriage", "q3": "voting rights"} MIN_SENATE_VOTES = 100 HOME_LOCALITY = "ARLINGTON COUNTY" COMPLETE_SHARE = 0.50 INCOMPLETE_SHARE = 0.40 BOOTSTRAP_ROUNDS = 400 BLACK_SHARE_GROUPS = [("under 10%", 0, .10), ("10-25%", .10, .25), ("25-50%", .25, .50), ("over 50%", .50, 1.01)] INCOME_GROUPS = [("under $60k", 0, 60), ("$60-80k", 60, 80), ("$80-110k", 80, 110), ("over $110k", 110, 1e9)] INCOME_SPLIT_GROUPS = [("median income under $80k", 0, 80), ("median income $80k or more", 80, 1e9)] def votes(s): return int(s or 0) def read_csv(path): return list(csv.DictReader(open(path, encoding="latin-1"))) def load_precincts(): precincts = {} for r in read_csv(DATA / "results.csv"): key = (r["LocalityName"], r["PrecinctName"]) p = precincts.setdefault(key, dict( locality=r["LocalityName"], name=r["PrecinctName"], county=r["LocalityCode"].strip().zfill(3), id=r["PrecinctId"].strip().zfill(4), sen=0, w=0, **{q + s: 0 for q in AMENDMENTS for s in "yn"})) n = votes(r["TOTAL_VOTES"]) if r["OfficeTitle"] == SENATE: p["sen"] += n if "Warner" in r["CandidateName"]: p["w"] += n elif r["OfficeTitle"].startswith("Proposed Constitutional Amendment Question"): q = "q" + r["OfficeTitle"][-1] p[q + ("y" if r["CandidateName"] == "Yes" else "n")] += n return list(precincts.values()) def yes_share(p, q): total = p[q + "y"] + p[q + "n"] return p[q + "y"] / total if total else None def two_group_fit(precincts, q): """Vote-weighted Goodman regression: yes = r*(1-W) + d*W. Returns (R yes, D no).""" sw = swx = swy = swxx = swxy = 0.0 for p in precincts: y = yes_share(p, q) if y is None: continue x, wt = p["w"] / p["sen"], p["sen"] sw += wt; swx += wt * x; swy += wt * y; swxx += wt * x * x; swxy += wt * x * y slope = (sw * swxy - swx * swy) / (sw * swxx - swx * swx) intercept = (swy - slope * swx) / sw return intercept, 1 - (intercept + slope) def estimates(precincts): return {q: dict(zip(("r_yes", "d_no"), two_group_fit(precincts, q))) for q in AMENDMENTS} def bootstrap_ranges(precincts): by_locality = collections.defaultdict(list) for p in precincts: by_locality[p["locality"]].append(p) groups = list(by_locality.values()) rng = random.Random(2026) draws = collections.defaultdict(list) for _ in range(BOOTSTRAP_ROUNDS): sample = [p for g in rng.choices(groups, k=len(groups)) for p in g] for q, e in estimates(sample).items(): for k, v in e.items(): draws[(q, k)].append(v) ranges = {} for (q, k), v in draws.items(): v.sort() ranges.setdefault(q, {})[k] = (v[int(.025 * len(v))], v[int(.975 * len(v)) - 1]) return ranges def totals(precincts): t = collections.Counter() for p in precincts: for k in ["sen", "w"] + [q + s for q in AMENDMENTS for s in "yn"]: t[k] += p[k] return dict(votes=t["sen"], warner=t["w"] / t["sen"], **{q: t[q + "y"] / (t[q + "y"] + t[q + "n"]) for q in AMENDMENTS}) def locality_completeness(precincts): registered = collections.Counter() for r in read_csv(DATA / "turnout.csv"): registered[r["locality"]] += votes(r["ActiveRegisteredVoters"]) early = collections.Counter() for r in read_csv(DATA / "absentee.csv"): if r["ReceiptType"]: early[r["LocalityName"]] += votes(r["TheCount"]) senate = collections.Counter() for p in precincts: senate[p["locality"]] += p["sen"] return {loc: dict(senate=senate[loc], registered=registered[loc], early_returned=early[loc], share=senate[loc] / registered[loc] if registered[loc] else None) for loc in senate} def simplify_name(name): name = re.sub(r"^\s*\d+\s*-\s*", "", name.lower()) name = re.sub(r"voting district|precinct", "", name) return re.sub(r"[^a-z0-9]", "", name) def census_race_by_precinct(precincts): rows = json.load(open(DATA / "census-vtd.json")) header = rows[0] by_name = collections.defaultdict(list) for row in rows[1:]: d = dict(zip(header, row)) by_name[(d["county"], simplify_name(d["NAME"].split(",")[0]))].append(d) matched, stats = [], collections.Counter() for p in precincts: candidates = by_name.get((p["county"], simplify_name(p["name"])), []) if len(candidates) != 1: stats["ambiguous" if candidates else "unmatched"] += 1 continue d = candidates[0] adults = int(d["P4_001N"]) if not adults or not 0.1 < p["sen"] / adults < 1.0: stats["implausible size"] += 1 continue stats["matched"] += 1 matched.append(dict(p, black=int(d["P4_006N"]) / adults)) return matched, dict(stats) def county_income(): rows = json.load(open(DATA / "census-county.json")) header = rows[0] return {d["county"]: float(d["B19013_001E"]) / 1000 for d in (dict(zip(header, r)) for r in rows[1:])} def grouped(precincts, groups, value, with_ranges=False): out = [] for label, lo, hi in groups: members = [p for p in precincts if lo <= value(p) < hi] g = dict(label=label, precincts=len(members), localities=len({p["locality"] for p in members}), estimates=estimates(members)) if with_ranges: g["ranges"] = bootstrap_ranges(members) out.append(g) return out def one_in(rate): return "~0" if rate < 0.01 else f"1 in {round(1 / rate)}" def report(a): print(f"Results publish: {a['publish']}") c = a["completeness"] print(f"Localities with early votes apparently loaded: {c['complete']} complete, {c['incomplete']} Election-Day-only, {c['unclear']} unclear") for label, t in (("All precincts", a["totals"]["all"]), ("Complete localities", a["totals"]["complete"]), (HOME_LOCALITY.title(), a["totals"]["home"])): print(f"{label}: {t['votes']:,} Senate votes, Warner {100*t['warner']:.1f}%, " + ", ".join(f"{AMENDMENTS[q]} {100*t[q]:.1f}%" for q in AMENDMENTS)) print("\nCrossovers (all precincts, 95% range from resampling counties):") for q, name in AMENDMENTS.items(): e, r = a["estimates"]["all"][q], a["ranges"][q] print(f" {name:14s} Republicans yes {one_in(e['r_yes']):>9s} ({100*e['r_yes']:.1f}%, {100*r['r_yes'][0]:.1f}-{100*r['r_yes'][1]:.1f})" f" Democrats no {one_in(e['d_no']):>9s} ({100*e['d_no']:.1f}%, {100*r['d_no'][0]:.1f}-{100*r['d_no'][1]:.1f})") print("\nSame, complete localities only:") for q, name in AMENDMENTS.items(): e = a["estimates"]["complete"][q] print(f" {name:14s} Republicans yes {100*e['r_yes']:.1f}% Democrats no {100*e['d_no']:.1f}%") print(f"\nDemocrats voting no, by Black share of adults in precinct (match: {a['race_match']}):") for g in a["by_black_share"]: print(f" {g['label']:10s} n={g['precincts']:4d} " + " ".join(f"{AMENDMENTS[q]} {100*g['estimates'][q]['d_no']:.1f}%" for q in AMENDMENTS)) print("\nRepublicans voting yes, by county median household income:") for g in a["by_income"]: print(f" {g['label']:11s} n={g['precincts']:4d} " + " ".join(f"{AMENDMENTS[q]} {100*g['estimates'][q]['r_yes']:.1f}%" for q in AMENDMENTS)) print("\nRepublicans voting yes, by locality median household income:") for g in a["by_income_split"]: print(f" {g['label']:33s} {g['localities']:3d} localities " + " ".join(f"{AMENDMENTS[q]} {100*g['estimates'][q]['r_yes']:.1f}% " f"({100*g['ranges'][q]['r_yes'][0]:.0f}-{100*g['ranges'][q]['r_yes'][1]:.0f})" for q in AMENDMENTS)) def main(): all_precincts = load_precincts() precincts = [p for p in all_precincts if p["sen"] >= MIN_SENATE_VOTES] completeness = locality_completeness(all_precincts) share = {loc: c["share"] or 0 for loc, c in completeness.items()} complete = [p for p in precincts if share[p["locality"]] >= COMPLETE_SHARE] race, race_stats = census_race_by_precinct(precincts) income = county_income() a = dict( publish=(DATA / "publish.txt").read_text().strip(), completeness=dict( complete=sum(s >= COMPLETE_SHARE for s in share.values()), incomplete=sum(s < INCOMPLETE_SHARE for s in share.values()), unclear=sum(INCOMPLETE_SHARE <= s < COMPLETE_SHARE for s in share.values()), localities=completeness), totals=dict(all=totals(precincts), complete=totals(complete), home=totals([p for p in all_precincts if p["locality"] == HOME_LOCALITY])), estimates=dict(all=estimates(precincts), complete=estimates(complete)), ranges=bootstrap_ranges(precincts), race_match=race_stats, by_black_share=grouped(race, BLACK_SHARE_GROUPS, lambda p: p["black"]), by_income=grouped(precincts, INCOME_GROUPS, lambda p: income[p["county"]]), by_income_split=grouped(precincts, INCOME_SPLIT_GROUPS, lambda p: income[p["county"]], with_ranges=True), ) OUT.mkdir(exist_ok=True) json.dump(a, open(OUT / "analysis.json", "w"), indent=1) report(a) if __name__ == "__main__": sys.exit(main())