# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca """Transaction standardisation, mutually exclusive property classification and exclusion ledger. Layers ------ standardized : one row per raw transaction, typed, joined to the roll unit and geography cleaned : + classification + every flag; nothing deleted, `exclude_reason` carries the first (highest-priority) reason, `in_headline_sample` the verdict analytic : headline sample only, with the modelling columns Classification precedence (first match wins): R1 built_after_sale roll building year > sale year (lot or unfinished structure at sale) R2 land roll type 'terrain' or CUBF land/road codes R3 non_residential CUBF outside the residential group {1000, 1100, 1211} R4 class from roll 1000 & n_units==1 & Intégré -> condo ; 1000 & n_units==1 & detached/semi/row -> house ; 1000 & 2<=n_units<=5 -> plex ; n_units>=6 -> multi ; 1100 -> cottage ; 1211 -> mobile R5 provider fallback when the roll is missing or incomplete: provider label in {unifamilial, condo, plex} conflicts between a complete roll record and the provider label are kept as `type_conflict` (headline keeps the roll class; sensitivity drops conflicts). """ from __future__ import annotations import polars as pl HOUSE_LINKS = {"Détaché", "Jumelé", "En rangée 1 côté", "En rangée plus de 1 côté"} PROVIDER_MAP = {"unifamilial": "house", "condo": "condo", "plex": "plex"} LAND_CUBF_PREFIX = ("9", "4") # 9xxx vacant land / forest ; 4xxx transport-utility (roads) RESIDENTIAL_CUBF = {1000, 1100, 1211} MAX_PARCEL_SALES = 10 # parcels with more sales in 2021-2026 are shared / developer parcels RATIO_LOG_BAND = 1.3862943611198906 # log(4): assessment-ratio deviation from the year×type×region median PRICE_CAP = 10_000_000.0 def standardize(tx: pl.DataFrame, units: pl.DataFrame, geo: pl.DataFrame) -> pl.DataFrame: """Join raw transactions to the roll unit and the geography layer; typed columns only.""" u = units.select( "id_provinc", pl.col("cubf"), pl.col("cubf_libelle"), pl.col("type_prop").alias("roll_type"), pl.col("annee_construction").alias("roll_year_built"), pl.col("annee_estimee").alias("roll_year_estimated"), pl.col("aire_etages_m2").alias("roll_floor_area_m2"), pl.col("superficie_terrain_m2").alias("roll_land_area_m2"), pl.col("front_terrain_m").alias("roll_frontage_m"), pl.col("nb_etages").alias("roll_storeys"), pl.col("nb_logements").alias("roll_n_units"), pl.col("nb_locaux_non_resid").alias("roll_n_nonres"), pl.col("lien_physique").alias("roll_link"), pl.col("genre_construction").alias("roll_genre"), pl.col("valeur_terrain").alias("roll_land_value"), pl.col("valeur_batiment").alias("roll_building_value"), pl.col("valeur_role").alias("roll_total_value"), pl.col("valeur_anterieure").alias("roll_previous_value"), pl.col("dat_cond_marche").alias("roll_reference_date"), pl.col("n_adresses").alias("roll_n_addresses"), pl.lit(True).alias("roll_matched"), ) g = geo.select(["id", "mun_code", "mun_name", "mun_design", "mrc_code", "mrc_name", "region_code", "region_name", "comet_code", "comet_name", "borough_code", "borough_name", "mamh_population", "x_m", "y_m", "grid500", "grid1k", "grid2k", "grid5k", "roll_mun_agrees"]) df = ( tx.rename({"id": "tx_id", "id_provinc": "parcel_id", "amount": "price", "property_type": "provider_type", "year_built": "provider_year_built", "floor_area": "provider_floor_area", "building_type": "provider_building_type", "land_area": "provider_land_area", "valeur_role": "provider_assessment"}) .with_columns(pl.when(pl.col("parcel_id") == "").then(None).otherwise(pl.col("parcel_id")).alias("parcel_id")) .join(u, left_on="parcel_id", right_on="id_provinc", how="left") .join(g, left_on="tx_id", right_on="id", how="left") .with_columns( pl.col("date").dt.strftime("%Y-%m").alias("month"), pl.col("date").dt.year().alias("year"), pl.col("price").log().alias("log_price"), pl.col("roll_matched").fill_null(False), ) ) return df def classify(df: pl.DataFrame) -> pl.DataFrame: roll_complete = pl.col("cubf").is_not_null() & pl.col("roll_n_units").is_not_null() cubf_s = pl.col("cubf").cast(pl.Utf8) is_land = (pl.col("roll_type") == "terrain") | cubf_s.str.starts_with("9") | cubf_s.str.starts_with("4") roll_class = ( pl.when(pl.col("cubf") == 1100).then(pl.lit("cottage")) .when(pl.col("cubf") == 1211).then(pl.lit("mobile")) .when((pl.col("cubf") == 1000) & (pl.col("roll_n_units") >= 6)).then(pl.lit("multi")) .when((pl.col("cubf") == 1000) & (pl.col("roll_n_units").is_between(2, 5))).then(pl.lit("plex")) .when((pl.col("cubf") == 1000) & (pl.col("roll_n_units") <= 1) & (pl.col("roll_link") == "Intégré")).then(pl.lit("condo")) .when((pl.col("cubf") == 1000) & (pl.col("roll_n_units") <= 1) & pl.col("roll_link").is_in(list(HOUSE_LINKS))).then(pl.lit("house")) .otherwise(None) ) provider_class = pl.col("provider_type").replace_strict(PROVIDER_MAP, default=None) df = df.with_columns( roll_class.alias("roll_class"), provider_class.alias("provider_class"), (pl.col("roll_year_built") > pl.col("year")).alias("built_after_sale"), (pl.col("roll_year_built") == pl.col("year")).alias("new_construction"), is_land.alias("roll_is_land"), (pl.col("cubf").is_not_null() & ~pl.col("cubf").is_in(list(RESIDENTIAL_CUBF)) & ~is_land).alias("roll_non_residential"), ) df = df.with_columns( pl.when(pl.col("roll_class").is_not_null()).then(pl.col("roll_class")).otherwise(pl.col("provider_class")).alias("property_class"), pl.when(pl.col("roll_class").is_not_null()).then(pl.lit("roll")).when(pl.col("provider_class").is_not_null()).then(pl.lit("provider")).otherwise(pl.lit("none")).alias("class_source"), (pl.col("roll_class").is_not_null() & pl.col("provider_class").is_not_null() & (pl.col("roll_class") != pl.col("provider_class"))).alias("type_conflict"), pl.when(pl.col("roll_class") == "house").then( pl.when(pl.col("roll_link") == "Détaché").then(pl.lit("detached")).when(pl.col("roll_link") == "Jumelé").then(pl.lit("semi")).otherwise(pl.lit("row")) ).when(pl.col("roll_class") == "condo").then(pl.lit("integrated")).otherwise(None).alias("subtype"), ) return df def add_flags(df: pl.DataFrame) -> pl.DataFrame: # duplicates and multi-parcel deeds df = df.with_columns( pl.col("street").str.replace(r"^\s*\d+[A-Za-z]?(?:-\d+)?[,\s]+", "").str.to_lowercase().alias("_street_name"), (pl.col("lat") * 100).round(0).alias("_lat_c"), (pl.col("lng") * 100).round(0).alias("_lng_c"), ) df = df.with_columns( (pl.len().over(["parcel_id", "date", "price"]) > 1).alias("_dup_exact_grp"), (pl.int_range(pl.len()).over(["parcel_id", "date", "price"]) > 0).alias("_dup_exact_later"), (pl.col("parcel_id").is_not_null() & (pl.col("price").n_unique().over(["parcel_id", "date"]) > 1)).alias("same_parcel_same_date_diff_price"), (pl.len().over(["date", "price", "city", "_street_name"]) > 1).alias("_mp_street"), (pl.len().over(["date", "price", "_lat_c", "_lng_c"]) > 1).alias("_mp_cell"), pl.len().over("parcel_id").alias("parcel_sales_2021_2026"), ).with_columns( (pl.col("parcel_id").is_not_null() & pl.col("_dup_exact_grp") & pl.col("_dup_exact_later")).alias("exact_duplicate"), (pl.col("_mp_street") | pl.col("_mp_cell")).alias("multi_parcel_candidate"), (pl.col("parcel_id").is_not_null() & (pl.col("parcel_sales_2021_2026") > MAX_PARCEL_SALES)).alias("shared_parcel"), pl.when(pl.col("parcel_id").is_null()).then(None).otherwise(pl.col("parcel_sales_2021_2026")).alias("parcel_sales_2021_2026"), ) # assessment-ratio deviation from the year × class × region median. Basis = the provider's # `valeur_role` (assessment in force at the sale date, see results/cleaning/provider_vs_roll_fields.csv), # falling back to the current roll value. df = df.with_columns(pl.coalesce(pl.col("provider_assessment"), pl.col("roll_total_value")).alias("assessment_at_sale")) df = df.with_columns((pl.col("price") / pl.col("assessment_at_sale")).log().alias("_lr")) df = df.with_columns(pl.col("_lr").median().over(["year", "property_class", "region_code"]).alias("_lr_med")) df = df.with_columns( pl.when(pl.col("assessment_at_sale") > 0).then((pl.col("_lr") - pl.col("_lr_med")).abs() > RATIO_LOG_BAND).otherwise(False).alias("assessment_ratio_outlier"), (pl.col("price") > PRICE_CAP).alias("price_above_cap"), (pl.col("roll_floor_area_m2").is_null() | (pl.col("roll_floor_area_m2") <= 0)).alias("floor_area_missing"), pl.col("roll_year_built").is_null().alias("year_built_missing"), ) reasons = [ ("exact_duplicate", "exact_duplicate"), ("same_parcel_same_date_diff_price", "same_parcel_same_date_diff_price"), ("built_after_sale", "built_after_sale"), ("land", "roll_is_land"), ("non_residential", "roll_non_residential"), ("shared_parcel", "shared_parcel"), ("multi_parcel_candidate", "multi_parcel_candidate"), ("price_above_cap", "price_above_cap"), ("assessment_ratio_outlier", "assessment_ratio_outlier"), ] expr = pl.lit(None, dtype=pl.Utf8) for label, col in reversed(reasons): expr = pl.when(pl.col(col).fill_null(False)).then(pl.lit(label)).otherwise(expr) df = df.with_columns(expr.alias("exclude_reason")) df = df.with_columns( pl.when(pl.col("exclude_reason").is_not_null()).then(pl.col("exclude_reason")) .when(~pl.col("property_class").fill_null("none").is_in(["house", "condo", "plex"])).then(pl.lit("class_not_in_headline")) .when(~pl.col("roll_matched")).then(pl.lit("no_roll_characteristics")) .otherwise(None).alias("exclude_reason_headline") ).with_columns(pl.col("exclude_reason_headline").is_null().alias("in_headline_sample")) return df.drop([c for c in df.columns if c.startswith("_")]) def analytic(df: pl.DataFrame) -> pl.DataFrame: """Headline analytic layer with modelling columns (ages, logs, imputation inputs).""" a = df.filter(pl.col("in_headline_sample")).with_columns( (pl.col("year") - pl.col("roll_year_built")).clip(0, None).alias("age"), pl.col("roll_floor_area_m2").alias("floor_area_m2"), pl.col("roll_land_area_m2").alias("land_area_m2"), pl.col("roll_storeys").alias("storeys"), pl.col("roll_n_units").alias("n_units"), pl.col("roll_genre").fill_null("unknown").alias("genre"), ) return a