user_preference_predict/local_process.py

from common.log_utils import logFactory
from common.database_utils import database_util
from common import constant
import pandas as pd
from tqdm import tqdm
from random import shuffle
from sklearn.model_selection import train_test_split
import lightgbm as lgb
from sklearn import metrics
import matplotlib.pyplot as plt
import seaborn as sns

logger = logFactory("local preprocess data").log

# click_client = database_util.get_client()
# # 获取正负样本数量，准备随机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_num_pos = 1
batch_num_neg = 13


# 根据每个批次的数据量计算出每个批次的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


# batch_num_pos = 10
# batch_num_neg = 5
total_pos = pd.read_pickle("data_pos_0218_06.pkl")
total_neg = pd.read_pickle("data_neg_part_1_0218_06.pkl")
positive_ids = range(0, total_pos.shape[0])
negative_ids = range(0, total_neg.shape[0])
pos_row_ids = partition_preserve_order(positive_ids, batch_num_pos)
neg_row_ids = partition_preserve_order(negative_ids, batch_num_neg)
train_tuple = gen_train_tuple(pos_row_ids, neg_row_ids)


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()


drop_columns = [
    'row_id',
    'month',
    'mark',
    'EVENT_FLUX_V',
    'EVENT_CONSUM_V',
    'EVENT_VIDEO_FLUX_V',
    # "EVENT_FLUX_C_0",
    # "EVENT_FLUX_C_1",
    # "EVENT_FLUX_C_2",
    # "EVENT_FLUX_C_3",
    # "EVENT_FLUX_C_4",
    # "EVENT_FLUX_C_5",
    # "EVENT_FLUX_C_6",
    # "EVENT_FLUX_C_7",
    # "EVENT_FLUX_C_8",
    # "EVENT_CONSUM_C_0",
    # "EVENT_CONSUM_C_1",
    # "EVENT_CONSUM_C_2",
    # "EVENT_CONSUM_C_3",
    # "EVENT_CONSUM_C_4",
    # "EVENT_CONSUM_C_5",
    # "EVENT_CONSUM_C_6",
    # "EVENT_CONSUM_C_7",
    # "EVENT_VIDEO_FLUX_C_0",
    # "EVENT_VIDEO_FLUX_C_1",
    # "EVENT_VIDEO_FLUX_C_2",
    # "EVENT_VIDEO_FLUX_C_3",
    # "EVENT_VIDEO_FLUX_C_4",
    # "EVENT_VIDEO_FLUX_C_5",
    # "EVENT_VIDEO_FLUX_C_6",
    # "EVENT_VIDEO_FLUX_C_7",
]


def start_train():
    lgb_model = None
    train_params = {
        'task': 'train',
        'objective': 'binary',
        'boosting_type': 'gbdt',
        'learning_rate': 0.1,
        'num_leaves': 10,
        'tree_learner': 'serial',
        'metric': {'binary_logloss', 'auc', 'average_precision'},  # l1:mae, l2:mse
        'max_bin': 20,  # 较小的max_bin会导致更快的速度，较大的值会提高准确性
        'max_depth': 6,
        # 'min_child_samples': 5,
        # "min_data_in_leaf": 10,
        "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,
    }

    for i, data_tuple in enumerate(tqdm(train_tuple)):
        step_num = str(i)
        logger.info(f"开始第{step_num}轮")
        pos_ids = data_tuple[0]
        neg_ids = data_tuple[1]

        train_data_frame_pos = total_pos[total_pos.row_id.isin(pos_ids)]
        train_data_frame_neg = total_neg[total_neg.row_id.isin(neg_ids)]
        train_data_frame_pos['mark'] = 0
        train_data_frame_neg['mark'] = 1
        total_train_data = pd.concat([train_data_frame_pos, train_data_frame_neg], axis=0)
        # 划分训练集和测试集
        data_train, data_test = train_test_split(total_train_data, train_size=0.6)

        train_y_data = data_train[['mark']]
        train_x_data = data_train.drop(columns=drop_columns)

        test_y_data = data_test[['mark']]
        test_x_data = data_test.drop(columns=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)

        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=train_x_data.columns.tolist(),
                              early_stopping_rounds=20,
                              verbose_eval=False, keep_training_booster=True)

        # 输出模型评估分数
        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"第{step_num}轮的结果如下:")
        logger.info(f"在训练集上:{score_train}")
        logger.info(f"在测试集上:{score_valid}")

        if i == len(train_tuple) - 1:
            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_roc_auc(test_y_data.values.reshape([-1, ]), test_predict)
            sns.set(rc={'figure.figsize': (50, 50)})
            sns.barplot(y=train_x_data.columns, x=lgb_model.feature_importance())
            plt.show()

    lgb_model.save_model("temp0218-2.model")


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 test_data_valid():
    logger.info("准备开始加载数据")
    # # 获取正负样本数量，准备随机id
    tb_pos = constant.insert_pos_tb_test_name
    tb_neg = constant.insert_neg_tb_test_name

    dataf_pos = pd.read_pickle('data_pos_0218_07.pkl')
    dataf_pos['mark'] = 0
    dataf_neg = pd.read_pickle('data_neg_part_1_0218_07.pkl')
    dataf_neg['mark'] = 1

    test_all_data = pd.concat([dataf_pos, dataf_neg], axis=0)

    test_x_data = test_all_data
    test_y_data = test_all_data[['mark']]
    test_x_data.drop(columns=drop_columns, inplace=True)

    lgb_model = lgb.Booster(model_file='temp0218-2.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)


if __name__ == "__main__":
    # start_train()
    test_data_valid()
    # data1 = pd.read_pickle("./data_neg_part_1_2013.pkl")
    # data2 = pd.read_pickle("./data_neg_part_2_2013.pkl")
    # data3 = pd.read_pickle("./data_neg_part_3_2013.pkl")
    # data4 = pd.read_pickle("./data_neg_part_4_2013.pkl")
    # data5 = pd.read_pickle("./data_neg_part_5_2013.pkl")
    # data_all = pd.concat([data1, data2, data3, data4], axis=0)
    # data_all.to_pickle("./data_neg_2013.pkl")
    pass