基于Spark和机器学习银行卡欺诈交易行为预测

一、spark介绍

     spark是开放源码的集群运算框架，由加州大学伯克利分校的AMPLab开发。Spark是一个弹性的运算框架，适合进行Spark Streaming数据流处理、Spark SQL互动分析、MLlib机器学习等应用，因此Spark可作为一个用途广泛的大数据运算平台。Spark 允许用户将数据加载到cluster集群的内存中储存，并多次重复运算，非常适合用于机器学习的算法，Spark 的核心是RDD ( Resilient Distributed Dataset)弹性分布式数据集，是由AMPLab实验室所提出的概念,属于- -种分布式的内容。Spark主要的优势来自RDD本身的特性，RDD能与其他系统兼容，可以导入外部存储系统的数据集，例如HDFS、HBase或其他Hadoop数据源。

二、机器学习

  机器学习:是计算机科学与统计学交叉的学科，其基本目标是学习一个x->y的函数（映射），用来做分类、聚类或者回归的工作。现在好多与数据挖掘有关的工作是通过机器学习提供的算法工具实现的，比如PB级别的点击日志通过典型的机器学习流程可以得到一个预估模型，从而提高互联网广告的点击率和回报率；个性化推荐，还是通过机器学习的一些算法分析平台上的各种购买，浏览和收藏日志得到的一个推荐模型，然后预测你喜欢的商品，还有现在的抖音你点赞关注什么类型的视频，就会给你推荐相似类型的视频等等。

三、行为预测以及可视化分析

（一）、用pandas做探索性分析

1、导入所需库并且输出前五行

import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.ml import Pipeline
from pyspark.ml.feature import StandardScaler, VectorAssembler
from pyspark.ml.evaluation import BinaryClassificationEvaluator
import matplotlib.pyplot as plt
import numpy as np
from sklearn.metrics import roc_curve, precision_recall_curve
from pyspark.ml.classification import LogisticRegression, DecisionTreeClassifier
#载入数据到pandas的DataFrame
df=pd.read_csv('./data/creditcard.csv')
df.head()

2、查看数据大小、类型及缺失值等信息

#查看数据大小
df.shape
#查看数据类型及缺失值等信息
df.info()

 3、数据预处理

#删除有缺失值的行
creditdata=df.dropna()
creditdata

 4、数据统计分析可视化

（1）、以是否“正常交易”为自变量的数量统计可视化

import matplotlib.pyplot as plt
import seaborn as sns
plt.figure(figsize=(14,6))
sns.set_style('darkgrid')
sns.countplot(x='Class',data=creditdata)
plt.title('Number of Spam and Ham')

（2）、交易时间分布图 

import pandas as pd
import matplotlib.pyplot as plt
%matplotlib inline
## 设置绘图时的中文字体
from matplotlib.font_manager import FontProperties
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus']=False
myfont = FontProperties(fname = 'FangSong.TTF', size = 15)
# 读取数据集
data = pd.read_csv('creditcard.csv')
# 将时间戳转换为日期格式
data['Time'] = pd.to_datetime(data['Time'], unit='s')
# 交易时间分布图
plt.figure(figsize=(10, 6))
data['Time'].dt.date.value_counts().sort_index().plot(kind='line')
plt.xlabel('数据',fontsize=12)
plt.ylabel('交易数量',fontsize=12)
plt.title('交易时间分布',fontsize=15)
plt.grid(True)
plt.show()

（3）、欺诈交易与正常交易比例饼图

plt.figure(figsize=(6, 6))
data['Class'].value_counts().plot(kind='pie', labels=['正常交易', '欺诈交易'], autopct='%1.1f%%')
plt.title('欺诈与正常交易',fontsize=15)
plt.show()

 （4）、交易金额分布图

plt.figure(figsize=(10, 6))
data['Amount'].plot(kind='hist', bins=50)
plt.xlabel('交易金额',fontsize=12)
plt.ylabel('频率',fontsize=12)
plt.title('交易金额分布',fontsize=15)
plt.grid(True)
plt.show()

（5）、特征变量相关性热力图 

 

import pandas as pd
import numpy as np
import seaborn as sns
import matplotlib.pyplot as plt
%matplotlib inline
## 设置绘图时的中文字体
from matplotlib.font_manager import FontProperties
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus']=False
myfont = FontProperties(fname = 'FangSong.TTF', size = 15)
# 读取数据集
data = pd.read_csv('creditcard.csv')
# 绘制特征变量相关性热力图
plt.figure(figsize=(12, 8))
correlation_matrix = data.corr()
sns.heatmap(correlation_matrix, annot=True)
plt.title('特征变量相关性热图',fontsize=15)
plt.show()

（6）、欺诈与正常交易时间趋势 

# 数据预处理：分离欺诈交易和正常交易数据
fraud_data = data[data['Class'] == 1]
normal_data = data[data['Class'] == 0]
# 欺诈交易与正常交易的时间趋势图
plt.figure(figsize=(12, 6))
sns.kdeplot(fraud_data['Time'], label='Fraud')
sns.kdeplot(normal_data['Time'], label='Normal')
plt.xlabel('交易时间',fontsize=12)
plt.ylabel('密度',fontsize=12)
plt.title('欺诈与正常交易时间趋势',fontsize=15)
plt.legend()
plt.show()

（二）、加载数据到Spark的DataFrame

 

from pyspark.sql import SparkSession
import os
os.environ['JAVA_HOME'] = '/opt/module/jdk1.8.0_171'
spark = SparkSession.builder.master("local").appName("credit").getOrCreate()
spark

1、将pandas的DataFrame转换为Spark的DataFrame

credit=spark.createDataFrame(creditdata)
credit.show()

2、 从文件读入数据

  Spark可以读取多种文件格式（csv、json、parquet、txt等），只需读取格式的参数与文件格式保持一致即可。 a.对于以制表符为分隔符的文件，需要在去取文件时传递一个额外参数以指定分隔符（sep='\t'); b.inferSchema设置为Ture，则Spark将自动识别数据集中值的数据类型； c.header表示第一行为标题行

#读入csv文件
df=spark.read.csv('./data/creditcard.csv',inferSchema=True,header=True)
type(df)
#显示前10行
df.show(10)
#显示列名
credit.columns
#统计列的数量：len函数
len(credit.columns)
#统计总行数：conut函数
credit.count()
#显示数据的形状结构
print(credit.count(),len(credit.columns))
#显示数据类型及列名：printSchema
credit.printSchema()
df.printSchema()
df.describe().show()

3、将 Amount 和 V1 到 V28 两个列合并为一个列 features：

assembler = VectorAssembler(inputCols=df.columns[:-1], outputCol="features")
output = assembler.transform(df)

4、标准化 features 列：

scaler = StandardScaler(inputCol="features", outputCol="scaledFeatures")
scalerModel = scaler.fit(output)
output = scalerModel.transform(output)
# 将列名为 "features" 的列重命名为 "new_features"
output = output.withColumnRenamed("features", "new_features")
#将列名为 "scaledFeatures" 的列重命名为 "new_scaledFeatures"
output = output.withColumnRenamed("scaledFeatures", "new_scaledFeatures")

5、拆分数据集为训练集和测试集：

train, test = output.randomSplit([0.7, 0.3], seed=12345)

6、使用逻辑回归模型进行训练和预测：

lr = LogisticRegression(featuresCol="scaledFeatures", labelCol="Class", maxIter=10)
pipeline = Pipeline(stages=[assembler, scaler, lr])
model = pipeline.fit(train)
pred = model.transform(test)

7、 使用二元分类评估器来评估模型：

evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", labelCol="Class")
auPRC = evaluator.evaluate(pred, {evaluator.metricName: "areaUnderPR"})
auROC = evaluator.evaluate(pred)
print("Area under ROC: {:.2f}%".format(auROC*100))
print("Area under PR: {:.2f}%".format(auPRC*100))

（1）、评估结果可视化

# 将 Spark DataFrame 的列转换为 Numpy 数组，并展平数组
y_true = np.array(pred.select("Class").collect()).ravel()
y_score = np.array(pred.select("rawPrediction").collect()).reshape(-1, 2)
# 计算 ROC 曲线的假阳性率、真阳性率和阈值
fpr, tpr, thresholds = roc_curve(y_true, y_score[:, 1])
# 计算 PR 曲线的准确率、召回率和阈值
precision, recall, thresholds = precision_recall_curve(y_true, y_score[:, 1])
# 绘制 ROC 曲线
plt.figure()
plt.plot(fpr, tpr, label="ROC curve")
plt.plot([0, 1], [0, 1], "k--")  # 对角线
plt.xlabel("False Positive Rate")
plt.ylabel("True Positive Rate")
plt.title("Receiver Operating Characteristic (ROC) Curve")
plt.legend(loc="lower right")
plt.show()
# 绘制 PR 曲线
plt.figure()
plt.plot(recall, precision, label="Precision-Recall curve")
plt.xlabel("Recall")
plt.ylabel("Precision")
plt.title("Precision-Recall Curve")
plt.legend(loc="upper right")
plt.show()

8、 决策树算法进行训练和预测

lr = LogisticRegression(featuresCol="scaledFeatures", labelCol="Class", maxIter=10)
dt = DecisionTreeClassifier(featuresCol="scaledFeatures", labelCol="Class", maxDepth=5)
lrPipeline = Pipeline(stages=[assembler, scaler, lr])
dtPipeline = Pipeline(stages=[assembler, scaler, dt])
lrModel = lrPipeline.fit(train)
dtModel = dtPipeline.fit(train)
lrPred = lrModel.transform(test)
dtPred = dtModel.transform(test)

9、使用二元分类评估器来评估逻辑回归和决策树模型及可视化：

lrEvaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", labelCol="Class")
dtEvaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", labelCol="Class")
lrAuROC = lrEvaluator.evaluate(lrPred)
dtAuROC = dtEvaluator.evaluate(dtPred)
lrAuPRC = lrEvaluator.evaluate(lrPred, {lrEvaluator.metricName: "areaUnderPR"})
dtAuPRC = dtEvaluator.evaluate(dtPred, {dtEvaluator.metricName: "areaUnderPR"})
print("逻辑回归模型的 ROC 曲线下面积：{:.2f}%".format(lrAuROC*100))
print("逻辑回归模型的 PR 曲线下面积：{:.2f}%".format(lrAuPRC*100))
print("决策树模型的 ROC 曲线下面积：{:.2f}%".format(dtAuROC*100))
print("决策树模型的 PR 曲线下面积：{:.2f}%".format(dtAuPRC*100))
import matplotlib.pyplot as plt
%matplotlib inline
import numpy as np 
import pandas as pd 
import os
from matplotlib.font_manager import FontProperties
import matplotlib.pyplot as plt
import seaborn as sns
plt.rcParams['font.sans-serif'] = ['DejaVu Sans']  # 使用DejaVu Sans字体
plt.rcParams['axes.unicode_minus'] = False  # 解决负号显示问题
myfont1 = FontProperties(fname = "simhei.ttf",size=18)
myfont2 = FontProperties(fname = "simhei.ttf",size=12)
# 假设你已经计算得到了 lrAuROC、lrAuPRC、dtAuROC 和 dtAuPRC 这四个评估指标的值
# 创建一个条形图，并设置 x 轴标签和评估指标值
labels = ['ROC-AUC', 'PR-AUC']
lr_values = [lrAuROC, lrAuPRC]
dt_values = [dtAuROC, dtAuPRC]
x = range(len(labels))
width = 0.35
fig, ax = plt.subplots()
rects1 = ax.bar(x=x, height=lr_values, width=width, label='LogisticRegression')
rects2 = ax.bar(x=[i + width for i in x], height=dt_values, width=width, label='DecisionTreeClassifier')
ax.set_ylabel('Vaule')
ax.set_title('Evaluation metrics for logistic regression models and decision tree models')
ax.set_xticks([i + width/2 for i in x])
ax.set_xticklabels(labels)
ax.legend()
def autolabel(rects):
    for rect in rects:
        height = rect.get_height()
        ax.annotate('{:.2f}%'.format(height*100),
                    xy=(rect.get_x() + rect.get_width() / 2, height),
                    xytext=(0, 3),  
                    textcoords="offset points",
                    ha='center', va='bottom')
autolabel(rects1)
autolabel(rects2)
plt.ylim(0, 1.2)
plt.show()

 

 四、结束

（一）总结

本文通过对银行卡交易欺诈行为数据的训练预测和分析，为我国打击预测银行卡犯罪方面提供重要的理论支持和实践参考。以机器学习为核心的建模方法能够大幅提升预测的准确程度，是新一代风险防控一体化平台建设的技术支持[]。在本文研究中，针对银行卡交易欺诈行为预测问题，使用逻辑回归和决策树两种模型进行了建模和预测，并使用二元分类评估器进行了结果评估。银行卡交易欺诈预测模型可以帮助警方提高犯罪打击效率。逻辑回归模型的较高ROC-AUC值表明它可以非常准确地分类正常和欺诈交易，这意味着该模型可以为警务部门提供有用的线索，帮助他们更快速地定位欺诈嫌疑人，并在不必要的情况下避免对无辜人的侵害。此外，逻辑回归模型可以对前期银行卡交易数据的背景分析预测和建模，突出重点进行监控，从而有效地缩小警方的调查范围和成本，提高调查效率，增强犯罪打击的实际成效。综上所述，银行卡交易欺诈预测模型可以为警务工作提供有力支持，使银行卡欺诈行为更容易被识别和定位，进而提高犯罪打击效率和成效，减小对社会的危害。

（二）期望

 在未来，可以继续探究更多的模型和方法，以提高银行卡交易欺诈行为预测的准确性和可靠性。例如，可以尝试使用神经网络等深度学习算法，或者使用集成学习方法，如随机森林、梯度提升树等。此外，也可以继续寻求更多的特征工程和数据处理方法，以提高模型的泛化能力和鲁棒性，帮助银行和金融机构更好地识别和预测银行卡交易欺诈行为，为保障金融安全做出更大的贡献。