#!/usr/bin/env python3 """Plan a CatBoost model for a CSV: which columns are cat_features / text_features, the loss and metric for the target, starting parameters, and (with --code) a runnable training script. Column handling: cat_features non-numeric low/medium-cardinality columns, numeric-looking codes (zip, *_code, *_id with repeats); NaN must become a string ("__nan__") - CatBoost rejects NaN here text_features long free-text columns (mean length >= 30 chars, many words); CatBoost tokenises them numeric everything numeric; NaN allowed (nan_mode="Min" by default) dropped near-unique IDs, constant columns, date columns (derive parts first), the time column (used for ordering only) Target: binary -> Logloss / AUC + PRAUC; multiclass -> MultiClass / TotalF1; numeric -> RMSE (or Tweedie / Poisson / Quantile when --loss says so). Usage: python catboost_plan.py data.csv --target churned --drop customer_id python catboost_plan.py tickets.csv --target queue --code -o train_catboost.py python catboost_plan.py data.csv --target price --loss "Tweedie:variance_power=1.5" --time-col sold_at --code Exit codes: 0 printed/written, 2 usage or read error. """ import argparse import csv import math import os import re import sys from itertools import islice MISSING = {"", "na", "n/a", "nan", "null", "none", "?"} def is_num(v): try: return math.isfinite(float(v)) except ValueError: return False def plan(header, rows, target, drop, time_col): cats, texts, nums, dropped = [], [], [], [] for c in header: if c in (target, time_col) or c in drop: if c == time_col: dropped.append((c, "time column: ordering / has_time, not a feature")) continue vals = [(r.get(c) or "").strip() for r in rows] present = [v for v in vals if v.lower() not in MISSING] if not present or len(set(present)) == 1: dropped.append((c, "empty or constant")) continue k = len(set(present)) if sum(bool(re.match(r"^\d{4}[-/]\d{1,2}[-/]\d{1,2}", v)) for v in present) / len(present) > 0.9: dropped.append((c, "date: derive year/month/day-of-week/days-since features first")) continue numeric = all(is_num(v) for v in present) code_like = bool(re.search(r"(zip|postcode|_code$|^code$|_id$)", c, re.I)) if numeric and not code_like: nums.append(c) elif numeric and code_like: if k > 0.9 * len(present): dropped.append((c, "near-unique identifier (for codes such as zip, use a prefix like zip3 as a cat feature)")) else: cats.append(c) else: mean_len = sum(len(v) for v in present) / len(present) words = sum(len(v.split()) for v in present) / len(present) if mean_len >= 30 and words >= 4: texts.append(c) elif k > 0.9 * len(present) and len(present) > 50: dropped.append((c, "near-unique text (ID)")) else: cats.append(c) return cats, texts, nums, dropped SCRIPT = '''"""CatBoost model for {target!r}. Generated by catboost_plan.py - review, then version. Run: python {script} (pandas, catboost >= 1.2, scikit-learn) """ import numpy as np import pandas as pd from catboost import CatBoostClassifier, CatBoostRegressor, Pool from sklearn.model_selection import train_test_split DATA, TARGET = {data!r}, {target!r} CAT = {cats!r} TEXT = {texts!r} NUM = {nums!r} SEED = 0 df = pd.read_csv(DATA) df = df[df[TARGET].notna()] {order}for c in CAT: # CatBoost needs strings/ints for cat features, never NaN df[c] = df[c].astype("string").fillna("__nan__").astype(str) for c in TEXT: df[c] = df[c].fillna("").astype(str) X, y = df[CAT + TEXT + NUM], df[TARGET] {split} train_pool = Pool(X_tr, y_tr, cat_features=CAT, text_features=TEXT or None) valid_pool = Pool(X_va, y_va, cat_features=CAT, text_features=TEXT or None) test_pool = Pool(X_te, y_te, cat_features=CAT, text_features=TEXT or None) model = {cls}( loss_function={loss!r}, eval_metric={metric!r},{custom} iterations=5000, learning_rate={lr}, depth=6, l2_leaf_reg=3.0, od_type="Iter", od_wait=200, random_seed=SEED, allow_writing_files=False, verbose=500,{extra} ) model.fit(train_pool, eval_set=valid_pool, use_best_model=True) print("best iteration:", model.get_best_iteration(), "best valid score:", model.get_best_score()["validation"]) print("test:", model.eval_metrics(test_pool, metrics=[{metric!r}], ntree_start=model.get_best_iteration() - 1)[{metric!r}][-1]) imp = model.get_feature_importance(train_pool, prettified=True) # PredictionValuesChange print(imp.head(15).to_string(index=False)) shap = model.get_feature_importance(test_pool, type="ShapValues") # rows x (features + 1 expected value) print("SHAP shape:", np.shape(shap)) model.save_model("catboost_model.cbm") ''' def main(): ap = argparse.ArgumentParser(description="Plan CatBoost features and parameters for a CSV.", epilog=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("csv") ap.add_argument("--target", required=True) ap.add_argument("--drop", nargs="*", default=[]) ap.add_argument("--time-col", help="split by time and set has_time=True (ordered boosting follows row order)") ap.add_argument("--loss", help="override loss_function, e.g. 'Tweedie:variance_power=1.5', 'Quantile:alpha=0.9'") ap.add_argument("--code", action="store_true", help="emit a runnable training script") ap.add_argument("-o", "--output") a = ap.parse_args() try: with open(a.csv, newline="", encoding="utf-8-sig") as fh: reader = csv.DictReader(fh) header = reader.fieldnames or [] rows = list(islice(reader, 20000)) except OSError as e: print(f"error: {e}", file=sys.stderr) return 2 for c in (a.target, a.time_col): if c and c not in header: print(f"error: column {c!r} not found", file=sys.stderr) return 2 cats, texts, nums, dropped = plan(header, rows, a.target, set(a.drop), a.time_col) tv = [(r.get(a.target) or "").strip() for r in rows if (r.get(a.target) or "").strip()] k = len(set(tv)) if k == 2: task, cls, loss, metric, custom = "binary", "CatBoostClassifier", "Logloss", "AUC", '\n custom_metric=["PRAUC", "Logloss"],' elif k <= 50 and not all(is_num(v) for v in tv) or (k <= 20 and all(is_num(v) and float(v).is_integer() for v in tv)): task, cls, loss, metric, custom = "multiclass", "CatBoostClassifier", "MultiClass", "TotalF1:average=Macro", '\n custom_metric=["Accuracy"],' else: task, cls, loss, metric, custom = "regression", "CatBoostRegressor", "RMSE", "RMSE", '\n custom_metric=["MAE", "R2"],' if a.loss: loss = a.loss if task == "regression": metric = a.loss.split(":")[0] if a.loss.split(":")[0] in ("Tweedie", "Poisson", "Quantile", "MAE", "RMSE") else metric if metric == "Tweedie": metric = a.loss if metric == "Quantile": metric = a.loss n = len(rows) lr = 0.05 if n < 100_000 else 0.08 extra = "" if a.time_col: extra += "\n has_time=True, # keep the time order for ordered boosting / statistics" if task == "binary": pos = min(set(tv), key=tv.count) rate = tv.count(pos) / len(tv) if rate < 0.05: extra += ("\n # rare positives: keep probabilities calibrated (no weights) and watch PRAUC;" "\n # auto_class_weights='Balanced' only if recall at the default threshold matters") if texts: extra += "\n # text_features are tokenised into bag-of-words + naive Bayes / BM25 estimates by default" if a.time_col: split = (f"df_sorted = df.sort_values({a.time_col!r})\n" "n = len(df_sorted)\n" "idx_tr, idx_va, idx_te = df_sorted.index[:int(.7 * n)], df_sorted.index[int(.7 * n):int(.85 * n)], df_sorted.index[int(.85 * n):]\n" "X_tr, X_va, X_te = X.loc[idx_tr], X.loc[idx_va], X.loc[idx_te]\n" "y_tr, y_va, y_te = y.loc[idx_tr], y.loc[idx_va], y.loc[idx_te]") order = "" else: strat = "stratify=y, " if task != "regression" else "" split = (f"X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.15, {strat}random_state=SEED)\n" f"X_tr, X_va, y_tr, y_va = train_test_split(X_tr, y_tr, test_size=0.15, " f"{'stratify=y_tr, ' if strat else ''}random_state=SEED)") order = "" if a.code: text = SCRIPT.format(target=a.target, script=os.path.basename(a.output or "train_catboost.py"), data=a.csv, cats=cats, texts=texts, nums=nums, order=order, split=split, cls=cls, loss=loss, metric=metric, custom=custom, lr=lr, extra=extra) if a.output: with open(a.output, "w", encoding="utf-8") as f: f.write(text) print(f"wrote {a.output}", file=sys.stderr) else: sys.stdout.write(text) return 0 print(f"CatBoost plan for {a.target!r} ({task}, {n:,} rows sampled)") print(f" cat_features ({len(cats)}): {', '.join(cats) or '-'}") print(f" text_features ({len(texts)}): {', '.join(texts) or '-'}") print(f" numeric ({len(nums)}): {', '.join(nums) or '-'}") for c, why in dropped: print(f" dropped: {c} ({why})") print(f"\n {cls}(loss_function={loss!r}, eval_metric={metric!r}, iterations=5000, learning_rate={lr}, depth=6,") print(" l2_leaf_reg=3, od_type='Iter', od_wait=200, random_seed=0" + (", has_time=True" if a.time_col else "") + ")") print(" fit(Pool(X_tr, y_tr, cat_features=..., text_features=...), eval_set=valid_pool, use_best_model=True)") print("\n Cat columns must be str/int (fill NaN with '__nan__'); no one-hot or target encoding beforehand -") print(" CatBoost's ordered target statistics do that without leakage. Run with --code for the full script.") return 0 if __name__ == "__main__": sys.exit(main())