Source code · Python
pd_model.py
"""Probability of default (PD) scorecard. Steps, in the order a credit risk team would review them:1. Bin each behaviour feature and measure its Weight of Evidence (WoE) and Information Value (IV) on the training sample.2. Keep features by a stated rule: IV of at least 0.02, and drop the weaker of any pair whose WoE values correlate above 0.7.3. Fit a logistic regression on the WoE values. Every coefficient must be negative (higher WoE = safer borrower); a feature that breaks this is removed and the model refitted.4. Validate on the held out 30%: discrimination (AUC, Gini, KS), calibration (deciles, Brier score, Hosmer-Lemeshow) and stability (PSI).5. Scale the model to scorecard points: 600 points at odds of 50 good to 1 bad, 20 points to double the odds. Demographic fields never enter the model; see sql/03_features.sql."""from dataclasses import dataclass import numpy as npimport pandas as pdimport statsmodels.api as smfrom scipy import stats SEED = 42TEST_SHARE = 0.30IV_MIN = 0.02CORR_MAX = 0.70BASE_SCORE, BASE_ODDS, PDO = 600, 50, 20 # Explicit bins and labels for the discrete features; continuous ones use# training deciles. Bins are kept in risk order, not alphabetical order.DISCRETE_BINS = { "worst_status_6m": ([-np.inf, -2, -1, 0, 1, 2, np.inf], ["No use in any month", "Paid duly at worst", "Revolving at worst", "1 month late", "2 months late", "3+ months late"]), "months_delayed_6m": ([-np.inf, 0, 1, 2, 4, 6], ["None", "1", "2", "3 to 4", "5 to 6"]),}CONTINUOUS = ["credit_limit", "utilisation_sep", "avg_utilisation_6m", "payment_ratio_sep"]CATEGORICAL = ["status_sep_group"]STATUS_ORDER = ["No use (-2)", "Paid duly (-1)", "Revolving (0)", "1 month late", "2 months late", "3+ months late"]TARGET = "default_next_month" @dataclassclass Binning: feature: str table: pd.DataFrame # bin label, accounts, defaults, default rate, WoE, IV contribution iv: float edges: list | None # numeric bin edges, None for categorical labels: list[str] # bin labels in order def split(features: pd.DataFrame) -> pd.Series: """Stratified 70/30 split with a fixed seed, so defaults are equally represented.""" rng = np.random.default_rng(SEED) is_test = pd.Series(False, index=features.index) for _, group in features.groupby(TARGET): chosen = rng.choice(group.index.to_numpy(), size=round(len(group) * TEST_SHARE), replace=False) is_test[chosen] = True return is_test def _edge_labels(feature: str, edges: list) -> list[str]: """Readable labels for decile bins: NT$ thousands for limits, percent for ratios.""" fmt = (lambda v: f"{v / 1000:,.0f}k") if feature == "credit_limit" else (lambda v: f"{v:.1%}") inner = edges[1:-1] labels = [f"Up to {fmt(inner[0])}"] labels += [f"{fmt(a)} to {fmt(b)}" for a, b in zip(inner, inner[1:])] labels.append(f"Over {fmt(inner[-1])}") if len(set(labels)) != len(labels): raise ValueError(f"bin labels for {feature} are not unique: {labels}") return labels def _bin_labels(values: pd.Series, edges: list | None, labels: list[str]) -> pd.Series: """Ordered categorical bin of each value, so tables keep the bins in order.""" if edges is None: return pd.Series(pd.Categorical(values, categories=labels, ordered=True), index=values.index) return pd.cut(values, bins=edges, labels=labels, include_lowest=True) def fit_binning(train: pd.DataFrame, feature: str) -> Binning: """WoE = ln(share of goods / share of bads) per bin, with 0.5 smoothing.""" if feature in DISCRETE_BINS: edges, names = DISCRETE_BINS[feature] elif feature in CONTINUOUS: qs = np.unique(np.quantile(train[feature], np.linspace(0, 1, 11))) edges = [-np.inf, *qs[1:-1], np.inf] names = _edge_labels(feature, edges) else: edges, names = None, STATUS_ORDER labels = _bin_labels(train[feature], edges, names) grouped = train.groupby(labels, observed=True)[TARGET].agg(accounts="size", defaults="sum") grouped.index = grouped.index.astype(str) goods = grouped["accounts"] - grouped["defaults"] share_good = (goods + 0.5) / (goods.sum() + 0.5) share_bad = (grouped["defaults"] + 0.5) / (grouped["defaults"].sum() + 0.5) table = grouped.assign( default_rate=grouped["defaults"] / grouped["accounts"], woe=np.log(share_good / share_bad), iv=(share_good - share_bad) * np.log(share_good / share_bad), ) return Binning(feature, table, float(table["iv"].sum()), edges, names) def apply_woe(df: pd.DataFrame, binning: Binning) -> pd.Series: labels = _bin_labels(df[binning.feature], binning.edges, binning.labels).astype(str) return labels.map(binning.table["woe"]).fillna(0.0).astype(float) def select_features(train: pd.DataFrame, binnings: dict[str, Binning]) -> tuple[list[str], list[dict]]: """Apply the IV and correlation rules and record why each feature was kept or dropped.""" log = [] ranked = sorted(binnings.values(), key=lambda b: b.iv, reverse=True) kept: list[str] = [] woe = pd.DataFrame({b.feature: apply_woe(train, b) for b in ranked}) for b in ranked: if b.iv < IV_MIN: log.append({"feature": b.feature, "iv": b.iv, "decision": f"Dropped: IV below {IV_MIN}"}) continue clash = next((k for k in kept if abs(woe[b.feature].corr(woe[k])) > CORR_MAX), None) if clash: r = woe[b.feature].corr(woe[clash]) log.append({"feature": b.feature, "iv": b.iv, "decision": f"Dropped: WoE correlation {r:.2f} with {clash}, which has higher IV"}) continue kept.append(b.feature) log.append({"feature": b.feature, "iv": b.iv, "decision": "Kept"}) return kept, log def fit_logit(x: pd.DataFrame, y: pd.Series, log: list[dict]): """Fit, then remove any feature whose coefficient has the wrong sign and refit.""" cols = list(x.columns) while True: model = sm.Logit(y, sm.add_constant(x[cols])).fit(disp=False) wrong = [c for c in cols if model.params[c] >= 0] if not wrong: return model, cols worst = max(wrong, key=lambda c: model.params[c]) for entry in log: if entry["feature"] == worst: entry["decision"] = "Dropped: positive coefficient after fitting (unstable sign)" cols.remove(worst) def auc(y: np.ndarray, p: np.ndarray) -> float: """Area under the ROC curve via the Mann-Whitney rank statistic.""" ranks = stats.rankdata(p) n1 = y.sum() n0 = len(y) - n1 return float((ranks[y == 1].sum() - n1 * (n1 + 1) / 2) / (n1 * n0)) def ks(y: np.ndarray, p: np.ndarray) -> float: order = np.argsort(-p) cum_bad = np.cumsum(y[order]) / y.sum() cum_good = np.cumsum(1 - y[order]) / (len(y) - y.sum()) return float(np.max(np.abs(cum_bad - cum_good))) def roc_points(y: np.ndarray, p: np.ndarray, n: int = 41) -> list[dict]: order = np.argsort(-p) tpr = np.concatenate([[0], np.cumsum(y[order]) / y.sum()]) fpr = np.concatenate([[0], np.cumsum(1 - y[order]) / (len(y) - y.sum())]) idx = np.unique(np.linspace(0, len(tpr) - 1, n).round().astype(int)) return [{"fpr": round(float(fpr[i]), 4), "tpr": round(float(tpr[i]), 4)} for i in idx] def calibration(y: np.ndarray, p: np.ndarray) -> tuple[list[dict], float, float]: """Predicted against observed default rate by decile of PD, plus Hosmer-Lemeshow.""" deciles = pd.qcut(pd.Series(p).rank(method="first"), 10, labels=False) rows, hl = [], 0.0 for d in range(10): mask = (deciles == d).to_numpy() n, obs, exp = mask.sum(), y[mask].sum(), p[mask].sum() hl += (obs - exp) ** 2 / (exp * (1 - exp / n)) rows.append({"decile": d + 1, "accounts": int(n), "predicted": round(float(exp / n), 4), "observed": round(float(obs / n), 4)}) return rows, float(hl), float(1 - stats.chi2.cdf(hl, 8)) def psi(expected: np.ndarray, actual: np.ndarray) -> float: """Population stability index of the score between training and test samples.""" edges = np.unique(np.quantile(expected, np.linspace(0, 1, 11))) edges[0], edges[-1] = -np.inf, np.inf e = np.histogram(expected, edges)[0] / len(expected) a = np.histogram(actual, edges)[0] / len(actual) e, a = np.clip(e, 1e-6, None), np.clip(a, 1e-6, None) return float(np.sum((a - e) * np.log(a / e))) def to_score(pd_values: np.ndarray) -> np.ndarray: factor = PDO / np.log(2) offset = BASE_SCORE - factor * np.log(BASE_ODDS) return offset + factor * np.log((1 - pd_values) / pd_values) def scorecard_points(model, cols: list[str], binnings: dict[str, Binning]) -> list[dict]: """Points per bin: the intercept is shared equally across the features.""" factor = PDO / np.log(2) offset = BASE_SCORE - factor * np.log(BASE_ODDS) k = len(cols) rows = [] for c in cols: for label, r in binnings[c].table.iterrows(): points = -(model.params[c] * r["woe"] + model.params["const"] / k) * factor + offset / k rows.append({"feature": c, "bin": label, "accounts": int(r["accounts"]), "default_rate": round(float(r["default_rate"]), 4), "woe": round(float(r["woe"]), 4), "points": round(float(points))}) return rows def score_bands(y: np.ndarray, score: np.ndarray) -> list[dict]: """Ten equal-population score bands, riskiest first, with cumulative default capture.""" band = pd.qcut(pd.Series(score).rank(method="first"), 10, labels=False).to_numpy() rows, captured = [], 0 for b in range(10): mask = band == b captured += y[mask].sum() rows.append({"band": b + 1, "score_min": int(score[mask].min()), "score_max": int(score[mask].max()), "accounts": int(mask.sum()), "default_rate": round(float(y[mask].mean()), 4), "cumulative_capture": round(float(captured / y.sum()), 4)}) return rows def run(features: pd.DataFrame) -> dict: is_test = split(features) train, test = features[~is_test], features[is_test] candidates = list(DISCRETE_BINS) + CONTINUOUS + CATEGORICAL binnings = {f: fit_binning(train, f) for f in candidates} kept, log = select_features(train, binnings) woe_all = pd.DataFrame({f: apply_woe(features, binnings[f]) for f in candidates}) model, cols = fit_logit(woe_all.loc[~is_test, kept], train[TARGET], log) pd_all = model.predict(sm.add_constant(woe_all[cols])).to_numpy() y = features[TARGET].to_numpy() tr, te = (~is_test).to_numpy(), is_test.to_numpy() score = to_score(pd_all) cal_rows, hl, hl_p = calibration(y[te], pd_all[te]) validation = { "train": {"accounts": int(tr.sum()), "default_rate": float(y[tr].mean()), "auc": auc(y[tr], pd_all[tr]), "ks": ks(y[tr], pd_all[tr])}, "test": {"accounts": int(te.sum()), "default_rate": float(y[te].mean()), "auc": auc(y[te], pd_all[te]), "ks": ks(y[te], pd_all[te]), "brier": float(np.mean((pd_all[te] - y[te]) ** 2)), "hosmer_lemeshow": hl, "hosmer_lemeshow_p": hl_p}, "psi_train_test": psi(score[tr], score[te]), } for s in ("train", "test"): validation[s]["gini"] = 2 * validation[s]["auc"] - 1 coefficients = [{"term": t, "coefficient": float(model.params[t]), "std_error": float(model.bse[t]), "p_value": float(model.pvalues[t])} for t in ["const", *cols]] model_data = woe_all[cols].assign(default=y, test=te.astype(int), pd_python=pd_all) return { "is_test": is_test, "pd": pd_all, "score": score, "model_data": model_data, "result": { "features_used": cols, "selection": [{**e, "iv": round(e["iv"], 4)} for e in log], "iv": {f: round(b.iv, 4) for f, b in binnings.items()}, "coefficients": coefficients, "validation": validation, "calibration": cal_rows, "roc": roc_points(y[te], pd_all[te]), "scorecard": scorecard_points(model, cols, binnings), "score_bands": score_bands(y[te], score[te]), "settings": {"seed": SEED, "test_share": TEST_SHARE, "iv_min": IV_MIN, "corr_max": CORR_MAX, "base_score": BASE_SCORE, "base_odds": BASE_ODDS, "pdo": PDO}, }, }