1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Shared helpers: feature selection from the feature matrix, prediction records, interval calibration."""3from __future__ import annotations45from dataclasses import dataclass67import numpy as np8import polars as pl910ID_COLS = {"tx_id", "split", "log_price", "price", "date", "parcel_id", "stratum"}11CATEGORICAL = ["region_name", "region_group", "mun_code", "property_class", "subtype", "grid1k", "grid5k", "urban_class"]12QUANTILES = (0.05, 0.25, 0.50, 0.75, 0.95)13QCOLS = ("q05", "q25", "q50", "q75", "q95")141516def numeric_features(df: pl.DataFrame) -> list[str]:17 out = []18 for c, dt in df.schema.items():19 if c in ID_COLS or c in CATEGORICAL:20 continue21 if dt.is_numeric() or dt == pl.Boolean:22 out.append(c)23 return out242526def categorical_features(df: pl.DataFrame) -> list[str]:27 return [c for c in CATEGORICAL if c in df.columns]282930@dataclass31class Prediction:32 tx_id: np.ndarray33 point: np.ndarray # CAD34 quantiles: np.ndarray | None # shape (n, 5) CAD, in QUANTILES order3536 def to_records(self, split: str, arm: str, run_id: str, latency_s: float = 0.0) -> pl.DataFrame:37 n = len(self.tx_id)38 q = self.quantiles if self.quantiles is not None else np.full((n, 5), np.nan)39 return pl.DataFrame(40 {41 "tx_id": self.tx_id, "split": [split] * n, "arm": [arm] * n, "committee": ["none"] * n, "aggregation": ["model"] * n,42 "point": self.point.astype(float), "q05": q[:, 0], "q25": q[:, 1], "q50": q[:, 2], "q75": q[:, 3], "q95": q[:, 4],43 "cost_usd": [0.0] * n, "latency_s": [latency_s / max(n, 1)] * n, "tokens": [0] * n, "run_id": [run_id] * n,44 }45 )464748def residual_quantile_intervals(point_log: np.ndarray, resid_valid: np.ndarray, scale: np.ndarray | None = None) -> np.ndarray:49 """Intervals from empirical validation residual quantiles (optionally scaled per observation)."""50 qs = np.quantile(resid_valid, QUANTILES)51 s = np.ones_like(point_log) if scale is None else scale52 return np.exp(point_log[:, None] + s[:, None] * qs[None, :])535455def conformal_scale(valid_point_log: np.ndarray, valid_q_log: np.ndarray, y_valid_log: np.ndarray, nominal: float = 0.90) -> float:56 """Multiplicative factor on the half-width of the [q05,q95] log interval achieving `nominal` coverage on validation."""57 lo, hi = valid_q_log[:, 0], valid_q_log[:, 4]58 half = np.maximum((hi - lo) / 2.0, 1e-6)59 centre = (hi + lo) / 2.060 score = np.abs(y_valid_log - centre) / half61 k = int(np.ceil((len(score) + 1) * nominal)) - 162 k = min(max(k, 0), len(score) - 1)63 return float(np.sort(score)[k])646566def apply_scale(q_log: np.ndarray, factor: float) -> np.ndarray:67 centre = (q_log[:, 4] + q_log[:, 0]) / 2.068 return centre[:, None] + factor * (q_log - centre[:, None])69