1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Arm C: LightGBM on log price (CatBoost / XGBoost as robustness), tuned on validation only (random search +3early stopping). Arm D: AVM = LightGBM point model on the full feature set (including as-of comparable features)4+ LightGBM quantile models for q05..q95 + conformal rescaling of the 90 % interval on validation."""5from __future__ import annotations67import json8import random910import numpy as np11import polars as pl1213from .common import Prediction, QUANTILES, apply_scale, categorical_features, conformal_scale, numeric_features1415SEARCH_SPACE = {16 "num_leaves": [31, 63, 127, 255],17 "learning_rate": [0.02, 0.03, 0.05, 0.08],18 "min_child_samples": [20, 50, 100, 200],19 "feature_fraction": [0.6, 0.8, 1.0],20 "bagging_fraction": [0.7, 0.9, 1.0],21 "lambda_l2": [0.0, 1.0, 10.0],22}232425def _to_pandas(df: pl.DataFrame, num: list[str], cat: list[str]):26 pdf = df.select(num + cat).to_pandas()27 for c in cat:28 pdf[c] = pdf[c].astype("category")29 for c in num:30 pdf[c] = pdf[c].astype(float)31 return pdf323334class LightGBMModel:35 def __init__(self, n_trials: int = 12, seed: int = 20261027, objective: str = "regression", alpha: float | None = None, max_rounds: int = 4000, exclude_prefixes: tuple[str, ...] = ()):36 self.exclude_prefixes = exclude_prefixes37 self.n_trials = n_trials38 self.seed = seed39 self.objective = objective40 self.alpha = alpha41 self.max_rounds = max_rounds42 self.params: dict = {}43 self.model = None44 self.num: list[str] = []45 self.cat: list[str] = []46 self.search_log: list[dict] = []4748 def _base_params(self, p: dict) -> dict:49 out = {"objective": self.objective, "verbose": -1, "seed": self.seed, "num_threads": 0, "bagging_freq": 1, **p}50 if self.objective == "quantile":51 out["alpha"] = self.alpha52 return out5354 def fit(self, train: pl.DataFrame, valid: pl.DataFrame, params: dict | None = None) -> "LightGBMModel":55 import lightgbm as lgb5657 self.num = [c for c in numeric_features(train) if not c.startswith(self.exclude_prefixes)]58 self.cat = categorical_features(train)59 Xtr, Xva = _to_pandas(train, self.num, self.cat), _to_pandas(valid, self.num, self.cat)60 # align categories61 for c in self.cat:62 cats = Xtr[c].cat.categories63 Xva[c] = Xva[c].cat.set_categories(cats)64 ytr, yva = train["log_price"].to_numpy(), valid["log_price"].to_numpy()65 dtr = lgb.Dataset(Xtr, ytr, categorical_feature=self.cat, free_raw_data=False, params={"feature_pre_filter": False})66 dva = lgb.Dataset(Xva, yva, reference=dtr, categorical_feature=self.cat, free_raw_data=False)67 rng = random.Random(self.seed)68 candidates = [params] if params else [{k: rng.choice(v) for k, v in SEARCH_SPACE.items()} for _ in range(self.n_trials)]69 best = None70 for p in candidates:71 m = lgb.train(self._base_params(p), dtr, num_boost_round=self.max_rounds, valid_sets=[dva], callbacks=[lgb.early_stopping(100, verbose=False)])72 pred = m.predict(Xva, num_iteration=m.best_iteration)73 score = float(np.mean(np.abs(pred - yva))) if self.objective != "quantile" else float(_pinball(yva, pred, self.alpha))74 self.search_log.append({"params": p, "best_iteration": m.best_iteration, "valid_score": score})75 if best is None or score < best[0]:76 best = (score, p, m)77 self.params = {**best[1], "num_boost_round": best[2].best_iteration}78 self.model = best[2]79 return self8081 def predict_log(self, df: pl.DataFrame) -> np.ndarray:82 X = _to_pandas(df, self.num, self.cat)83 for c in self.cat:84 X[c] = X[c].cat.set_categories(self.model.pandas_categorical[self.cat.index(c)])85 return self.model.predict(X, num_iteration=self.model.best_iteration)8687 def predict(self, df: pl.DataFrame) -> Prediction:88 return Prediction(tx_id=df["tx_id"].to_numpy(), point=np.exp(self.predict_log(df)), quantiles=None)899091def _pinball(y, q, a):92 d = y - q93 return np.mean(np.maximum(a * d, (a - 1) * d))949596class AVM:97 """Point model + five quantile models + conformal scaling (validation)."""9899 def __init__(self, n_trials: int = 12, seed: int = 20261027):100 self.point = LightGBMModel(n_trials=n_trials, seed=seed)101 self.qmodels: dict[float, LightGBMModel] = {}102 self.scale = 1.0103 self.n_trials = n_trials104 self.seed = seed105106 def fit(self, train: pl.DataFrame, valid: pl.DataFrame) -> "AVM":107 self.point.fit(train, valid)108 p = {k: v for k, v in self.point.params.items() if k != "num_boost_round"}109 for a in QUANTILES:110 m = LightGBMModel(n_trials=1, seed=self.seed, objective="quantile", alpha=a)111 m.fit(train, valid, params=p)112 self.qmodels[a] = m113 qv = self._q_log(valid)114 self.scale = conformal_scale(self.point.predict_log(valid), qv, valid["log_price"].to_numpy(), 0.90)115 return self116117 def _q_log(self, df: pl.DataFrame) -> np.ndarray:118 q = np.column_stack([self.qmodels[a].predict_log(df) for a in QUANTILES])119 return np.sort(q, axis=1) # enforce monotone quantiles120121 def predict(self, df: pl.DataFrame) -> Prediction:122 p = self.point.predict_log(df)123 q = apply_scale(self._q_log(df), self.scale)124 # keep the point inside [q05, q95]125 q[:, 0] = np.minimum(q[:, 0], p - 1e-4)126 q[:, 4] = np.maximum(q[:, 4], p + 1e-4)127 q = np.sort(q, axis=1)128 return Prediction(tx_id=df["tx_id"].to_numpy(), point=np.exp(p), quantiles=np.exp(q))129130131class CatBoostModel:132 def __init__(self, seed: int = 20261027, iterations: int = 4000, exclude_prefixes: tuple[str, ...] = ()):133 self.exclude_prefixes = exclude_prefixes134 self.seed, self.iterations = seed, iterations135 self.model = None136 self.num: list[str] = []137 self.cat: list[str] = []138139 def fit(self, train: pl.DataFrame, valid: pl.DataFrame) -> "CatBoostModel":140 from catboost import CatBoostRegressor, Pool141142 self.num, self.cat = [c for c in numeric_features(train) if not c.startswith(self.exclude_prefixes)], categorical_features(train)143 Xtr = train.select(self.num + self.cat).to_pandas()144 Xva = valid.select(self.num + self.cat).to_pandas()145 for c in self.cat:146 Xtr[c] = Xtr[c].astype(str)147 Xva[c] = Xva[c].astype(str)148 self.model = CatBoostRegressor(iterations=self.iterations, learning_rate=0.05, depth=8, loss_function="MAE", random_seed=self.seed, verbose=0, od_type="Iter", od_wait=100, thread_count=-1)149 self.model.fit(Pool(Xtr, train["log_price"].to_numpy(), cat_features=self.cat), eval_set=Pool(Xva, valid["log_price"].to_numpy(), cat_features=self.cat), use_best_model=True)150 return self151152 def predict(self, df: pl.DataFrame) -> Prediction:153 X = df.select(self.num + self.cat).to_pandas()154 for c in self.cat:155 X[c] = X[c].astype(str)156 return Prediction(tx_id=df["tx_id"].to_numpy(), point=np.exp(self.model.predict(X)), quantiles=None)157158159class XGBoostModel:160 def __init__(self, seed: int = 20261027, rounds: int = 4000, exclude_prefixes: tuple[str, ...] = ()):161 self.exclude_prefixes = exclude_prefixes162 self.seed, self.rounds = seed, rounds163 self.model = None164 self.num: list[str] = []165 self.cat: list[str] = []166167 def fit(self, train: pl.DataFrame, valid: pl.DataFrame) -> "XGBoostModel":168 import xgboost as xgb169170 self.num, self.cat = [c for c in numeric_features(train) if not c.startswith(self.exclude_prefixes)], categorical_features(train)171 Xtr, Xva = _to_pandas(train, self.num, self.cat), _to_pandas(valid, self.num, self.cat)172 for c in self.cat:173 Xva[c] = Xva[c].cat.set_categories(Xtr[c].cat.categories)174 self.model = xgb.XGBRegressor(n_estimators=self.rounds, learning_rate=0.05, max_depth=8, subsample=0.8, colsample_bytree=0.8, enable_categorical=True, tree_method="hist", random_state=self.seed, early_stopping_rounds=100, objective="reg:absoluteerror")175 self.model.fit(Xtr, train["log_price"].to_numpy(), eval_set=[(Xva, valid["log_price"].to_numpy())], verbose=False)176 self._cats = {c: Xtr[c].cat.categories for c in self.cat}177 return self178179 def predict(self, df: pl.DataFrame) -> Prediction:180 X = _to_pandas(df, self.num, self.cat)181 for c in self.cat:182 X[c] = X[c].cat.set_categories(self._cats[c])183 return Prediction(tx_id=df["tx_id"].to_numpy(), point=np.exp(self.model.predict(X)), quantiles=None)184185186def dump_search_log(model: LightGBMModel, path) -> None:187 with open(path, "w") as fh:188 json.dump({"params": model.params, "search_log": model.search_log}, fh, indent=1, default=float)189