How to pull any account balance for every US credit union with Python

from pathlib import Path
import io, zipfile, requests

DATA_DIR = Path("data")


def fetch_quarter(zip_url: str, label: str) -> Path:
    """Download and extract one quarterly Call Report archive.

    label is whatever you want the folder called, e.g. "2026-03".
    """
    target = DATA_DIR / label
    if target.exists() and any(target.iterdir()):
        print(f"{label} already present, skipping download")
        return target

    target.mkdir(parents=True, exist_ok=True)
    print(f"downloading {label} ...")

    resp = requests.get(zip_url, timeout=180)
    resp.raise_for_status()

    with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
        zf.extractall(target)

    print(f"extracted {len(list(target.iterdir()))} files to {target}")
    return target

import pandas as pd
from pathlib import Path


def load_dictionary(quarter_dir: Path) -> pd.DataFrame:
    """Load the account description file, whatever it's called this quarter."""
    candidates = [
        p for p in quarter_dir.iterdir()
        if "acctdesc" in p.name.lower() and p.suffix.lower() in {".csv", ".txt"}
    ]
    if not candidates:
        raise FileNotFoundError(
            f"No account description file in {quarter_dir}. "
            f"Found: {[p.name for p in quarter_dir.iterdir()]}"
        )

    return pd.read_csv(candidates[0], dtype=str, encoding="latin-1")


def find_codes(dictionary: pd.DataFrame, *terms: str) -> pd.DataFrame:
    """Return dictionary rows whose description contains ALL search terms.

    Case-insensitive. Pass the words you'd expect in the line item's
    official description, not marketing language.
    """
    text_cols = [c for c in dictionary.columns if dictionary[c].dtype == object]
    blob = dictionary[text_cols].fillna("").agg(" ".join, axis=1).str.lower()

    mask = pd.Series(True, index=dictionary.index)
    for term in terms:
        mask &= blob.str.contains(term.lower(), regex=False)

    return dictionary[mask]


# Usage: search, read what comes back, THEN pick your code.
#   matches = find_codes(dictionary, "charitable", "donation")
#   print(matches.to_string())
#
# Do not skip reading the output. Similar descriptions exist and
# picking the wrong one gives you a plausible, entirely wrong list.

import pandas as pd
from pathlib import Path

ID_COL = "CU_NUMBER"


def _read(path: Path, **kw) -> pd.DataFrame:
    return pd.read_csv(path, dtype=str, encoding="latin-1",
                       low_memory=False, **kw)


def extract_account(quarter_dir: Path, acct_code: str) -> pd.DataFrame:
    """Find whichever file holds acct_code and return CU_NUMBER + that column."""
    for path in sorted(quarter_dir.iterdir()):
        if path.suffix.lower() not in {".csv", ".txt"}:
            continue

        # peek at headers only — don't parse whole files looking for one column
        try:
            header = _read(path, nrows=0)
        except Exception:
            continue

        cols = {c.strip().upper() for c in header.columns}
        if acct_code.upper() in cols and ID_COL in cols:
            print(f"found {acct_code} in {path.name}")
            df = _read(path, usecols=lambda c: c.strip().upper()
                       in {ID_COL, acct_code.upper()})
            df.columns = [c.strip().upper() for c in df.columns]
            df[acct_code.upper()] = pd.to_numeric(
                df[acct_code.upper()], errors="coerce"
            ).fillna(0)
            return df.rename(columns={acct_code.upper(): "balance"})

    raise KeyError(f"{acct_code} not found in any file in {quarter_dir}")


def attach_institutions(balances: pd.DataFrame,
                        quarter_dir: Path) -> pd.DataFrame:
    """Join balances to the institution master file for names and location."""
    master = None
    for path in sorted(quarter_dir.iterdir()):
        if path.suffix.lower() not in {".csv", ".txt"}:
            continue
        try:
            header = _read(path, nrows=0)
        except Exception:
            continue

        cols = {c.strip().upper() for c in header.columns}
        # the master file is the one carrying the institution name
        if ID_COL in cols and any("NAME" in c for c in cols):
            master = _read(path)
            master.columns = [c.strip().upper() for c in master.columns]
            break

    if master is None:
        raise FileNotFoundError("No institution master file found")

    name_col = next(c for c in master.columns if "NAME" in c)
    keep = [ID_COL, name_col] + [
        c for c in ("STATE", "CITY", "ZIP_CODE") if c in master.columns
    ]

    out = balances.merge(master[keep], on=ID_COL, how="left")
    return out.rename(columns={name_col: "institution"})

import re
import pandas as pd
from rapidfuzz import process, fuzz

# suffixes that carry no identifying information
NOISE = [
    "federal credit union", "credit union", "fcu", "cu",
    "incorporated", "inc", "llc", "the",
]


def normalize(name: str) -> str:
    """Reduce an institution name to its identifying core."""
    if not isinstance(name, str):
        return ""
    s = name.lower()
    s = re.sub(r"[^a-z0-9\s]", " ", s)          # punctuation out
    s = re.sub(r"\s+", " ", s).strip()
    for suffix in NOISE:                          # longest first, see below
        s = re.sub(rf"\b{re.escape(suffix)}\b", " ", s)
    return re.sub(r"\s+", " ", s).strip()


# NOISE is ordered longest-first on purpose: strip "credit union" before
# "cu", or "cu" fires inside it and leaves debris behind.
NOISE.sort(key=len, reverse=True)


def anti_join(universe: pd.DataFrame,
              existing: pd.DataFrame,
              threshold: int = 88) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Split universe into (net_new, matched) against existing records.

    Returns matched rows too — you want to eyeball scores near the
    threshold before trusting either pile.
    """
    universe = universe.copy()
    universe["_key"] = universe["institution"].map(normalize)

    known = [k for k in existing["company"].map(normalize).unique() if k]

    scores, hits = [], []
    for key in universe["_key"]:
        if not key:
            scores.append(0)
            hits.append("")
            continue
        match = process.extractOne(
            key, known, scorer=fuzz.token_set_ratio, score_cutoff=threshold
        )
        scores.append(match[1] if match else 0)
        hits.append(match[0] if match else "")

    universe["_score"] = scores
    universe["_matched_to"] = hits

    matched = universe[universe["_score"] >= threshold]
    net_new = universe[universe["_score"] < threshold]
    return net_new.drop(columns="_key"), matched.drop(columns="_key")