spb/wp22_uqo

1#!/usr/bin/env python32# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca3"""Phase 1 — standardized, cleaned and analytic layers + exclusion ledger + classification audit."""4from __future__ import annotations56import sys7from pathlib import Path89sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))10import polars as pl1112from sharp_hpi import config as C13from sharp_hpi.cleaning import add_flags, analytic, classify, standardize1415OUT = C.RESULTS / "cleaning"16OUT.mkdir(parents=True, exist_ok=True)171819def main() -> None:20    tx = pl.read_parquet(C.RAW_TX)21    units = pl.read_parquet(C.RAW_UNITS_MATCHED)22    geo = pl.read_parquet(C.GEO_TX)23    std = standardize(tx, units, geo)24    std.write_parquet(C.STD_TX)25    # provider fields vs roll fields (are the provider characteristics simply the roll's?)26    pl.DataFrame({27        "pair": ["year_built", "floor_area", "land_area", "assessment_vs_current_roll", "assessment_vs_previous_roll"],28        "both_present": [std.filter(pl.col(a).is_not_null() & pl.col(b).is_not_null()).height for a, b in29                         [("provider_year_built", "roll_year_built"), ("provider_floor_area", "roll_floor_area_m2"), ("provider_land_area", "roll_land_area_m2"), ("provider_assessment", "roll_total_value"), ("provider_assessment", "roll_previous_value")]],30        "share_equal": [float((std[a] == std[b]).mean()) for a, b in31                        [("provider_year_built", "roll_year_built"), ("provider_floor_area", "roll_floor_area_m2"), ("provider_land_area", "roll_land_area_m2"), ("provider_assessment", "roll_total_value"), ("provider_assessment", "roll_previous_value")]],32    }).write_csv(OUT / "provider_vs_roll_fields.csv")3334    clean = add_flags(classify(std))35    clean.write_parquet(C.CLEAN_TX)36    n = clean.height37    # exclusion ledger (priority order, counts, share of raw)38    ledger = clean.group_by("exclude_reason_headline").agg(pl.len().alias("n")).with_columns((pl.col("n") / n).alias("share_of_raw")).sort("n", descending=True)39    ledger.write_csv(OUT / "exclusion_ledger.csv")40    # every flag independently (overlaps allowed)41    flags = ["exact_duplicate", "same_parcel_same_date_diff_price", "built_after_sale", "new_construction", "roll_is_land", "roll_non_residential",42             "shared_parcel", "multi_parcel_candidate", "price_above_cap", "assessment_ratio_outlier", "type_conflict", "floor_area_missing", "year_built_missing", "roll_matched"]43    pl.DataFrame({"flag": flags, "n": [int(clean[f].fill_null(False).sum()) for f in flags]}).with_columns((pl.col("n") / n).alias("share_of_raw")).write_csv(OUT / "flag_counts.csv")44    # classification audit45    clean.group_by(["provider_type", "property_class", "class_source"]).agg(pl.len().alias("n")).sort("n", descending=True).write_csv(OUT / "classification_crosstab.csv")46    clean.filter(pl.col("in_headline_sample")).group_by(["year", "property_class"]).agg(pl.len().alias("n"), pl.col("type_conflict").mean().alias("share_type_conflict"),47        pl.col("new_construction").mean().alias("share_new_construction"), pl.col("floor_area_missing").mean().alias("share_floor_area_missing"),48        pl.col("year_built_missing").mean().alias("share_year_built_missing")).sort(["year", "property_class"]).write_csv(OUT / "headline_sample_by_year_class.csv")49    clean.group_by(["year", "in_headline_sample"]).agg(pl.len().alias("n")).sort(["year", "in_headline_sample"]).write_csv(OUT / "headline_share_by_year.csv")50    ana = analytic(clean)51    ana.write_parquet(C.ANALYTIC_TX)52    print(ledger)53    print("headline sample:", ana.height, "of", n)545556if __name__ == "__main__":57    main()58