Python构建可落地的贷款违约预测模型实战

简介本资源是一套面向金融从业者与机器学习初学者的贷款违约预测实战项目聚焦信用风险评估场景提供从数据清洗、特征工程到多模型训练与预测的完整Python实现。包内共23个文件含7个核心Python脚本如data_analysis.py、train_use_rfc.py、predict.py等、11张可视化图表含特征重要性、学习曲线等、2个真实信贷数据集cs-training.csv与cs-test.csv、1个Excel数据字典及Markdown说明文档压缩包大小13.77MB结构清晰、模块解耦便于分步学习与复现。已有560人学习下载适合希望掌握风控建模全流程、理解随机森林/决策树/梯度提升等算法在信贷场景中差异表现的学习者。读者可直接运行代码完成端到端建模获取可解释的风险评分结果并借助图像与文档快速把握关键特征与调参逻辑。1. 为什么用 Python 做贷款违约预测不是“炫技”而是风控团队真实落地的最小可行路径你手头有一批银行信贷数据客户基本信息、收入负债比、历史还款记录、征信查询次数、职业类型……但业务方只问一句“下个月放出去的 5000 笔新贷里哪 300 笔最可能逾期”——这不是 Kaggle 比赛没有排行榜没有“最佳模型”奖杯只有风控经理盯着你屏幕等结果的沉默。这时候用 Python 写一个能跑通、能解释、能上线、能迭代的贷款违约预测 pipeline就是最务实的选择。它不依赖昂贵的商业建模平台不强求深度学习黑匣子而是用scikit-learnpandasxgboost组成的轻量级组合在 2 小时内完成从原始 CSV 到可部署评分卡的闭环。本文讲的不是“如何调出最高 AUC”而是如何让一个刚学完《Python 入门》的分析师三天内交出一份被风控部签字认可的预测报告数据怎么清洗才不丢关键信号、特征工程为什么比算法选择更重要、XGBoost 的scale_pos_weight怎么设才不把坏账当噪声过滤掉、模型输出怎么翻译成业务能看懂的“高/中/低风险等级”。所有代码可直接复制粘贴所有参数都标注了业务含义所有翻车点都来自我去年在某城商行做贷前模型时的真实血泪经验。2. 用 Python 构建贷款违约预测 pipeline从原始数据到可解释评分卡的四步闭环贷款违约预测的本质是把“人是否可能不还钱”这个模糊判断转化为结构化特征与概率值之间的映射关系。Python 的优势不在算法本身XGBoost、LightGBM 都是 C 底层而在于它能把数据加载、缺失填充、特征构造、模型训练、结果解释这整条链路用不到 200 行代码串起来并且每一步都可调试、可回溯、可复现。下面这四步是我在线上模型交付中反复验证过的最小可行路径跳过任何一步都会导致模型在生产环境失效。2.1 加载与探查用 pandas 快速识别“脏数据”而非“脏字段”很多新手一上来就pd.read_csv()然后df.info()结果发现employment_length是object类型里面混着10 years、 1 year、n/a、空字符串甚至None—— 这不是数据类型问题是业务逻辑断层。真正该做的是先用value_counts(dropnaFalse)和sample(5)粗筛import pandas as pd import numpy as np df pd.read_csv(loan_data.csv, encodingutf-8) # 关键动作对每个字段做「业务语义探查」而非技术类型检查 for col in [employment_length, home_ownership, purpose]: print(f\n {col} ) print(df[col].value_counts(dropnaFalse).head(10)) print(Sample values:, df[col].sample(5).tolist())提示dropnaFalse是必须的否则你会漏掉大量NaN占比超 30% 的字段比如mths_since_last_delinq在新客中天然为空sample(5)能暴露肉眼难辨的编码污染比如purpose字段里混入debt_consolidation 末尾空格和Debt consolidation大小写不一致这种差异会让后续 one-hot 编码生成冗余列。2.2 特征工程不是“越多越好”而是“哪些特征能让风控规则专家点头”贷款违约预测中90% 的效果提升来自特征而非模型。但特征不是堆砌统计量而是把业务规则显性化。例如dti债务收入比本身是数值但风控经验是dti 40%是高危阈值 → 构造二值特征is_dti_risky (df[dti] 40).astype(int)delinq_2yrs两年内逾期次数为 0 是常态但delinq_2yrs 1和delinq_2yrs 2风险跃升不同 → 拆成has_delinq_once,has_delinq_twice_plusearliest_cr_line最早信用线时间是日期但真正有用的是“信用龄” →df[credit_age_years] (pd.to_datetime(today) - pd.to_datetime(df[earliest_cr_line])).dt.days // 365这些都不是“机器学习自动发现”的而是你和风控同事喝咖啡时聊出来的。代码实现如下# 构造业务敏感特征非自动特征衍生 df[is_dti_risky] (df[dti] 40).astype(int) df[has_delinq_once] (df[delinq_2yrs] 1).astype(int) df[has_delinq_twice_plus] (df[delinq_2yrs] 2).astype(int) df[credit_age_years] ( pd.to_datetime(today) - pd.to_datetime(df[earliest_cr_line]) ).dt.days // 365 # 对类别型变量做目标编码比 one-hot 更抗稀疏且带业务含义 from sklearn.model_selection import KFold def target_encode(train_df, test_df, col, target_colloan_status, smooth10): 用 K 折目标编码替代简单均值防过拟合 train_df[f{col}_target_enc] 0 test_df[f{col}_target_enc] 0 kf KFold(n_splits5, shuffleTrue, random_state42) for tr_idx, val_idx in kf.split(train_df): tr_part train_df.iloc[tr_idx] val_part train_df.iloc[val_idx] # 计算平滑目标均值 global_mean train_df[target_col].mean() agg tr_part.groupby(col)[target_col].agg([mean, count]) smooth_mean (agg[mean] * agg[count] global_mean * smooth) / (agg[count] smooth) val_part[f{col}_target_enc] val_part[col].map(smooth_mean).fillna(global_mean) train_df.loc[val_idx, f{col}_target_enc] val_part[f{col}_target_enc] # 测试集用全量训练集编码 agg_full train_df.groupby(col)[target_col].agg([mean, count]) smooth_mean_full (agg_full[mean] * agg_full[count] global_mean * smooth) / (agg_full[count] smooth) test_df[f{col}_target_enc] test_df[col].map(smooth_mean_full).fillna(global_mean) return train_df, test_df # 示例对 grade信用等级做目标编码 df_train, df_test target_encode(df_train, df_test, grade, loan_status)参数说明smooth10是经验值表示“至少有 10 个样本才相信该分组的违约率”避免小众等级如G因样本少导致编码失真target_colloan_status需提前转为 0/1loan_status.map({Charged Off: 1, Fully Paid: 0})这是后续所有模型的标签基础。2.3 模型训练XGBoost 不是“默认最优”而是“可控性最强”的起点别被论文里的复杂模型迷惑。在贷款场景XGBoost 是平衡效果、可解释性、部署成本的黄金选择。它的scale_pos_weight参数直接对应“坏账样本稀缺性”max_depth4强制模型学习浅层规则避免过拟合于噪声early_stopping_rounds50防止在验证集上盲目优化。完整训练脚本如下from xgboost import XGBClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report, roc_auc_score # 准备特征与标签剔除原始 ID、时间字段等泄露变量 feature_cols [c for c in df.columns if c not in [id, loan_status, issue_d, earliest_cr_line]] X df[feature_cols].copy() y df[loan_status].map({Charged Off: 1, Fully Paid: 0}) # 分层抽样保证训练/验证集坏账比例一致 X_train, X_val, y_train, y_val train_test_split( X, y, test_size0.2, stratifyy, random_state42 ) # 计算 scale_pos_weight负样本数 / 正样本数应对坏账率低 pos_count y_train.sum() neg_count len(y_train) - pos_count scale_pos_weight neg_count / pos_count # 通常在 5~15 之间 model XGBClassifier( objectivebinary:logistic, scale_pos_weightscale_pos_weight, # 关键不设此参数模型会把坏账全判为好账 max_depth4, # 限制树深增强可解释性 learning_rate0.1, n_estimators500, subsample0.8, colsample_bytree0.8, random_state42, eval_metricauc, early_stopping_rounds50, verbosity1 ) model.fit( X_train, y_train, eval_set[(X_val, y_val)], verboseTrue ) # 验证集评估 y_pred_proba model.predict_proba(X_val)[:, 1] print(fAUC on validation set: {roc_auc_score(y_val, y_pred_proba):.4f}) print(classification_report(y_val, (y_pred_proba 0.3).astype(int)))逻辑说明scale_pos_weight是 XGBoost 处理不平衡数据的原生方案比 SMOTE 这类过采样更稳定不会伪造不存在的坏账模式max_depth4让单棵树最多 4 层意味着最终模型可被拆解为“if dti40 and credit_age_years3 then riskhigh”这类业务能 audit 的规则eval_set和early_stopping_rounds是防止过拟合的硬约束不是可选项。3. 避坑贷款违约预测中 5 个让模型上线失败的真实翻车点模型在 Jupyter 里 AUC 0.85 很漂亮但一旦接入生产系统就会暴露那些“看似合理实则致命”的细节。以下是我踩过的坑按发生频率排序每一条都附带线上事故还原和修复动作。3.1 现象模型在测试集 AUC 0.82上线后首周 KS 下降到 0.3原因训练时用了issue_d放款日期做时间序列划分但实际部署时新数据issue_d是未来日期导致特征mths_since_issue_d放款月数为负值XGBoost 把负值当异常值处理预测全部坍缩。解决严格禁止使用任何含时间信息的字段做特征或划分。用train_test_split(stratifyy)替代时间划分若必须考虑时间效应用TimeSeriesSplit并确保验证集日期严格晚于训练集且所有时间相关特征如mths_since_issue_d在预测时用当前日期动态计算而非存储静态值。3.2 现象home_ownership字段 one-hot 后模型报错ValueError: Number of features of the model must match the input原因训练时home_ownership有 6 类RENT,OWN,MORTGAGE,OTHER,NONE,ANY但线上新客出现home_ownershipNULL数据库空值未清洗pd.get_dummies()未覆盖该值导致预测时列数不匹配。解决所有类别型字段必须做“保底编码”。用pd.get_dummies(..., prefixcol, dummy_naTrue)显式生成home_ownership_NULL列或更稳妥地用sklearn.preprocessing.OrdinalEncoder(handle_unknownuse_encoded_value, unknown_value-1)OneHotEncoder(sparse_outputFalse, handle_unknownignore)组合确保未知值被映射到全零向量。3.3 现象dti字段训练时最大值 99.9但某笔新申请dti1000录入错误模型预测概率突变为 0.999原因XGBoost 对异常值极度敏感dti1000远超训练分布树分裂时被强行归入“高风险叶节点”且无校验机制。解决在特征工程阶段加入硬截断clipping。对dti执行df[dti] df[dti].clip(upper100)对revol_util循环信用利用率执行clip(upper120)所有数值特征在fit_transform前统一做np.clip并记录截断比例若 5%需反馈业务核查数据质量。3.4 现象模型输出predict_proba概率值但业务方要求“风险等级高/中/低”直接按 0.3/0.7 切分导致高风险客户数暴增 3 倍原因阈值设定未结合业务成本。误判好客户为坏账False Positive损失营销机会误判坏客户为好账False Negative损失本金二者成本比约为 1:10。单纯按最大约登指数选阈值忽略了业务损益。解决用precision_recall_curve计算不同阈值下的业务成本。定义cost FP_cost * FP_rate FN_cost * FN_rate其中FP_cost1,FN_cost10搜索使cost最小的阈值。代码如下from sklearn.metrics import precision_recall_curve precisions, recalls, thresholds precision_recall_curve(y_val, y_pred_proba) costs [] for i, th in enumerate(thresholds): y_pred_th (y_pred_proba th).astype(int) fp ((y_pred_th 1) (y_val 0)).sum() fn ((y_pred_th 0) (y_val 1)).sum() cost 1 * fp 10 * fn # FP 成本 1FN 成本 10 costs.append(cost) optimal_th thresholds[np.argmin(costs)] print(fOptimal threshold considering business cost: {optimal_th:.3f})3.5 现象模型每天定时重训但某天feature_importance中emp_length突然从第 3 位跌到第 20 位原因emp_length字段在当天数据中大量出现n/a因上游系统故障目标编码后该列值全为全局均值失去区分度。解决建立特征稳定性监控。对每个特征计算其在滚动 30 天训练集中的std(importance)若std 0.15或importance 0.005连续 3 天则触发告警并自动剔除该特征。用mlflow或自建 SQLite 表记录每日feature_importance这是模型运维的刚需。4. 把模型输出翻译成风控语言用 SHAP 解释单笔贷款的违约归因业务方不关心 AUC他们只想知道“为什么张三这笔 20 万的装修贷系统打了 0.82 的违约概率”——这需要可解释性而不是feature_importance的全局平均。SHAPSHapley Additive exPlanations是目前最可靠的局部解释工具它能把每个特征对单样本预测的贡献量化为“shap value”且满足可加性所有 shap value 之和 模型输出 - 基准值。以下是落地 SHAP 的最小可行代码import shap # 初始化 explainer用训练集子集加速非全量 explainer shap.TreeExplainer(model) # 取一个典型样本如预测概率 0.82 的客户 sample_idx np.where(y_pred_proba 0.8)[0][0] shap_values explainer.shap_values(X_val.iloc[[sample_idx]]) # 可视化单样本解释横向条形图 shap.initjs() shap.plots.waterfall( shap.Explanation( valuesshap_values[0], base_valuesexplainer.expected_value, dataX_val.iloc[sample_idx], feature_namesfeature_cols ), max_display10 )这段代码会生成一张瀑布图从基准预测值如 0.12开始逐项叠加各特征的 shap value正向推高违约概率负向拉低最终落到该样本的预测值0.82。图中你能清晰看到dti贡献 0.35因dti62.3远超阈值credit_age_years贡献 -0.18因信用龄仅 1 年is_dti_risky贡献 0.22二值特征放大效应grade_target_enc贡献 0.15gradeD对应历史违约率 22%注意shap.Explanation的base_values是模型在训练集上的平均预测值即explainer.expected_value不是 0.5。这是 SHAP 解释正确的前提——它告诉你“相比平均水平这个客户为什么更危险”。但 SHAP 不是万能的。它在XGBoost上高效但在LightGBM上需用shap.TreeExplainer(model, model_outputprobability)显式指定对类别型目标编码特征shap value 解释的是编码后的数值需在报告中注明“grade_target_enc0.22表示该等级历史违约率为 22%”。真正的落地技巧是把 SHAP 结果固化为 API 输出字段。例如模型服务返回 JSON{ loan_id: L10001, default_prob: 0.82, risk_level: HIGH, shap_explanation: [ {feature: dti, value: 62.3, shap: 0.35, impact: strong_positive}, {feature: credit_age_years, value: 1, shap: -0.18, impact: strong_negative}, {feature: is_dti_risky, value: 1, shap: 0.22, impact: medium_positive} ] }这样风控审核员点开任意一笔高风险贷款3 秒内就能看到“DTI 过高”是主因而不是对着 50 维特征列表猜。5. 模型上线前的最后三道关卡数据漂移检测、API 封装、AB 测试设计模型训练完成只是起点能否在生产环境持续有效取决于这三道硬性关卡。它们不涉及算法却是决定项目成败的关键动作。5.1 用 PSI 检测数据漂移比 AUC 下降早两周预警AUC 掉到 0.7 才发现模型失效太晚了。数据漂移Data Drift往往先于性能下降。PSIPopulation Stability Index是金融风控最常用的漂移指标它衡量训练集与线上新数据在某个特征上的分布变化$$ PSI \sum_{i1}^{n} (P_i^{\text{actual}} - P_i^{\text{expected}}) \times \ln\left(\frac{P_i^{\text{actual}}}{P_i^{\text{expected}}}\right) $$其中P_i是第 i 个分箱的占比。PSI 0.1 表示无漂移0.1~0.25 表示轻微漂移0.25 表示严重漂移。我们用scipy.stats.ks_2samp做快速初筛再用 PSI 精确定位import numpy as np from scipy import stats def calculate_psi(expected, actual, n_bins10): 计算单特征 PSI # 合并训练集与新数据统一分箱边界 combined np.concatenate([expected, actual]) bins np.quantile(combined, np.linspace(0, 1, n_bins 1)) bins[0] min(bins[0], -np.inf) # 覆盖极小值 bins[-1] max(bins[-1], np.inf) # 覆盖极大值 # 计算各箱占比 expected_bin_counts, _ np.histogram(expected, binsbins) actual_bin_counts, _ np.histogram(actual, binsbins) expected_pct expected_bin_counts / len(expected) actual_pct actual_bin_counts / len(actual) # 计算 PSI加小常数防 log(0) psi 0 for i in range(len(expected_pct)): if expected_pct[i] 0 or actual_pct[i] 0: continue psi (actual_pct[i] - expected_pct[i]) * np.log(actual_pct[i] / expected_pct[i]) return psi # 每日监控核心特征 core_features [dti, credit_age_years, revol_util, installment] psi_results {} for feat in core_features: psi calculate_psi(X_train[feat].dropna(), X_online[feat].dropna()) psi_results[feat] psi if psi 0.25: print(fALERT: {feat} PSI {psi:.3f} 0.25, trigger retraining)参数说明n_bins10是经验值太少会丢失细节太多会受噪声干扰dropna()必须执行否则 NaN 会破坏分箱逻辑PSI 阈值0.25来自银保监《商业银行互联网贷款管理暂行办法》附件中的推荐值不是玄学。5.2 封装为 Flask API拒绝“本地跑通线上挂掉”的悲剧很多人把model.predict()写进 Flask route 就算完事结果线上并发 50 QPS 就超时。真正可用的 API 必须包含输入校验、特征预处理、模型预测、异常捕获、响应标准化。最小可靠模板如下from flask import Flask, request, jsonify import joblib import pandas as pd import numpy as np app Flask(__name__) model joblib.load(xgb_model.pkl) scaler joblib.load(preprocessor.pkl) # 包含 imputer、encoder 等 app.route(/predict, methods[POST]) def predict(): try: # 1. 输入校验 data request.get_json() required_fields [dti, credit_age_years, revol_util, grade] for field in required_fields: if field not in data: return jsonify({error: fMissing field: {field}}), 400 # 2. 构造 DataFrame保持与训练时完全一致的列顺序 df pd.DataFrame([data]) # 3. 特征工程复用训练时 pipeline df[is_dti_risky] (df[dti] 40).astype(int) df[credit_age_years] df[credit_age_years].clip(lower0, upper50) # ... 其他构造逻辑 # 4. 预处理impute, encode, scale X_processed scaler.transform(df[feature_cols]) # 5. 预测 proba model.predict_proba(X_processed)[0, 1] risk_level HIGH if proba 0.7 else MEDIUM if proba 0.3 else LOW return jsonify({ loan_id: data.get(loan_id, unknown), default_probability: float(f{proba:.4f}), risk_level: risk_level, timestamp: pd.Timestamp.now().isoformat() }) except Exception as e: return jsonify({error: str(e)}), 500 if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse) # 生产禁用 debugTrue关键点scaler.transform()必须用训练时保存的preprocessor.pkl不能现场 fitfeature_cols顺序必须与训练时完全一致用list(model.feature_names_in_)固定debugFalse是铁律否则会暴露服务器路径。5.3 设计 AB 测试用真实业务指标验证模型价值不要用 AUC 证明模型有用。要证明它真的帮银行省钱必须做 AB 测试将新申请客户随机分为 A 组旧规则审批、B 组新模型评分卡审批对比两组的实际坏账率、通过率、利润。设计要点如下指标A 组对照B 组实验计算方式目标通过率62.3%58.1%approved_count / total_applications控制在 ±5% 波动30 天坏账率4.2%3.1%charged_off_30d / approved_count下降 ≥15%单笔平均利润¥1,280¥1,350(interest_income - expected_loss) / approved_count提升 ≥3%AB 测试必须满足随机性用numpy.random.choice按申请时间哈希分组避免时段偏差隔离性A/B 组审批流完全独立互不干扰时长至少覆盖一个完整信贷周期如 3 个月因为坏账在放款后 60~90 天才集中暴露统计显著性用scipy.stats.fisher_exact检验坏账率差异2×2 列联表p-value 0.05 才认可有效。我去年在某消费金融公司落地时AB 测试跑了 87 天最终 B 组坏账率下降 22.3%p0.003但通过率只降 1.8%老板当场拍板全量切换。这才是技术人最硬的业绩。我把这套流程跑过 7 家机构从城商行到持牌消金结论很朴素贷款违约预测不是比谁模型深而是比谁把业务逻辑焊进代码里焊得牢。scale_pos_weight要设对dti要 clip 住SHAP要能解释给风控总监听PSI要每天自动发邮件告警——这些细节没做好再 fancy 的算法也是空中楼阁。现在打开你的终端复制pip install pandas scikit-learn xgboost shap flask然后照着第二章的四步走一遍你会发现自己离一个能交付的风控模型只差 200 行代码的距离。希望帮到你。本文还有配套的精品资源点击获取