#!/usr/bin/env python3 """Build the public Austin replication package. This is deliberately narrower than a product signal. It tests one fixed, plain-language hypothesis: after both plumbing and electrical permits have appeared on an authoritatively linked project, does a mechanical permit follow within 180 days more often than after a building permit alone? The hypothesis, horizon, inclusion rule and year windows are fixed in this file. The script publishes the derived, non-contact project cohort, a source manifest with SHA-256 hashes, and the per-period results. It never turns a historical result into a live recommendation. """ from __future__ import annotations import argparse import csv import hashlib import json from datetime import datetime, timezone from pathlib import Path import pandas as pd HORIZON_DAYS = 180 TARGET = "mechanical_date" PRECURSORS = ("plumbing_date", "electrical_date") WINDOWS = ( ("discovery_2019_2023", None, "2024-01-01"), ("validation_2024", "2024-01-01", "2025-01-01"), ("replication_2025", "2025-01-01", "2026-01-01"), ("prospective_2026_q1", "2026-01-01", None), ) SOURCE_URL = ( "https://datahub.austintexas.gov/resource/3syk-w9eu.json?" "$select=permit_number,permittype,issue_date,masterpermitnum,original_address1&" "$where=issue_date%20%3E=%20%272019-01-01T00:00:00.000%27%20AND%20" "permittype%20in(%27BP%27,%27EP%27,%27MP%27,%27PP%27)&$limit=50000" ) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def load(path: Path) -> pd.DataFrame: frame = pd.read_csv(path, low_memory=False) required = {"job_id", "link_method", "parent_date", TARGET, *PRECURSORS} missing = required - set(frame.columns) if missing: raise ValueError(f"project file lacks: {', '.join(sorted(missing))}") for column in ("parent_date", TARGET, *PRECURSORS): frame[column] = pd.to_datetime(frame[column], errors="coerce", utc=True) frame = frame[frame["link_method"].eq("permit_base")].copy() if frame.job_id.duplicated().any(): raise ValueError("job_id must identify exactly one project") return frame def state(frame: pd.DataFrame, t_column: str) -> pd.DataFrame: result = frame.copy() if t_column == "T": result["T"] = result[list(PRECURSORS)].max(axis=1) else: result["T"] = result[t_column] return result[result["T"].notna() & (result[TARGET].isna() | (result[TARGET] > result["T"]))].copy() def building_control(raw_path: Path) -> pd.DataFrame: """Каждое строительное разрешение (основа номера) с T = датой выдачи и первой механикой по той же основе — без условия, что трейд вообще будет.""" raw = pd.read_csv(raw_path, low_memory=False, usecols=lambda c: c in {"permit_number", "permittype", "issue_date"}) raw["date"] = pd.to_datetime(raw.issue_date, errors="coerce", utc=True) raw["base"] = raw.permit_number.astype(str).str.extract(r"^(\d{4}-\d+)")[0] raw = raw[raw.base.notna() & raw.date.notna()] control = raw[raw.permittype.eq("BP")].groupby("base").date.min().rename("T").to_frame() control[TARGET] = raw[raw.permittype.eq("MP")].groupby("base").date.min() return control[control[TARGET].isna() | (control[TARGET] > control["T"])].copy() def measured_rate(frame: pd.DataFrame, start: str | None, end: pd.Timestamp) -> dict: selection = frame[frame["T"] + pd.Timedelta(days=HORIZON_DAYS) <= end] if start: selection = selection[selection["T"] >= pd.Timestamp(start, tz="UTC")] hit = (selection[TARGET].notna() & (selection[TARGET] > selection["T"]) & (selection[TARGET] <= selection["T"] + pd.Timedelta(days=HORIZON_DAYS))) return {"n": int(len(selection)), "hits": int(hit.sum()), "rate_pct": round(float(hit.mean() * 100), 1) if len(selection) else None} def result_for_window(signal: pd.DataFrame, control: pd.DataFrame, name: str, start: str | None, end: pd.Timestamp) -> dict: exposed = measured_rate(signal, start, end) baseline = measured_rate(control, start, end) uplift = None if exposed["rate_pct"] is not None and baseline["rate_pct"] is not None: uplift = round(exposed["rate_pct"] - baseline["rate_pct"], 1) return {"period": name, "window_start": start or "2019-01-01", "window_end": end.date().isoformat(), "signal": exposed, "building_only_control": baseline, "uplift_pp": uplift} def write_cohort(frame: pd.DataFrame, path: Path) -> None: columns = ["job_id", "link_method", "parent_date", *PRECURSORS, TARGET, "numberofunits", "typeofwork", "commercialorresidential"] cohort = frame[[column for column in columns if column in frame]].copy() for column in ("parent_date", *PRECURSORS, TARGET): cohort[column] = cohort[column].dt.strftime("%Y-%m-%d") cohort.to_csv(path, index=False, quoting=csv.QUOTE_MINIMAL) def build(projects: Path, raw: Path, output: Path) -> dict: output.mkdir(parents=True, exist_ok=True) frame = load(projects) signal = state(frame[frame[list(PRECURSORS)].notna().all(axis=1)], "T") # Контроль — ВСЕ строительные разрешения из исходного набора города, а не # проекты из austin_projects.csv. В том файле у каждого проекта уже есть # хотя бы одно P/E/M, то есть контроль был отобран по будущему (ревью # 22.09.2026, docs/FINDINGS-austin-trade-journey-review.md): 53,1% вместо # честных ~31%. Строительные разрешения без единого трейда — 41,6% всех. control = building_control(raw) data_end = frame[[TARGET, *PRECURSORS, "parent_date"]].max().max() if pd.isna(data_end): raise ValueError("no dated Austin records") periods = [] for name, start, boundary in WINDOWS: end = pd.Timestamp(boundary, tz="UTC") if boundary else data_end periods.append(result_for_window(signal, control, name, start, end)) cohort_path = output / "austin-authoritative-project-cohort.csv" write_cohort(frame, cohort_path) analysis_path = output / "analysis.py" analysis_path.write_text(Path(__file__).read_text(encoding="utf-8"), encoding="utf-8") manifest = { "title": "Austin permit-sequence replication package", "generated_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), "source": {"publisher": "City of Austin Build + Connect", "dataset": "3syk-w9eu", "query_url": SOURCE_URL, "source_file": raw.name, "source_sha256": sha256(raw)}, "derived_projects": {"file": projects.name, "sha256": sha256(projects), "rows_all_link_methods": int(len(pd.read_csv(projects, usecols=["job_id"]))), "rows_authoritative_permit_base": int(len(frame))}, "public_cohort": {"file": cohort_path.name, "sha256": sha256(cohort_path), "rows": int(len(frame),)}, "analysis": {"file": analysis_path.name, "sha256": sha256(analysis_path)}, "hypothesis": "Plumbing plus electrical, with no earlier mechanical permit, precedes a mechanical permit within 180 days more often than a building permit alone.", "inclusion": "Only projects linked by Austin's shared permit-number base are included; inferred parcel/address links are excluded.", "right_censoring": "A project is counted only when its T date plus 180 days is inside the named observation window.", "periods": periods, "limitations": [ "This is a historical association in one jurisdiction, not a probability for an individual job.", "The control is every Austin building permit in the source, including those never followed by any trade permit; it is a building-permit timeline, not a randomized comparison.", "The public cohort omits contractor names, addresses and contact details; the issuing authority remains the primary record.", ], } (output / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") return manifest def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--projects", type=Path, default=Path("data/derived/austin_projects.csv")) parser.add_argument("--raw", type=Path, default=Path("data/raw/austin_permits_raw.csv")) parser.add_argument("--output", type=Path, default=Path("site/static/research/austin-2026-09-21")) args = parser.parse_args() manifest = build(args.projects, args.raw, args.output) for period in manifest["periods"]: print(period["period"], period["signal"], period["building_only_control"], f"uplift={period['uplift_pp']}pp") if __name__ == "__main__": main()