screen.py

"""Build the distress screening list from the universe and tax pulls.

Usage: uv run python screen.py [date]

Defaults to the most recent date-stamped pull in data/raw/. Writes
data/raw/distress_screen_<date>.csv, which carries real owner names and real mailing
addresses. Everything under data/raw/ is gitignored: it is the operational copy, and the
notebook never reads it.
"""
import json
import re
import sys
from pathlib import Path

import pandas as pd

import distress
import ownership

DATA = Path("data") / "raw"
ID_FORMAT_FLOOR = 0.95
DATE_PAT = re.compile(r"\d{8}")

COLUMNS = ["Property ID", "tier", "rank_in_tier", "years_delinquent", "delinquent_balance",
           "ratio", "taxable_value", "Parcel Size", "Neighborhood", "last_sale_date", "tenure",
           "Owner Name", "Mail Address 1", "Mail City", "Mail State", "Mail Zip",
           "out_of_state", "individual", "ratio_undefined", "delq_year_missing",
           "blank_mail_state"]


def load(stem, date):
    path = DATA / f"{stem}_{date}.json"
    return pd.DataFrame(json.loads(path.read_text(encoding="utf-8")))


def latest_date():
    dates = sorted(p.stem.rsplit("_", 1)[1] for p in DATA.glob("tax_delinquent_*.json"))
    if not dates:
        raise SystemExit("no tax_delinquent_<date>.json in data/; run pull_tax.py first")
    return dates[-1]


def build(universe_raw, tax_raw, as_of_year):
    parcels = distress.latest_per_parcel(universe_raw)
    rural = distress.exclude_near_town(parcels)

    rate = distress.check_id_format(tax_raw)
    if rate < ID_FORMAT_FLOOR:
        raise SystemExit(
            f"only {rate:.1%} of Real Estate tax accounts carry an assessor Property ID "
            f"(floor {ID_FORMAT_FLOOR:.0%}). The join key namespace has changed; a silent "
            f"mismatch would return a plausible-looking screen covering a fraction of the county.")

    screened = distress.assign_tiers(
        distress.derive(rural, tax_raw, today_year=as_of_year))
    return parcels, rural, screened, rate


def main(date):
    universe_raw = load("universe_sales_40_79", date)
    tax_raw = load("tax_delinquent", date)
    parcels, rural, screened, rate = build(universe_raw, tax_raw, int(date[:4]))

    hits = screened[screened["tier"].notna()].sort_values(["tier", "rank_in_tier"])
    out = DATA / f"distress_screen_{date}.csv"
    hits[COLUMNS].to_csv(out, index=False)

    print(f"sale records            {len(universe_raw):>7,}")
    print(f"distinct parcels        {len(parcels):>7,}")
    print(f"after SC/EL exclusion   {len(rural):>7,}")
    print(f"  delinquent            {int(screened['is_delinquent'].sum()):>7,}")
    print(f"tax id format valid     {rate:>7.1%}")
    for tier, n in screened["tier"].value_counts().sort_index().items():
        print(f"  tier {tier}              {n:>7,}")
    print(f"screened total          {len(hits):>7,}")
    print(f"wrote {out}")

    reg = ownership.register(universe_raw, int(date[:4]))
    interests = ownership.consolidate(reg)
    reg_out = DATA / f"ownership_register_{date}.csv"
    interests.assign(member_names=interests.member_names.map("; ".join)).to_csv(
        reg_out, index=False)
    print(f"wrote {reg_out}: {len(interests):,} interests over {len(reg):,} parcels")


def parse_args(argv):
    """The pull date to build for, taken from argv, or the latest pull when none is given.

    The date is matched before it is used because main interpolates it into a filename and
    into int(date[:4]): an unchecked argument surfaced as a FileNotFoundError naming a path
    the user never typed, and --help produced that same traceback rather than the usage
    line this module's docstring already carries.
    """
    if len(argv) > 1:
        if argv[1] in ("-h", "--help"):
            print(__doc__.strip())
            raise SystemExit(0)
        if not DATE_PAT.fullmatch(argv[1]):
            raise SystemExit(
                f"not a pull date: {argv[1]!r}. Expected YYYYMMDD, the stamp in the "
                f"data/raw/<stem>_<date>.json filenames. Run with no argument to use "
                f"the most recent pull.")
        return argv[1]
    return latest_date()


if __name__ == "__main__":
    main(parse_args(sys.argv))