Python异常检测案例如何实现？

wen python案例 2026-06-07 01:00:00 2

本文目录导读：

基于统计方法（Z-Score）
IQR（四分位距法）
Isolation Forest（孤立森林）
DBSCAN聚类方法
完整案例：时间序列异常检测
实时数据流异常检测（简化版）
选择合适的检测方法
总结建议

我来介绍几种Python异常检测的常见实现方法,从简单到复杂：

基于统计方法（Z-Score）

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# 生成示例数据
np.random.seed(42)
normal_data = np.random.normal(50, 10, 1000)
# 添加异常值
anomalies = np.array([5, 95, 100, 2, 98])
data = np.concatenate([normal_data, anomalies])
def z_score_detection(data, threshold=3):
    """使用Z-Score检测异常值"""
    mean = np.mean(data)
    std = np.std(data)
    z_scores = (data - mean) / std
    outliers = np.abs(z_scores) > threshold
    return outliers, z_scores
# 检测异常
outliers, z_scores = z_score_detection(data)
print(f"检测到 {np.sum(outliers)} 个异常值")
print(f"异常值索引: {np.where(outliers)[0]}")

IQR（四分位距法）

def iqr_detection(data, multiplier=1.5):
    """使用IQR方法检测异常值"""
    Q1 = np.percentile(data, 25)
    Q3 = np.percentile(data, 75)
    IQR = Q3 - Q1
    lower_bound = Q1 - multiplier * IQR
    upper_bound = Q3 + multiplier * IQR
    outliers = (data < lower_bound) | (data > upper_bound)
    return outliers, (lower_bound, upper_bound)
# 检测异常
outliers_iqr, bounds = iqr_detection(data)
print(f"IQR方法检测到 {np.sum(outliers_iqr)} 个异常值")

Isolation Forest（孤立森林）

from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
# 生成多维数据
np.random.seed(42)
X_normal = np.random.randn(200, 2) * 0.3
X_outliers = np.random.uniform(low=-4, high=4, size=(20, 2))
X = np.vstack([X_normal, X_outliers])
# 标准化
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
# 孤立森林检测
iso_forest = IsolationForest(
    contamination=0.1,  # 预期异常比例
    random_state=42
)
predictions = iso_forest.fit_predict(X_scaled)
# -1表示异常，1表示正常
outliers_if = predictions == -1
print(f"孤立森林检测到 {np.sum(outliers_if)} 个异常值")

DBSCAN聚类方法

from sklearn.cluster import DBSCAN
def dbscan_detection(X, eps=0.5, min_samples=5):
    """使用DBSCAN检测异常点"""
    db = DBSCAN(eps=eps, min_samples=min_samples)
    clusters = db.fit_predict(X)
    # -1表示异常点（未分配到任何簇）
    outliers = clusters == -1
    return outliers, clusters
# 使用上面的数据
outliers_db, clusters = dbscan_detection(X_scaled, eps=0.3, min_samples=3)
print(f"DBSCAN检测到 {np.sum(outliers_db)} 个异常值")

完整案例：时间序列异常检测

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
import matplotlib.pyplot as plt
# 生成时间序列数据
def generate_time_series():
    dates = pd.date_range('2024-01-01', periods=500, freq='D')
    trend = np.linspace(0, 10, 500)
    seasonal = 5 * np.sin(2 * np.pi * np.arange(500) / 30)
    noise = np.random.normal(0, 1, 500)
    data = trend + seasonal + noise
    # 添加异常值
    data[100] += 15
    data[200] += -12
    data[300] += 20
    data[400] += -18
    return pd.DataFrame({'date': dates, 'value': data})
# 准备数据
df = generate_time_series()
df.set_index('date', inplace=True)
# 特征工程：添加时间特征
df['hour'] = df.index.hour
df['dayofweek'] = df.index.dayofweek
df['day'] = df.index.day
# 异常检测
features = ['value', 'dayofweek']
model = IsolationForest(contamination=0.05, random_state=42)
df['anomaly'] = model.fit_predict(df[features])
# 可视化
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
# 原始数据
ax1.plot(df.index, df['value'], label='正常值', color='blue', alpha=0.6)
anomaly_points = df[df['anomaly'] == -1]
ax1.scatter(anomaly_points.index, anomaly_points['value'], 
           color='red', s=100, label='异常值', marker='x')
ax1.set_title('时间序列异常检测')
ax1.set_xlabel('日期')
ax1.set_ylabel('数值')
ax1.legend()
ax1.grid(True)
# 异常值分布
ax2.hist(df['value'], bins=30, alpha=0.7, label='所有数据')
ax2.hist(anomaly_points['value'], bins=10, alpha=0.7, 
         color='red', label='异常值')
ax2.set_xlabel('数值')
ax2.set_ylabel('频率')
ax2.legend()
ax2.grid(True)
plt.tight_layout()
plt.show()
print(f"检测到 {len(anomaly_points)} 个异常值")
print("异常值位置:")
print(anomaly_points.index)

实时数据流异常检测（简化版）

from collections import deque
import numpy as np
class RealtimeAnomalyDetector:
    def __init__(self, window_size=100, threshold=3):
        self.window = deque(maxlen=window_size)
        self.threshold = threshold
    def update(self, value):
        """添加新数据点并检测是否异常"""
        self.window.append(value)
        if len(self.window) < self.window.maxlen:
            return False
        mean = np.mean(self.window)
        std = np.std(self.window)
        if std == 0:
            return False
        z_score = (value - mean) / std
        return abs(z_score) > self.threshold
# 使用示例
detector = RealtimeAnomalyDetector(window_size=50)
# 模拟数据流
np.random.seed(42)
for i in range(200):
    normal_value = np.random.normal(50, 5)
    # 在第100个点插入异常
    if i == 100:
        normal_value = 100
    is_anomaly = detector.update(normal_value)
    if is_anomaly:
        print(f"检测到异常！在第{i}个数据点: {normal_value:.2f}")

选择合适的检测方法

# 方法选择指南
def choose_anomaly_detection_method(data_description):
    """
    根据数据特点推荐检测方法
    Parameters:
    - data_description: dict, 包含数据特征描述
      - 'data_type': str, 'univariate'/'multivariate'
      - 'distribution': str, 'normal'/'unknown'
      - 'dataset_size': str, 'small'/'large'
      - 'requires_training': bool
    """
    method_map = {
        ('univariate', 'normal', 'small'): 'Z-Score或IQR',
        ('univariate', 'unknown', 'small'): 'IQR',
        ('multivariate', 'normal', 'large'): 'Isolation Forest',
        ('multivariate', 'unknown', 'large'): 'Isolation Forest或DBSCAN',
    }
    key = (
        data_description['data_type'],
        data_description['distribution'],
        data_description['dataset_size']
    )
    return method_map.get(key, '建议使用Isolation Forest（通用方法）')
# 使用示例
data_info = {
    'data_type': 'multivariate',
    'distribution': 'unknown',
    'dataset_size': 'large'
}
recommended = choose_anomaly_detection_method(data_info)
print(f"推荐方法: {recommended}")