机器学习实战:交易数据异常检测

数据集介绍

项目来源:python数据分析与机器学习实战-唐宇迪 https://study.163.com/course/introduction.htm?courseId=1003590004&share=1&shareId=1030301518#/courseDetail?tab=1 。
数据集下载地址:
链接：https://pan.baidu.com/s/1Rmcutb1gYHCKaBoPCtYZ-A 。
提取码：lbef

数据集是一批经过处理后的信用卡用户交易数据，我们需要通过这些数据学习一个模型，可以用来预测新的一条交易数据是否涉嫌信用卡欺诈。

项目实战过程

Time表示交易时间，Amount表示交易金额总量，Class为数据类别，表示此条交易数据是否存在信用卡欺诈，0为正常，1为异常。
Amount这个特征尺度明显和其他特征不在同一数量级，将其归一化，并删除无用的Time特征项。
接下来我们发现正常样本有284315个，而异常样本只有492个，存在严重的类不平衡问题。
我们尝试用下采样来重新组建一个数据集，学习一个逻辑回归模型。逻辑回归模型采用l1正则化项。使用k折交叉验证来找到最合适的l1正则化项前的参数C。
然后我们用刚学习的模型分别预测我们的下采样数据集中的测试集和初始数据集中的测试集，我们发现这样学习到的模型对初始数据集中的测试集的召回率和准确率都不错，但是精确率太低。这是因为我们只使用了284315个正常样本中的492个样本来学习模型，导致模型泛化能力太差。
我们改用SMOTE算法来处理数据集。SMOTE算法过程如下:

• 设训练集的一个少数类的样本数为T，SMOTE算法将为这个少数类合成NT个新样本；
• N必须是正整数,如果给定的N<1,那么算法将”认为”少数类的样本数T=NT,并将强制N=1；
• 考虑该少数类的一个样本i,首先从少数类样本中找到i的k个近邻(用欧式距离计算),然后从这k个近邻中随机选择一个样本xi(nn)
• 再生成一个0-1之间的随机数ζ1,合成一个新样本xi1:xi1=xi+ζ1(xi(nn)-xi)；
• 上面步骤重复N次,合成N个新样本,对全部的T个少数类样本进行上述操作,便可为该少数类合成NT个新样本。

使用SMOTE算法后我们再查看初始数据集中的异常/正常样本个数分别为:199019，199019。这时异常/正常样本个数均衡了，我们再用这个数据集学习一个模型，再观察模型对测试集的泛化能力。这时可以发现模型的性能变得好多了。

项目代码

import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import KFold
from sklearn.metrics import confusion_matrix, recall_score
# pip install imblearn
from imblearn.over_sampling import SMOTE

# 现在有一个信用卡用户交易数据集
# 我们通过这个数据集学习一个模型,用来预测新的一条交易数据是否涉嫌信用卡欺诈
data = pd.read_csv("creditcard.csv")
# print(data.head())

# Time表示交易时间,Amount表示交易金额总量,Class表示本条数据的类别,即表示此条交易行为是否存在信用卡欺诈,0为正常,1为异常
# Amount这个特征范围与其他特征差距较大,需要进行特征缩放
# fit_transform先对数据拟合fit,得到均值和方差等,然后对数据进行转换transform,从而实现数据的归一化
data["normAmount"] = StandardScaler().fit_transform(data["Amount"].values.reshape(-1, 1))
# 删除Time列和Amount列
data = data.drop(["Time", "Amount"], axis=1)
# print(data.head())

# 将样本按类别排列,观察正常样本(类别为0)和异常样本(类别为1)的个数,value_counts计算相同特征值出现的次数
count_class = pd.value_counts(data["Class"], sort=True).sort_index()
# print(count_class)

# 分离出特征X和类别y
X = data.iloc[:, data.columns != "Class"]
Y = data.iloc[:, data.columns == "Class"]
# print(X.head())
# print(Y.head())
# print(X.shape, Y.shape)

# 代表异常的样本个数
abnormal_sample_count = len(data[data["Class"] == 1])
print("异常样本个数为:{}".format(abnormal_sample_count))

# 异常样本所对应的索引为
abnormal_sample_index = np.array(data[data["Class"] == 1].index)
print("异常样本在数据集中的索引(打印前5个):{}".format(abnormal_sample_index[:5]))
# 正常样本所对应的索引
normal_sample_index = data[data["Class"] == 0].index
# np.random.choice表示从normal_sample_index中随机取出abnormal_sample_count个样本
# replace为True表示每次都随机生成,false表示只随机生成一次
random_normal_sample_index = np.array(np.random.choice(normal_sample_index, abnormal_sample_count, replace=False))
print("正常样本在数据集中的索引(打印前5个):{}".format(random_normal_sample_index[:5]))
# 正常样本数量较多,对正常样本进行下采样
# 只有492个异常样本,我们就从正常样本(284315个)中,随机取出492个,然后和之前的492个异常样本重新组成一个新的训练数据集
# np.concatenate是拼接函数,把选出的492个异常样本的index和492个正常样本的index拼接在一起
under_sample_index = np.concatenate([abnormal_sample_index, random_normal_sample_index])
under_sample_data = data.iloc[under_sample_index, :]
# 得到下采样后的样本特征和样本类别
x_under_sample = under_sample_data.iloc[:, under_sample_data.columns != "Class"]
y_under_sample = under_sample_data.iloc[:, under_sample_data.columns == "Class"]

print("下采样后新数据集中异常样本所占比例:{}".format(len(under_sample_data[under_sample_data.Class == 1]) / len(under_sample_data)))
print("下采样后新数据集中正常样本所占比例:{}".format(len(under_sample_data[under_sample_data.Class == 0]) / len(under_sample_data)))
print("下采样后新数据集样本个数为:{}".format(len(under_sample_data)))

# 对初始数据集进行训练集和测试集划分
X_train, X_test, y_train, y_test = train_test_split(X, Y, test_size=0.3, random_state=0)
# 下采样后新数据集进行训练集和测试集的划分
x_train_under_sample, x_test_under_sample, y_train_under_sample, y_test_under_sample = train_test_split(x_under_sample,
                                                                                                        y_under_sample,
                                                                                                        test_size=0.3,
                                                                                                        random_state=0)
# 下采样后新数据集的训练集和测试集样本数
print("训练集样本数:{},测试集样本数:{}\n".format(len(x_train_under_sample), len(x_test_under_sample)))


# 对训练集进行k折交叉验证
# 将训练集随机分为K份,依次选择其中的k-1份来进行训练,剩下的1份用来进行测试,循环k次,最后取平均精度,作为当前训练出的模型的精度
def Kfold_for_TrainModel(x_train_data, y_train_data):
   # 将训练集划分n_splits个互斥子集,每次用其中一个子集当作验证集,剩下的n_splits-1个作为训练集
   # 进行n_splits次训练和测试，得到n_splits个结果
   fold = KFold(n_splits=5, shuffle=False)
   # 正则化项前的C参数
   c_params = [0.01, 0.1, 1, 10, 100]
   # 查准率=TP/(TP+FP),TP为预测为真,实际为也真;FP为预测为真,实际为假。
   # 召回率=TP/(TP+FN),TP为预测为真,实际为也真;FN为预测为假,实际为真。
   # 简单来说就是查准率是预测为真的里面有多少实际为真,召回率为实际为真的里面有多少预测为真。
   # 生成一个DataFrame用来保存不同的C参数下对应的召回率
   result_tables = pd.DataFrame(columns=["C_parameter", "Mean recall score"])
   result_tables["C_parameter"] = c_params
   j = 0
   for c_param in c_params:
      print("C参数为:{}\n".format(c_param))
      # 记录每个c_param参数下k折交叉验证时每次的召回率
      recall_accs = []
      for iteration, indices in enumerate(fold.split(x_train_data)):
         # 逻辑回归,采用l1正则化,solver="liblinear"防止出现future warning
         lr = LogisticRegression(C=c_param, solver="liblinear", penalty="l1")
         # indices[0] 保存的是这个k=5次训练中的某一次的用来训练的数据的索引
         # indices[1] 保存的是这个k=5次训练中的某一次的用来验证的数据的索引
         # .fit学习一个模型,.ravel将输出降到一维
         lr.fit(x_train_data.iloc[indices[0], :], y_train_data.iloc[indices[0], :].values.ravel())
         # 用剩下的indices[1]验证数据进行验证测试模型
         y_undersample_pred = lr.predict(x_train_data.iloc[indices[1], :].values)
         recall_acc = recall_score(y_train_data.iloc[indices[1], :].values, y_undersample_pred)
         # 记录k折中每一次的召回率
         recall_accs.append(recall_acc)
         print("k:{},召回率:{}".format(iteration, recall_acc))
      print("平均召回率为:{}\n".format(np.mean(recall_accs)))
      # 记录每个c_param参数的平均召回率
      result_tables.loc[j, "Mean recall score"] = np.mean(recall_accs)
      j = j + 1

   result_tables["Mean recall score"] = result_tables["Mean recall score"].astype("float64")
   best_c_param = result_tables.loc[result_tables["Mean recall score"].idxmax(), "C_parameter"]
   # 得到最佳的C参数
   print("最佳模型对应的C参数为:{}\n".format(best_c_param))
   return best_c_param


the_best_c_param = Kfold_for_TrainModel(x_train_under_sample, y_train_under_sample)


# 绘制混淆矩阵
# 混淆矩阵(confusion matrix)是一种评价分类模型好坏的图像,其中行代表的是实际类别,列代表的是预测类别
def plot_confusion_matrix(confusion_matrix, classes):
   plt.figure()
   plt.imshow(confusion_matrix, interpolation="nearest", cmap=plt.cm.Blues)
   plt.title("confusion matrix")
   plt.colorbar()
   tick_marks = np.arange(len(classes))
   plt.xticks(tick_marks, classes, rotation=0)
   plt.yticks(tick_marks, classes)
   thresh = confusion_matrix.max() / 2.
   for i, j in itertools.product(range(confusion_matrix.shape[0]), range(confusion_matrix.shape[1])):
      plt.text(j, i, confusion_matrix[i, j], horizontalalignment="center",
               color="white" if confusion_matrix[i, j] > thresh else "black")

   plt.tight_layout()
   plt.ylabel("True label")
   plt.xlabel("Predicted label")
   plt.show()
   print("使用最佳C参数:{}".format(the_best_c_param))
   # 这里标签为1认为是为真
   # 查准率即预测为真的样本中，实际为真所占的比例
   # 召回率即预测为真的样本占所有实际为真样本的比例
   # 混淆矩阵中第一个标签是真实标签,第二个标签是预测标签
   print("查准率为：", confusion_matrix[1, 1] / (confusion_matrix[1, 1] + confusion_matrix[0, 1]))
   print("召回率为：", confusion_matrix[1, 1] / (confusion_matrix[1, 1] + confusion_matrix[1, 0]))
   print("准确率为：", (confusion_matrix[0, 0] + confusion_matrix[1, 1]) / (
         confusion_matrix[0, 0] + confusion_matrix[0, 1] + confusion_matrix[1, 1] + confusion_matrix[1, 0]))
   print("")


# 使用下采样后得到的数据集学习一个模型
lr = LogisticRegression(C=the_best_c_param, solver="liblinear", penalty="l1")
lr.fit(x_train_under_sample, y_train_under_sample.values.ravel())
# 得到模型对下采样数据集中的测试集的预测结果
y_undersample_pred = lr.predict(x_test_under_sample.values)
# 创建混淆矩阵,第一个标签是真实标签,第二个标签是预测标签
conf_matrix = confusion_matrix(y_test_under_sample, y_undersample_pred)
class_names = [0, 1]
# 绘制混淆矩阵
plot_confusion_matrix(conf_matrix, classes=class_names)

# 得到模型对初始数据集划分的测试集的预测结果
y_pred = lr.predict(X_test.values)
# 创建混淆矩阵
conf_matrix = confusion_matrix(y_test, y_pred)
class_names = [0, 1]
# 绘制混淆矩阵
# 对于整个数据集划分出的测试集,虽然召回率和准确率都不错，但是精确率太低
# 虽然把147个异常样本中的134个都已正确预测出,但同时把10894个正常样本预测为异常样本
# 出现这个问题的原因是我们只选取了284315个正常样本中的492个来训练模型,导致模型泛化能力差,这也是下采样的缺点
plot_confusion_matrix(conf_matrix, classes=class_names)

# SMOTE即Synthetic Minority Oversampling Technique,合成少数类过采样技术
oversampler = SMOTE(random_state=0)
# 设训练集的一个少数类的样本数为T,SMOTE算法将为这个少数类合成NT个新样本
# N必须是正整数,如果给定的N<1,那么算法将"认为"少数类的样本数T=NT,并将强制N=1。
# 考虑该少数类的一个样本i,首先从少数类样本中找到i的k个近邻(用欧式距离计算),然后从这k个近邻中随机选择一个样本xi(nn)
# 再生成一个0-1之间的随机数ζ1,合成一个新样本xi1:xi1=xi+ζ1(xi(nn)-xi)
# 上面步骤重复N次,合成N个新样本,对全部的T个少数类样本进行上述操作,便可为该少数类合成NT个新样本
X_over_samples, y_over_samples = oversampler.fit_sample(X_train, y_train.values.ravel())
# 使用SMOTE算法后异常样本和正常样本的个数
print(len(y_over_samples[y_over_samples == 1]), len(y_over_samples[y_over_samples == 0]))

# 使用SMOTE算法处理后的数据集进行k折交叉验证,找到最佳C参数
the_best_c_param = Kfold_for_TrainModel(pd.DataFrame(X_over_samples), pd.DataFrame(y_over_samples))
# 使用最佳C参数,用SMOTE算法处理后的数据集学习一个模型
lr = LogisticRegression(C=the_best_c_param, solver="liblinear", penalty="l1")
lr.fit(X_over_samples, y_over_samples)
# 获得初始数据集划分的测试集的预测结果
y_pred = lr.predict(X_test.values)
# 构建稀疏矩阵
conf_matrix = confusion_matrix(y_test, y_pred)
class_names = [0, 1]
# 绘制混淆矩阵
plot_confusion_matrix(conf_matrix, classes=class_names)
# 显然使用SMOTE算法处理后的数据集所学习的模型泛化能力要强得多