1#!/usr/bin/env python32# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca3"""Phase 6 — prospective evaluation set → analytic contract + runner bundle.45Input : data/raw/prospective_<date>.parquet (+ prospective_units_<date>.parquet) written by scripts/60_fetch_prospective.py6 (sales present in the live Vrai Prix DB but absent from the 2026-10-21 snapshot, joined to their roll unit).7Output: data/processed/prospective_analytic.parquet — ANALYTIC_COLUMNS contract, every row validated8 results/data/prospective_summary.json — funnel, date range, class / region counts9 data/processed/prospective_bundle/{analytic,splits,llm_subsamples,geo_features}.parquet10 analytic = snapshot analytic ∪ prospective rows (history needed for comparables / market context)11 splits = snapshot splits ∪ prospective rows labelled "prospective"12 llm_subsamples = snapshot subsamples ∪ stratified draw of the prospective rows (same strata rule and seed)13 geo_features = snapshot geo cache ∪ geo features of the prospective rows14 The bundle is consumed by `scripts/30_run_arm.py --data-dir data/processed/prospective_bundle` and by15 scripts/62_prospective_baselines.py; the frozen protocol (prompts, configs, roles, aggregation) is untouched.1617Cleaning re-implements, minimally, the WP 22 rules (sharp-hpi-quebec/src/sharp_hpi/cleaning.py) on these rows:18classification from the roll (CUBF 1000 × n_units × physical link; provider label as fallback), exclusions19(duplicates, same-parcel/same-date price conflicts, built after sale, land, non-residential, shared parcel,20multi-parcel deed candidates, price cap, assessment-ratio outliers), then the WP 26 funnel of data/build.py21(classes house/condo/plex, coordinates, parcel id, price band, roll match). Geography is looked up from the snapshot22analytic table by municipal code (the snapshot used a spatial join; the roll's code agrees in > 99 % of cases).23No address is ever written or printed.24"""25from __future__ import annotations2627import argparse28import json29import sys30import time31from datetime import date32from pathlib import Path33from typing import Any3435import numpy as np36import polars as pl3738sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))39from agentic_valuation.config import PROCESSED, RAW, RESULTS, SEED, load_config # noqa: E40240from agentic_valuation.data.analytic_schema import ( # noqa: E40241 coerce_analytic, grid_id_expr, project_xy, region_group_expr, validate_analytic,42)43from agentic_valuation.data.build import roll_reference_at_sale_expr # noqa: E40244from agentic_valuation.data.splits import stratified_subsample # noqa: E4024546# WP 22 constants (cleaning.py)47HOUSE_LINKS = {"Détaché", "Jumelé", "En rangée 1 côté", "En rangée plus de 1 côté"}48PROVIDER_MAP = {"unifamilial": "house", "condo": "condo", "plex": "plex"}49RESIDENTIAL_CUBF = {1000, 1100, 1211}50MAX_PARCEL_SALES = 1051RATIO_LOG_BAND = 1.3862943611198906 # log(4)52PRICE_CAP = 10_000_000.053HEADLINE_CLASSES = ("house", "condo", "plex")54BUNDLE = PROCESSED / "prospective_bundle"55PROSPECTIVE_SUBSAMPLE_N = 40565758def latest_stamp() -> str:59 mans = sorted(RAW.glob("prospective_*.manifest.json"))60 if not mans:61 sys.exit("no data/raw/prospective_<date>.manifest.json: run scripts/60_fetch_prospective.py first")62 return mans[-1].name[len("prospective_") : -len(".manifest.json")]636465# ----------------------------------------------------------------------------------------------------------------66# WP 22 standardisation + classification (subset needed for the contract)67# ----------------------------------------------------------------------------------------------------------------68def standardize(tx: pl.DataFrame, units: pl.DataFrame) -> pl.DataFrame:69 u = units.select(70 "id_provinc",71 pl.col("cubf").cast(pl.Int64), pl.col("type_prop").alias("roll_type"),72 pl.col("annee_construction").cast(pl.Int64).alias("roll_year_built"),73 pl.col("aire_etages_m2").cast(pl.Float64).alias("roll_floor_area_m2"),74 pl.col("superficie_terrain_m2").cast(pl.Float64).alias("roll_land_area_m2"),75 pl.col("nb_etages").cast(pl.Float64).alias("roll_storeys"), pl.col("nb_logements").cast(pl.Int64).alias("roll_n_units"),76 pl.col("lien_physique").alias("roll_link"), pl.col("genre_construction").alias("roll_genre"),77 pl.col("valeur_terrain").cast(pl.Float64).alias("roll_land_value"), pl.col("valeur_batiment").cast(pl.Float64).alias("roll_building_value"),78 pl.col("valeur_role").cast(pl.Float64).alias("roll_total_value"), pl.col("dat_cond_marche").alias("roll_reference_date"),79 pl.col("code_mun").alias("mun_code"), pl.lit(True).alias("roll_matched"),80 ).unique(subset=["id_provinc"], keep="first")81 df = (82 tx.rename({"id": "tx_id", "id_provinc": "parcel_id", "amount": "price", "property_type": "provider_type",83 "year_built": "provider_year_built", "floor_area": "provider_floor_area", "land_area": "provider_land_area",84 "valeur_role": "provider_assessment"})85 .with_columns(86 pl.when(pl.col("parcel_id") == "").then(None).otherwise(pl.col("parcel_id")).alias("parcel_id"),87 pl.col("date").str.to_date("%Y-%m-%d"), pl.col("price").cast(pl.Float64),88 )89 .join(u, left_on="parcel_id", right_on="id_provinc", how="left")90 .with_columns(pl.col("date").dt.year().alias("year"), pl.col("roll_matched").fill_null(False))91 )92 return df939495def classify(df: pl.DataFrame) -> pl.DataFrame:96 """Verbatim WP 22 precedence: roll class first, provider label as fallback; subtype from the physical link."""97 cubf_s = pl.col("cubf").cast(pl.Utf8)98 is_land = (pl.col("roll_type") == "terrain") | cubf_s.str.starts_with("9") | cubf_s.str.starts_with("4")99 roll_class = (100 pl.when(pl.col("cubf") == 1100).then(pl.lit("cottage"))101 .when(pl.col("cubf") == 1211).then(pl.lit("mobile"))102 .when((pl.col("cubf") == 1000) & (pl.col("roll_n_units") >= 6)).then(pl.lit("multi"))103 .when((pl.col("cubf") == 1000) & (pl.col("roll_n_units").is_between(2, 5))).then(pl.lit("plex"))104 .when((pl.col("cubf") == 1000) & (pl.col("roll_n_units") <= 1) & (pl.col("roll_link") == "Intégré")).then(pl.lit("condo"))105 .when((pl.col("cubf") == 1000) & (pl.col("roll_n_units") <= 1) & pl.col("roll_link").is_in(list(HOUSE_LINKS))).then(pl.lit("house"))106 .otherwise(None)107 )108 provider_class = pl.col("provider_type").replace_strict(PROVIDER_MAP, default=None)109 df = df.with_columns(110 roll_class.alias("roll_class"), provider_class.alias("provider_class"),111 (pl.col("roll_year_built") > pl.col("year")).alias("built_after_sale"),112 is_land.alias("roll_is_land"),113 (pl.col("cubf").is_not_null() & ~pl.col("cubf").is_in(list(RESIDENTIAL_CUBF)) & ~is_land).alias("roll_non_residential"),114 )115 return df.with_columns(116 pl.when(pl.col("roll_class").is_not_null()).then(pl.col("roll_class")).otherwise(pl.col("provider_class")).alias("property_class"),117 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"),118 pl.when(pl.col("roll_class") == "house").then(119 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"))120 ).when(pl.col("roll_class") == "condo").then(pl.lit("integrated")).otherwise(pl.lit("unknown")).alias("subtype"),121 )122123124def add_flags(df: pl.DataFrame, snap: pl.DataFrame) -> pl.DataFrame:125 """WP 22 exclusion flags, computed on prospective ∪ snapshot rows of the same dates (a multi-parcel deed or a126 same-day price conflict can straddle the snapshot date), plus the cross-snapshot duplicate check."""127 cols = ["tx_id", "parcel_id", "date", "price", "street", "city", "lat", "lng"]128 recent = snap.select(cols).filter(pl.col("date") >= df["date"].min()).with_columns(pl.lit("snapshot").alias("_src"))129 both = pl.concat([df.select(cols).with_columns(pl.lit("prospective").alias("_src")), recent], how="vertical_relaxed")130 both = both.with_columns(131 pl.col("street").str.replace(r"^\s*\d+[A-Za-z]?(?:-\d+)?[,\s]+", "").str.to_lowercase().alias("_street_name"),132 (pl.col("lat") * 100).round(0).alias("_lat_c"), (pl.col("lng") * 100).round(0).alias("_lng_c"),133 ).with_columns(134 (pl.col("parcel_id").is_not_null() & (pl.col("price").n_unique().over(["parcel_id", "date"]) > 1)).alias("same_parcel_same_date_diff_price"),135 ((pl.len().over(["date", "price", "city", "_street_name"]) > 1) | (pl.len().over(["date", "price", "_lat_c", "_lng_c"]) > 1)).alias("multi_parcel_candidate"),136 ).filter(pl.col("_src") == "prospective").select("tx_id", "same_parcel_same_date_diff_price", "multi_parcel_candidate")137 df = df.join(both, on="tx_id", how="left")138 # exact duplicates inside the prospective set (later rows) and against the whole snapshot (same parcel / date / price)139 df = df.sort(["parcel_id", "date", "price", "tx_id"]).with_columns(140 (pl.col("parcel_id").is_not_null() & (pl.int_range(pl.len()).over(["parcel_id", "date", "price"]) > 0)).alias("exact_duplicate"),141 )142 snap_keys = snap.select("parcel_id", "date", "price").drop_nulls().unique().with_columns(pl.lit(True).alias("duplicate_of_snapshot"))143 df = df.join(snap_keys, on=["parcel_id", "date", "price"], how="left").with_columns(pl.col("duplicate_of_snapshot").fill_null(False))144 # shared parcels: sales of the parcel over 2021-2026 (snapshot ∪ prospective)145 n_snap = snap.group_by("parcel_id").len().rename({"len": "_n_snap"})146 df = df.join(n_snap, on="parcel_id", how="left").with_columns(147 (pl.col("_n_snap").fill_null(0) + pl.len().over("parcel_id")).alias("parcel_sales_2021_2026")148 ).with_columns((pl.col("parcel_id").is_not_null() & (pl.col("parcel_sales_2021_2026") > MAX_PARCEL_SALES)).alias("shared_parcel"))149 # assessment ratio vs the snapshot's 2026 median by class × region (WP 22: year × class × region)150 df = df.with_columns(pl.coalesce(pl.col("provider_assessment"), pl.col("roll_total_value")).alias("assessment_at_sale"))151 med = (152 snap.filter(pl.col("year") == df["year"].max()).filter(pl.col("assessment_at_sale") > 0)153 .with_columns((pl.col("price") / pl.col("assessment_at_sale")).log().alias("_lr"))154 .group_by(["property_class", "region_name"]).agg(pl.col("_lr").median().alias("_lr_med"))155 )156 med_cls = snap.filter(pl.col("assessment_at_sale") > 0).with_columns((pl.col("price") / pl.col("assessment_at_sale")).log().alias("_lr")).group_by("property_class").agg(pl.col("_lr").median().alias("_lr_med_cls"))157 df = df.join(med, on=["property_class", "region_name"], how="left").join(med_cls, on="property_class", how="left")158 df = df.with_columns(159 pl.when(pl.col("assessment_at_sale") > 0).then(((pl.col("price") / pl.col("assessment_at_sale")).log() - pl.coalesce(pl.col("_lr_med"), pl.col("_lr_med_cls"))).abs() > RATIO_LOG_BAND).otherwise(False).fill_null(False).alias("assessment_ratio_outlier"),160 (pl.col("price") > PRICE_CAP).alias("price_above_cap"),161 )162 reasons = [163 ("exact_duplicate", "exact_duplicate"), ("duplicate_of_snapshot", "duplicate_of_snapshot"),164 ("same_parcel_same_date_diff_price", "same_parcel_same_date_diff_price"), ("built_after_sale", "built_after_sale"),165 ("land", "roll_is_land"), ("non_residential", "roll_non_residential"), ("shared_parcel", "shared_parcel"),166 ("multi_parcel_candidate", "multi_parcel_candidate"), ("price_above_cap", "price_above_cap"),167 ("assessment_ratio_outlier", "assessment_ratio_outlier"),168 ]169 expr = pl.lit(None, dtype=pl.Utf8)170 for label, col in reversed(reasons):171 expr = pl.when(pl.col(col).fill_null(False)).then(pl.lit(label)).otherwise(expr)172 df = df.with_columns(expr.alias("wp22_exclude_reason"))173 return df.drop([c for c in df.columns if c.startswith("_")])174175176# ----------------------------------------------------------------------------------------------------------------177# WP 26 contract178# ----------------------------------------------------------------------------------------------------------------179def build(stamp: str, cfg: dict[str, Any]) -> tuple[pl.DataFrame, dict[str, Any]]:180 tx = pl.read_parquet(RAW / f"prospective_{stamp}.parquet")181 units_path = RAW / f"prospective_units_{stamp}.parquet"182 units = pl.read_parquet(units_path) if units_path.exists() else pl.DataFrame(schema={"id_provinc": pl.String})183 snap = pl.read_parquet(RAW / "wp22_clean_transactions.parquet", columns=[184 "tx_id", "parcel_id", "date", "price", "street", "city", "lat", "lng", "year", "property_class", "region_name", "assessment_at_sale",185 ])186 analytic = pl.read_parquet(PROCESSED / "analytic.parquet")187 s = cfg["sample"]188 funnel: list[dict[str, Any]] = [{"step": "raw_new_rows", "rows": tx.height, "dropped": 0}]189190 df = standardize(tx, units)191 assert not df["tx_id"].is_in(snap["tx_id"].implode()).any(), "prospective tx_id present in the snapshot"192 # geography lookup by municipal code (unique mapping in the snapshot analytic table, verified)193 lookup = analytic.group_by("mun_code").agg(pl.col("mun_name").first(), pl.col("region_name").first(), pl.col("urban_class").first())194 df = df.join(lookup, on="mun_code", how="left")195 df = classify(df)196 df = add_flags(df, snap)197198 def step(name: str, expr: pl.Expr) -> None:199 nonlocal df200 before = df.height201 df = df.filter(expr)202 funnel.append({"step": name, "rows": df.height, "dropped": before - df.height})203204 excl = {k: int(v) for k, v in df.group_by("wp22_exclude_reason").len().iter_rows() if k is not None}205 step("roll_matched", pl.col("roll_matched"))206 step("wp22_no_exclusion_flag", pl.col("wp22_exclude_reason").is_null())207 step("property_class_in_scope", pl.col("property_class").is_in(s["property_classes"]))208 step("coordinates_present", pl.col("lat").is_not_null() & pl.col("lng").is_not_null())209 step("parcel_id_present", pl.col("parcel_id").is_not_null())210 step("price_in_range", (pl.col("price") >= s["price_min"]) & (pl.col("price") <= s["price_max"]))211 step("municipality_in_snapshot_lookup", pl.col("mun_name").is_not_null() & pl.col("region_name").is_not_null() & pl.col("urban_class").is_not_null())212213 # roll attributes as in data/build.py (assessment in force at the sale; land/building split rescaled; triennial rule)214 scale = pl.when(pl.col("roll_total_value") > 0).then(pl.col("assessment_at_sale") / pl.col("roll_total_value")).otherwise(None)215 df = df.with_columns(216 pl.coalesce(pl.col("roll_floor_area_m2"), pl.col("provider_floor_area")).alias("floor_area_m2"),217 pl.coalesce(pl.col("roll_land_area_m2"), pl.col("provider_land_area")).alias("land_area_m2"),218 pl.coalesce(pl.col("roll_year_built"), pl.col("provider_year_built")).cast(pl.Int64).alias("year_built"),219 pl.col("roll_storeys").alias("storeys"), pl.col("roll_n_units").alias("n_units"),220 (pl.col("roll_land_value") * scale).round(0).alias("roll_land_value"),221 (pl.col("roll_building_value") * scale).round(0).alias("roll_building_value"),222 pl.coalesce(pl.col("assessment_at_sale"), pl.col("roll_total_value")).alias("roll_total_value"),223 roll_reference_at_sale_expr().alias("roll_reference_date"),224 pl.lit(None, dtype=pl.String).alias("exclude_reason"),225 ).with_columns((pl.col("year_built") >= pl.col("year")).fill_null(False).alias("new_construction"))226 # projection, grids, region group, borough (mode of the snapshot sales in the same municipality × 1 km cell)227 x, y = project_xy(df["lat"].to_numpy(), df["lng"].to_numpy())228 df = df.with_columns(pl.Series("x_m", x, dtype=pl.Float64), pl.Series("y_m", y, dtype=pl.Float64)).with_columns(229 grid_id_expr("x_m", "y_m", 1000).alias("grid1k"), grid_id_expr("x_m", "y_m", 5000).alias("grid5k"),230 region_group_expr().alias("region_group"), pl.col("price").log().alias("log_price"),231 )232 borough = (233 analytic.filter(pl.col("borough_name").is_not_null()).group_by(["mun_code", "grid1k", "borough_name"]).len()234 .sort(["mun_code", "grid1k", "len", "borough_name"], descending=[False, False, True, False])235 .unique(subset=["mun_code", "grid1k"], keep="first").select("mun_code", "grid1k", "borough_name")236 )237 df = df.join(borough, on=["mun_code", "grid1k"], how="left")238 # prior sale: strictly earlier sale of the same parcel in the full snapshot ∪ the prospective rows239 hist = pl.concat([snap.select("parcel_id", "date", "price"), df.select("parcel_id", "date", "price")]).drop_nulls("parcel_id")240 hist = hist.sort(["parcel_id", "date", "price"]).rename({"date": "_pdate", "price": "_pprice"})241 df = df.with_row_index("_row").sort(["parcel_id", "date"])242 df = df.join_asof(hist, left_on="date", right_on="_pdate", by="parcel_id", strategy="backward", allow_exact_matches=False, check_sortedness=False)243 df = df.sort("_row").drop("_row").rename({"_pprice": "prior_sale_price", "_pdate": "prior_sale_date"})244 # price tercile from the snapshot's class-specific cut points (linear quantiles of the snapshot analytic table)245 cuts = analytic.group_by("property_class").agg(pl.col("price").quantile(1 / 3, interpolation="linear").alias("q33"), pl.col("price").quantile(2 / 3, interpolation="linear").alias("q67"))246 df = df.join(cuts, on="property_class", how="left").with_columns(247 pl.when(pl.col("price") <= pl.col("q33")).then(1).when(pl.col("price") <= pl.col("q67")).then(2).otherwise(3).cast(pl.Int8).alias("price_tercile")248 )249 out = coerce_analytic(df)250 validate_analytic(out)251252 def counts(col: str) -> dict[str, int]:253 return {str(k): int(v) for k, v in out.group_by(col).len().sort(col).iter_rows()}254255 summary = {256 "stamp": stamp, "built_at": date.today().isoformat(),257 "manifest": json.loads((RAW / f"prospective_{stamp}.manifest.json").read_text()),258 "units_rows": units.height, "funnel": funnel, "wp22_exclusion_flags": excl,259 "class_source": {str(k): int(v) for k, v in df.group_by("class_source").len().iter_rows()},260 "rows": out.height, "parcels": int(out["parcel_id"].n_unique()),261 "date_min": out["date"].min().isoformat() if out.height else None, "date_max": out["date"].max().isoformat() if out.height else None,262 "by_property_class": counts("property_class"), "by_subtype": counts("subtype"), "by_region_group": counts("region_group"),263 "by_region_name": counts("region_name"), "by_urban_class": counts("urban_class"), "by_price_tercile": counts("price_tercile"),264 "by_roll_reference_date": counts("roll_reference_date"),265 "price_tercile_cuts_snapshot": {r["property_class"]: [r["q33"], r["q67"]] for r in cuts.sort("property_class").iter_rows(named=True)},266 "rows_with_prior_sale": int(out["prior_sale_price"].is_not_null().sum()),267 "rows_new_construction": int(out["new_construction"].sum()),268 "share_floor_area_from_provider": float((df["roll_floor_area_m2"].is_null() & df["provider_floor_area"].is_not_null()).mean()) if df.height else None,269 "share_borough_imputed_from_grid": float(out["borough_name"].is_not_null().mean()) if out.height else None,270 "price_median": float(out["price"].median()) if out.height else None,271 "missingness_pct": {c: round(100.0 * out[c].null_count() / max(out.height, 1), 3) for c in out.columns},272 "snapshot_analytic_rows": analytic.height,273 }274 return out, summary275276277def write_bundle(pros: pl.DataFrame, cfg: dict[str, Any]) -> dict[str, Any]:278 BUNDLE.mkdir(parents=True, exist_ok=True)279 analytic = pl.read_parquet(PROCESSED / "analytic.parquet")280 assert not pros["tx_id"].is_in(analytic["tx_id"].implode()).any()281 union = pl.concat([analytic, pros.select(analytic.columns)]).sort(["date", "tx_id"])282 validate_analytic(union)283 union.write_parquet(BUNDLE / "analytic.parquet", compression="zstd")284 splits = pl.read_parquet(PROCESSED / "splits.parquet")285 regions = list(cfg.get("spatial_holdout", {}).get("regions", []))286 p_splits = pros.select("tx_id", pl.lit("prospective").alias("split"), pl.col("region_name").is_in(regions).alias("spatial_holdout_region"))287 u_splits = pl.concat([splits, p_splits.select(splits.columns)])288 u_splits.write_parquet(BUNDLE / "splits.parquet")289 subs = pl.read_parquet(PROCESSED / "llm_subsamples.parquet")290 n = min(PROSPECTIVE_SUBSAMPLE_N, pros.height)291 p_subs = stratified_subsample(union, u_splits, cfg, n_by_split={"prospective": n}, seed=cfg.get("seed", SEED))292 u_subs = pl.concat([subs, p_subs.select(subs.columns)])293 u_subs.write_parquet(BUNDLE / "llm_subsamples.parquet")294 geo_info: dict[str, Any] = {"geo_rows_prospective": 0}295 geo_path = PROCESSED / "geo_features.parquet"296 try:297 from agentic_valuation.features.geo import GeoLayers, geo_features_batch298299 if GeoLayers.available() and geo_path.exists():300 layers = GeoLayers.load()301 g = geo_features_batch(pros.select("tx_id", "lat", "lng", "x_m", "y_m"), layers)302 snap_geo = pl.read_parquet(geo_path)303 pl.concat([snap_geo, g.select(snap_geo.columns)]).write_parquet(BUNDLE / "geo_features.parquet", compression="zstd")304 geo_info = {"geo_rows_prospective": g.height, "share_flood_mapped": float(g["flood_mapped"].mean()), "share_with_da": float(g["dauid"].is_not_null().mean())}305 except Exception as e: # geo is optional for the runner; report instead of failing306 geo_info = {"geo_rows_prospective": 0, "geo_error": f"{type(e).__name__}: {e}"[:200]}307 return {308 "dir": str(BUNDLE.relative_to(PROCESSED.parents[1])), "analytic_rows": union.height,309 "splits": {k: int(v) for k, v in u_splits.group_by("split").len().iter_rows()},310 "llm_subsamples": {k: int(v) for k, v in u_subs.group_by("split").len().iter_rows()},311 "prospective_subsample_n": p_subs.height, "prospective_subsample_strata": int(p_subs["stratum"].n_unique()),312 "prospective_subsample_by_stratum": {k: int(v) for k, v in p_subs.group_by("stratum").len().sort("stratum").iter_rows()},313 **geo_info,314 }315316317def main() -> None:318 ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)319 ap.add_argument("--date", default=None, help="stamp of data/raw/prospective_<date>.parquet (default: latest)")320 ap.add_argument("--no-bundle", action="store_true")321 a = ap.parse_args()322 t0 = time.perf_counter()323 cfg = load_config("data")324 stamp = a.date or latest_stamp()325 out, summary = build(stamp, cfg)326 out.write_parquet(PROCESSED / "prospective_analytic.parquet", compression="zstd")327 summary["output"] = "data/processed/prospective_analytic.parquet"328 if not a.no_bundle and out.height:329 summary["bundle"] = write_bundle(out, cfg)330 summary["seconds"] = round(time.perf_counter() - t0, 1)331 (RESULTS / "data").mkdir(parents=True, exist_ok=True)332 (RESULTS / "data" / "prospective_summary.json").write_text(json.dumps(summary, indent=1, ensure_ascii=False, default=str))333 print(json.dumps({k: summary[k] for k in ("rows", "date_min", "date_max", "funnel", "wp22_exclusion_flags", "by_property_class", "by_region_group", "bundle", "seconds") if k in summary}, ensure_ascii=False, indent=1))334335336if __name__ == "__main__":337 main()338