Overview1. XGBoost2. LightGBM3. TensorFlow2.0 Overview 之前的文章介绍了一些推荐系统的理论知识，理论虽然很容易理解，但是实际项目当中会有一些细节问题是需要我们去解决的。比如在很多流量很大的电商以及资讯网站的推荐系统中，每天的数据其实是增长很快的，所以模型迭代的频率也是非常高的，甚至于需要每天更新。那么我们就要面临一个很现实的问题，我们每天更新模型，那么每次都要用全量数据去训练模型吗？显然这是不可能的。一来耗费资源，二来时间也不允许。那么，增量学习Incremental Learning就派上用场了。增量学习是这几年才出来的概念，其目的在于训练新数据的同时也能保留以前模型的效果，即在之前的模型基础上对新增的数据进行训练。 主流的几种机器学习框架，已经实现了增量学习的功能，接下来看一下他们是怎么做到的。 1. XGBoost 我们一般情况下用XGBoost训练一个全新的模型，其相关方法是这样的： watchlist = [(xgb_train, 'train'), (xgb_val, 'val'), (xgb_test, 'test')] # training model_path = 'init.model' model = xgb.train(params, xgb_train, num_boost, watchlist) model.save_model(model_path) 如果我们线上已经有一个效果很好的模型，需要在此基础上对新增的数据进行增量训练，那么XGBoost是这样实现的： # 指定基础模型参数，xgb_model # 传入新的增量训练数据，xgb_train_new model_new = xgb.train(params, xgb_train_new, num_boost, xgb_model=model_path) 我们来看这个参数的作用： xgb_model : file name of stored xgb model or 'Booster' instance Xgb model to be loaded before training (allows training continuation). 即在指定的模型上继续训练。 更多相关知识请参考： How can I implement incremental training for xgboost? dmlc/xgboost 2. LightGBM model = lgb.train(params, lgb_train, num_boost_round=1000, valid_sets=lgb_eval, feature_name=x_cols, early_stopping_rounds=10, verbose_eval=False, init_model=model, # 如果init_model不为None，那么就是在init_model基础上接着训练 keep_training_booster=True) # 增量训练 LightGBM实现增量训练也很简单，只需要设置init_model为当前线上的模型，且keep_training_booster设置为True即可。 更多相关知识请参考： Incremental learning using Dataset.subset LightGBM 2.1.1 python API Standard outputs infos even when disabled lightgbm.train 3. TensorFlow2.0 我们知道，TensorFlow2.0已经集成好了Keras高级API，所以，我们看看增量学习在Keras当中是如何实现的。 # 加载线上的模型 model_path = 'init.model' loaded_model = tf.keras.models.load_model(model_path) # 新数据上接着训练 history = loaded_model.fit( train_data_gen, steps_per_epoch=total_train // batch_size, epochs=epochs, validation_data=val_data_gen, validation_steps=total_val // batch_size ) 可以看到，TensorFlow2.0对于增量训练支持也是很完备的。
PySpark笔记之五：lightGBM调参之PySpark + mmlspark + Hyperopt
Overview1. 加载相关包2. 加载数据3. 定义参数空间和代价函数4. 随机搜索调参和贝叶斯优化调参 Overview 之前的文章中记录了大数据平台上lightGBM分类器的Grid Search调参方法的应用。这次我们继续用lightGBM分类器，看看另外两种常用的调参方法随机搜索Random Search和贝叶斯优化Bayesian Optimization怎么在Spark平台上使用。 1. 加载相关包 import numpy as np import pyspark spark = pyspark.sql.SparkSession.builder.appName("spark lightgbm") \ .master("spark://***.***.***.***:7077") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:0.18.1") \ .config("spark.cores.max", "20") \ .config("spark.driver.memory", "6G") \ .config("spark.executor.memory", "6G") \ .config("spark.executor.cores", "6") \ .getOrCreate() import mmlspark from mmlspark.lightgbm import LightGBMClassifier from pyspark.ml.feature import VectorAssembler from hyperopt import fmin, rand, tpe, hp, space_eval, Trials 其中hyperopt是一款强大的调参工具，实现了随机搜索和贝叶斯优化。 如果没有安装hyperopt,那么先在master节点安装hyperopt： pip3 install hyperopt 2. 加载数据 df_train = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/yangbingjiao/训练集特征.csv") df_val = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/yangbingjiao/验证集特征.csv") df_test = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/yangbingjiao/测试集特征.csv") 处理特征： feature_cols = list(df_train.columns) feature_cols.remove("label") assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") df_train = assembler.transform(df_train) df_val = assembler.transform(df_val) df_test = assembler.transform(df_test) 3. 定义参数空间和代价函数 定义参数搜索空间，以lambdaL1和lambdaL2为例： space = { 'objective':"binary", 'boostingType':'gbdt', 'isUnbalance':True, 'featuresCol':'features', 'labelCol':'label', 'maxBin':60, 'baggingFreq':1, 'baggingSeed':696, 'earlyStoppingRound':20, 'learningRate':0.1, 'maxDepth':3, 'numLeaves':128, 'baggingFraction':0.7, 'featureFraction':0.7, 'minSumHessianInLeaf':0.001, 'numIterations':800, 'verbosity':1, 'lambdaL1': hp.uniform('lambdaL1', 0.0, 200.0), 'lambdaL2': hp.uniform('lambdaL2', 0.0, 200.0), } 定义ks计算函数： import numpy as np from sklearn.metrics import roc_curve, auc def ks_score(y, y_pred, threshold=None): """ 计算KS """ fpr, tpr, thresholds = roc_curve(y, y_pred) if threshold is None: score = np.max(np.abs(tpr - fpr)) else: idx = np.digitize(threshold, thresholds) - 1 score = np.abs(tpr[idx] - fpr[idx]) return score 训练过程记录： def train_evaluate(df_train, df_val, df_test, params): lgb = LightGBMClassifier() model = lgb.fit(df_train, params) val_preds = model.transform(df_val) test_preds = model.transform(df_test) val_prob_list = [row.probability[0] for row in val_preds.select('probability').collect()] val_label_list = [row.label for row in val_preds.select('label').collect()] test_prob_list = [row.probability[0] for row in test_preds.select('probability').collect()] test_label_list = [row.label for row in test_preds.select('label').collect()] val_ks = ks_score(1-np.array(val_label_list), np.array(val_prob_list)) test_ks = ks_score(1-np.array(test_label_list), np.array(test_prob_list)) print("val ks = %f" %(val_ks)) print("test ks = %f" %(test_ks)) print("-------------------------------") return val_ks 定义代价函数： def objective(params): return -train_evaluate(df_train, df_val, df_test, params) 定义代价函数的时候一定要注意，hyperopt会求目标函数的最小值，如果目标函数是loss，那么train_evaluate之前是不用加负号-的；如果我们要求AUC或者KS等最大值，那么train_evaluate之前一定要加上负号-，这样才能得到我们想要的结果。 4. 随机搜索调参和贝叶斯优化调参 随机搜索调参优化： rand_trials = Trials() rand_best = fmin(fn=objective, space=space, algo=rand.suggest, trials=rand_trials, max_evals=5) space_eval(space, rand_best) 输出如下： val ks = 0.380907 test ks = 0.365298 ------------------------------- val ks = 0.372782 test ks = 0.367067 ------------------------------- val ks = 0.370621 test ks = 0.368570 ------------------------------- val ks = 0.383066 test ks = 0.370889 ------------------------------- val ks = 0.378644 test ks = 0.367473 ------------------------------- 100%|██████████| 5/5 [12:04<00:00, 144.83s/it, best loss: -0.38306588065580294] {'baggingFraction': 0.7, 'baggingFreq': 1, 'baggingSeed': 696, 'boostingType': 'gbdt', 'featureFraction': 0.7, 'featuresCol': 'features', 'isUnbalance': True, 'labelCol': 'label', 'lambdaL1': 73.35909229652955, 'lambdaL2': 198.46023902528685, 'learningRate': 0.1, 'maxBin': 60, 'maxDepth': 3, 'minSumHessianInLeaf': 0.001, 'numLeaves': 128, 'objective': 'binary', 'verbosity': 1} 也就是说，随机搜索调参时，KS最大可以达到0.38306588065580294。space_eval这个函数是为了找出最好的那组参数，而Trials类可以记录整个训练过程。 如果要是想用贝叶斯优化，那么这段代码应该是： tpe_trials = Trials() tpe_best = fmin(fn=objective, space=space, algo=tpe.suggest, trials=tpe_trials, max_evals=5) space_eval(space, tpe_best) 输出过程： val ks = 0.392768 test ks = 0.373026 ------------------------------- val ks = 0.383776 test ks = 0.367786 ------------------------------- val ks = 0.385806 test ks = 0.370686 ------------------------------- val ks = 0.375033 test ks = 0.370436 ------------------------------- val ks = 0.388076 test ks = 0.371288 ------------------------------- 100%|██████████| 5/5 [10:51<00:00, 130.20s/it, best loss: -0.39276827239870765] {'baggingFraction': 0.7, 'baggingFreq': 1, 'baggingSeed': 696, 'boostingType': 'gbdt', 'featureFraction': 0.7, 'featuresCol': 'features', 'isUnbalance': True, 'labelCol': 'label', 'lambdaL1': 52.782517457829606, 'lambdaL2': 20.799935111367063, 'learningRate': 0.1, 'maxBin': 60, 'maxDepth': 3, 'minSumHessianInLeaf': 0.001, 'numLeaves': 128, 'objective': 'binary', 'verbosity': 1} 使用贝叶斯优化时，KS可以达到0.39276827239870765。可以看到，贝叶斯优化还是要好于随机搜索算法的。 我们在拿到最优参数后，就可以训练最佳的模型了。 很不幸，mmlspark当中实现的lightGBM算法，当前很可能是无法复现的，也就是说少了一个控制随机性的seed。我们可以看到，LightGBMClassifier这个类，接收的参数当中，只有一个baggingSeed，用来控制baggingFraction时每次抽取哪些样本；而featureFraction显然也需要一个参数来控制每次抽取哪些特征。当然，原生的lightGBM当中的seed参数是可以控制这两个随机性的。这个问题我已经在GitHub上进行反馈了，有兴趣可以去该页面留言 lightgbm run twice with the same parameters, but got different result in validation #564，催微软更新，或者自己贡献代码。我会继续关注这个问题。 本文参考了： Parameter Tuning with Hyperopt Introduction: Bayesian Optimization using Hyperopt An Introductory Example of Bayesian Optimization in Python with Hyperopt Hyperopt tutorial for Optimizing Neural Networks’ Hyperparameters Hyperopt
PySpark笔记之四：lightGBM调参之PySpark + mmlspark + Grid Search
Overview1. 加载PySpark和mmlspark相关包2.加载数据3.建模4.预测测试集 Overview 上一篇文章PySpark笔记之三：lightGBM调参之PySpark + Grid Search，我们记录了分别用PySpark中自带的CrossValidator和更通用的生成多个分类器同时执行训练预测的方式选取最好的模型。其中CrossValidator并不能得到验证集上最佳的分类器，而是得到训练集上最佳的效果。而mmlspark当中却有更为简单的方式，既可以得到验证集上最佳的效果，也可以方便地记录我们每一组参数对应的结果，是一种很好的方式。 1. 加载PySpark和mmlspark相关包 import numpy as np import pyspark spark = pyspark.sql.SparkSession.builder.appName("spark lightgbm") \ .master("spark://***.***.***.***:7077") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:0.18.1") \ .config("spark.cores.max", "20") \ .config("spark.driver.memory", "6G") \ .config("spark.executor.memory", "6G") \ .config("spark.executor.cores", "6") \ .getOrCreate() from pyspark.ml.feature import VectorAssembler from pyspark.ml.evaluation import BinaryClassificationEvaluator import mmlspark from mmlspark.lightgbm import LightGBMClassifier from mmlspark.automl import * from mmlspark.train import TrainClassifier, ComputeModelStatistics 2.加载数据 df_train = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/训练集特征.csv") df_val = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/验证集特征.csv") df_test = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/测试集特征.csv") 处理训练集、验证集和测试集数据： feature_cols = list(df_train.columns) feature_cols.remove("label") assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") df_train = assembler.transform(df_train) df_val = assembler.transform(df_val) df_test = assembler.transform(df_test) 3.建模 lgb = LightGBMClassifier( objective="binary", boostingType='gbdt', isUnbalance=True, featuresCol='features', labelCol='label', maxBin=60, baggingFreq=1, baggingSeed=696, earlyStoppingRound=20, learningRate=0.1, #lambdaL1=1.0, #lambdaL2=45.0, maxDepth=3, numLeaves=128, baggingFraction=0.7, featureFraction=0.7, minSumHessianInLeaf=0.001, numIterations=800, verbosity=1 ) 以lambdaL1和lambdaL2为例，设置4组不同的参数，这里可以自己记录每组参数，方便和后面各模型效果对应，就不再赘述了： lightGBMs = list() for lambdaL1 in list(np.arange(1.0, 3.0, 1.0)): for lambdaL2 in list(np.arange(1.0, 3.0, 1.0)): lightGBMs.append(lgb.setLambdaL1(lambdaL1).setLambdaL2(lambdaL2)) 当然，这里可以用这种方式来设置参数，更方便： import itertools lightGBMs = list() params = itertools.product([1.0, 2.0], [1.0, 2.0]) for param in params: lightGBMs.append(lgb.setLambdaL1(param[0]).setLambdaL2(param[1])) 利用mmlspark.train模块当中的TrainClassifier类训练模型： lgb_models = [TrainClassifier(model=lgb, labelCol="label").fit(df_train) for lgb in lightGBMs] 利用mmlspark.automl当中的FindBestModel类，寻找在验证集上效果最好的模型： best_model = FindBestModel(evaluationMetric='AUC', models=lgb_models).fit(df_val) 我们可以看看最好的模型效果： best_model.getBestModelMetrics().collect() 结果如下，是一个长度为1的list： [Row(evaluation_type='Classification', confusion_matrix=DenseMatrix(2, 2, [6647.0, 578.0, 2774.0, 1219.0], False), accuracy=0.7011945088251025, precision=0.30528424743300775, recall=0.6783528102392877, AUC=0.7554637200060459)] 也可以看看所有4个模型在验证集上的效果： best_model.getAllModelMetrics().collect() 结果是个list，长度为4： [Row(model_name='TrainClassifier_64edf29511fb', metric=0.7544038268737058, parameters='featuresCol: TrainClassifier_31b56e2ce33a_features, labelCol: label, predictionCol: prediction, probabilityCol: probability, rawPredictionCol: rawPrediction'), Row(model_name='TrainClassifier_b09abcd6c7e5', metric=0.7541219526558816, parameters='featuresCol: TrainClassifier_9c85465a6f6d_features, labelCol: label, predictionCol: prediction, probabilityCol: probability, rawPredictionCol: rawPrediction'), Row(model_name='TrainClassifier_5a1bd3015154', metric=0.7528073862858738, parameters='featuresCol: TrainClassifier_ba6c57087959_features, labelCol: label, predictionCol: prediction, probabilityCol: probability, rawPredictionCol: rawPrediction'), Row(model_name='TrainClassifier_323901015bd2', metric=0.7554575473623404, parameters='featuresCol: TrainClassifier_59842f00f628_features, labelCol: label, predictionCol: prediction, probabilityCol: probability, rawPredictionCol: rawPrediction')] 4.预测测试集 predictions = bestModel.transform(df_test) metrics = ComputeModelStatistics().transform(predictions) print("Best model's AUC on test set = " + "{0:.2f}%".format(metrics.first()["AUC"] * 100)) 我们可以看看metrics里面都是什么： metrics.collect() 结果也是一个list，包含混淆矩阵，accuracy,precision和recall： [Row(evaluation_type='Classification', confusion_matrix=DenseMatrix(2, 2, [11323.0, 1777.0, 83.0, 115.0], False), accuracy=0.8601293427583095, precision=0.5808080808080808, recall=0.060782241014799156, AUC=0.7411781436942614)] 总体来说，这种方式来进行分布式训练，比PySpark自带的api更方便一些，推荐mmlspark方式。 本文参考了以下文章： Simplifying Machine Learning Pipelines with mmlspark
PySpark笔记之三：lightGBM调参之PySpark + Grid Search
Overview1. 引入依赖包2. 加载数据3. 模型构建4. 交叉验证选择模型5. 官方调参探讨 Overview 在实际机器学习工作当中，调参是我们一个重要的内容。PySpark当中就实现了一个最常用的调参方法Grid Search，我们结合lightGBM使用一下PySpark的调参。这个程序需要安装的依赖的安装方式，可以参考上一篇博客。 1. 引入依赖包 import numpy as np import pyspark spark = pyspark.sql.SparkSession.builder.appName("spark lightgbm") \ .master("spark://***.***.***.***:7077") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:0.18.1") \ .config("spark.cores.max", "20") \ .config("spark.driver.memory", "6G") \ .config("spark.executor.memory", "6G") \ .config("spark.executor.cores", "6") \ .getOrCreate() import mmlspark from mmlspark.lightgbm import LightGBMClassifier from pyspark.ml.feature import VectorAssembler from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import BinaryClassificationEvaluator from pyspark.ml import Pipeline 其中，pyspark.ml.tuning.ParamGridBuilder就是用以实现Grid Search的包。 2. 加载数据 df_train = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/训练集特征.csv") df_val = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/验证集特征.csv") df_test = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/测试集特征.csv") 处理训练集特征： feature_cols = list(df_train.columns) feature_cols.remove("label") # 从列名当中删除label才是真正的特征列表 assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") df_train = assembler.transform(df_train) 3. 模型构建 lgb = LightGBMClassifier( objective="binary", boostingType='gbdt', isUnbalance=True, featuresCol='features', labelCol='label', maxBin=60, baggingFreq=1, baggingSeed=696, earlyStoppingRound=30, learningRate=0.1, # lambdaL1=1.0, # lambdaL2=45.0, maxDepth=3, numLeaves=128, baggingFraction=0.7, featureFraction=0.7, minSumHessianInLeaf=0.001, numIterations=800, verbosity=1 ) 设置Grid Search参数组： paramGrid = ParamGridBuilder() \ .addGrid(lgb.lambdaL1, list(np.arange(1.0, 3.0, 1.0))) \ .addGrid(lgb.lambdaL2, list(np.arange(1.0, 4.0, 1.0))) \ .build() 设置完成之后，我们可以看一下参数都是哪些： for param in paramGrid: print(param.values()) 可以看到共6组参数，左边为lambdaL1，右边为lambdaL2： dict_values([1.0, 1.0]) dict_values([1.0, 2.0]) dict_values([1.0, 3.0]) dict_values([2.0, 1.0]) dict_values([2.0, 2.0]) dict_values([2.0, 3.0]) 4. 交叉验证选择模型 官方提供了两种模型选择的方式：CrossValidator和TrainValidationSplit，可以参考官方文档。CrossValidator和TrainValidationSplit的区别在于：CrossValidator会每次选取一部分训练集建模，去预测另外一部分训练集，这样会有K个预测的分数（K折交叉验证），最后模型的预测分数为K个分数的平均；而TrainValidationSplit则只会训练预测一次。这里，我们试着给出一个CrossValidator的例子。 evaluator = BinaryClassificationEvaluator(labelCol="label", metricName="areaUnderROC") cross_vallidator = CrossValidator(estimator=lgb, estimatorParamMaps=paramGrid, evaluator=evaluator, numFolds=3) model = cross_vallidator.fit(df_train) 最后，我们可以得到最好的模型，并看最好的模型的特征重要性： model.bestModel model.bestModel.getFeatureImportances() 但是，官方api查看每个结果对应的超参数却是非常不友好，我们只好自己想办法，这里我们参考了这篇博客LightGBM Hyper Parameters Tuning in Spark： def params_extract(model): """ function extact hyperparameter information from a CrossValidatorModel input: a CrossValidatorModel instance, model fit by CrossValidator in pyspark.ml.tuning output: a dictionary with key(hyperparameters setting), value(evaluator's metrics, r2, auc,...) """ length = len(model.avgMetrics) res = {} for i in range(length): s = "" paraDict = model.extractParamMap()[model.estimatorParamMaps][i] for j in paraDict.keys(): s += str(j).split("__")[1] + " " s += str(paraDict[j]) + " " res[s.strip()] = model.avgMetrics[i] return {k: v for k, v in sorted(res.items(), key=lambda item: item[1])} 我们试一下： params_extract(model) 输出如下： {'lambdaL1 1.0 lambdaL2 2.0': 0.7300710699287217, 'lambdaL1 1.0 lambdaL2 1.0': 0.7307078416518147, 'lambdaL1 1.0 lambdaL2 3.0': 0.7310860740685388, 'lambdaL1 2.0 lambdaL2 1.0': 0.7312930665848859, 'lambdaL1 2.0 lambdaL2 2.0': 0.7317563737747359, 'lambdaL1 2.0 lambdaL2 3.0': 0.7327463775422981} 5. 官方调参探讨 我们通过CrossValidator可以获得最佳的模型，但是会有一个问题：这个最佳是拟合训练集的最佳，而不是我们给出的验证集的最佳；即使是TrainValidationSplit，我们也不能自定义验证集并传入，只能随机选择验证集。这样对于那些样本时间先后顺序不敏感的数据是影响不大的，比如图像等，但是对于交易类数据，我们希望可以根据时间先后顺序自定义训练集，验证集和测试集，并且根据验证集的效果来确定最佳参数和模型。因此，我们就需要换种方式达到目的。 import numpy as np import pyspark spark = pyspark.sql.SparkSession.builder.appName("spark lightgbm") \ .master("spark://***.***.***.***:7077") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:0.18.1") \ .config("spark.cores.max", "20") \ .config("spark.driver.memory", "6G") \ .config("spark.executor.memory", "6G") \ .config("spark.executor.cores", "6") \ .getOrCreate() import mmlspark from mmlspark.lightgbm import LightGBMClassifier from pyspark.ml.feature import VectorAssembler from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import BinaryClassificationEvaluator # 加载数据 df_train = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/训练集特征.csv") df_val = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/验证集特征.csv") df_test = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/测试集特征.csv") feature_cols = list(df_train.columns) feature_cols.remove("label") assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") df_train = assembler.transform(df_train) df_val = assembler.transform(df_val) df_test = assembler.transform(df_test) lgb = LightGBMClassifier( objective="binary", boostingType='gbdt', isUnbalance=True, featuresCol='features', labelCol='label', maxBin=60, baggingFreq=1, baggingSeed=696, earlyStoppingRound=20, learningRate=0.1, #lambdaL1=1.0, #lambdaL2=45.0, maxDepth=3, numLeaves=128, baggingFraction=0.7, featureFraction=0.7, minSumHessianInLeaf=0.001, numIterations=800, verbosity=1 ) lightGBMs = list() for lambdaL1 in list(np.arange(1.0, 3.0, 1.0)): for lambdaL2 in list(np.arange(1.0, 4.0, 1.0)): lightGBMs.append(lgb.setLambdaL1(lambdaL1).setLambdaL2(lambdaL2)) evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", metricName="areaUnderROC") metrics = [] models = [] # 选择验证集效果最好的模型 for learner in lightGBMs: model = learner.fit(df_train) models.append(model) scoredData = model.transform(df_val) metrics.append(evaluator.evaluate(scoredData)) best_metric = max(metrics) best_model = models[metrics.index(best_metric)] # 得到测试集上AUC scored_test = best_model.transform(df_test) print(evaluator.evaluate(scored_test)) 当然，用这种方式的话，我们就可以很方便地记录每组参数对应的结果了，在评价验证集效果时，也可以用KS或者Gini等指标了，这里就不再赘述。 本文主要参考了：How to build machine learning model at large scale with Apache Spark and LightGBM for credit card fraud detection? Simplifying Machine Learning Pipelines with mmlspark
Overview1. 安装mmlspark2. 引入依赖包3. 加载数据4. 模型构建 Overview 就目前的PySpark版本2.4.5来说，虽有pyspark.ml这个模块可以进行机器学习，但是都是一些工业界不太常用的算法，而XGBoost和LightGBM这样的常用算法还没有集成。幸好微软前几年发布了mmlspark这个包，其中包含了深度学习和LightGBM等算法，可以和PySpark无缝对接。下面我们看看怎么用PySpark和mmlspark来运行LightGBM。 1. 安装mmlspark 首先，我们默认已经安装好了PySpark，如果没有安装，那么安装命令如下： pip3 install pyspark 然后，安装mmlspark的方式如下，命令行输入： pyspark --packages com.microsoft.ml.spark:mmlspark_2.11:0.18.1 或者直接在jupyter notebook中，使用PySpark时，这样启动： import pyspark spark = pyspark.sql.SparkSession.builder.appName("MyApp") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:1.0.0-rc1") \ .config("spark.jars.repositories", "https://mmlspark.azureedge.net/maven") \ .getOrCreate() import mmlspark 这个包比较大，第一次安装需要较长时间。我们服务器上maven仓库当中，mmlspark版本是0.18.1，而这个版本训练时会导致一个bug，下面会提到。 详细安装方式可参考官方文档：Azure / mmlspark，最新的mmlspark版本是1.0.0-rc1。 2. 引入依赖包 import numpy as np import pyspark spark = pyspark.sql.SparkSession.builder.appName("spark lightgbm") \ .master("spark://***.***.***.***:7077") \ .config("spark.jars.packages", "com.microsoft.ml.spark:mmlspark_2.11:0.18.1") \ .config("spark.cores.max", "20") \ .config("spark.driver.memory", "6G") \ .config("spark.executor.memory", "6G") \ .config("spark.executor.cores", "6") \ .getOrCreate() import mmlspark from mmlspark.lightgbm import LightGBMClassifier from pyspark.ml.feature import VectorAssembler # from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import BinaryClassificationEvaluator from pyspark.ml import Pipeline 其中，VectorAssembler这个类，可以将特征转换成一个向量，作为分类器的输入。BinaryClassificationEvaluator可以评价预测效果。Pipeline相当于Spark的流水线，将各个步骤连接在一起后，一起由Spark运行。 3. 加载数据 df_train = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/训练集特征.csv") df_val = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/验证集特征.csv") df_test = spark.read.format("csv") \ .option("inferSchema", "true") \ .option("header", "true") \ .option("sep", ",") \ .load("hdfs://***.***.***.***:39000/young/测试集特征.csv") 首先，要保证Hadoop上有我们需要的数据集文件。本地文件上传到Hadoop的命令很简单： hdfs dfs -copyFromLocal 训练集特征.csv 验证集特征.csv 测试集特征.csv /young 处理训练集特征： feature_cols = list(df_train.columns) feature_cols.remove("label") # 从列名当中删除label才是真正的特征列表 assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") VectorAssembler之后，原数据DataFrame当中会多出一列，列名为features，这个字段内容其实就是一个行向量，里面是该样本的每一个特征。 4. 模型构建 lgb = LightGBMClassifier( objective="binary", boostingType='gbdt', isUnbalance=True, featuresCol='features', labelCol='label', maxBin=60, baggingFreq=1, baggingSeed=696, earlyStoppingRound=30, learningRate=0.1, lambdaL1=1.0, lambdaL2=45.0, maxDepth=3, numLeaves=128, baggingFraction=0.7, featureFraction=0.7, # minSumHessianInLeaf=1, numIterations=800, verbosity=30 ) 其中，minSumHessianInLeaf这个参数，在设置为整型数值时，会抛ClassCastException异常， Py4JJavaError: An error occurred while calling o1592.w. : java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.Double at scala.runtime.BoxesRunTime.unboxToDouble(BoxesRunTime.java:114) at org.apache.spark.ml.param.DoubleParam.w(params.scala:330) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) at py4j.Gateway.invoke(Gateway.java:282) at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at py4j.commands.CallCommand.execute(CallCommand.java:79) at py4j.GatewayConnection.run(GatewayConnection.java:238) at java.lang.Thread.run(Thread.java:748) 这是一个bug，在最新版本当中，这个bug应该已经改过来了。不过，对0.18.1这样比较老的版本，我们可以将这个变量设置为double或者float类型，例如1写成1.0，那么这个异常就不会出现了。 装载各个阶段到Pipeline流水线中，执行训练： stages = [assembler, lgb] pipeline_model = Pipeline(stages=stages) model = pipeline_model.fit(df_train) 预测三个数据集: train_preds = model.transform(df_train) val_preds = model.transform(df_val) val_preds = model.transform(df_val) 我们看看模型训练效果： binaryEvaluator = BinaryClassificationEvaluator() print ("Train AUC: " + str(binaryEvaluator.evaluate(train_preds, {binaryEvaluator.metricName: "areaUnderROC"}))) print ("Val AUC: " + str(binaryEvaluator.evaluate(val_preds, {binaryEvaluator.metricName: "areaUnderROC"}))) print ("Test AUC: " + str(binaryEvaluator.evaluate(test_preds, {binaryEvaluator.metricName: "areaUnderROC"}))) 输出三个AUC： Train AUC: 0.8236477163798177 Val AUC: 0.7548691969544107 Test AUC: 0.749396806843621 当然，我们可以把预测概率结果和真实label取出来，方便进行计算其他自定义指标，例如KS等。 train_prob_list = [row.probability[0] for row in train_preds.select('probability').collect()] train_label_list = [row.label for row in train_preds.select('label').collect()] val_prob_list = [row.probability[0] for row in val_preds.select('probability').collect()] val_label_list = [row.label for row in val_preds.select('label').collect()] test_prob_list = [row.probability[0] for row in test_preds.select('probability').collect()] test_label_list = [row.label for row in test_preds.select('label').collect()] DataFrame在经过一系列transform之后，会多出4列，分别为features|rawPrediction|probability|prediction|，rawPrediction是树模型最后的一对儿得分，和为0；在经过sigmoid之后，得到probability里的一对儿概率值，其和为1，分别表示模型判定该样本为两个分类的可能性；而prediction则是模型预测的样本类别。 本文参考了以下文章： 在集群上训练一个机器学习模型（以lightGBM为例） 利用pyspark.ml训练lightgbm模型的流程 Get Started with PySpark and LightGBM Build XGBoost / LightGBM models on large datasets — what are the possible solutions?
Overview1. 特征挖掘2. 特征处理2.1 样本不均衡问题2.2 空值问题2.3 类别特征处理2.4 归一化和标准化2.5 特征选择和降维2.6 特征评估3. 模型构建3.1 数据集准备3.2 机器学习算法选择3.3 过拟合与欠拟合3.3.1 过拟合3.3.2 欠拟合3.4 XGBoost调参训练demo3.5 模型预测3.6 多模型融合3.6.1 不同类型的样本产生的多个模型进行融合3.6.2 不同类型的机器学习算法产生的多个模型进行融合 Overview 互联网金融行业主要涉及以下方面：保险、理财、基金、信贷、虚拟信用卡、催收等，主要用到以下类型的模型： 营销模型、召回模型、排序模型（支付宝蚂蚁财富基金推荐红包发放等） 新用户反欺诈模型 新老用户信用额度模型（蚂蚁借呗，腾讯微粒贷，京东金条，微博钱包，美团借钱，各类小贷等） 虚拟信用卡贷中风控模型（蚂蚁花呗，京东白条等） 贷后催收模型 我在互联网金融行业已经工作两年了，也积累了不少信贷数据特征挖掘和信贷模型开发经验。现在将这些经验记录一下，算是对两年时间的一个总结。总体上，这个领域和其他领域在机器学习模型构建的流程上区别不大，主要精力都是在特征挖掘和模型构建上。 1. 特征挖掘 显而易见，这个领域，用户的资产和负债数据是最有效的数据，但是除了银行或者蚂蚁金服和京东金融等大金融平台之外，我们一般情况下很难拿到用户的资产数据，我们最多能拿到用户的第三方征信数据，即用户负债数据（国内常见为同盾，百融，百行征信等），结合用户授权我们爬取的用户的通讯录，APP安装列表，淘宝京东APP订单数据等，以及用户在本公司的借贷历史行为记录和用户登录APP的埋点数据，我们可以做出很多特征，三方多头征信特征是最有效的，其次对于老用户，用户的借贷历史行为也很有效。其中，时间序列数据是特征工程的重头戏，对于和时间相关的订单类，行为类等数据，我们可以用类似的方法开发出一大批时间序列特征，其形式均为“某时间段内 + 某种行为发生的次数/金额/比例/时长间隔/持续时间 + 最值/q1/q2/q3/均值/众数/中位数”，例如"三个月内用户在本平台贷款金额序列的均值"。这部分特征由三部分组成： 时间长度 行为类型 统计量 另外，用户授权我们拿到他们的通讯录之后，我们可以通过他的一二度人脉在本公司的表现，来做关系图特征。我们取得用户的APP安装列表之后，可以将APP按照风险等级给分为3或者4类：714高炮类，小贷类，普通消费金融与正常金融类，学习以及其他正向类。另外，如果是贷后催收模型，还需要加入用户的催收表现特征。 2. 特征处理 2.1 样本不均衡问题 我们在业务上遇到的很多数据，几乎都是正负样本不平衡的，一般情况下是好样本数量远大于坏样本，用户客群质量越高，样本差异越大。小贷平台好坏样本比例一般在3:1到20:1之间，大平台差异更大，这样设计一个课题：样本不均衡问题。 样本不均衡时，我们有以下这些处理方式（通用方式，不只信贷领域），但基本思想就是"再缩放（rescaling）"： 较多类欠采样 较少类过采样 预测时阈值移动 合成数据或无label样本学习 权重缩放或惩罚权重 异常检测的方式（如果好坏样本数量悬殊，例如欺诈检测） 组合/集成学习 个人以为，我们在信贷工作当中，业务样本真实性还是很重要的，因此，不建议用SMOTE等方法进行插值产生坏样本。当然像图像，语音，文本等具象的数据，人可以轻松判断正负样本的领域除外。 2.2 空值问题 我们的特征中是有不少空值的，我们可以进行零值填充、均值填充或者众数填充，不过我们一般不会对这些空值进行填充，因为填充大概率会引入误差。另外，某些机器学习算法，其思想是计算距离来划分类别，例如SVM和KNN，对于这些算法，有空值的情况是不适合的，填充空值也是不适合的，因为填充不同的数据会对结果产生较大影响。而且，我们业务上特征维度是非常高的，即便进行了特征选择和降维，因此有很多算法效率是很低的，不推荐使用，这是后话。 因此，我们原则上，保留空值不处理。 2.3 类别特征处理 类别特征大家一般都是做one-hot处理，因为大部分算法是只支持数字型特征的。但是有少部分算法是支持类别型特征，而不用做独热编码的，例如LightGBM。 2.4 归一化和标准化 归一化和标准化都属于无量纲化，其主要目的就是为了避免数值问题和加快模型收敛速度。有些模型最好要进行无量纲化，例如线性回归，SVM，kNN，逻辑回归，k-Means，神经网络等；而XGBoost，lightGBM，朴素贝叶斯，决策树和随机森林等算法通常是不需要无量纲化的。 归一化和标准化之前的文章数据挖掘之数据标准化(Normalization)写过，这里就不赘述了。 2.5 特征选择和降维 特征选择的主要作用是去掉冗余的特征加速学习过程和减轻过拟合，降维也有类似的目的。但是某些降维算法使用之后，得到的低维度特征和降维之前的特征的量纲就发生了变化，模型解释性会受到影响，因为我们不知道得到的特征是什么意义。这是降维和特征选择的一点区别。而且，我们在业务当中，可能同类型的三方数据会有好几种，这些数据效果相差无几，同时使用也不会有更多增益，所以我们需要选择合适的特征，以节约数据成本。 特征选择主要有以下三类： 过滤式 包裹式 嵌入式 降维有两大类: 无监督降维（PCA等） 有监督降维（LDA等） 降维这个领域，Chris博士有较深研究。 2.6 特征评估 我们如果选择好了特征，想知道哪些特征对模型作用更大，除了训练模型之后查看gain和weight，也可以在训练模型之前用以下几个指标去初步判断一下： WOE IV 单特征AUC 单特征KS 3. 模型构建 3.1 数据集准备 互联网金融领域受政策，市场，经济等外部因素影响较大，因此客群变化也是很快的。例如2018年1月左右和2019年3月15日左右，受政策影响，互联网金融市场发生两次“风暴期”，导致坏账率急剧提高，因为我们选用样本的时候要考虑这些因素。 我们离线跑机器学习模型，一般情况下会将数据集按时间先后顺序划分为：训练集，验证集和测试集。此外，还要准备最新的业务数据（尚未产生表现的样本）作为回归集以作为监控模型稳定性的benchmark。原则上，我们最好保持几个数据集的客群分布一致，即其间未发生重大市场震荡和较大业务调整。如果不能满足以上条件，我们至少需要满足验证集，测试集，回归集和当前线上业务的用户分布一致。且训练集可以抽样，验证集和测试集最好不要抽样。 经典的训练集，验证集和测试集的划分为7 : 1.5 : 1.5，但是如果我们有几十万到几千万条样本，再这样按比例划分就不太合适了。我们最好使验证集和测试集保证足够的正样本和负样本，有稳定的统计学表现，将偶然性降至合理范围。且原则上如果机器性能足够，我们建议保留所有训练集而不采样，这样会尽最大可能保留样本的多样性，带来的不平衡问题可以用合适的机器学习算法包来解决（例如SVM和XGBoost算法工程实现中分别有class_weight和scale_pos_weight等参数调整样本权重）。 另外，信贷领域label通常是这样定义的：该笔贷款逾期天数是否超过n天，n可以选择0,3,7,30等等。 3.2 机器学习算法选择 如果业务不是新开展的，那么我们一般要沿用前人使用的算法，之后再慢慢迭代尝试同类型更好的算法（XGBoost替换为lightGBM或CatBoost），或者数据量积累足够大时尝试神经网络类算法。如果是新开展的业务，那我们可以这么选择： 样本规模较小，选SVM（不适合稀疏数据） 样本规模适中，XGBoost或lightGBM或CatBoost 样本规模较大，XGBoost或lightGBM或CatBoost或神经网络 互联网金融领域，XGBoost还是比较流行的算法，包括蚂蚁金服，京东金融，360金融都在大规模使用。我们下面就以此算法为例，做记录。 3.3 过拟合与欠拟合 过拟合和欠拟合是建模当中经常出现又非常重要的知识点，过拟合出现更多一些。我们主要有以下手段去缓解过拟合和欠拟合的影响。 3.3.1 过拟合 过拟合就是模型能够很好的拟合训练集，但是对于验证集和测试集的泛化效果非常差，这种情况也叫高方差。 更多训练样本 特征选择 减小模型规模，修改模型架构 L1和L2正则化 early stop dropout 权重衰减 加入噪音 bagging集成 3.3.2 欠拟合 欠拟合就是模型对训练集不能很好拟合，这种情况也叫高偏差。 增加模型规模，修改模型架构 挖掘更好的特征 增加特征 较小正则化 值得注意的是，欠拟合和过拟合在实际任务当中有可能会同时存在的。 3.4 XGBoost调参训练demo 本demo不含任何敏感数据。 # 引入数据处理和模型训练必须的包 import numpy as np import pandas as pd import xgboost as xgb # 加载特征列表 name_list = pd.read_csv('feature_list.txt', header=None, index_col=0) my_feature_names = list(name_list.transpose()) print(len(my_feature_names)) # 加载样本 df_total = pd.read_csv('data_total.csv') print(df_total.shape) # 划分数据集 print(df_total.apply_time.min()) print(df_total.apply_time.max()) df_train = df_total[df_total.apply_time < '2020-01-21 00:00:00'] df_val = df_total[(df_total.apply_time >= '2020-01-21 00:00:00') & (df_total.apply_time < '2020-02-01 00:00:00')] df_test = df_total[df_total.apply_time >= '2020-02-01 00:00:00'] df_train.label.mean() df_val.label.mean() df_test.label.mean() # 数据处理 train_x = df_train[my_feature_names] train_y = df_train.get('label') val_x = df_val[my_feature_names] val_y = df_val.get('label') test_x = df_test[my_feature_names] test_y = df_test.get('label') class_weight = (1-train_y.mean())/train_y.mean() #负样本(0样本)与正样本(1样本)的数量比 data_train_X = train_x.astype(float) data_val_X = val_x.astype(float) data_test_X = test_x.astype(float) xgb_train = xgb.DMatrix(data_train_X, label=train_y) xgb_val = xgb.DMatrix(data_val_X, label=val_y) xgb_test = xgb.DMatrix(data_test_X, label=test_y) 将我们需要的数据处理完成之后，就可以导入模型进行调参了。 调参方式有很多种，比如网格搜索和贝叶斯调参等，我这里就用我经常用的贝叶斯调参，因为XGBoost参数较多，用网格搜索不太划算。 import functools from bayes_opt import BayesianOptimization scale_pos_weight_ratio = class_weight from sklearn.metrics import roc_curve # 计算KS def ks_measure(preds, label): fpr, tpr, thresholds = roc_curve(label, preds) threshold = None if threshold is None: score = np.max(np.abs(tpr - fpr)) else: idx = np.digitize(threshold, thresholds) - 1 score = np.abs(tpr[idx] - fpr[idx]) return score def new_eval_xgb_ks(is_cv, metrics, X_train, y_train, X_validation, y_validation, X_test, y_test,eta, gamma, max_depth, min_child_weight, subsample, colsample_bytree, alpha,lambda_s): ''' 我们用验证集效果最好的参数作为最终训练的参数 ''' params = dict() params['eta'] = eta params['gamma'] = gamma params['max_depth'] = int(max_depth) params['min_child_weight'] = min_child_weight params['subsample'] = subsample params['colsample_bytree'] = colsample_bytree params['alpha'] = alpha params['lambda'] = lambda_s params['silent'] = 1 params['seed'] = 696 params['objective'] = 'binary:logistic' params['scale_pos_weight'] = scale_pos_weight_ratio params['eval_metric'] = ['auc'] n_estimators = 800 print(params) # 是否对这组参数做cross validation if is_cv: pass else: dtr = X_train dva = X_validation eval_list = [(dtr,"train"), (dva,"validation")] clf = xgb.train(params, dtr, n_estimators, eval_list, verbose_eval=30, early_stopping_rounds=20) y_pred = clf.predict(dva, ntree_limit=clf.best_ntree_limit) dtest = X_test y_pred_test = clf.predict(dtest, ntree_limit=clf.best_ntree_limit) # 用ks选参 if metrics == "ks": result = ks_measure(y_pred, y_validation) print("test ks: ", ks_measure(y_pred_test, y_test)) return result def new_model_selection_bayes_opt(X_train, y_train, X_validation, y_validation, X_test, y_test,classifier='xgb', metrics="ks", is_cv=False): num_iter = 150 init_points = 50 if classifier == 'xgb': eval_xgb_ks_part = functools.partial(new_eval_xgb_ks, is_cv, metrics, X_train, y_train, X_validation, y_validation, X_test, y_test) classifierBO = BayesianOptimization(eval_xgb_ks_part, {'eta': (0.01, 0.5), 'gamma': (0, 20), 'max_depth': (3, 5), 'min_child_weight': (1, 200), 'subsample': (0.6, 1), 'colsample_bytree': (0.6, 1), 'alpha': (1, 100), 'lambda_s':(1,200)}) classifierBO.maximize(init_points=init_points, n_iter=num_iter) best_params = classifierBO.max['params'] best_val = classifierBO.max['target'] print(best_val) print(best_params) return classifierBO classifier_max = new_model_selection_bayes_opt(xgb_train, train_y, xgb_val, val_y, xgb_test, test_y, classifier='xgb') 调参结束之后，我们查看验证集上最好的表现和对应的参数。 target_bst = classifier_max.max['target'] params_bst = classifier_max.max['params'] 然后开始训练模型： # 参数 params = { 'booster': 'gbtree', 'objective': 'binary:logistic', 'alpha': params_bst['alpha'], 'colsample_bytree': params_bst['colsample_bytree'], 'eta': params_bst['eta'], 'gamma': params_bst['gamma'], 'lambda': params_bst['lambda_s'], 'max_depth': int(params_bst['max_depth']), 'min_child_weight': params_bst['min_child_weight'], 'subsample': params_bst['subsample'], 'silent': 1, 'seed': 696, 'scale_pos_weight': scale_pos_weight_ratio, 'eval_metric': ['auc'] } num_boost = bst_num_boost # 从输出日志中可以找到最好的树棵数 watchlist = [(xgb_train, 'train'), (xgb_val, 'val'), (xgb_test, 'test')] # 训练 model = xgb.train(params, xgb_train, num_boost, watchlist) # 保存模型 model.save_model('result_model.model') # 用于存储训练出的模型 评估模型，我们一般用KS，其实AUC也可以，但是不如KS通用。KS的主要意义就是评估模型对好坏用户的区分度，值越高，区分度越好。相应的，我们的信用额度策略同事会利用我们的模型打分，对不同区间的用户赋予不同的额度，以实现业务利润最大化。 评估函数如此定义 import numpy as np import pandas as pd from sklearn.metrics import roc_curve, auc def model_evaluation(y, y_pred): """ 输入真实的y和预测的y的概率 输出ks和auc """ result = {} result['ks']= ks_score(y, y_pred) result['auc'] = auc_score(y, y_pred) return result def ks_score(y, y_pred, threshold=None): """ 计算KS """ fpr, tpr, thresholds = roc_curve(y, y_pred) if threshold is None: score = np.max(np.abs(tpr - fpr)) else: idx = np.digitize(threshold, thresholds) - 1 score = np.abs(tpr[idx] - fpr[idx]) return score def auc_score(y, y_pred): """ 计算AUC """ fpr, tpr, thresholds = roc_curve(y, y_pred) roc_auc = auc(fpr, tpr) return roc_auc train_pred = model.predict(xgb_train) val_pred = model.predict(xgb_val) test_pred = model.predict(xgb_test) model_evaluation(train_y, train_pred) model_evaluation(val_y, val_pred) model_evaluation(test_y, test_pred) 如果验证集和测试集KS接近，在0.2以上，且和训练集的KS差距不大，那么这个模型的效果就是可以接受的。当然实际工作当中，我们评估的步骤更加严格，我们需要看特征重要性，看看排名在前面的特征是不是符合我们的认知，单特征的表现以及单特征分bin表现等等。 3.5 模型预测 我们模型保存之后，在线上应用，需要这样导入并预测。 model = xgb.Booster() model.load_model('result_model.model') reg_pred = model.predict(xgb_reg) # 假设xgb_reg是回归集处理过的数据 online_pred = model.predict(xgb_online) # 假设xgb_online是上线后的数据集处理过的数据 我们可以将回归集的预测结果reg_pred保存下来，然后在线上监控模型预测的稳定性，主要指标是PSI。 def psi(bench, target, group=None): if group is None: group = int(len(bench)**0.5) labels_q = np.percentile(bench,[(100.0//group)*i for i in range(group + 1)],interpolation = "nearest") ben_pct = (pd.cut(bench, bins = np.unique(labels_q),include_lowest = True).value_counts())/len(bench) target_pct = (pd.cut(target,bins = np.unique(labels_q),include_lowest = True).value_counts())/len(target) target_pct = target_pct.sort_index() ben_pct = ben_pct.sort_index() psi = sum((target_pct - ben_pct)*np.log(target_pct/ben_pct)) return psi 评估模型稳定性： psi(reg_pred, online_pred, group=10) # 10或者20都可以，但是得保证每个bin的数量充足 如果psi < 0.1那么表示模型稳定，一般会远小于0.1。 3.6 多模型融合 实际业务当中，模型的效果还是可以有更大的提升的，这种技术手段就是多模型的融合(stacking)。关于模型融合，可以有这样两种方式： 3.6.1 不同类型的样本产生的多个模型进行融合 例如我们的业务当中，多条不同的产品线有可能会有一定的用户重合，我们可以将其他不同产品线用户样本作为不同的训练集，验证集和测试集则使用相同的本产品线的样本。这样，产生的不同的模型，对相同的测试集进行预测，生成的预测结果（概率值）作为一个新的逻辑回归模型（或者别的模型）的训练集，再进行模型训练。 3.6.2 不同类型的机器学习算法产生的多个模型进行融合 这种情况可能在kaggle等比赛当中更常见一些。相同的训练集，用不同的机器学习算法建模（最好是不同类型的算法，例如用了XGBoost之后最好就不要用LightGBM了），然后用训练好的模型对相同的测试集进行预测，生成的预测结果（概率值）作为一个新的逻辑回归模型（或者别的模型）的训练集，再进行模型训练。 当然以上两种方式可以结合使用，进一步提升模型效果。