#!/usr/bin/env python3
# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca
"""Phase 6 — prospective evaluation set → analytic contract + runner bundle.

Input : data/raw/prospective_<date>.parquet (+ prospective_units_<date>.parquet) written by scripts/60_fetch_prospective.py
        (sales present in the live Vrai Prix DB but absent from the 2026-10-21 snapshot, joined to their roll unit).
Output: data/processed/prospective_analytic.parquet       — ANALYTIC_COLUMNS contract, every row validated
        results/data/prospective_summary.json             — funnel, date range, class / region counts
        data/processed/prospective_bundle/{analytic,splits,llm_subsamples,geo_features}.parquet
            analytic       = snapshot analytic ∪ prospective rows (history needed for comparables / market context)
            splits         = snapshot splits ∪ prospective rows labelled "prospective"
            llm_subsamples = snapshot subsamples ∪ stratified draw of the prospective rows (same strata rule and seed)
            geo_features   = snapshot geo cache ∪ geo features of the prospective rows
        The bundle is consumed by `scripts/30_run_arm.py --data-dir data/processed/prospective_bundle` and by
        scripts/62_prospective_baselines.py; the frozen protocol (prompts, configs, roles, aggregation) is untouched.

Cleaning re-implements, minimally, the WP 22 rules (sharp-hpi-quebec/src/sharp_hpi/cleaning.py) on these rows:
classification from the roll (CUBF 1000 × n_units × physical link; provider label as fallback), exclusions
(duplicates, same-parcel/same-date price conflicts, built after sale, land, non-residential, shared parcel,
multi-parcel deed candidates, price cap, assessment-ratio outliers), then the WP 26 funnel of data/build.py
(classes house/condo/plex, coordinates, parcel id, price band, roll match). Geography is looked up from the snapshot
analytic table by municipal code (the snapshot used a spatial join; the roll's code agrees in > 99 % of cases).
No address is ever written or printed.
"""
from __future__ import annotations

import argparse
import json
import sys
import time
from datetime import date
from pathlib import Path
from typing import Any

import numpy as np
import polars as pl

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from agentic_valuation.config import PROCESSED, RAW, RESULTS, SEED, load_config  # noqa: E402
from agentic_valuation.data.analytic_schema import (  # noqa: E402
    coerce_analytic, grid_id_expr, project_xy, region_group_expr, validate_analytic,
)
from agentic_valuation.data.build import roll_reference_at_sale_expr  # noqa: E402
from agentic_valuation.data.splits import stratified_subsample  # noqa: E402

# WP 22 constants (cleaning.py)
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"}
RESIDENTIAL_CUBF = {1000, 1100, 1211}
MAX_PARCEL_SALES = 10
RATIO_LOG_BAND = 1.3862943611198906  # log(4)
PRICE_CAP = 10_000_000.0
HEADLINE_CLASSES = ("house", "condo", "plex")
BUNDLE = PROCESSED / "prospective_bundle"
PROSPECTIVE_SUBSAMPLE_N = 40


def latest_stamp() -> str:
    mans = sorted(RAW.glob("prospective_*.manifest.json"))
    if not mans:
        sys.exit("no data/raw/prospective_<date>.manifest.json: run scripts/60_fetch_prospective.py first")
    return mans[-1].name[len("prospective_") : -len(".manifest.json")]


# ----------------------------------------------------------------------------------------------------------------
# WP 22 standardisation + classification (subset needed for the contract)
# ----------------------------------------------------------------------------------------------------------------
def standardize(tx: pl.DataFrame, units: pl.DataFrame) -> pl.DataFrame:
    u = units.select(
        "id_provinc",
        pl.col("cubf").cast(pl.Int64), pl.col("type_prop").alias("roll_type"),
        pl.col("annee_construction").cast(pl.Int64).alias("roll_year_built"),
        pl.col("aire_etages_m2").cast(pl.Float64).alias("roll_floor_area_m2"),
        pl.col("superficie_terrain_m2").cast(pl.Float64).alias("roll_land_area_m2"),
        pl.col("nb_etages").cast(pl.Float64).alias("roll_storeys"), pl.col("nb_logements").cast(pl.Int64).alias("roll_n_units"),
        pl.col("lien_physique").alias("roll_link"), pl.col("genre_construction").alias("roll_genre"),
        pl.col("valeur_terrain").cast(pl.Float64).alias("roll_land_value"), pl.col("valeur_batiment").cast(pl.Float64).alias("roll_building_value"),
        pl.col("valeur_role").cast(pl.Float64).alias("roll_total_value"), pl.col("dat_cond_marche").alias("roll_reference_date"),
        pl.col("code_mun").alias("mun_code"), pl.lit(True).alias("roll_matched"),
    ).unique(subset=["id_provinc"], keep="first")
    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", "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"),
            pl.col("date").str.to_date("%Y-%m-%d"), pl.col("price").cast(pl.Float64),
        )
        .join(u, left_on="parcel_id", right_on="id_provinc", how="left")
        .with_columns(pl.col("date").dt.year().alias("year"), pl.col("roll_matched").fill_null(False))
    )
    return df


def classify(df: pl.DataFrame) -> pl.DataFrame:
    """Verbatim WP 22 precedence: roll class first, provider label as fallback; subtype from the physical link."""
    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"),
        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"),
    )
    return 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.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(pl.lit("unknown")).alias("subtype"),
    )


def add_flags(df: pl.DataFrame, snap: pl.DataFrame) -> pl.DataFrame:
    """WP 22 exclusion flags, computed on prospective ∪ snapshot rows of the same dates (a multi-parcel deed or a
    same-day price conflict can straddle the snapshot date), plus the cross-snapshot duplicate check."""
    cols = ["tx_id", "parcel_id", "date", "price", "street", "city", "lat", "lng"]
    recent = snap.select(cols).filter(pl.col("date") >= df["date"].min()).with_columns(pl.lit("snapshot").alias("_src"))
    both = pl.concat([df.select(cols).with_columns(pl.lit("prospective").alias("_src")), recent], how="vertical_relaxed")
    both = both.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"),
    ).with_columns(
        (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) | (pl.len().over(["date", "price", "_lat_c", "_lng_c"]) > 1)).alias("multi_parcel_candidate"),
    ).filter(pl.col("_src") == "prospective").select("tx_id", "same_parcel_same_date_diff_price", "multi_parcel_candidate")
    df = df.join(both, on="tx_id", how="left")
    # exact duplicates inside the prospective set (later rows) and against the whole snapshot (same parcel / date / price)
    df = df.sort(["parcel_id", "date", "price", "tx_id"]).with_columns(
        (pl.col("parcel_id").is_not_null() & (pl.int_range(pl.len()).over(["parcel_id", "date", "price"]) > 0)).alias("exact_duplicate"),
    )
    snap_keys = snap.select("parcel_id", "date", "price").drop_nulls().unique().with_columns(pl.lit(True).alias("duplicate_of_snapshot"))
    df = df.join(snap_keys, on=["parcel_id", "date", "price"], how="left").with_columns(pl.col("duplicate_of_snapshot").fill_null(False))
    # shared parcels: sales of the parcel over 2021-2026 (snapshot ∪ prospective)
    n_snap = snap.group_by("parcel_id").len().rename({"len": "_n_snap"})
    df = df.join(n_snap, on="parcel_id", how="left").with_columns(
        (pl.col("_n_snap").fill_null(0) + pl.len().over("parcel_id")).alias("parcel_sales_2021_2026")
    ).with_columns((pl.col("parcel_id").is_not_null() & (pl.col("parcel_sales_2021_2026") > MAX_PARCEL_SALES)).alias("shared_parcel"))
    # assessment ratio vs the snapshot's 2026 median by class × region (WP 22: year × class × region)
    df = df.with_columns(pl.coalesce(pl.col("provider_assessment"), pl.col("roll_total_value")).alias("assessment_at_sale"))
    med = (
        snap.filter(pl.col("year") == df["year"].max()).filter(pl.col("assessment_at_sale") > 0)
        .with_columns((pl.col("price") / pl.col("assessment_at_sale")).log().alias("_lr"))
        .group_by(["property_class", "region_name"]).agg(pl.col("_lr").median().alias("_lr_med"))
    )
    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"))
    df = df.join(med, on=["property_class", "region_name"], how="left").join(med_cls, on="property_class", how="left")
    df = df.with_columns(
        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"),
        (pl.col("price") > PRICE_CAP).alias("price_above_cap"),
    )
    reasons = [
        ("exact_duplicate", "exact_duplicate"), ("duplicate_of_snapshot", "duplicate_of_snapshot"),
        ("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("wp22_exclude_reason"))
    return df.drop([c for c in df.columns if c.startswith("_")])


# ----------------------------------------------------------------------------------------------------------------
# WP 26 contract
# ----------------------------------------------------------------------------------------------------------------
def build(stamp: str, cfg: dict[str, Any]) -> tuple[pl.DataFrame, dict[str, Any]]:
    tx = pl.read_parquet(RAW / f"prospective_{stamp}.parquet")
    units_path = RAW / f"prospective_units_{stamp}.parquet"
    units = pl.read_parquet(units_path) if units_path.exists() else pl.DataFrame(schema={"id_provinc": pl.String})
    snap = pl.read_parquet(RAW / "wp22_clean_transactions.parquet", columns=[
        "tx_id", "parcel_id", "date", "price", "street", "city", "lat", "lng", "year", "property_class", "region_name", "assessment_at_sale",
    ])
    analytic = pl.read_parquet(PROCESSED / "analytic.parquet")
    s = cfg["sample"]
    funnel: list[dict[str, Any]] = [{"step": "raw_new_rows", "rows": tx.height, "dropped": 0}]

    df = standardize(tx, units)
    assert not df["tx_id"].is_in(snap["tx_id"].implode()).any(), "prospective tx_id present in the snapshot"
    # geography lookup by municipal code (unique mapping in the snapshot analytic table, verified)
    lookup = analytic.group_by("mun_code").agg(pl.col("mun_name").first(), pl.col("region_name").first(), pl.col("urban_class").first())
    df = df.join(lookup, on="mun_code", how="left")
    df = classify(df)
    df = add_flags(df, snap)

    def step(name: str, expr: pl.Expr) -> None:
        nonlocal df
        before = df.height
        df = df.filter(expr)
        funnel.append({"step": name, "rows": df.height, "dropped": before - df.height})

    excl = {k: int(v) for k, v in df.group_by("wp22_exclude_reason").len().iter_rows() if k is not None}
    step("roll_matched", pl.col("roll_matched"))
    step("wp22_no_exclusion_flag", pl.col("wp22_exclude_reason").is_null())
    step("property_class_in_scope", pl.col("property_class").is_in(s["property_classes"]))
    step("coordinates_present", pl.col("lat").is_not_null() & pl.col("lng").is_not_null())
    step("parcel_id_present", pl.col("parcel_id").is_not_null())
    step("price_in_range", (pl.col("price") >= s["price_min"]) & (pl.col("price") <= s["price_max"]))
    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())

    # roll attributes as in data/build.py (assessment in force at the sale; land/building split rescaled; triennial rule)
    scale = pl.when(pl.col("roll_total_value") > 0).then(pl.col("assessment_at_sale") / pl.col("roll_total_value")).otherwise(None)
    df = df.with_columns(
        pl.coalesce(pl.col("roll_floor_area_m2"), pl.col("provider_floor_area")).alias("floor_area_m2"),
        pl.coalesce(pl.col("roll_land_area_m2"), pl.col("provider_land_area")).alias("land_area_m2"),
        pl.coalesce(pl.col("roll_year_built"), pl.col("provider_year_built")).cast(pl.Int64).alias("year_built"),
        pl.col("roll_storeys").alias("storeys"), pl.col("roll_n_units").alias("n_units"),
        (pl.col("roll_land_value") * scale).round(0).alias("roll_land_value"),
        (pl.col("roll_building_value") * scale).round(0).alias("roll_building_value"),
        pl.coalesce(pl.col("assessment_at_sale"), pl.col("roll_total_value")).alias("roll_total_value"),
        roll_reference_at_sale_expr().alias("roll_reference_date"),
        pl.lit(None, dtype=pl.String).alias("exclude_reason"),
    ).with_columns((pl.col("year_built") >= pl.col("year")).fill_null(False).alias("new_construction"))
    # projection, grids, region group, borough (mode of the snapshot sales in the same municipality × 1 km cell)
    x, y = project_xy(df["lat"].to_numpy(), df["lng"].to_numpy())
    df = df.with_columns(pl.Series("x_m", x, dtype=pl.Float64), pl.Series("y_m", y, dtype=pl.Float64)).with_columns(
        grid_id_expr("x_m", "y_m", 1000).alias("grid1k"), grid_id_expr("x_m", "y_m", 5000).alias("grid5k"),
        region_group_expr().alias("region_group"), pl.col("price").log().alias("log_price"),
    )
    borough = (
        analytic.filter(pl.col("borough_name").is_not_null()).group_by(["mun_code", "grid1k", "borough_name"]).len()
        .sort(["mun_code", "grid1k", "len", "borough_name"], descending=[False, False, True, False])
        .unique(subset=["mun_code", "grid1k"], keep="first").select("mun_code", "grid1k", "borough_name")
    )
    df = df.join(borough, on=["mun_code", "grid1k"], how="left")
    # prior sale: strictly earlier sale of the same parcel in the full snapshot ∪ the prospective rows
    hist = pl.concat([snap.select("parcel_id", "date", "price"), df.select("parcel_id", "date", "price")]).drop_nulls("parcel_id")
    hist = hist.sort(["parcel_id", "date", "price"]).rename({"date": "_pdate", "price": "_pprice"})
    df = df.with_row_index("_row").sort(["parcel_id", "date"])
    df = df.join_asof(hist, left_on="date", right_on="_pdate", by="parcel_id", strategy="backward", allow_exact_matches=False, check_sortedness=False)
    df = df.sort("_row").drop("_row").rename({"_pprice": "prior_sale_price", "_pdate": "prior_sale_date"})
    # price tercile from the snapshot's class-specific cut points (linear quantiles of the snapshot analytic table)
    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"))
    df = df.join(cuts, on="property_class", how="left").with_columns(
        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")
    )
    out = coerce_analytic(df)
    validate_analytic(out)

    def counts(col: str) -> dict[str, int]:
        return {str(k): int(v) for k, v in out.group_by(col).len().sort(col).iter_rows()}

    summary = {
        "stamp": stamp, "built_at": date.today().isoformat(),
        "manifest": json.loads((RAW / f"prospective_{stamp}.manifest.json").read_text()),
        "units_rows": units.height, "funnel": funnel, "wp22_exclusion_flags": excl,
        "class_source": {str(k): int(v) for k, v in df.group_by("class_source").len().iter_rows()},
        "rows": out.height, "parcels": int(out["parcel_id"].n_unique()),
        "date_min": out["date"].min().isoformat() if out.height else None, "date_max": out["date"].max().isoformat() if out.height else None,
        "by_property_class": counts("property_class"), "by_subtype": counts("subtype"), "by_region_group": counts("region_group"),
        "by_region_name": counts("region_name"), "by_urban_class": counts("urban_class"), "by_price_tercile": counts("price_tercile"),
        "by_roll_reference_date": counts("roll_reference_date"),
        "price_tercile_cuts_snapshot": {r["property_class"]: [r["q33"], r["q67"]] for r in cuts.sort("property_class").iter_rows(named=True)},
        "rows_with_prior_sale": int(out["prior_sale_price"].is_not_null().sum()),
        "rows_new_construction": int(out["new_construction"].sum()),
        "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,
        "share_borough_imputed_from_grid": float(out["borough_name"].is_not_null().mean()) if out.height else None,
        "price_median": float(out["price"].median()) if out.height else None,
        "missingness_pct": {c: round(100.0 * out[c].null_count() / max(out.height, 1), 3) for c in out.columns},
        "snapshot_analytic_rows": analytic.height,
    }
    return out, summary


def write_bundle(pros: pl.DataFrame, cfg: dict[str, Any]) -> dict[str, Any]:
    BUNDLE.mkdir(parents=True, exist_ok=True)
    analytic = pl.read_parquet(PROCESSED / "analytic.parquet")
    assert not pros["tx_id"].is_in(analytic["tx_id"].implode()).any()
    union = pl.concat([analytic, pros.select(analytic.columns)]).sort(["date", "tx_id"])
    validate_analytic(union)
    union.write_parquet(BUNDLE / "analytic.parquet", compression="zstd")
    splits = pl.read_parquet(PROCESSED / "splits.parquet")
    regions = list(cfg.get("spatial_holdout", {}).get("regions", []))
    p_splits = pros.select("tx_id", pl.lit("prospective").alias("split"), pl.col("region_name").is_in(regions).alias("spatial_holdout_region"))
    u_splits = pl.concat([splits, p_splits.select(splits.columns)])
    u_splits.write_parquet(BUNDLE / "splits.parquet")
    subs = pl.read_parquet(PROCESSED / "llm_subsamples.parquet")
    n = min(PROSPECTIVE_SUBSAMPLE_N, pros.height)
    p_subs = stratified_subsample(union, u_splits, cfg, n_by_split={"prospective": n}, seed=cfg.get("seed", SEED))
    u_subs = pl.concat([subs, p_subs.select(subs.columns)])
    u_subs.write_parquet(BUNDLE / "llm_subsamples.parquet")
    geo_info: dict[str, Any] = {"geo_rows_prospective": 0}
    geo_path = PROCESSED / "geo_features.parquet"
    try:
        from agentic_valuation.features.geo import GeoLayers, geo_features_batch

        if GeoLayers.available() and geo_path.exists():
            layers = GeoLayers.load()
            g = geo_features_batch(pros.select("tx_id", "lat", "lng", "x_m", "y_m"), layers)
            snap_geo = pl.read_parquet(geo_path)
            pl.concat([snap_geo, g.select(snap_geo.columns)]).write_parquet(BUNDLE / "geo_features.parquet", compression="zstd")
            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())}
    except Exception as e:  # geo is optional for the runner; report instead of failing
        geo_info = {"geo_rows_prospective": 0, "geo_error": f"{type(e).__name__}: {e}"[:200]}
    return {
        "dir": str(BUNDLE.relative_to(PROCESSED.parents[1])), "analytic_rows": union.height,
        "splits": {k: int(v) for k, v in u_splits.group_by("split").len().iter_rows()},
        "llm_subsamples": {k: int(v) for k, v in u_subs.group_by("split").len().iter_rows()},
        "prospective_subsample_n": p_subs.height, "prospective_subsample_strata": int(p_subs["stratum"].n_unique()),
        "prospective_subsample_by_stratum": {k: int(v) for k, v in p_subs.group_by("stratum").len().sort("stratum").iter_rows()},
        **geo_info,
    }


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--date", default=None, help="stamp of data/raw/prospective_<date>.parquet (default: latest)")
    ap.add_argument("--no-bundle", action="store_true")
    a = ap.parse_args()
    t0 = time.perf_counter()
    cfg = load_config("data")
    stamp = a.date or latest_stamp()
    out, summary = build(stamp, cfg)
    out.write_parquet(PROCESSED / "prospective_analytic.parquet", compression="zstd")
    summary["output"] = "data/processed/prospective_analytic.parquet"
    if not a.no_bundle and out.height:
        summary["bundle"] = write_bundle(out, cfg)
    summary["seconds"] = round(time.perf_counter() - t0, 1)
    (RESULTS / "data").mkdir(parents=True, exist_ok=True)
    (RESULTS / "data" / "prospective_summary.json").write_text(json.dumps(summary, indent=1, ensure_ascii=False, default=str))
    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))


if __name__ == "__main__":
    main()