1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Standard strata and stratified metric tables with the minimum-cell-size rule (n ≥ 30, else suppressed).3Definitions: docs/evaluation_protocol.md §6."""4from __future__ import annotations56from collections.abc import Sequence7from typing import Any89import numpy as np10import pandas as pd1112from ._common import REGION_GROUP_ORDER, derive_price_tercile, derive_region_group13from .metrics import PRIMARY_METRIC, per_record, summarise_records1415MIN_CELL_N = 3016STANDARD_STRATA = ["region_group", "urban_class", "property_class", "price_tercile", "year_built_decade", "density_tercile"]17STRATUM_LABELS = {18 "region_group": "Region", "urban_class": "Urban class", "property_class": "Property class",19 "price_tercile": "Price tercile", "year_built_decade": "Construction decade", "density_tercile": "Transaction density",20}21LEVEL_ORDER = {22 "region_group": REGION_GROUP_ORDER,23 "price_tercile": ["low", "mid", "high"],24 "density_tercile": ["thin", "mid", "liquid"],25 "property_class": ["house", "condo", "plex"],26 "urban_class": ["metro", "suburban", "rural"],27}28DEFAULT_STRATA_METRICS = [PRIMARY_METRIC, "mdape", "p95_abs_log_error", "severe_rate", "coverage_90", "bias_log"]293031def year_built_decade(year_built: pd.Series) -> pd.Series:32 """Decade bins: <1950, 1950s … 2000s, ≥2010; NaN → NA (row excluded from this stratum)."""33 y = pd.to_numeric(year_built, errors="coerce")34 lab = np.where(y < 1950, "<1950", np.where(y >= 2010, "2010+", (np.floor(y / 10) * 10).astype("float").astype("Int64").astype(str) + "s"))35 lab = pd.Series(lab, index=year_built.index, dtype="string")36 lab[y.isna()] = pd.NA # missing construction year: excluded from the decade stratum (as for density)37 return lab383940def density_tercile(n_sales: pd.Series) -> pd.Series:41 """Terciles of local liquidity (`n_sales_3km_12m`) → thin / mid / liquid."""42 v = pd.to_numeric(n_sales, errors="coerce")43 q1, q2 = np.nanquantile(v, [1 / 3, 2 / 3])44 lab = pd.Series(np.where(v <= q1, "thin", np.where(v <= q2, "mid", "liquid")), index=n_sales.index, dtype="string")45 lab[v.isna()] = pd.NA46 return lab474849def add_strata(df: pd.DataFrame) -> pd.DataFrame:50 """Add every standard stratum column that can be derived from the available columns."""51 out = df.copy()52 if "region_group" not in out.columns and "region_name" in out.columns:53 out["region_group"] = derive_region_group(out["region_name"])54 if "price_tercile" not in out.columns and "price" in out.columns:55 out["price_tercile"] = derive_price_tercile(out["price"])56 if "year_built_decade" not in out.columns and "year_built" in out.columns:57 out["year_built_decade"] = year_built_decade(out["year_built"])58 if "density_tercile" not in out.columns and "n_sales_3km_12m" in out.columns:59 out["density_tercile"] = density_tercile(out["n_sales_3km_12m"])60 return out616263def _sorted_levels(stratum: str, levels: list) -> list:64 order = LEVEL_ORDER.get(stratum)65 if order:66 return sorted(levels, key=lambda v: (order.index(v) if v in order else len(order), str(v)))67 return sorted(levels, key=str)686970def stratified_metrics(71 preds: Any,72 truth: Any,73 strata: Sequence[str] = STANDARD_STRATA,74 metrics: Sequence[str] = DEFAULT_STRATA_METRICS,75 *,76 group_cols: Sequence[str] = ("arm", "committee", "aggregation"),77 min_n: int = MIN_CELL_N,78) -> pd.DataFrame:79 """Long table: group_cols…, stratum, level, n, metric, value, suppressed.8081 A cell is suppressed (value = NaN, suppressed = True) when it has fewer than `min_n` records, so that82 unsupported granular estimates never reach the tables. Strata absent from the data are skipped.83 """84 d = add_strata(per_record(preds, truth))85 group_cols = list(group_cols)86 rows = []87 for stratum in strata:88 if stratum not in d.columns:89 continue90 for gkey, g in d.groupby(group_cols, sort=True, dropna=False):91 gkey = gkey if isinstance(gkey, tuple) else (gkey,)92 base = dict(zip(group_cols, gkey))93 levels = _sorted_levels(stratum, [v for v in g[stratum].dropna().unique()])94 for level in levels:95 cell = g[g[stratum] == level]96 n = len(cell)97 summ = summarise_records(cell) if n >= min_n else {}98 for m in metrics:99 row = dict(base)100 row.update({"stratum": stratum, "level": str(level), "n": n, "metric": m,101 "value": float(summ.get(m, np.nan)) if n >= min_n else np.nan, "suppressed": n < min_n})102 rows.append(row)103 cols = group_cols + ["stratum", "level", "n", "metric", "value", "suppressed"]104 return pd.DataFrame(rows, columns=cols)105