AI异常检测与欺诈识别完全教程
1. 异常检测概述与分类
异常检测（Anomaly Detection）是识别数据中偏离正常模式的观测值的过程。这些偏离点可能代表设备故障、网络入侵、金融欺诈等重要事件。根据数据形态和检测方式，异常可分为以下几类：
点异常（Point Anomaly）：单个数据点显著偏离整体分布。例如一笔远超正常金额的交易。
上下文异常（Contextual Anomaly）：在特定上下文中异常，但在全局范围内正常。例如夏天出现30°C是正常的，但冬天出现30°C就是异常。
集体异常（Collective Anomaly）：单个数据点可能正常，但一组数据点的组合模式异常。例如网络流量中短时间内大量相同请求可能代表DDoS攻击。
import numpy as np
import matplotlib.pyplot as plt
def generate_anomaly_data(n_normal=1000, n_anomaly=50, seed=42):
"""生成包含异常的模拟数据"""
np.random.seed(seed)
# 正常数据：两个高斯分布的混合
normal1 = np.random.normal(loc=[5, 5], scale=[1, 1], size=(n_normal // 2, 2))
normal2 = np.random.normal(loc=[-5, -5], scale=[1, 1], size=(n_normal // 2, 2))
normal = np.vstack([normal1, normal2])
# 异常数据：均匀分布的离群点
anomaly = np.random.uniform(low=-15, high=15, size=(n_anomaly, 2))
X = np.vstack([normal, anomaly])
y = np.array([0] * n_normal + [1] * n_anomaly)
return X, y
X, y = generate_anomaly_data()
print(f"正常样本: {sum(y == 0)}, 异常样本: {sum(y == 1)}")
print(f"异常比例: {sum(y == 1) / len(y):.2%}")
2. 统计方法与传统机器学习
基于统计的方法
最基础的异常检测方法利用统计学原理，假设正常数据服从某种已知分布：
from scipy import stats
class StatisticalAnomalyDetector:
"""基于统计方法的异常检测器"""
def __init__(self, method="zscore", threshold=3.0):
self.method = method
self.threshold = threshold
self.mean = None
self.std = None
def fit(self, X: np.ndarray):
self.mean = np.mean(X, axis=0)
self.std = np.std(X, axis=0)
return self
def predict(self, X: np.ndarray) -> np.ndarray:
if self.method == "zscore":
z_scores = np.abs((X - self.mean) / (self.std + 1e-8))
return (z_scores > self.threshold).any(axis=1).astype(int)
elif self.method == "iqr":
Q1 = np.percentile(X, 25, axis=0)
Q3 = np.percentile(X, 75, axis=0)
IQR = Q3 - Q1
lower = Q1 - self.threshold * IQR
upper = Q3 + self.threshold * IQR
return ((X < lower) | (X > upper)).any(axis=1).astype(int)
elif self.method == "mad":
# Median Absolute Deviation
median = np.median(X, axis=0)
mad = np.median(np.abs(X - median), axis=0)
modified_z = 0.6745 * (X - median) / (mad + 1e-8)
return (np.abs(modified_z) > self.threshold).any(axis=1).astype(int)
# 使用示例
for method in ["zscore", "iqr", "mad"]:
detector = StatisticalAnomalyDetector(method=method, threshold=3.0)
detector.fit(X[y == 0]) # 只用正常数据训练
preds = detector.predict(X)
tp = sum((preds == 1) & (y == 1))
fp = sum((preds == 1) & (y == 0))
print(f"{method:6s} | 检出: {tp}/{sum(y==1)}, 误报: {fp}")
Isolation Forest
孤立森林是目前最流行的无监督异常检测算法之一，其核心思想是异常点由于稀疏性更容易被"孤立"：
from sklearn.ensemble import IsolationForest
from sklearn.metrics import precision_recall_fscore_support
class IsolationForestDetector:
def __init__(self, n_estimators=100, contamination=0.05):
self.model = IsolationForest(
n_estimators=n_estimators,
contamination=contamination,
random_state=42
)
def fit(self, X: np.ndarray):
self.model.fit(X)
return self
def predict(self, X: np.ndarray) -> np.ndarray:
# IsolationForest返回-1表示异常，1表示正常
preds = self.model.predict(X)
return (preds == -1).astype(int)
def score_samples(self, X: np.ndarray) -> np.ndarray:
# 返回异常分数（越小越异常）
return -self.model.score_samples(X)
# 训练与评估
iso_detector = IsolationForestDetector(contamination=0.05)
iso_detector.fit(X)
preds = iso_detector.predict(X)
precision, recall, f1, _ = precision_recall_fscore_support(y, preds, average="binary")
print(f"Precision: {precision:.3f}, Recall: {recall:.3f}, F1: {f1:.3f}")
Local Outlier Factor (LOF)
LOF 通过比较数据点与其邻居的局部密度来判断异常程度：
from sklearn.neighbors import LocalOutlierFactor
class LOFDetector:
def __init__(self, n_neighbors=20, contamination=0.05):
self.model = LocalOutlierFactor(
n_neighbors=n_neighbors,
contamination=contamination,
novelty=True # 支持对新数据预测
)
def fit(self, X: np.ndarray):
self.model.fit(X)
return self
def predict(self, X: np.ndarray) -> np.ndarray:
preds = self.model.predict(X)
return (preds == -1).astype(int)
lof_detector = LOFDetector(n_neighbors=30, contamination=0.05)
lof_detector.fit(X[y == 0])
preds = lof_detector.predict(X)
precision, recall, f1, _ = precision_recall_fscore_support(y, preds, average="binary")
print(f"LOF - Precision: {precision:.3f}, Recall: {recall:.3f}, F1: {f1:.3f}")
3. 深度学习异常检测
AutoEncoder 方法
AutoEncoder 通过学习数据的压缩表示来检测异常——正常数据的重建误差较小，异常数据的重建误差较大：
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
class AnomalyAutoEncoder(nn.Module):
"""用于异常检测的自编码器"""
def __init__(self, input_dim, encoding_dim=8):
super().__init__()
self.encoder = nn.Sequential(
nn.Linear(input_dim, 64),
nn.ReLU(),
nn.BatchNorm1d(64),
nn.Dropout(0.2),
nn.Linear(64, 32),
nn.ReLU(),
nn.Linear(32, encoding_dim),
)
self.decoder = nn.Sequential(
nn.Linear(encoding_dim, 32),
nn.ReLU(),
nn.Linear(32, 64),
nn.ReLU(),
nn.Linear(64, input_dim),
)
def forward(self, x):
encoded = self.encoder(x)
decoded = self.decoder(encoded)
return decoded
def get_anomaly_score(self, x: torch.Tensor) -> torch.Tensor:
"""计算重建误差作为异常分数"""
with torch.no_grad():
reconstructed = self.forward(x)
mse = torch.mean((x - reconstructed) ** 2, dim=1)
return mse
def train_autoencoder(model, data, epochs=50, lr=1e-3, batch_size=64):
"""训练自编码器"""
dataset = TensorDataset(torch.FloatTensor(data))
loader = DataLoader(dataset, batch_size=batch_size, shuffle=True)
optimizer = torch.optim.Adam(model.parameters(), lr=lr)
criterion = nn.MSELoss()
model.train()
for epoch in range(epochs):
total_loss = 0
for batch in loader:
x = batch[0]
reconstructed = model(x)
loss = criterion(reconstructed, x)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item()
if (epoch + 1) % 10 == 0:
print(f"Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(loader):.6f}")
# 训练模型（只用正常数据）
normal_data = X[y == 0]
ae_model = AnomalyAutoEncoder(input_dim=2, encoding_dim=4)
train_autoencoder(ae_model, normal_data, epochs=50)
# 检测异常
scores = ae_model.get_anomaly_score(torch.FloatTensor(X))
threshold = scores[y == 0].mean() + 3 * scores[y == 0].std()
preds = (scores > threshold).int().numpy()
precision, recall, f1, _ = precision_recall_fscore_support(y, preds, average="binary")
print(f"AutoEncoder - Precision: {precision:.3f}, Recall: {recall:.3f}, F1: {f1:.3f}")
VAE（变分自编码器）
VAE 在 AutoEncoder 基础上引入了概率建模，可以更好地捕捉数据分布：
class VariationalAutoEncoder(nn.Module):
"""变分自编码器用于异常检测"""
def __init__(self, input_dim, latent_dim=8):
super().__init__()
# 编码器
self.encoder = nn.Sequential(
nn.Linear(input_dim, 64),
nn.ReLU(),
nn.Linear(64, 32),
nn.ReLU(),
)
self.fc_mu = nn.Linear(32, latent_dim)
self.fc_logvar = nn.Linear(32, latent_dim)
# 解码器
self.decoder = nn.Sequential(
nn.Linear(latent_dim, 32),
nn.ReLU(),
nn.Linear(32, 64),
nn.ReLU(),
nn.Linear(64, input_dim),
)
def encode(self, x):
h = self.encoder(x)
return self.fc_mu(h), self.fc_logvar(h)
def reparameterize(self, mu, logvar):
std = torch.exp(0.5 * logvar)
eps = torch.randn_like(std)
return mu + eps * std
def forward(self, x):
mu, logvar = self.encode(x)
z = self.reparameterize(mu, logvar)
reconstructed = self.decoder(z)
return reconstructed, mu, logvar
def vae_loss(self, x, reconstructed, mu, logvar):
"""VAE损失 = 重建损失 + KL散度"""
recon_loss = nn.functional.mse_loss(reconstructed, x, reduction='sum')
kl_loss = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp())
return recon_loss + kl_loss
def get_anomaly_score(self, x: torch.Tensor) -> torch.Tensor:
"""综合重建误差和KL散度作为异常分数"""
with torch.no_grad():
reconstructed, mu, logvar = self.forward(x)
recon_err = torch.mean((x - reconstructed) ** 2, dim=1)
kl = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp(), dim=1)
return recon_err + 0.1 * kl
GAN-based 异常检测
AnoGAN 利用 GAN 生成正常数据的能力来检测异常：
class AnomalyGAN(nn.Module):
"""基于GAN的异常检测（简化版AnoGAN）"""
def __init__(self, latent_dim=32, output_dim=2):
super().__init__()
self.generator = nn.Sequential(
nn.Linear(latent_dim, 64),
nn.LeakyReLU(0.2),
nn.Linear(64, 32),
nn.LeakyReLU(0.2),
nn.Linear(32, output_dim),
)
self.discriminator = nn.Sequential(
nn.Linear(output_dim, 32),
nn.LeakyReLU(0.2),
nn.Linear(32, 64),
nn.LeakyReLU(0.2),
nn.Linear(64, 1),
nn.Sigmoid(),
)
def find_closest_z(self, x: torch.Tensor, n_steps=1000, lr=0.01):
"""AnoGAN核心：找到最能重建输入x的潜在向量z"""
z = torch.randn(x.shape[0], 32, requires_grad=True)
optimizer = torch.optim.Adam([z], lr=lr)
for _ in range(n_steps):
generated = self.generator(z)
recon_loss = torch.mean((x - generated) ** 2)
optimizer.zero_grad()
recon_loss.backward()
optimizer.step()
final_generated = self.generator(z)
anomaly_score = torch.mean((x - final_generated) ** 2, dim=1)
return anomaly_score.detach()
4. 时序异常检测
时间序列数据的异常检测需要考虑时间维度上的依赖关系。
基于 STL 分解
STL（Seasonal and Trend decomposition using Loess）将时序分解为趋势、季节性和残差，残差中的异常值即为时序异常：
import pandas as pd
class STLAnomalyDetector:
"""基于STL分解的时序异常检测"""
def __init__(self, period=7, threshold=3.0):
self.period = period
self.threshold = threshold
self.residual_stats = None
def fit(self, ts: pd.Series):
"""拟合正常数据的残差分布"""
# 简化版STL：移动平均去趋势和季节性
trend = ts.rolling(window=self.period, center=True).mean()
detrended = ts - trend
seasonal = detrended.rolling(window=self.period).mean()
residual = detrended - seasonal
residual = residual.dropna()
self.residual_stats = {
"mean": residual.mean(),
"std": residual.std(),
}
return self
def predict(self, ts: pd.Series) -> pd.Series:
trend = ts.rolling(window=self.period, center=True).mean()
detrended = ts - trend
seasonal = detrended.rolling(window=self.period).mean()
residual = detrended - seasonal
z_scores = np.abs((residual - self.residual_stats["mean"]) /
(self.residual_stats["std"] + 1e-8))
return (z_scores > self.threshold).fillna(False)
# 生成模拟时序数据
np.random.seed(42)
dates = pd.date_range("2024-01-01", periods=365, freq="D")
trend = np.linspace(10, 30, 365)
seasonal = 5 * np.sin(2 * np.pi * np.arange(365) / 7)
noise = np.random.normal(0, 1, 365)
values = trend + seasonal + noise
# 注入异常
values[100] += 20 # 突增异常
values[200] -= 15 # 突降异常
ts = pd.Series(values, index=dates)
detector = STLAnomalyDetector(period=7, threshold=3.0)
detector.fit(ts)
anomalies = detector.predict(ts)
print(f"检测到 {anomalies.sum()} 个异常点")
print(f"异常位置: {ts[anomalies].index.tolist()}")
LSTM 时序异常检测
使用 LSTM 学习正常时序模式，当实际值与预测值偏差过大时判定为异常：
class LSTMAnomalyDetector(nn.Module):
"""基于LSTM的时序异常检测"""
def __init__(self, input_dim=1, hidden_dim=64, num_layers=2):
super().__init__()
self.lstm = nn.LSTM(input_dim, hidden_dim, num_layers,
batch_first=True, dropout=0.2)
self.fc = nn.Linear(hidden_dim, input_dim)
def forward(self, x):
# x: [batch, seq_len, input_dim]
lstm_out, _ = self.lstm(x)
predictions = self.fc(lstm_out)
return predictions
def create_sequences(data: np.ndarray, seq_length: int = 30):
"""创建滑动窗口序列"""
sequences = []
targets = []
for i in range(len(data) - seq_length):
sequences.append(data[i:i + seq_length])
targets.append(data[i + seq_length])
return np.array(sequences), np.array(targets)
# 准备数据
seq_length = 30
X_seq, y_seq = create_sequences(values.reshape(-1, 1), seq_length)
# 训练（使用前80%数据）
split = int(0.8 * len(X_seq))
X_train = torch.FloatTensor(X_seq[:split])
y_train = torch.FloatTensor(y_seq[:split])
model = LSTMAnomalyDetector(input_dim=1, hidden_dim=64)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.MSELoss()
model.train()
for epoch in range(50):
predictions = model(X_train).squeeze(-1)
# 取序列最后一个时间步的预测
loss = criterion(predictions[:, -1, :], y_train)
optimizer.zero_grad()
loss.backward()
optimizer.step()
# 异常检测
model.eval()
with torch.no_grad():
X_all = torch.FloatTensor(X_seq)
preds = model(X_all).squeeze(-1)[:, -1, :].numpy()
errors = np.abs(preds.flatten() - y_seq.flatten())
threshold = np.mean(errors[:split]) + 3 * np.std(errors[:split])
anomalies = errors > threshold
print(f"LSTM检测到 {anomalies.sum()} 个异常点")
5. 图异常检测
图结构数据中的异常检测需要考虑节点之间的关系：
import numpy as np
from collections import defaultdict
class SimpleGraphAnomalyDetector:
"""简化的图异常检测：基于节点度和社区结构"""
def __init__(self):
self.graph = defaultdict(set)
self.node_features = {}
def add_edge(self, u: str, v: str):
self.graph[u].add(v)
self.graph[v].add(u)
def set_node_features(self, node: str, features: np.ndarray):
self.node_features[node] = features
def detect_degree_anomaly(self, threshold_factor: float = 3.0) -> list:
"""基于节点度的异常检测"""
degrees = {node: len(neighbors) for node, neighbors in self.graph.items()}
mean_deg = np.mean(list(degrees.values()))
std_deg = np.std(list(degrees.values()))
anomalies = []
for node, deg in degrees.items():
if abs(deg - mean_deg) > threshold_factor * std_deg:
anomalies.append((node, deg, "度异常"))
return anomalies
def detect_structural_anomaly(self) -> list:
"""基于结构特征的异常检测（如聚类系数）"""
anomalies = []
for node in self.graph:
neighbors = self.graph[node]
if len(neighbors) < 2:
continue
# 计算聚类系数
possible_triangles = len(neighbors) * (len(neighbors) - 1) / 2
actual_triangles = 0
for n1 in neighbors:
for n2 in neighbors:
if n1 != n2 and n2 in self.graph[n1]:
actual_triangles += 1
actual_triangles //= 2 # 去重
clustering = actual_triangles / possible_triangles if possible_triangles > 0 else 0
# 聚类系数极低可能是异常（如桥接节点）
if clustering < 0.01 and len(neighbors) > 5:
anomalies.append((node, clustering, "低聚类系数"))
return anomalies
# 构建示例图
detector = SimpleGraphAnomalyDetector()
normal_nodes = [f"n{i}" for i in range(50)]
for i in range(len(normal_nodes) - 1):
detector.add_edge(normal_nodes[i], normal_nodes[i + 1])
detector.add_edge(normal_nodes[0], normal_nodes[-1]) # 形成环
# 添加异常节点：度极高
anomaly_node = "anomaly_hub"
for i in range(20):
detector.add_edge(anomaly_node, normal_nodes[i])
degree_anomalies = detector.detect_degree_anomaly()
for node, deg, reason in degree_anomalies:
print(f"节点 {node}: 度={deg}, 原因={reason}")
6. 金融欺诈检测系统
金融欺诈检测是异常检测最重要的应用场景之一，面临极端不平衡、实时性要求高、概念漂移等挑战：
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
class FraudDetectionPipeline:
"""金融欺诈检测流水线"""
def __init__(self):
self.scaler = StandardScaler()
self.models = {}
self.feature_names = []
def engineer_features(self, transactions: list) -> np.ndarray:
"""特征工程：从原始交易数据提取欺诈相关特征"""
features = []
for tx in transactions:
feat = [
tx["amount"],
tx["amount"] / (tx.get("avg_amount", 1) + 1e-8), # 金额偏离度
tx.get("hour_of_day", 12),
tx.get("day_of_week", 3),
tx.get("distance_from_home", 0), # 交易地点与常驻地距离
tx.get("time_since_last_tx", 0), # 距上次交易时间
tx.get("tx_count_1h", 0), # 1小时内交易次数
tx.get("tx_count_24h", 0), # 24小时内交易次数
tx.get("unique_merchants_24h", 0), # 24小时商户数
tx.get("is_foreign", 0), # 是否境外交易
tx.get("is_online", 0), # 是否线上交易
tx.get("merchant_risk_score", 0), # 商户风险评分
]
features.append(feat)
return np.array(features)
def handle_imbalance(self, X, y, strategy="smote"):
"""处理类别不平衡"""
if strategy == "smote":
# 简化版SMOTE：对少数类进行过采样
minority_idx = np.where(y == 1)[0]
majority_idx = np.where(y == 0)[0]
n_synthetic = len(majority_idx) - len(minority_idx)
synthetic_samples = []
for _ in range(n_synthetic):
idx = np.random.choice(minority_idx)
neighbor_idx = np.random.choice(minority_idx)
alpha = np.random.random()
sample = alpha * X[idx] + (1 - alpha) * X[neighbor_idx]
synthetic_samples.append(sample)
X_resampled = np.vstack([X, np.array(synthetic_samples)])
y_resampled = np.hstack([y, np.ones(n_synthetic)])
# 打乱顺序
shuffle_idx = np.random.permutation(len(y_resampled))
return X_resampled[shuffle_idx], y_resampled[shuffle_idx].astype(int)
elif strategy == "undersample":
minority_idx = np.where(y == 1)[0]
majority_idx = np.where(y == 0)[0]
selected_majority = np.random.choice(majority_idx,
size=len(minority_idx) * 5,
replace=False)
selected_idx = np.concatenate([minority_idx, selected_majority])
return X[selected_idx], y[selected_idx]
def train_ensemble(self, X_train, y_train):
"""训练集成模型"""
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
self.models = {
"rf": RandomForestClassifier(n_estimators=100, random_state=42,
class_weight="balanced"),
"gb": GradientBoostingClassifier(n_estimators=100, random_state=42),
"lr": LogisticRegression(class_weight="balanced", max_iter=1000),
}
for name, model in self.models.items():
model.fit(X_train, y_train)
print(f"模型 {name} 训练完成")
def predict(self, X: np.ndarray, method="voting") -> np.ndarray:
"""集成预测"""
predictions = {}
probabilities = {}
for name, model in self.models.items():
predictions[name] = model.predict(X)
if hasattr(model, "predict_proba"):
probabilities[name] = model.predict_proba(X)[:, 1]
if method == "voting":
# 加权投票
weights = {"rf": 0.4, "gb": 0.4, "lr": 0.2}
weighted_proba = sum(
probabilities[name] * weights[name]
for name in self.models
)
return (weighted_proba > 0.5).astype(int)
return predictions["rf"] # 默认使用随机森林
# 模拟欺诈检测场景
np.random.seed(42)
n_normal = 10000
n_fraud = 100 # 极端不平衡：1%欺诈率
# 生成模拟数据
X_normal = np.random.randn(n_normal, 12) * 0.5
X_fraud = np.random.randn(n_fraud, 12) * 2 + 1 # 欺诈数据分布不同
X = np.vstack([X_normal, X_fraud])
y = np.array([0] * n_normal + [1] * n_fraud)
# 训练流水线
pipeline = FraudDetectionPipeline()
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3,
random_state=42, stratify=y)
X_train_balanced, y_train_balanced = pipeline.handle_imbalance(X_train, y_train)
pipeline.train_ensemble(X_train_balanced, y_train_balanced)
# 评估
from sklearn.metrics import classification_report
preds = pipeline.predict(X_test)
print("\n分类报告:")
print(classification_report(y_test, preds, target_names=["正常", "欺诈"]))
7. 实时流式异常检测
生产环境中的异常检测通常需要处理实时数据流：
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Callable
@dataclass
class StreamingAnomalyDetector:
"""流式异常检测器：基于滑动窗口"""
window_size: int = 1000
alert_threshold: float = 3.0
window: deque = field(default_factory=lambda: deque(maxlen=1000))
alert_callbacks: list = field(default_factory=list)
def register_alert(self, callback: Callable):
"""注册告警回调"""
self.alert_callbacks.append(callback)
def process_event(self, event: dict) -> dict:
"""处理单个事件"""
value = event.get("value", 0)
timestamp = event.get("timestamp", time.time())
# 添加到窗口
self.window.append(value)
# 至少需要一定数据量才能判断
if len(self.window) < 30:
return {"is_anomaly": False, "score": 0, "reason": "数据不足"}
# 计算异常分数
window_array = np.array(self.window)
mean = np.mean(window_array[:-1]) # 排除当前值
std = np.std(window_array[:-1]) + 1e-8
z_score = abs(value - mean) / std
is_anomaly = z_score > self.alert_threshold
result = {
"is_anomaly": is_anomaly,
"score": z_score,
"value": value,
"mean": mean,
"std": std,
"timestamp": timestamp,
}
if is_anomaly:
result["reason"] = f"Z-score {z_score:.2f} 超过阈值 {self.alert_threshold}"
for callback in self.alert_callbacks:
callback(result)
return result
# 使用示例
detector = StreamingAnomalyDetector(window_size=500, alert_threshold=3.0)
def alert_handler(result: dict):
print(f"⚠️ 异常告警! 值={result['value']:.2f}, "
f"Z-score={result['score']:.2f}")
detector.register_alert(alert_handler)
# 模拟实时数据流
np.random.seed(42)
for i in range(200):
if i in [50, 100, 150]: # 注入异常
value = np.random.normal(100, 5)
else:
value = np.random.normal(0, 1)
result = detector.process_event({"value": value, "timestamp": time.time()})
if result["is_anomaly"]:
print(f" 位置 {i}: {result}")
8. 无监督 vs 半监督方法
无监督方法
无监督方法不需要标签，直接从数据结构中发现异常模式。常用方法包括聚类、密度估计和降维：
from sklearn.cluster import DBSCAN
from sklearn.decomposition import PCA
class UnsupervisedAnomalyDetector:
"""基于多种无监督方法的异常检测"""
def __init__(self):
self.models = {}
def dbscan_detect(self, X: np.ndarray, eps=0.5, min_samples=5) -> np.ndarray:
"""DBSCAN聚类：噪声点即为异常"""
clustering = DBSCAN(eps=eps, min_samples=min_samples)
labels = clustering.fit_predict(X)
# label为-1的点是噪声（异常）
return (labels == -1).astype(int)
def pca_reconstruct_detect(self, X: np.ndarray, n_components=2,
threshold_percentile=95) -> np.ndarray:
"""PCA重建误差法"""
pca = PCA(n_components=n_components)
X_reduced = pca.fit_transform(X)
X_reconstructed = pca.inverse_transform(X_reduced)
recon_error = np.sum((X - X_reconstructed) ** 2, axis=1)
threshold = np.percentile(recon_error, threshold_percentile)
return (recon_error > threshold).astype(int)
# 对比两种方法
unsup = UnsupervisedAnomalyDetector()
dbscan_preds = unsup.dbscan_detect(X, eps=2.0, min_samples=10)
pca_preds = unsup.pca_reconstruct_detect(X, n_components=6, threshold_percentile=95)
print(f"DBSCAN检出: {dbscan_preds.sum()}, PCA检出: {pca_preds.sum()}")
半监督方法
半监督方法利用少量标签来提升检测效果，在实际业务中非常实用：
class SemiSupervisedAnomalyDetector:
"""半监督异常检测：自训练框架"""
def __init__(self, base_model, threshold=0.5):
self.base_model = base_model
self.threshold = threshold
def fit_iterative(self, X_labeled, y_labeled, X_unlabeled,
n_iterations=5, confidence_threshold=0.9):
"""自训练迭代：逐步将高置信度预测加入训练集"""
X_train = X_labeled.copy()
y_train = y_labeled.copy()
for iteration in range(n_iterations):
self.base_model.fit(X_train, y_train)
if len(X_unlabeled) == 0:
break
# 对无标签数据预测
proba = self.base_model.predict_proba(X_unlabeled)
max_confidence = np.max(proba, axis=1)
predictions = np.argmax(proba, axis=1)
# 选择高置信度样本
high_conf_mask = max_confidence > confidence_threshold
n_new = high_conf_mask.sum()
if n_new == 0:
print(f"迭代 {iteration}: 无高置信度样本，停止")
break
# 加入训练集
X_train = np.vstack([X_train, X_unlabeled[high_conf_mask]])
y_train = np.hstack([y_train, predictions[high_conf_mask]])
# 移除已标注的无标签数据
X_unlabeled = X_unlabeled[~high_conf_mask]
print(f"迭代 {iteration}: 新增 {n_new} 个伪标签样本, "
f"剩余无标签 {len(X_unlabeled)}")
return self
9. 可解释性与误报管理
在金融等高风险领域，异常检测结果的可解释性至关重要：
class AnomalyExplainer:
"""异常检测结果解释器"""
def __init__(self, feature_names: list):
self.feature_names = feature_names
def explain_point(self, x: np.ndarray, x_normal_mean: np.ndarray,
x_normal_std: np.ndarray) -> dict:
"""解释单个样本为什么被判定为异常"""
z_scores = np.abs((x - x_normal_mean) / (x_normal_std + 1e-8))
# 按异常程度排序
sorted_indices = np.argsort(z_scores)[::-1]
explanations = []
for idx in sorted_indices[:5]: # 取前5个最异常的特征
if z_scores[idx] > 1.5: # 只报告显著异常的特征
direction = "高于" if x[idx] > x_normal_mean[idx] else "低于"
explanations.append({
"feature": self.feature_names[idx],
"value": float(x[idx]),
"normal_mean": float(x_normal_mean[idx]),
"z_score": float(z_scores[idx]),
"description": (
f"{self.feature_names[idx]} = {x[idx]:.2f}, "
f"比正常均值{direction} {z_scores[idx]:.1f}个标准差"
),
})
return {
"is_anomaly": any(e["z_score"] > 3 for e in explanations),
"overall_score": float(np.max(z_scores)),
"top_explanations": explanations,
}
def generate_report(self, X_anomalies: np.ndarray,
X_normal_mean: np.ndarray,
X_normal_std: np.ndarray) -> str:
"""生成异常检测报告"""
report_lines = ["=== 异常检测报告 ===\n"]
for i, x in enumerate(X_anomalies):
result = self.explain_point(x, X_normal_mean, X_normal_std)
report_lines.append(f"样本 #{i+1}:")
report_lines.append(f" 异常分数: {result['overall_score']:.2f}")
report_lines.append(f" 主要原因:")
for exp in result["top_explanations"]:
report_lines.append(f" - {exp['description']}")
report_lines.append("")
return "\n".join(report_lines)
# 使用示例
feature_names = [
"金额", "金额偏离度", "小时", "星期几", "距离常驻地",
"距上次交易间隔", "1h交易次数", "24h交易次数",
"24h商户数", "是否境外", "是否线上", "商户风险分"
]
explainer = AnomalyExplainer(feature_names)
normal_mean = np.mean(X[y == 0], axis=0)
normal_std = np.std(X[y == 0], axis=0)
# 解释一个异常样本
anomaly_sample = X[y == 1][0]
explanation = explainer.explain_point(anomaly_sample, normal_mean, normal_std)
for exp in explanation["top_explanations"]:
print(f" {exp['description']}")
误报管理策略
降低误报率是欺诈检测系统的核心挑战：
class FalsePositiveManager:
"""误报管理器"""
def __init__(self):
self.feedback_log = []
self.whitelist = set()
self.threshold_adjustments = {}
def add_feedback(self, transaction_id: str, predicted: int,
actual: int, reason: str = ""):
"""记录人工审核反馈"""
self.feedback_log.append({
"tx_id": transaction_id,
"predicted": predicted,
"actual": actual,
"correct": predicted == actual,
"reason": reason,
})
def analyze_fp_patterns(self) -> dict:
"""分析误报模式"""
fps = [f for f in self.feedback_log
if f["predicted"] == 1 and f["actual"] == 0]
if not fps:
return {"total_fp": 0, "patterns": []}
# 分析误报原因分布
reason_counts = {}
for fp in fps:
reason = fp.get("reason", "unknown")
reason_counts[reason] = reason_counts.get(reason, 0) + 1
return {
"total_fp": len(fps),
"total_feedback": len(self.feedback_log),
"fp_rate": len(fps) / max(len(self.feedback_log), 1),
"top_reasons": sorted(reason_counts.items(),
key=lambda x: x[1], reverse=True)[:5],
}
def adaptive_threshold(self, target_precision: float = 0.95) -> float:
"""根据反馈自适应调整阈值"""
if len(self.feedback_log) < 100:
return 0.5 # 数据不足时使用默认阈值
scores = []
labels = []
for fb in self.feedback_log:
scores.append(fb.get("score", 0.5))
labels.append(fb["actual"])
scores = np.array(scores)
labels = np.array(labels)
# 搜索满足目标精确率的最优阈值
best_threshold = 0.5
for threshold in np.arange(0.1, 1.0, 0.01):
preds = (scores > threshold).astype(int)
if preds.sum() == 0:
continue
precision = sum((preds == 1) & (labels == 1)) / max(preds.sum(), 1)
if precision >= target_precision:
best_threshold = threshold
break
return best_threshold
10. 实战案例：交易欺诈检测系统
下面整合前面所有模块，构建一个完整的交易欺诈检测系统：
import json
import time
from datetime import datetime
class TransactionFraudSystem:
"""交易欺诈检测完整系统"""
def __init__(self):
self.rule_engine = RuleEngine()
self.ml_model = None # 训练好的ML模型
self.feature_store = {} # 用户特征存储
self.alert_queue = []
self.stats = {"total": 0, "flagged": 0, "confirmed_fraud": 0}
def process_transaction(self, tx: dict) -> dict:
"""处理单笔交易的完整流程"""
self.stats["total"] += 1
user_id = tx["user_id"]
# 1. 更新用户特征
self._update_features(user_id, tx)
features = self._get_features(user_id, tx)
# 2. 规则引擎（快速过滤）
rule_result = self.rule_engine.evaluate(tx, features)
if rule_result["blocked"]:
return self._create_alert(tx, "rule", rule_result, score=1.0)
# 3. ML模型评分
ml_score = self._ml_score(features)
# 4. 综合决策
final_score = 0.6 * ml_score + 0.4 * rule_result.get("risk_score", 0)
if final_score > 0.8:
return self._create_alert(tx, "ml+rule", {
"ml_score": ml_score,
"rule_score": rule_result.get("risk_score", 0),
"final_score": final_score,
}, score=final_score)
elif final_score > 0.5:
return {"action": "review", "score": final_score, "tx_id": tx["id"]}
else:
return {"action": "approve", "score": final_score, "tx_id": tx["id"]}
def _update_features(self, user_id: str, tx: dict):
"""实时更新用户特征"""
if user_id not in self.feature_store:
self.feature_store[user_id] = {
"tx_history": [],
"total_amount_24h": 0,
"tx_count_1h": 0,
"tx_count_24h": 0,
"unique_merchants": set(),
"last_tx_time": 0,
}
store = self.feature_store[user_id]
now = time.time()
store["tx_history"].append(tx)
store["total_amount_24h"] += tx["amount"]
store["tx_count_24h"] += 1
store["tx_count_1h"] += 1
store["unique_merchants"].add(tx.get("merchant_id", ""))
store["last_tx_time"] = now
def _get_features(self, user_id: str, tx: dict) -> np.ndarray:
"""提取特征向量"""
store = self.feature_store[user_id]
features = [
tx["amount"],
store["total_amount_24h"],
store["tx_count_1h"],
store["tx_count_24h"],
len(store["unique_merchants"]),
time.time() - store["last_tx_time"],
int(tx.get("is_foreign", 0)),
int(tx.get("is_online", 0)),
]
return np.array(features).reshape(1, -1)
def _ml_score(self, features: np.ndarray) -> float:
"""ML模型评分"""
if self.ml_model is None:
return 0.0
proba = self.ml_model.predict_proba(features)[0][1]
return float(proba)
def _create_alert(self, tx: dict, source: str, details: dict,
score: float) -> dict:
"""创建告警"""
self.stats["flagged"] += 1
alert = {
"alert_id": f"ALT-{self.stats['flagged']:06d}",
"tx_id": tx["id"],
"user_id": tx["user_id"],
"amount": tx["amount"],
"score": score,
"source": source,
"details": details,
"timestamp": datetime.now().isoformat(),
"status": "pending",
}
self.alert_queue.append(alert)
return {"action": "block", "alert": alert}
class RuleEngine:
"""规则引擎"""
def __init__(self):
self.rules = [
self._check_amount_limit,
self._check_frequency,
self._check_time_anomaly,
]
def evaluate(self, tx: dict, features: np.ndarray) -> dict:
risk_score = 0
reasons = []
for rule in self.rules:
result = rule(tx, features)
if result["triggered"]:
risk_score += result["weight"]
reasons.append(result["reason"])
return {
"blocked": risk_score > 0.9,
"risk_score": min(risk_score, 1.0),
"reasons": reasons,
}
def _check_amount_limit(self, tx: dict, features: np.ndarray) -> dict:
if tx["amount"] > 50000:
return {"triggered": True, "weight": 0.5,
"reason": f"单笔金额 {tx['amount']} 超过5万"}
return {"triggered": False, "weight": 0, "reason": ""}
def _check_frequency(self, tx: dict, features: np.ndarray) -> dict:
if features[0][2] > 10: # 1小时内超过10笔
return {"triggered": True, "weight": 0.4,
"reason": "1小时内交易次数过多"}
return {"triggered": False, "weight": 0, "reason": ""}
def _check_time_anomaly(self, tx: dict, features: np.ndarray) -> dict:
hour = datetime.now().hour
if hour >= 1 and hour <= 5:
return {"triggered": True, "weight": 0.3,
"reason": "凌晨时段交易"}
return {"triggered": False, "weight": 0, "reason": ""}
# 运行示例
system = TransactionFraudSystem()
test_transactions = [
{"id": "TX001", "user_id": "U100", "amount": 200, "is_online": 1},
{"id": "TX002", "user_id": "U100", "amount": 80000, "is_foreign": 1},
{"id": "TX003", "user_id": "U101", "amount": 50, "is_online": 0},
]
for tx in test_transactions:
result = system.process_transaction(tx)
print(f"交易 {tx['id']}: {result['action']}, "
f"分数: {result.get('score', 'N/A'):.2f}")
11. 模型监控与更新策略
部署后的异常检测模型需要持续监控和更新，以应对概念漂移和新型攻击：
class ModelMonitor:
"""异常检测模型监控器"""
def __init__(self, window_size=10000):
self.window_size = window_size
self.predictions = []
self.ground_truth = []
self.feature_distributions = {}
self.alert_history = []
def log_prediction(self, features: np.ndarray, prediction: int,
score: float, actual: int = None):
"""记录预测结果"""
self.predictions.append({
"prediction": prediction,
"score": score,
"actual": actual,
"timestamp": time.time(),
"features": features.tolist(),
})
if actual is not None:
self.ground_truth.append({
"predicted": prediction,
"actual": actual,
"timestamp": time.time(),
})
def check_data_drift(self, reference_data: np.ndarray,
current_data: np.ndarray,
threshold=0.1) -> dict:
"""检测数据漂移（使用PSI - Population Stability Index）"""
drift_results = {}
for col in range(reference_data.shape[1]):
# 计算PSI
ref_hist, bins = np.histogram(reference_data[:, col], bins=20)
cur_hist, _ = np.histogram(current_data[:, col], bins=bins)
# 归一化
ref_pct = ref_hist / ref_hist.sum() + 1e-8
cur_pct = cur_hist / cur_hist.sum() + 1e-8
psi = np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))
drift_results[f"feature_{col}"] = {
"psi": float(psi),
"drifted": psi > threshold,
}
total_drifted = sum(1 for v in drift_results.values() if v["drifted"])
return {
"features": drift_results,
"total_drifted": total_drifted,
"needs_retrain": total_drifted > len(drift_results) * 0.3,
}
def performance_report(self) -> dict:
"""生成性能报告"""
labeled = [p for p in self.predictions if p["actual"] is not None]
if not labeled:
return {"status": "no_labeled_data"}
y_true = [p["actual"] for p in labeled]
y_pred = [p["prediction"] for p in labeled]
tp = sum(1 for t, p in zip(y_true, y_pred) if t == 1 and p == 1)
fp = sum(1 for t, p in zip(y_true, y_pred) if t == 0 and p == 1)
fn = sum(1 for t, p in zip(y_true, y_pred) if t == 1 and p == 0)
tn = sum(1 for t, p in zip(y_true, y_pred) if t == 0 and p == 0)
precision = tp / max(tp + fp, 1)
recall = tp / max(tp + fn, 1)
f1 = 2 * precision * recall / max(precision + recall, 1e-8)
return {
"total_labeled": len(labeled),
"precision": precision,
"recall": recall,
"f1": f1,
"confusion_matrix": {"tp": tp, "fp": fp, "fn": fn, "tn": tn},
"alert_rate": (tp + fp) / max(len(labeled), 1),
}
def should_retrain(self) -> dict:
"""判断是否需要重新训练"""
reasons = []
# 检查性能下降
report = self.performance_report()
if report.get("f1", 1.0) < 0.7:
reasons.append(f"F1分数下降至 {report['f1']:.3f}")
if report.get("precision", 1.0) < 0.8:
reasons.append(f"精确率下降至 {report['precision']:.3f}")
# 检查误报率上升
if report.get("alert_rate", 0) > 0.1:
reasons.append(f"告警率过高: {report['alert_rate']:.2%}")
return {
"should_retrain": len(reasons) > 0,
"reasons": reasons,
"current_performance": report,
}
class AutoRetrainer:
"""自动重训练管理器"""
def __init__(self, model_class, monitor: ModelMonitor):
self.model_class = model_class
self.monitor = monitor
self.model_versions = []
self.current_version = 0
def retrain(self, X_new: np.ndarray, y_new: np.ndarray,
X_old: np.ndarray = None, y_old: np.ndarray = None):
"""重训练模型"""
self.current_version += 1
# 合并新旧数据（可配置比例）
if X_old is not None:
# 旧数据采样，避免数据量过大
old_sample_idx = np.random.choice(
len(X_old), size=min(len(X_old), len(X_new) * 3), replace=False
)
X_combined = np.vstack([X_old[old_sample_idx], X_new])
y_combined = np.hstack([y_old[old_sample_idx], y_new])
else:
X_combined = X_new
y_combined = y_new
# 训练新模型
new_model = self.model_class()
new_model.fit(X_combined, y_combined)
# 评估新模型
from sklearn.metrics import f1_score
new_preds = new_model.predict(X_combined)
new_f1 = f1_score(y_combined, new_preds)
version_info = {
"version": self.current_version,
"model": new_model,
"f1": new_f1,
"train_size": len(X_combined),
"timestamp": datetime.now().isoformat(),
}
self.model_versions.append(version_info)
print(f"模型 v{self.current_version} 训练完成, F1={new_f1:.3f}")
return new_model
# 监控与重训练流程示例
from sklearn.ensemble import RandomForestClassifier
monitor = ModelMonitor()
retrainer = AutoRetrainer(RandomForestClassifier, monitor)
# 模拟生产环境监控
for i in range(100):
features = np.random.randn(1, 12)
prediction = np.random.choice([0, 1], p=[0.95, 0.05])
score = np.random.random()
actual = np.random.choice([0, 1], p=[0.95, 0.05])
monitor.log_prediction(features, prediction, score, actual)
# 检查是否需要重训练
decision = monitor.should_retrain()
print(f"是否需要重训练: {decision['should_retrain']}")
if decision["should_retrain"]:
print(f"原因: {decision['reasons']}")
异常检测与欺诈识别是一个持续演进的领域。随着深度学习和图计算技术的发展，检测精度和实时性不断提升。在实际业务中，最关键的不是选择最复杂的模型，而是建立完善的数据流水线、监控体系和反馈机制，让系统能够在对抗环境中持续进化。
