#!/usr/bin/env python3 """Generate a scikit-learn learning-curve script whose output feeds diagnose_curve.py. The script it writes: - loads the CSV, drops ID/leaky columns, builds a preprocessing + model pipeline (logreg | rf | hgb | lgbm) - calls sklearn.model_selection.learning_curve with geometrically spaced training sizes (small sizes are where the curve bends, so linear spacing wastes points), shuffle=True with a seed (subsets of sorted data are otherwise not random), the right splitter (stratified, grouped or time-ordered) and return_times - writes learning_curve.csv (train_size, fold, train_score, val_score, fit_time) in scorer sign convention (higher is better, neg_* scorers stay negative) and learning_curve.png with ±1 std bands when matplotlib is installed Then run: python diagnose_curve.py learning_curve.csv Usage: python gen_learning_curve.py data.csv --target churned --model hgb --scoring roc_auc -o lc.py python gen_learning_curve.py data.csv --target price --model rf --scoring neg_root_mean_squared_error python gen_learning_curve.py data.csv --target churned --group-col customer_id --sizes 8 --min-frac 0.05 Exit codes: 0 written, 2 usage or read error. """ import argparse import csv import os import sys MODELS = { "logreg": ("sklearn.linear_model", "LogisticRegression(max_iter=5000)", "Ridge(alpha=1.0)", "Ridge"), "rf": ("sklearn.ensemble", "RandomForestClassifier(n_estimators=300, min_samples_leaf=1, n_jobs=-1, random_state=0)", "RandomForestRegressor(n_estimators=300, min_samples_leaf=1, n_jobs=-1, random_state=0)", None), "hgb": ("sklearn.ensemble", "HistGradientBoostingClassifier(random_state=0)", "HistGradientBoostingRegressor(random_state=0)", None), "lgbm": ("lightgbm", "LGBMClassifier(n_estimators=300, learning_rate=0.05, verbose=-1, random_state=0)", "LGBMRegressor(n_estimators=300, learning_rate=0.05, verbose=-1, random_state=0)", None), } CLASSIFICATION_SCORERS = {"roc_auc", "average_precision", "accuracy", "balanced_accuracy", "f1", "neg_log_loss", "neg_brier_score", "roc_auc_ovr"} REGRESSION_SCORERS = {"r2", "neg_mean_absolute_error", "neg_root_mean_squared_error", "neg_mean_squared_error", "neg_mean_absolute_percentage_error"} TEMPLATE = '''"""Learning curve for {target!r} ({model}, {scoring}). Generated by gen_learning_curve.py. Run: python {script} Then: python diagnose_curve.py learning_curve.csv{diag_flags} """ import numpy as np import pandas as pd from sklearn.compose import ColumnTransformer from sklearn.impute import SimpleImputer from sklearn.model_selection import {splitter}, learning_curve from sklearn.pipeline import make_pipeline from sklearn.preprocessing import OneHotEncoder, OrdinalEncoder, StandardScaler from {module} import {estimator_class} DATA, TARGET, DROP = {data!r}, {target!r}, {drop!r} SCORING, SEED = {scoring!r}, 0 df = pd.read_csv(DATA) df = df[df[TARGET].notna()].reset_index(drop=True) {order_line}{groups_line}{target_line}X = df.drop(columns=[TARGET] + [c for c in DROP if c in df.columns]) CAT = [c for c in X.columns if not pd.api.types.is_numeric_dtype(X[c])] NUM = [c for c in X.columns if c not in CAT] X[CAT] = X[CAT].astype("object").where(X[CAT].notna(), "__nan__") prep = {prep} model = make_pipeline(prep, {estimator}) cv = {cv} # geometric sizes: the curve bends at small n, so spend points there sizes = np.unique(np.geomspace({min_frac}, 1.0, {n_sizes})) n, train_scores, val_scores, fit_times, _ = learning_curve( model, X, y, {groups_kw}train_sizes=sizes, cv=cv, scoring=SCORING, shuffle={shuffle}, random_state=SEED, return_times=True, n_jobs=-1, error_score="raise") rows = [] for i, size in enumerate(n): for k in range(train_scores.shape[1]): rows.append((int(size), k, train_scores[i, k], val_scores[i, k], fit_times[i, k])) out = pd.DataFrame(rows, columns=["train_size", "fold", "train_score", "val_score", "fit_time"]) out.to_csv("learning_curve.csv", index=False) print(out.groupby("train_size")[["train_score", "val_score", "fit_time"]].mean().round(4).to_string()) print("wrote learning_curve.csv") try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt except ImportError: plt = None if plt is not None: fig, ax = plt.subplots(figsize=(6, 4)) for scores, label in ((train_scores, "train"), (val_scores, "validation")): m, s = scores.mean(axis=1), scores.std(axis=1) ax.plot(n, m, marker="o", label=label) ax.fill_between(n, m - s, m + s, alpha=0.2) ax.set_xscale("log") ax.set_xlabel("training rows (log scale)") ax.set_ylabel(SCORING) ax.set_title("Learning curve") ax.legend() fig.tight_layout() fig.savefig("learning_curve.png", dpi=120) print("wrote learning_curve.png") ''' def main(): ap = argparse.ArgumentParser(description="Generate a learning-curve script.", epilog=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("csv") ap.add_argument("--target", required=True) ap.add_argument("--model", choices=sorted(MODELS), default="hgb") ap.add_argument("--scoring", default=None, help="sklearn scorer name (default roc_auc / r2)") ap.add_argument("--task", choices=["auto", "classification", "regression"], default="auto") ap.add_argument("--drop", nargs="*", default=[]) ap.add_argument("--group-col") ap.add_argument("--time-col") ap.add_argument("--folds", type=int, default=5) ap.add_argument("--sizes", type=int, default=6, help="number of training sizes") ap.add_argument("--min-frac", type=float, default=0.1, help="smallest size as a fraction of the training fold") ap.add_argument("-o", "--output") a = ap.parse_args() if not 0 < a.min_frac < 1 or a.sizes < 3: print("error: need 0 < --min-frac < 1 and --sizes >= 3", file=sys.stderr) return 2 try: with open(a.csv, newline="", encoding="utf-8-sig") as fh: rd = csv.DictReader(fh) header = rd.fieldnames or [] values = [r.get(a.target, "") for _, r in zip(range(20000), rd)] except OSError as e: print(f"error: {e}", file=sys.stderr) return 2 for c in (a.target, a.group_col, a.time_col): if c and c not in header: print(f"error: column {c!r} not found", file=sys.stderr) return 2 distinct = {v for v in values if v != ""} task = a.task if task == "auto": try: [float(v) for v in distinct] numeric = True except ValueError: numeric = False task = "classification" if (not numeric or len(distinct) <= 10) else "regression" scoring = a.scoring or ("roc_auc" if task == "classification" else "r2") if task == "classification" and scoring in REGRESSION_SCORERS or task == "regression" and scoring in CLASSIFICATION_SCORERS: print(f"error: scorer {scoring!r} does not fit a {task} target", file=sys.stderr) return 2 binary = len(distinct) == 2 if task == "classification" and scoring in ("roc_auc", "average_precision", "f1") and not binary: scoring = {"roc_auc": "roc_auc_ovr", "f1": "f1_macro"}.get(scoring, scoring) module, clf, reg, _ = MODELS[a.model] est = clf if task == "classification" else reg estimator_class = est.split("(")[0] if a.model == "logreg": prep = ('ColumnTransformer([("num", make_pipeline(SimpleImputer(strategy="median"), StandardScaler()), NUM),\n' ' ("cat", OneHotEncoder(handle_unknown="ignore", min_frequency=5), CAT)])') else: prep = ('ColumnTransformer([("cat", OrdinalEncoder(handle_unknown="use_encoded_value", unknown_value=-1), CAT)],\n' ' remainder={})'.format('SimpleImputer(strategy="median")' if a.model == "rf" else '"passthrough"')) if a.model == "hgb": est = est.replace("(random_state=0)", "(categorical_features=list(range(len(CAT))) or None, random_state=0)") drop = list(a.drop) + [c for c in (a.group_col, a.time_col) if c and c not in a.drop] order_line, groups_line, groups_kw, shuffle = "", "", "", "True" if a.time_col: splitter, cv = "TimeSeriesSplit", f"TimeSeriesSplit(n_splits={a.folds})" order_line = f"df = df.sort_values({a.time_col!r}).reset_index(drop=True)\n" shuffle = "False" # keep the earliest rows for the smallest sizes elif a.group_col: splitter = "StratifiedGroupKFold" if task == "classification" else "GroupKFold" cv = (f"StratifiedGroupKFold(n_splits={a.folds}, shuffle=True, random_state=SEED)" if task == "classification" else f"GroupKFold(n_splits={a.folds})") groups_line = f"groups = df[{a.group_col!r}].to_numpy()\n" groups_kw = "groups=groups, " else: splitter = "StratifiedKFold" if task == "classification" else "KFold" cv = f"{splitter}(n_splits={a.folds}, shuffle=True, random_state=SEED)" target_line = ("y = df[TARGET].astype(str)\n" if task == "classification" and not binary else "positive = df[TARGET].value_counts().idxmin()\ny = (df[TARGET] == positive).astype(int)\n" if task == "classification" else "y = df[TARGET].astype(float)\n") text = TEMPLATE.format(target=a.target, model=a.model, scoring=scoring, script=os.path.basename(a.output or "lc.py"), diag_flags="", splitter=splitter, module=module, estimator_class=estimator_class, data=a.csv, drop=drop, order_line=order_line, groups_line=groups_line, target_line=target_line, prep=prep, estimator=est, cv=cv, min_frac=a.min_frac, n_sizes=a.sizes, groups_kw=groups_kw, shuffle=shuffle) if a.output: with open(a.output, "w", encoding="utf-8") as f: f.write(text) print(f"wrote {a.output}: {a.model} {task}, {scoring}, {splitter}, {a.sizes} sizes from {a.min_frac:g}", file=sys.stderr) else: sys.stdout.write(text) return 0 if __name__ == "__main__": sys.exit(main())