congress-analyzer/analyze.py

#!/usr/bin/env python3
"""Pure analytical functions over the unified votes schema.

No I/O. Given a list of parsed vote dicts (see parse.py) and a target
member id + party, produce a metrics dict ready for rendering.

Classification rules — see DOCUMENTATION.md §6.
"""
from collections import Counter, defaultdict

# Lone-wolf defection threshold (max fellow same-party defectors).
# Smaller chamber => tighter threshold.
LONE_WOLF_THRESHOLD = {"house": 5, "senate": 3}

def _norm_vote(v):
    """Normalize Aye/No (procedural) to Yea/Nay."""
    if v in ("Yea", "Aye"): return "Yea"
    if v in ("Nay", "No"):  return "Nay"
    return v  # Present, Not Voting, None, ""

def _majority_position(party_totals):
    y, n = party_totals["yea"], party_totals["nay"]
    if y > n: return "Yea"
    if n > y: return "Nay"
    return "Split"

def classify_vote(record, member_id):
    """Return dict with: member_vote, member_vote_norm, r_pos, d_pos,
    alignment, blocked. Single-vote analysis, no aggregation."""
    raw = record["votes"].get(member_id)
    norm = _norm_vote(raw)
    r_pos = _majority_position(record["totals"]["R"])
    d_pos = _majority_position(record["totals"]["D"])
    if norm not in ("Yea", "Nay"):
        return {"member_vote": raw, "member_vote_norm": norm,
                "r_pos": r_pos, "d_pos": d_pos,
                "alignment": "N/A: " + (raw or "absent"),
                "blocked": None}
    helped_r = (r_pos != "Split" and norm == r_pos)
    helped_d = (d_pos != "Split" and norm == d_pos)
    if helped_r and helped_d: align = "Helped Both"
    elif helped_r:            align = "Helped Republicans"
    elif helped_d:            align = "Helped Democrats"
    else:                     align = "Helped Neither"
    result = (record.get("result") or "").lower()
    failed = any(s in result for s in ("fail", "reject", "not agreed", "not passed", "not invoked"))
    blocked = None
    if d_pos == "Yea" and failed and norm == "Nay" and r_pos != "Yea":
        blocked = "Democrat"
    if r_pos == "Yea" and failed and norm == "Nay" and d_pos != "Yea":
        blocked = "Republican"
    return {"member_vote": raw, "member_vote_norm": norm,
            "r_pos": r_pos, "d_pos": d_pos,
            "alignment": align, "blocked": blocked}

def aggregate(records, member_id, member_party, chamber):
    """records: iterable of parsed vote dicts. Returns the metrics dict
    consumed by render.py."""
    align_counts = Counter()
    raw_vote_counts = Counter()
    blocked_counts = Counter()
    month_align = defaultdict(Counter)
    rows = []
    va_gop = vw_gop = va_dem = vw_dem = 0
    lone_wolf = 0
    threshold = LONE_WOLF_THRESHOLD.get(chamber, 5)
    yeas = nays = nv = present = voting = 0
    for rec in records:
        cls = classify_vote(rec, member_id)
        raw = cls["member_vote"]
        norm = cls["member_vote_norm"]
        align_counts[cls["alignment"]] += 1
        raw_vote_counts[raw or "Absent"] += 1
        if cls["blocked"]:
            blocked_counts[cls["blocked"]] += 1
        if rec.get("date"):
            month_align[rec["date"][:7]][cls["alignment"]] += 1
        # vote-distribution counters
        if norm == "Yea": yeas += 1; voting += 1
        elif norm == "Nay": nays += 1; voting += 1
        elif raw == "Not Voting": nv += 1
        elif raw == "Present": present += 1
        # voted with/against each party majority
        if norm in ("Yea","Nay"):
            r_pos, d_pos = cls["r_pos"], cls["d_pos"]
            if r_pos != "Split":
                if norm == r_pos: vw_gop += 1
                else: va_gop += 1
            if d_pos != "Split":
                if norm == d_pos: vw_dem += 1
                else: va_dem += 1
            # lone wolf vs own party
            own_pos = r_pos if member_party == "R" else d_pos if member_party == "D" else None
            if own_pos and own_pos != "Split" and norm != own_pos:
                own_totals = rec["totals"]["R"] if member_party == "R" else rec["totals"]["D"]
                defectors = own_totals["nay"] if own_pos == "Yea" else own_totals["yea"]
                if defectors <= threshold:
                    lone_wolf += 1
        # row for the table
        rows.append({
            "y": rec.get("year") or "",
            "r": rec.get("num") or 0,
            "d": rec.get("date_raw") or rec.get("date") or "",
            "ln": rec.get("bill") or "",
            "q": rec.get("question") or "",
            "ds": (rec.get("desc") or "")[:90],
            "rs": rec.get("result") or "",
            "m": raw or "",
            "ry": rec["totals"]["R"]["yea"], "rn": rec["totals"]["R"]["nay"],
            "dy": rec["totals"]["D"]["yea"], "dn": rec["totals"]["D"]["nay"],
            "a": cls["alignment"],
            "b": cls["blocked"] or "",
        })
    months_sorted = sorted(month_align.keys())
    align_labels = ["Helped Republicans","Helped Democrats","Helped Both","Helped Neither"]
    monthly = {lab: [month_align[m].get(lab, 0) for m in months_sorted]
               for lab in align_labels}
    total = len(rows)
    return {
        "chamber": chamber,
        "total": total, "voting": voting,
        "yeas": yeas, "nays": nays, "nv": nv, "present": present,
        "alignment": dict(align_counts),
        "member": dict(raw_vote_counts),
        "blocked": dict(blocked_counts),
        "months": months_sorted, "monthly": monthly,
        "rows": rows,
        "blocked_dem_count": blocked_counts.get("Democrat", 0),
        "blocked_rep_count": blocked_counts.get("Republican", 0),
        "lone_wolf": lone_wolf,
        "voted_with_gop": vw_gop, "voted_against_gop": va_gop,
        "voted_with_dem": vw_dem, "voted_against_dem": va_dem,
        "lone_wolf_threshold": threshold,
    }