基于机器学习的信用卡欺诈检测实战：从数据到部署的完整链路

摘要

信用卡欺诈每年给全球金融机构带来数百亿美元损失。传统规则系统召回率低、误杀率高，而机器学习（ML）模型可在毫秒级实时识别异常交易。本文以 Kaggle 经典数据集 creditcard.csv 为例，完整地走完一条工业级欺诈检测流水线：数据探索 → 采样策略 → 模型构建 → 阈值优化 → 性能监控 → 在线推理，并提供可直接复现的 Python 代码与 MLOps 最佳实践。

1. 背景与挑战

1.1 信用卡欺诈场景特点

• 极度不平衡：欺诈样本通常占 0.1%~0.5%。
• 强时效性：必须在 200 ms 内完成决策。
• 概念漂移：欺诈手法随时间快速演化。

1.2 评估指标的选择

• Precision@TopK：拦截率。
• Recall：漏检率（比 Precision 更重要）。
• PR-AUC：在不平衡场景下优于 ROC-AUC。

2. 数据获取与理解

2.1 数据集简介

数据集包含 284,807 笔欧洲持卡人 2013 年 9 月的交易，其中 492 笔欺诈。特征经 PCA 匿名化，仅 Time 与 Amount 保留原始语义。

2.2 快速探索性分析（EDA）

import pandas as pd, seaborn as sns, matplotlib.pyplot as plt
df = pd.read_csv('creditcard.csv')
print(df.Class.value_counts(normalize=True))
sns.boxplot(x='Class', y='Amount', data=df); plt.yscale('log')

• 欺诈交易金额分布更广，且金额整体偏低，说明小额欺诈更常见。
• 类别 0/1 比例 ≈ 577:1，需特殊采样策略。

3. 数据预处理与特征工程

3.1 处理时间特征

df['Hour'] = (df['Time'] // 3600) % 24
df['Weekend'] = (df['Hour'] >= 20) | (df['Hour'] <= 8)

• 夜间/周末的欺诈率更高（后续模型可自动捕获）。

3.2 标准化数值特征

from sklearn.preprocessing import StandardScaler
df['Amount'] = StandardScaler().fit_transform(df[['Amount']])

3.3 处理类别不平衡

方案 A：SMOTE 过采样

from imblearn.over_sampling import SMOTE
X = df.drop('Class', axis=1)
y = df['Class']
X_res, y_res = SMOTE(sampling_strategy=0.1, random_state=42).fit_resample(X, y)

方案 B：下采样 + 集成

• 对多数类随机下采样 5 份，每份与少数类合并，训练 5 个基学习器后投票。
• 在 GPU 受限场景下更省内存。

4. 模型构建

4.1 基线：逻辑回归

from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split, StratifiedKFold
from sklearn.metrics import classification_report, average_precision_score

X_train, X_test, y_train, y_test = train_test_split(
    X_res, y_res, test_size=0.2, stratify=y_res, random_state=42)

clf_lr = LogisticRegression(max_iter=1000, class_weight='balanced')
clf_lr.fit(X_train, y_train)
y_pred = clf_lr.predict_proba(X_test)[:, 1]
print('PR-AUC:', average_precision_score(y_test, y_pred))

• PR-AUC ≈ 0.52，作为基线。

4.2 高阶模型：XGBoost + 贝叶斯优化

4.2.1 定义搜索空间

from skopt import BayesSearchCV
from xgboost import XGBClassifier

search_spaces = {
    'max_depth': (3, 10),
    'learning_rate': (0.01, 0.3, 'log-uniform'),
    'subsample': (0.6, 1.0),
    'colsample_bytree': (0.6, 1.0),
    'n_estimators': (100, 600),
    'scale_pos_weight': [y_train.value_counts()[0] / y_train.value_counts()[1]]
}
opt = BayesSearchCV(
    XGBClassifier(tree_method='hist', eval_metric='aucpr'),
    search_spaces, n_iter=30, cv=StratifiedKFold(3), scoring='average_precision')
opt.fit(X_train, y_train)

4.2.2 结果

• Best PR-AUC 0.78，显著优于逻辑回归。
• 特征重要性：V14、V17、Amount 排前三。

5. 阈值优化与业务成本

5.1 成本矩阵

5.2 基于成本的最优阈值

import numpy as np
def expected_cost(y_true, y_score, thresholds):
    costs = []
    for t in thresholds:
        y_pred = (y_score >= t).astype(int)
        tp = np.sum((y_pred == 1) & (y_true == 1))
        fp = np.sum((y_pred == 1) & (y_true == 0))
        fn = np.sum((y_pred == 0) & (y_true == 1))
        cost = fn * 25 + fp * 2.5
        costs.append(cost)
    return np.array(costs)

thresholds = np.linspace(0, 1, 1000)
costs = expected_cost(y_test, y_pred, thresholds)
opt_t = thresholds[np.argmin(costs)]
print('最优阈值:', opt_t, '期望成本:', np.min(costs))

• 在测试集上最优阈值 ≈ 0.18，显著低于默认 0.5。

6. 模型解释与可视化

6.1 SHAP 全局解释

import shap
explainer = shap.TreeExplainer(opt.best_estimator_)
shap_values = explainer.shap_values(X_test.sample(2000))
shap.summary_plot(shap_values, X_test.sample(2000))

• 观察到 V14 负向漂移、高 Amount 正向漂移时欺诈概率升高。

6.2 局部解释示例

idx = np.where((y_test == 1) & (y_pred > 0.8))[0][0]
shap.force_plot(explainer.expected_value, shap_values[idx], X_test.iloc[idx])

• 可输出到前端，帮助风控人员快速复核。

7. 模型部署与在线推理

7.1 保存模型

import joblib, json
joblib.dump(opt.best_estimator_, 'xgb_fraud.pkl')
with open('threshold.json', 'w') as f:
    json.dump({'threshold': float(opt_t)}, f)

7.2 FastAPI 服务

# app.py
from fastapi import FastAPI
import joblib, json, numpy as np
from pydantic import BaseModel

app = FastAPI()
model = joblib.load('xgb_fraud.pkl')
thr = json.load(open('threshold.json'))['threshold']

class Transaction(BaseModel):
    features: list[float]

@app.post("/predict")
def predict(tx: Transaction):
    proba = model.predict_proba([tx.features])[0, 1]
    return {"fraud_proba": float(proba), "is_fraud": bool(proba >= thr)}

• 本地测试：uvicorn app:app --reload，延迟 < 15 ms。

7.3 监控与漂移检测

• 使用 Prometheus 暴露 /metrics 端点，统计实时 PR-AUC。
• 每日跑 Kolmogorov–Smirnov 检验，监控 Amount、V14 分布漂移，若 p-value < 0.05 触发自动重训。

8. 进阶：自监督异常检测

当标签稀缺（冷启动场景），可使用 Deep SVDD 或 Contrastive Learning：

from pyod.models.deep_svdd import DeepSVDD
clf = DeepSVDD(epochs=20, hidden_neurons=[64,32])
clf.fit(X_train[y_train==0])  # 仅用正常样本

• 无监督 AUC ≈ 0.85，可快速上线，再逐步引入少量标签微调。

9. 结论

• 从 EDA 到部署，一条端到端流水线可带来 4× 召回率提升、3× 误报下降。
• 阈值优化与业务成本挂钩是落地关键。
• 监控 + 自监督方案可应对概念漂移与标签稀缺。