spb/wp3_uqo

1#!/usr/bin/env python32# Author: Simon-Pierre Boucher — contact@spboucher.ai3#4"""Build the analytical sample from the raw Zillow DuckDB extract.56Rebuilt in the November 2026 review (WP03-DATA-01, EDIT-05, ADVE-07, ADVE-10,7ADVE-18, THEO-15): the original construction script was lost; the rules in8``src/wp3/build.py`` reproduce ``data/processed/analytical_sample.pkl`` exactly.910Outputs11  data/processed/analytical_sample_v2.parquet   788,842 x (68 + flag columns)12  data/ref/category_map.csv                     (with --export-lookup; frozen13                                                 raw string -> category lookup)14  results/build/attrition.csv                   Table 215  results/build/descriptives.csv                Table 5 (all numeric columns)16  results/build/missingness_rules.csv           Table 3, corrected rules17  results/build/lot_reconciliation.csv          Table 4 / ADVE-10 (ln lot = 0)18  results/build/listing_types.csv               listing_type / home_type shares19  results/build/duplicates.csv                  exact-duplicate groups (THEO-15)20  results/build/build_meta.json                 data window, counts, sha25621  results/build/sha256.txt2223Usage:  python scripts/00_build_sample.py [--export-lookup] [--check]24        --check asserts equality with analytical_sample.pkl (laptop only).25"""2627import argparse28import hashlib29import json30import sys31import time32from pathlib import Path3334import duckdb35import numpy as np36import pandas as pd3738sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))39from wp3 import build, config  # noqa: E4024041REF = config.PROJECT_ROOT / "data" / "ref"42OUT = config.RESULTS_DIR / "build"43SAMPLE_V2 = config.DATA_PROCESSED / "analytical_sample_v2.parquet"44DUP_KEYS = ["address_zipcode", "living_area_sqft", "bedrooms", "bathrooms", "year_built", "price"]454647def sha256(path, chunk=1 << 24):48    h = hashlib.sha256()49    with open(path, "rb") as f:50        for b in iter(lambda: f.read(chunk), b""):51            h.update(b)52    return h.hexdigest()535455def export_lookup(raw):56    """One-off: freeze the raw-string -> category mapping from the stored pickle."""57    pk = pd.read_pickle(config.ANALYTICAL_SAMPLE)58    assert (pk["zpid"].values == raw["zpid"].values).all()59    rows = []60    for rc, cc in build.CATEGORICALS.items():61        t = pd.DataFrame({"raw": raw[rc].values, "category": pk[cc].values}).dropna(subset=["raw"])62        amb = t.groupby("raw")["category"].nunique()63        assert (amb <= 1).all(), f"{rc}: ambiguous strings"64        t = t.drop_duplicates("raw").sort_values("raw")65        t.insert(0, "field", rc)66        rows.append(t)67    lk = pd.concat(rows, ignore_index=True)68    REF.mkdir(parents=True, exist_ok=True)69    lk.to_csv(REF / "category_map.csv", index=False)70    print(f"category_map.csv: {len(lk):,} strings "71          f"({lk.groupby('field').size().to_dict()})")72    return lk737475def attrition(con):76    rows, where = [], []77    for label, cond in build.FILTER_STEPS:78        if cond:79            where.append(cond)80        q = "SELECT COUNT(*) FROM properties" + (" WHERE " + " AND ".join(where) if where else "")81        n = con.execute(q).fetchone()[0]82        rows.append({"step": label, "n_remaining": n})83    a = pd.DataFrame(rows)84    a["n_dropped"] = -a["n_remaining"].diff().fillna(0).astype(int)85    a["share_of_raw"] = a["n_remaining"] / a["n_remaining"].iloc[0]86    return a878889def descriptives(df):90    num = df.select_dtypes(include=["number", "bool"]).drop(columns=["zpid"])91    d = num.describe(percentiles=[0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99]).T92    d["skew"] = num.skew()93    d["kurtosis"] = num.kurt()94    d["n_missing"] = num.isna().sum()95    d.index.name = "variable"96    return d979899def missingness_rules(raw, df):100    rules = {101        "year_built (age, age^2, interactions)": ("year_built", "state median year built (age only; year_built keeps NaN)"),102        "lot_size_sqft (ln lot)": ("lot_size_sqft", "state median in lot_size_sqft; ln(lot) coded 0 in v1.1 (v2: ln state-median lot + indicator)"),103        "avg_school_rating": ("avg_school_rating", "state median"),104        "school_count": ("school_count", "state median"),105        "nearest_school_distance": ("nearest_school_distance", "state median"),106        "walk_score": ("walk_score", "state median"),107        "bike_score": ("bike_score", "state median"),108        "transit_score": ("transit_score", "set to 0"),109        "has_garage": ("has_garage", "set to False"),110        "parking_spaces": ("parking_spaces", "state median (Table 3 of v1.2 said 'set to 0': corrected)"),111        "stories": ("stories", "state median (Table 3 of v1.2 said 'set to 1': corrected)"),112        "property_tax_rate": ("property_tax_rate", "state median"),113        "hoa_annual_cost (ln HOA)": ("hoa_annual_cost", "set to 0"),114        "has_pool / has_spa / has_basement / has_fireplace / on_waterfront": ("has_pool", "set to False"),115        "roof_type / construction_material / foundation_type": ("roof_type", "'Unknown' category"),116        "cooling_type (central-air dummy)": ("cooling_type", "dummy set to False"),117        "heating_type (forced-air dummy)": ("heating_type", "dummy set to False"),118        "flooring (hardwood dummy)": ("flooring", "dummy set to False"),119    }120    rows = []121    for label, (col, rule) in rules.items():122        imputed = raw[col].isna()123        rows.append({"input": label, "raw_column": col, "n_missing": int(imputed.sum()),124                     "missing_share": float(imputed.mean()), "imputation": rule})125    m = pd.DataFrame(rows).sort_values("missing_share", ascending=False)126    # values actually assigned to the two mis-described rules127    extra = []128    for col in ["stories", "parking_spaces"]:129        miss = raw[col].isna()130        vc = df.loc[miss, col].value_counts().head(5)131        extra.append({"column": col, "n_imputed": int(miss.sum()),132                      "imputed_values_top5": json.dumps({str(k): int(v) for k, v in vc.items()})})133    return m, pd.DataFrame(extra)134135136def lot_reconciliation(flags, df):137    t = flags["lot_class"].value_counts()138    rows = [139        ("lot_size_sqft missing (null)", int(t.get("null", 0))),140        ("lot_size_sqft = 0", int(t.get("zero", 0))),141        ("lot_size_sqft = 1 (ln = 0)", int(t.get("one", 0))),142        ("0 < lot_size_sqft < 1", int(t.get("between_0_and_1", 0))),143        ("lot_size_sqft < 0 (negative)", int(t.get("negative", 0))),144    ]145    ln0 = int((df["ln_lot"] == 0).sum())146    rows.append(("Total with ln(lot) = 0 in the v1.1 coding", ln0))147    rows.append(("Check: sum of the five classes", sum(r[1] for r in rows[:5])))148    r = pd.DataFrame(rows, columns=["class", "n"])149    r["share_of_sample"] = r["n"] / len(df)150    return r151152153def duplicates(df):154    key = pd.Series(pd.util.hash_pandas_object(df[DUP_KEYS].fillna(-1), index=False).values)155    size = key.map(key.value_counts())156    grp = pd.factorize(key)[0]157    dup_group = np.where(size > 1, grp, -1)158    n_rows = int((size > 1).sum())159    n_groups = int(pd.Series(grp[size > 1]).nunique())160    summary = pd.DataFrame([{"keys": ", ".join(DUP_KEYS), "n_rows_in_duplicate_groups": n_rows,161                             "n_duplicate_groups": n_groups,162                             "share_of_sample": n_rows / len(df)}])163    return dup_group, summary164165166def data_window(con):167    q = """SELECT max(date_sold) AS max_date_sold, max(ph_last_sold_date) AS max_ph_last_sold,168                  max(ph_first_listed_date) AS max_ph_first_listed,169                  (SELECT max(event_date) FROM price_history) AS max_event_date,170                  (SELECT min(event_date) FROM price_history) AS min_event_date,171                  median(days_on_zillow) AS med_days_on_zillow,172                  quantile_cont(days_on_zillow, 0.25) AS p25_days, quantile_cont(days_on_zillow, 0.75) AS p75_days,173                  count(*) AS n_raw, count(DISTINCT zpid) AS n_zpid,174                  sum(CASE WHEN home_status = 'FOR_SALE' THEN 1 ELSE 0 END) AS n_for_sale175           FROM properties"""176    try:177        r = con.execute(q).fetchdf().iloc[0].to_dict()178    except Exception as exc:  # price_history columns may differ179        r = {"error": str(exc)}180    return {k: (str(v) if not isinstance(v, (int, float)) else v) for k, v in r.items()}181182183def main():184    ap = argparse.ArgumentParser(description=__doc__)185    ap.add_argument("--export-lookup", action="store_true")186    ap.add_argument("--check", action="store_true", help="assert equality with analytical_sample.pkl")187    args = ap.parse_args()188    t0 = time.time()189    OUT.mkdir(parents=True, exist_ok=True)190191    con = duckdb.connect(str(config.DUCKDB_PATH), read_only=True)192    attr = attrition(con)193    attr.to_csv(OUT / "attrition.csv", index=False)194    print(attr.to_string(index=False))195    raw = con.execute(f"SELECT {', '.join(build.RAW_COLUMNS)} FROM properties WHERE {build.FILTER_SQL}").fetchdf()196    lt = con.execute("SELECT listing_type, home_type, count(*) AS n FROM properties GROUP BY 1, 2").fetchdf()197    window = data_window(con)198    con.close()199    assert len(raw) == 788_842, len(raw)200    assert raw["zpid"].is_unique201202    if args.export_lookup:203        lookup = export_lookup(raw)204    else:205        lookup = pd.read_csv(REF / "category_map.csv", keep_default_na=False, na_filter=False)206207    df = build.engineer(raw, lookup)208    flags = build.raw_flags(raw)209    print(f"sample: {df.shape[0]:,} rows x {df.shape[1]} columns ({time.time() - t0:.0f}s)")210211    if args.check:212        pk = pd.read_pickle(config.ANALYTICAL_SAMPLE)213        assert list(pk.columns) == list(df.columns), "column order differs"214        assert (pk["zpid"].values == df["zpid"].values).all(), "zpid order differs"215        worst = {}216        for c in df.columns:217            a, b = pk[c], df[c]218            if a.dtype.kind in "biufc" or str(a.dtype).startswith("Int"):219                aa, bb = a.astype(float).values, b.astype(float).values220                both_nan = np.isnan(aa) & np.isnan(bb)221                assert (np.isnan(aa) == np.isnan(bb)).all(), f"{c}: NaN pattern differs"222                worst[c] = float(np.nanmax(np.abs(aa[~both_nan] - bb[~both_nan]))) if (~both_nan).any() else 0.0223                assert worst[c] < 1e-9, f"{c}: max|d| = {worst[c]}"224            else:225                eq = ((a.isna() & b.isna()) | (a == b)).all()226                assert eq, f"{c}: string mismatch"227                worst[c] = 0.0228        print("CHECK PASSED: all 68 columns equal to analytical_sample.pkl "229              f"(max numeric |d| = {max(worst.values()):.2e})")230        pd.Series(worst, name="max_abs_diff").rename_axis("column").to_csv(OUT / "check_vs_pickle.csv")231232    # ------------------------------------------------------------------ extras233    dup_group, dup_summary = duplicates(df)234    dup_summary.to_csv(OUT / "duplicates.csv", index=False)235    print(dup_summary.to_string(index=False))236    out = pd.concat([df, flags], axis=1)237    out["dup_group_id"] = dup_group.astype("int32")238    out["zip3"] = out["address_zipcode"].fillna("000").astype(str).str[:3]239    out["implausible"] = ((df["parking_spaces"] > 20) | (df["stories"] > 10) |240                          (df["lot_size_sqft"] > 43560 * 640) | (df["age"] >= 200)).astype("int8")241    config.DATA_PROCESSED.mkdir(parents=True, exist_ok=True)242    out.to_parquet(SAMPLE_V2, index=False)243244    descriptives(df).to_csv(OUT / "descriptives.csv")245    m, extra = missingness_rules(raw, df)246    m.to_csv(OUT / "missingness_rules.csv", index=False)247    extra.to_csv(OUT / "missingness_imputed_values.csv", index=False)248    lot = lot_reconciliation(flags, df)249    lot.to_csv(OUT / "lot_reconciliation.csv", index=False)250    print(lot.to_string(index=False))251    lt["share_raw"] = lt["n"] / lt["n"].sum()252    lt.sort_values("n", ascending=False).to_csv(OUT / "listing_types.csv", index=False)253    lt1 = lt.groupby("listing_type", dropna=False)["n"].sum().sort_values(ascending=False).reset_index()254    lt1["share_raw"] = lt1["n"] / lt1["n"].sum()255    lt1.to_csv(OUT / "listing_types_raw_shares.csv", index=False)256    ht = out["home_type"].value_counts(dropna=False)257    pd.DataFrame({"home_type": ht.index.astype(str), "n": ht.values, "share_sample": ht.values / len(out)}) \258        .to_csv(OUT / "home_types_sample.csv", index=False)259260    meta = {261        "n_sample": int(len(df)), "n_columns_sample": int(df.shape[1]),262        "n_states": int(df["address_state"].nunique()), "n_zip5": int(df["address_zipcode"].nunique()),263        "n_zip3": int(out["zip3"].nunique()), "n_zip_missing": int(df["address_zipcode"].isna().sum()),264        "n_implausible": int(out["implausible"].sum()),265        "n_trimmed_sample": int((out["implausible"] == 0).sum()),266        "county_fips_null_share": float(out["county_fips"].isna().mean()),267        "data_window": window,268        "inputs_sha256": {"us_housing.duckdb": sha256(config.DUCKDB_PATH)},269        "lookup_strings": int(len(lookup)),270        "filters": build.FILTER_SQL,271        "elapsed_s": round(time.time() - t0, 1),272    }273    (OUT / "build_meta.json").write_text(json.dumps(meta, indent=2, default=str))274    with open(OUT / "sha256.txt", "w") as f:275        for p in [SAMPLE_V2, REF / "category_map.csv"]:276            f.write(f"{sha256(p)}  {p.relative_to(config.PROJECT_ROOT)}\n")277    print(json.dumps(meta, indent=1, default=str))278    print(f"Done in {time.time() - t0:.0f}s")279280281if __name__ == "__main__":282    main()283