Tracking PAC money through FEC data: entity resolution across 50 filing types at 2.4 million donor records

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class JFCAllocation:
    participant_committee_id: str    # FEC committee ID of the participant
    participant_name: str
    allocation_pct: float            # 0.0-1.0; must sum to 1.0 across participants
    allocation_cap_usd: Optional[float]  # per-donor aggregate limit (None = no cap)

@dataclass
class JointFundraisingCommittee:
    jfc_committee_id: str            # FEC ID of the JFC itself
    jfc_name: str
    registered_agent: str
    participants: list[JFCAllocation] = field(default_factory=list)
    # Computed fields populated after resolving Form 99 filings
    total_raised_usd: float = 0.0
    total_transferred_usd: float = 0.0

def resolve_jfc_participants(
    jfc_id: str,
    form99_df,      # DataFrame of FEC Form 99 filings
    disbursements_df,  # DataFrame of FEC disbursement records
) -> JointFundraisingCommittee:
    """Reconstruct JFC allocation from Form 99 + disbursement records."""
    # Form 99 filings contain the human-readable allocation memo
    # Disbursements from the JFC to participants give the realized allocation
    form99_rows = form99_df[form99_df['committee_id'] == jfc_id]
    allocation_text = ' '.join(form99_rows['memo_text'].dropna().tolist())

    # Parse "Candidate A: 75%, Candidate B: 25%" allocation strings
    participants = _parse_allocation_memo(allocation_text)

    # Cross-validate against actual disbursement records
    actual_disbursements = disbursements_df[
        (disbursements_df['filer_committee_id'] == jfc_id) &
        (disbursements_df['disbursement_type'].isin(['24K', '24Z']))
    ]

    # If memo parsing failed, infer from realized disbursements
    if not participants:
        total = actual_disbursements['disbursement_amount'].sum()
        for _, row in actual_disbursements.iterrows():
            participants.append(JFCAllocation(
                participant_committee_id=row['recipient_committee_id'],
                participant_name=row['recipient_name'],
                allocation_pct=row['disbursement_amount'] / total if total > 0 else 0.0,
                allocation_cap_usd=None,
            ))

    return JointFundraisingCommittee(
        jfc_committee_id=jfc_id,
        jfc_name=form99_rows.iloc[0]['committee_name'] if len(form99_rows) > 0 else jfc_id,
        registered_agent='',
        participants=participants,
    )

import re
import unicodedata

# Legal entity suffixes to strip from organization names
LEGAL_SUFFIXES = re.compile(
    r'(inc|incorporated|corp|corporation|co|company|llc|llp|lp|ltd|limited'
    r'|pc|pllc|pa|assoc|associates|grp|group|intl|international'
    r'|fund|pac|super pac|political action committee).?$',
    re.IGNORECASE,
)

# Individual name prefixes/suffixes
NAME_PREFIXES = re.compile(r'^(mr|mrs|ms|dr|prof|rev|hon|capt|col|gen).?s+', re.IGNORECASE)
NAME_SUFFIXES = re.compile(r's+(jr|sr|ii|iii|iv|esq|md|phd|cpa).?$', re.IGNORECASE)

def normalize_entity_name(raw: str, entity_type: str = 'ORG') -> str:
    """
    Normalize a FEC filer/recipient name to a canonical form.
    entity_type: 'ORG' or 'IND' (individual).
    """
    if not raw:
        return ''

    # Unicode normalization: decompose, strip diacritics, recompose
    nfkd = unicodedata.normalize('NFKD', raw)
    ascii_str = ''.join(c for c in nfkd if not unicodedata.combining(c))

    # Uppercase, collapse whitespace, strip punctuation except hyphens
    s = re.sub(r'[^ws-]', ' ', ascii_str.upper())
    s = re.sub(r's+', ' ', s).strip()

    if entity_type == 'ORG':
        # Strip trailing legal suffixes (iteratively; 'Apple Inc Corp' -> 'Apple')
        prev = None
        while prev != s:
            prev = s
            s = LEGAL_SUFFIXES.sub('', s).strip()
    elif entity_type == 'IND':
        s = NAME_PREFIXES.sub('', s)
        s = NAME_SUFFIXES.sub('', s).strip()

    return s

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

class FECEntityMatcher:
    """
    TF-IDF cosine matcher for FEC committee and organization names.
    Uses character 3-grams to handle abbreviations and partial matches.
    """
    def __init__(self, threshold: float = 0.82):
        self.threshold = threshold
        self.vectorizer = TfidfVectorizer(
            analyzer='char_wb',
            ngram_range=(3, 3),
            min_df=1,
            sublinear_tf=True,
        )
        self._fitted = False
        self._canonical_names: list[str] = []
        self._canonical_ids: list[str] = []

    def fit(self, canonical_names: list[str], canonical_ids: list[str]) -> None:
        self._canonical_names = canonical_names
        self._canonical_ids = canonical_ids
        self.vectorizer.fit(canonical_names)
        self._matrix = self.vectorizer.transform(canonical_names)
        self._fitted = True

    def match(self, query_name: str, top_k: int = 3) -> list[tuple[str, str, float]]:
        """Return [(canonical_id, canonical_name, score)] sorted descending."""
        if not self._fitted:
            raise RuntimeError('Call fit() first')
        q_vec = self.vectorizer.transform([query_name])
        scores = cosine_similarity(q_vec, self._matrix)[0]
        top_idxs = np.argsort(scores)[::-1][:top_k]
        results = []
        for idx in top_idxs:
            if scores[idx] >= self.threshold:
                results.append((
                    self._canonical_ids[idx],
                    self._canonical_names[idx],
                    float(scores[idx]),
                ))
        return results