from clickhouse_driver import Client from common.log_utils import logFactory from common.database_utils import database_util from common import constant import numpy as np import math import pandas as pd from random import shuffle from tqdm import tqdm import optuna from sklearn.metrics import log_loss from sklearn.model_selection import train_test_split from optuna.integration import LightGBMPruningCallback import lightgbm as lgb from lightgbm.sklearn import LGBMClassifier from lightgbm import plot_importance from sklearn import metrics import matplotlib.pyplot as plt import seaborn as sns import random click_client = database_util.get_client() logger = logFactory("preprocess data").log # # # 获取正负样本数量，准备随机id # tb_pos = constant.insert_pos_tb_name # tb_neg = constant.insert_neg_tb_name # positive_sql = f"select count(1) from {tb_pos}" # positive_sql_result = click_client.execute(positive_sql)[0][0] # positive_ids = range(0, positive_sql_result) # negative_sql = f"select count(1) from {tb_neg}" # negative_sql_result = click_client.execute(negative_sql)[0][0] # negative_ids = range(0, negative_sql_result) # # # 每个批次的数据量 # # batch_size = 100000 # batch_size = 10 # # 分组数量 # batch_num_pos = 1 # batch_num_neg = 1 # # # 根据每个批次的数据量计算出每个批次的row_id def partition_preserve_order(list_in, n): indices = list(range(len(list_in))) shuffle(indices) index_partitions = [sorted(indices[i::n]) for i in range(n)] return [[list_in[i] for i in index_partition] for index_partition in index_partitions] # # # def gen_train_tuple(pos_row_ids, neg_row_ids): # result = [] # for i, e, in enumerate(neg_row_ids): # pos_index = i % batch_num_pos # result.append((pos_row_ids[pos_index], neg_row_ids[i])) # return result # # # pos_row_ids = partition_preserve_order(positive_ids, batch_num_pos) # neg_row_ids = partition_preserve_order(negative_ids, batch_num_neg) # neg_row_ids = [random.sample(neg_row_ids[0], 1000000)] # pos_row_ids = [random.sample(pos_row_ids[0], 1000000)] # train_tuple = gen_train_tuple(pos_row_ids, neg_row_ids) DROP_COLUMNS = ['row_id', 'month', 'EVENT_CATEGORYNAME_C_0', 'EVENT_CATEGORYNAME_C_1', 'EVENT_CATEGORYNAME_C_2', 'EVENT_CATEGORYNAME_C_3', 'EVENT_CATEGORYNAME_C_4', 'EVENT_CATEGORYNAME_C_5', 'EVENT_CATEGORYNAME_C_6', 'EVENT_CATEGORYNAME_C_7', 'EVENT_CATEGORYNAME_C_8', 'EVENT_CATEGORYNAME_C_9', 'EVENT_CATEGORYNAME_C_10', "EVENT_CANCEL_DIFF_C_0", "EVENT_CANCEL_DIFF_C_1", "EVENT_CANCEL_DIFF_C_2", "EVENT_CANCEL_DIFF_C_3", "EVENT_CANCEL_DIFF_C_4", "EVENT_CANCEL_DIFF_C_5", "EVENT_CANCEL_DIFF_C_6" ] def draw_roc_auc(y_label, y_test): # ROC曲线绘制 fpr, tpr, thresholds = metrics.roc_curve(y_label, y_test) ##计算曲线下面积 roc_auc = metrics.auc(fpr, tpr) ##绘图 plt.clf() plt.plot(fpr, tpr, label='ROC curve (area = %0.2f)' % roc_auc) plt.plot([0, 1], [0, 1], 'k--') plt.xlim([0.0, 1.0]) plt.ylim([0.0, 1.0]) plt.legend(loc="lower right") plt.show() def draw_confusion_matrix(y_label, y_test): # 画出混淆矩阵 confusion_data = metrics.confusion_matrix(y_label, y_test) print(confusion_data) sns.heatmap(confusion_data, cmap="Greens", annot=True) plt.xlabel("Predicted labels") plt.ylabel("True labels") plt.tight_layout() plt.show() def start_train(): logger.info("准备开始加载数据") lgb_model = None train_params = { 'task': 'train', 'objective': 'binary', 'boosting_type': 'gbdt', 'learning_rate': 0.1, 'num_leaves': 30, 'tree_learner': 'serial', 'metric': {'binary_logloss', 'auc', 'average_precision'}, # l1:mae, l2:mse # 'max_bin': 3, # 较小的max_bin会导致更快的速度，较大的值会提高准确性 'max_depth': 6, # 'min_child_samples': 5, "bagging_fraction": 0.8, # 样本采样比例，同 XGBoost ，调小可以防止过拟合，加快运算速度 "feature_fraction": 0.8, # 样本采样比例，同 XGBoost ，调小可以防止过拟合，加快运算速度 "n_jobs": 8, "boost_from_average": False, 'seed': 2022, "lambda_l1": 1e-5, "lambda_l2": 1e-5, } total_train_data = pd.read_pickle('total_train_data.pkl') # 划分训练集和测试集 data_train, data_test = train_test_split(total_train_data, train_size=0.7) # data_train, data_test = train_test_split(total_train_data, total_train_data['mark'].values, train_size=0.8, # random_state=0, stratify=total_train_data['mark'].values) # df1 = data_train[data_train['mark']==0] # df2 = data_train[data_train['mark']==1] # df11 = data_test[data_test['mark']==0] # df12 = data_test[data_test['mark']==1] train_x_data = data_train[constant.feature_column_names] train_x_data.drop(columns=DROP_COLUMNS, inplace=True) train_y_data = data_train[['mark']] test_x_data = data_test[constant.feature_column_names] test_x_data.drop(columns=DROP_COLUMNS, inplace=True) test_y_data = data_test[['mark']] feature_col = [x for x in constant.feature_column_names if x not in DROP_COLUMNS] # 创建lgb的数据集 lgb_train = lgb.Dataset(train_x_data, train_y_data.values, silent=True) lgb_eval = lgb.Dataset(test_x_data, test_y_data.values, reference=lgb_train, silent=True) # clf = LGBMClassifier() # clf.fit(train_x_data, train_y_data.values) # train_predict = clf.predict(train_x_data) # test_predict = clf.predict(test_x_data) # train_predict_score = metrics.accuracy_score(train_y_data.values, train_predict) # test_predict_score = metrics.accuracy_score(test_y_data.values, test_predict) # print(train_predict_score) # print(test_predict_score) # confusion_data = metrics.confusion_matrix(test_y_data.values, test_predict) # print(confusion_data) # sns.heatmap(confusion_data, cmap="Greens", annot=True) # plt.xlabel("Predicted labels") # plt.ylabel("True labels") # plt.tight_layout() # plt.show() lgb_model = lgb.train(params=train_params, train_set=lgb_train, num_boost_round=100, valid_sets=lgb_eval, init_model=lgb_model, feature_name=feature_col, verbose_eval=False, keep_training_booster=True) # lgb_model = lgb.cv(params=train_params, train_set=lgb_train, num_boost_round=1000, nfold=5,stratified=False,shuffle=True, early_stopping_rounds=50,verbose_eval=50,show_stdv=True) lgb_model.save_model("temp.model") # 输出模型评估分数 score_train = str(dict([(s[1], s[2]) for s in lgb_model.eval_train()])) score_valid = str(dict([(s[1], s[2]) for s in lgb_model.eval_valid()])) logger.info(f"在训练集上:{score_train}") logger.info(f"在测试集上:{score_valid}") result = pd.DataFrame({ 'column': feature_col, 'importance': lgb_model.feature_importance(), }).sort_values(by='importance') sns.barplot(y=train_x_data.columns, x=lgb_model.feature_importance()) plt.tight_layout() plt.show() # pass test_predict = lgb_model.predict(test_x_data) test_result = [] for x in test_predict: if x < 0.5: test_result.append(0) else: test_result.append(1) result = metrics.classification_report(test_y_data.values, test_result) logger.info(result) draw_confusion_matrix(test_y_data.values.reshape([-1,]), test_result) draw_roc_auc(test_y_data.values.reshape([-1,]), test_predict) # # ##计算相关数据：注意返回的结果顺序 # fpr, tpr, thresholds = metrics.roc_curve(test_y_data.values.reshape([-1,]), test_predict) # ##计算曲线下面积 # roc_auc = metrics.auc(fpr, tpr) # ##绘图 # plt.clf() # plt.plot(fpr, tpr, label='ROC curve (area = %0.2f)' % roc_auc) # plt.plot([0, 1], [0, 1], 'k--') # plt.xlim([0.0, 1.0]) # plt.ylim([0.0, 1.0]) # plt.xlabel('False Positive Rate') # plt.ylabel('True Positive Rate') # plt.legend(loc="lower right") # plt.show() pass def test_data_valid(): logger.info("准备开始加载数据") # # 获取正负样本数量，准备随机id tb_pos = constant.insert_pos_tb_test_name tb_neg = constant.insert_neg_tb_test_name positive_sql = f"select t.* from {tb_pos} t limit 1000000" positive_sql_result = click_client.execute(positive_sql) negative_sql = f"select t.* from {tb_neg} t limit 1000000" negative_sql_result = click_client.execute(negative_sql) dataf_pos = pd.DataFrame(positive_sql_result, columns=constant.feature_column_names) dataf_pos['mark'] = 0 dataf_neg = pd.DataFrame(negative_sql_result, columns=constant.feature_column_names) dataf_neg['mark'] = 1 test_all_data = pd.concat([dataf_pos, dataf_neg], axis=0) test_x_data = test_all_data[constant.feature_column_names] test_x_data.drop(columns=DROP_COLUMNS, inplace=True) test_y_data = test_all_data[['mark']] lgb_model = lgb.Booster(model_file='temp.model') test_predict = lgb_model.predict(test_x_data, num_iteration=lgb_model.best_iteration) test_result = [] for x in test_predict: if x < 0.5: test_result.append(0) else: test_result.append(1) result = metrics.classification_report(test_y_data.values, test_result) logger.info(result) draw_confusion_matrix(test_y_data.values, test_result) draw_roc_auc(test_y_data.values.reshape([-1,]), test_predict) pass if __name__ == "__main__": # start_train() test_data_valid()