"""Train & compare salary-regression models (ML block) and write artifacts. Run: .venv\\Scripts\\python.exe src/train.py What it does (satisfies the ML block requirements end-to-end): 1. Loads the cleaned salary frame and builds the shared feature contract. 2. Holds out a test split, then **compares ≥2 models** with 5-fold CV on the training split: a naive mean baseline, Ridge (linear), RandomForest and HistGradientBoosting. The target is log-transformed (salaries are skewed). 3. Picks the best model by CV-RMSE, refits it, evaluates on the held-out set, and runs a short **error analysis** (by job title and salary quintile). 4. Writes ``artifacts/``: the fitted pipeline + metadata + metrics — everything the inference app needs (training and inference are fully separated). """ from __future__ import annotations import json import sys from datetime import date from pathlib import Path import numpy as np import pandas as pd ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) # Windows consoles default to cp1252; force UTF-8 so unicode in logs never crashes. try: sys.stdout.reconfigure(encoding="utf-8") except Exception: # noqa: BLE001 pass from src.data import load_salary_df # noqa: E402 from src.features import ( # noqa: E402 BOOL_COLS, CATEGORICAL_COLS, DEFAULTS, FEATURE_COLS, NUMERIC_COLS, SKILL_VOCAB, TARGET_COL, build_feature_frame, ) import joblib # noqa: E402 import sklearn # noqa: E402 from sklearn.compose import ColumnTransformer, TransformedTargetRegressor # noqa: E402 from sklearn.dummy import DummyRegressor # noqa: E402 from sklearn.ensemble import HistGradientBoostingRegressor, RandomForestRegressor # noqa: E402 from sklearn.linear_model import Ridge # noqa: E402 from sklearn.metrics import mean_absolute_error, r2_score, root_mean_squared_error # noqa: E402 from sklearn.model_selection import KFold, cross_validate, train_test_split # noqa: E402 from sklearn.pipeline import Pipeline # noqa: E402 from sklearn.preprocessing import OneHotEncoder, StandardScaler # noqa: E402 ARTIFACTS = ROOT / "artifacts" RANDOM_STATE = 42 N_FOLDS = 5 # --------------------------------------------------------------------------- # # Pipeline building blocks # --------------------------------------------------------------------------- # def make_preprocessor(scale_numeric: bool) -> ColumnTransformer: """One-hot the categoricals; pass through (or scale) the numeric/boolean cols. ``infrequent_if_exist`` buckets rare/unseen categories, which keeps the model robust to whatever the NLP block hands us at inference time. """ cat = OneHotEncoder(handle_unknown="infrequent_if_exist", min_frequency=25, sparse_output=False) num = StandardScaler() if scale_numeric else "passthrough" return ColumnTransformer( [ ("cat", cat, CATEGORICAL_COLS), ("num", num, BOOL_COLS + NUMERIC_COLS), ], remainder="drop", ) def log_target(reg) -> TransformedTargetRegressor: """Wrap a regressor so it trains on log1p(salary) and predicts in dollars.""" return TransformedTargetRegressor(regressor=reg, func=np.log1p, inverse_func=np.expm1) def build_models() -> dict[str, Pipeline]: """Return the candidate models (each a full preprocessing+estimator pipeline).""" return { "baseline_mean": Pipeline( [("prep", make_preprocessor(scale_numeric=False)), ("model", DummyRegressor(strategy="mean"))] ), "ridge": Pipeline( [("prep", make_preprocessor(scale_numeric=True)), ("model", log_target(Ridge(alpha=1.0, random_state=RANDOM_STATE)))] ), "random_forest": Pipeline( [("prep", make_preprocessor(scale_numeric=False)), ("model", log_target(RandomForestRegressor( n_estimators=300, min_samples_leaf=2, n_jobs=-1, random_state=RANDOM_STATE)))] ), "hist_gbr": Pipeline( [("prep", make_preprocessor(scale_numeric=False)), ("model", log_target(HistGradientBoostingRegressor( max_iter=400, learning_rate=0.06, max_leaf_nodes=31, l2_regularization=1.0, random_state=RANDOM_STATE)))] ), } # --------------------------------------------------------------------------- # # Helpers # --------------------------------------------------------------------------- # def _round(x, n: int = 2): return None if x is None or (isinstance(x, float) and np.isnan(x)) else round(float(x), n) def error_analysis(model: Pipeline, X_test: pd.DataFrame, y_test: pd.Series) -> dict: """Residual diagnostics on the held-out set.""" pred = model.predict(X_test) resid = pred - y_test.to_numpy() abs_err = np.abs(resid) by_title = ( pd.DataFrame({"title": X_test["job_title_short"].to_numpy(), "abs_err": abs_err}) .groupby("title")["abs_err"].agg(["count", "mean"]).sort_values("mean", ascending=False) ) quint = pd.qcut(y_test, 5, duplicates="drop") by_quint = ( pd.DataFrame({"q": quint.astype(str).to_numpy(), "abs_err": abs_err}) .groupby("q")["abs_err"].mean() ) return { "mean_abs_error": _round(abs_err.mean()), "median_abs_error": _round(np.median(abs_err)), "mean_signed_error_bias": _round(resid.mean()), "abs_error_by_title": {k: _round(v) for k, v in by_title["mean"].items()}, "count_by_title": {k: int(v) for k, v in by_title["count"].items()}, "abs_error_by_salary_quintile": {k: _round(v) for k, v in by_quint.items()}, } # --------------------------------------------------------------------------- # # Training entry points # --------------------------------------------------------------------------- # def train_salary_model() -> None: print("Loading salary data ...") df = load_salary_df() X = build_feature_frame(df) X[BOOL_COLS] = X[BOOL_COLS].astype(int) y = df[TARGET_COL].astype(float) print(f" rows={len(X):,} | features={len(FEATURE_COLS)} " f"({len(CATEGORICAL_COLS)} cat + {len(BOOL_COLS)} bool + {len(NUMERIC_COLS)} num)") X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=RANDOM_STATE ) print(f" train={len(X_train):,} | test={len(X_test):,}") cv = KFold(n_splits=N_FOLDS, shuffle=True, random_state=RANDOM_STATE) scoring = { "rmse": "neg_root_mean_squared_error", "mae": "neg_mean_absolute_error", "r2": "r2", } print(f"\nComparing {len(build_models())} models with {N_FOLDS}-fold CV " f"(metrics in USD/year) ...\n") cv_results: dict[str, dict] = {} rows = [] for name, pipe in build_models().items(): res = cross_validate(pipe, X_train, y_train, cv=cv, scoring=scoring, n_jobs=-1) rmse = -res["test_rmse"] mae = -res["test_mae"] r2 = res["test_r2"] cv_results[name] = { "rmse_mean": _round(rmse.mean()), "rmse_std": _round(rmse.std()), "mae_mean": _round(mae.mean()), "mae_std": _round(mae.std()), "r2_mean": _round(r2.mean(), 4), "r2_std": _round(r2.std(), 4), } rows.append((name, rmse.mean(), mae.mean(), r2.mean())) table = pd.DataFrame(rows, columns=["model", "CV_RMSE", "CV_MAE", "CV_R2"]).set_index("model") print(table.round(2).to_string()) # Best non-baseline model by CV-RMSE. ranked = sorted( [(n, v["rmse_mean"]) for n, v in cv_results.items() if n != "baseline_mean"], key=lambda t: t[1], ) best_name = ranked[0][0] print(f"\nBest model by CV-RMSE: {best_name}") # Refit best on full train, evaluate on held-out test. best = build_models()[best_name] best.fit(X_train, y_train) pred = best.predict(X_test) holdout = { "model": best_name, "rmse": _round(root_mean_squared_error(y_test, pred)), "mae": _round(mean_absolute_error(y_test, pred)), "r2": _round(r2_score(y_test, pred), 4), "n_test": int(len(y_test)), } base = build_models()["baseline_mean"] base.fit(X_train, y_train) bpred = base.predict(X_test) baseline_holdout = { "rmse": _round(root_mean_squared_error(y_test, bpred)), "mae": _round(mean_absolute_error(y_test, bpred)), "r2": _round(r2_score(y_test, bpred), 4), } uplift = _round(100 * (1 - holdout["rmse"] / baseline_holdout["rmse"]), 1) print(f"\nHeld-out test ({best_name}): " f"RMSE={holdout['rmse']:,.0f} MAE={holdout['mae']:,.0f} R2={holdout['r2']}") print(f"Held-out test (baseline) : RMSE={baseline_holdout['rmse']:,.0f} " f"MAE={baseline_holdout['mae']:,.0f} R2={baseline_holdout['r2']}") print(f"RMSE reduction vs. baseline: {uplift}%") ea = error_analysis(best, X_test, y_test) print(f"\nError analysis: MAE={ea['mean_abs_error']:,.0f} " f"bias(mean signed)={ea['mean_signed_error_bias']:,.0f}") print(" worst titles (MAE):", dict(list(ea["abs_error_by_title"].items())[:3])) # --- Persist artifacts (refit on ALL data for the shipped model) --------- # ARTIFACTS.mkdir(exist_ok=True) final = build_models()[best_name] final.fit(X, y) # compress=3: RandomForest tree arrays are highly redundant, so this shrinks # the artifact ~5x (≈190 MB → <40 MB) with no change to predictions — keeping # it under GitHub's 100 MB hard limit and friendly for the HF Space. joblib.dump(final, ARTIFACTS / "salary_model.joblib", compress=3) metadata = { "task": "salary_regression", "target": TARGET_COL, "target_unit": "USD per year", "target_transform": "log1p / expm1", "model": best_name, "feature_cols": FEATURE_COLS, "categorical_cols": CATEGORICAL_COLS, "bool_cols": BOOL_COLS, "numeric_cols": NUMERIC_COLS, "skill_vocab": SKILL_VOCAB, "defaults": DEFAULTS, "title_categories": sorted(df["job_title_short"].dropna().unique().tolist()), "n_rows_total": int(len(X)), "trained_at": date.today().isoformat(), "sklearn_version": sklearn.__version__, } (ARTIFACTS / "salary_metadata.json").write_text( json.dumps(metadata, indent=2, ensure_ascii=False), encoding="utf-8" ) metrics = { "cv": {"folds": N_FOLDS, "metric_unit": "USD/year", "models": cv_results}, "holdout": holdout, "baseline_holdout": baseline_holdout, "rmse_reduction_vs_baseline_pct": uplift, "error_analysis": ea, } (ARTIFACTS / "salary_metrics.json").write_text( json.dumps(metrics, indent=2, ensure_ascii=False), encoding="utf-8" ) print(f"\nSaved -> {ARTIFACTS / 'salary_model.joblib'}") print(f"Saved -> {ARTIFACTS / 'salary_metadata.json'}") print(f"Saved -> {ARTIFACTS / 'salary_metrics.json'}") def main() -> None: print("#" * 72 + "\n# ML block: salary regression\n" + "#" * 72) train_salary_model() # NLP block: the supervised résumé↔JD fit classifier (writes fit_* artifacts). # Imported (not run as __main__) so its custom transformers pickle correctly. print("\n" + "#" * 72 + "\n# NLP block: résumé↔JD fit classifier\n" + "#" * 72) from src.fit_model import train_fit_classifier train_fit_classifier() print("\nALL ARTIFACTS WRITTEN. DONE.") if __name__ == "__main__": main()