异常检测：识别数据中的离群点

import numpy as np
import matplotlib.pyplot as plt

# 模拟正常数据和异常点
np.random.seed(42)
n_normal = 500
n_outliers = 15

# 正常数据：二维高斯分布
normal = np.random.randn(n_normal, 2) * 0.5 + np.array([2, 3])

# 异常点：来自不同分布
outliers = np.random.uniform(low=-2, high=8, size=(n_outliers, 2))

plt.figure(figsize=(8, 6))
plt.scatter(normal[:, 0], normal[:, 1], c='steelblue', alpha=0.6, label='Normal')
plt.scatter(outliers[:, 0], outliers[:, 1], c='red', marker='x', s=100, label='Outlier')
plt.xlabel('Feature 1')
plt.ylabel('Feature 2')
plt.legend()
plt.title('Anomaly Detection: Finding Needles in a Haystack')
plt.show()

from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split

def generate_anomaly_dataset(n_samples=1000, contamination=0.05):
    """生成带有异常样本的分类数据集"""
    n_outliers = int(n_samples * contamination)
    n_inliers = n_samples - n_outliers

    # 正常样本
    inliers = np.random.randn(n_inliers, 2) * 0.3 + np.array([0, 0])

    # 异常样本（远离中心）
    angle = np.random.uniform(0, 2 * np.pi, n_outliers)
    radius = np.random.uniform(3, 6, n_outliers)
    outliers = np.column_stack([radius * np.cos(angle), radius * np.sin(angle)])

    X = np.vstack([inliers, outliers])
    y = np.array([0] * n_inliers + [1] * n_outliers)  # 1 = anomaly

    return X, y

X, y = generate_anomaly_dataset(n_samples=2000, contamination=0.03)
print(f"正常样本: {np.sum(y == 0)}, 异常样本: {np.sum(y == 1)}")
print(f"污染率: {np.mean(y):.4f}")

import numpy as np
from scipy import stats

def zscore_anomaly_detection(data, threshold=3.0):
    """基于 Z-Score 的异常检测"""
    z_scores = np.abs(stats.zscore(data))
    anomalies = np.where(z_scores > threshold)[0]
    return anomalies, z_scores

# 一维示例
data = np.random.randn(1000) * 10 + 50
# 注入 5 个异常值
data[[100, 250, 500, 750, 900]] = [200, -50, 150, -30, 180]

anomalies, scores = zscore_anomaly_detection(data, threshold=3.0)
print(f"检测到 {len(anomalies)} 个异常点")
print(f"异常索引: {anomalies}")
print(f"异常值: {data[anomalies]}")
print(f"对应 Z-Score: {scores[anomalies]:.2f}")

from scipy import stats
import numpy as np

def grubbs_test(data, alpha=0.05):
    """Grubbs 检验：迭代检测最极端异常值"""
    n = len(data)
    results = []

    while n > 3:  # 至少保留 3 个样本
        mean = np.mean(data)
        std = np.std(data, ddof=1)
        if std == 0:
            break

        # 找到最偏离的点
        deviations = np.abs(data - mean)
        idx = np.argmax(deviations)
        G = deviations[idx] / std

        # 临界值（基于 t 分布）
        t_crit = stats.t.ppf(1 - alpha / (2 * n), n - 2)
        G_crit = ((n - 1) / np.sqrt(n)) * np.sqrt(t_crit  2 / (n - 2 + t_crit  2))

        if G > G_crit:
            results.append({
                "index": idx, "value": data[idx],
                "G_statistic": G, "G_critical": G_crit,
                "significant": True
            })
            data = np.delete(data, idx)
            n -= 1
        else:
            break

    return results

# 测试
np.random.seed(42)
sample = np.random.randn(50) * 5 + 100
sample[10] = 200
sample[30] = -50

outliers = grubbs_test(sample, alpha=0.01)
for o in outliers:
    print(f"异常值: {o['value']:.1f} (G={o['G_statistic']:.3f} > {o['G_critical']:.3f})")

from sklearn.ensemble import IsolationForest
from sklearn.datasets import make_blobs
import numpy as np

# 生成测试数据
X, _ = make_blobs(n_samples=500, centers=1, cluster_std=0.5, random_state=42)
# 添加异常点
outliers_X = np.random.uniform(low=-6, high=6, size=(25, 2))
X_all = np.vstack([X, outliers_X])

# Isolation Forest
iso_forest = IsolationForest(
    n_estimators=100,
    contamination=0.05,
    max_samples='auto',
    random_state=42
)
iso_forest.fit(X_all)

# 预测：-1 为异常，1 为正常
predictions = iso_forest.predict(X_all)
scores = iso_forest.decision_function(X_all)

n_anomalies = np.sum(predictions == -1)
print(f"检测到 {n_anomalies} 个异常点")
print(f"异常得分范围: [{scores.min():.4f}, {scores.max():.4f}]")

# 查看异常点的平均路径长度
print(f"平均路径长度: {iso_forest.decision_function(X_all).mean():.4f}")

import numpy as np

class IsolationTree:
    """单棵隔离树的简化实现"""

    def __init__(self, height_limit):
        self.height_limit = height_limit
        self.split_feature = None
        self.split_value = None
        self.left = None
        self.right = None
        self.size = 0
        self.is_leaf = False

    def fit(self, X, height=0):
        self.size = len(X)

        if height >= self.height_limit or len(X) <= 1:
            self.is_leaf = True
            return self

        n_features = X.shape[1]
        self.split_feature = np.random.randint(0, n_features)
        f_min, f_max = X[:, self.split_feature].min(), X[:, self.split_feature].max()

        if f_min == f_max:
            self.is_leaf = True
            return self

        self.split_value = np.random.uniform(f_min, f_max)

        left_mask = X[:, self.split_feature] < self.split_value
        self.left = IsolationTree(self.height_limit)
        self.right = IsolationTree(self.height_limit)
        self.left.fit(X[left_mask], height + 1)
        self.right.fit(X[~left_mask], height + 1)
        return self

    def path_length(self, x, height=0):
        if self.is_leaf:
            return height + _c(self.size)
        if x[self.split_feature] < self.split_value:
            return self.left.path_length(x, height + 1)
        return self.right.path_length(x, height + 1)

def _c(n):
    """二叉搜索树平均路径长度估计"""
    if n <= 1:
        return 0
    if n == 2:
        return 1
    return 2.0 * (np.log(n - 1) + 0.5772156649) - 2.0 * (n - 1) / n

from sklearn.svm import OneClassSVM
from sklearn.preprocessing import StandardScaler
import numpy as np

# 生成正常训练数据（训练时不应包含异常样本）
np.random.seed(42)
X_train = np.random.randn(500, 2) * 0.5

# 标准化（OC-SVM 对尺度敏感）
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)

# 训练 One-Class SVM
oc_svm = OneClassSVM(
    kernel='rbf',
    gamma='auto',
    nu=0.05  # 期望的异常比例
)
oc_svm.fit(X_train_scaled)

# 测试数据（含异常点）
X_test = np.random.randn(100, 2) * 0.5
X_test = np.vstack([X_test, np.random.uniform(-3, 3, (10, 2))])
X_test_scaled = scaler.transform(X_test)

predictions = oc_svm.predict(X_test_scaled)
scores = oc_svm.decision_function(X_test_scaled)

print(f"正常点: {np.sum(predictions == 1)}, 异常点: {np.sum(predictions == -1)}")
print(f"支持向量数量: {np.sum(oc_svm.n_support_)}")
print(f"决策函数范围: [{scores.min():.4f}, {scores.max():.4f}]")

from sklearn.svm import OneClassSVM
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import GridSearchCV
import numpy as np

# 使用已标注的异常数据进行参数搜索（半监督设置）
np.random.seed(42)
X_normal = np.random.randn(300, 2) * 0.5
X_anomaly = np.random.uniform(-3, 3, (15, 2))
X = np.vstack([X_normal, X_anomaly])
y = np.array([1] * 300 + [-1] * 15)  # 1=正常, -1=异常

scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# 网格搜索最优参数
param_grid = {
    'nu': [0.01, 0.05, 0.1],
    'gamma': [0.01, 0.1, 1.0, 'auto', 'scale']
}

from sklearn.metrics import f1_score

best_score = 0
best_params = None

for nu in param_grid['nu']:
    for gamma in param_grid['gamma']:
        model = OneClassSVM(kernel='rbf', nu=nu, gamma=gamma)
        model.fit(X_scaled[:300])  # 仅用正常数据训练

        preds = model.predict(X_scaled)
        score = f1_score(y, preds, pos_label=-1)  # 以异常为正类

        if score > best_score:
            best_score = score
            best_params = {'nu': nu, 'gamma': gamma}

print(f"最优参数: {best_params}")
print(f"最优 F1 (异常类): {best_score:.4f}")

from sklearn.neighbors import LocalOutlierFactor
import numpy as np

# 创建局部异常场景：两个密度不同的簇
np.random.seed(42)
# 密集簇
dense_cluster = np.random.randn(200, 2) * 0.3 + np.array([0, 0])
# 稀疏簇
sparse_cluster = np.random.randn(100, 2) * 1.0 + np.array([5, 5])
# 稀疏簇中的异常点
local_outlier = np.array([[5.0, 5.0], [5.2, 4.8], [4.8, 5.2]])

X = np.vstack([dense_cluster, sparse_cluster, local_outlier])

# LOF 检测
lof = LocalOutlierFactor(
    n_neighbors=20,
    contamination=0.02,
    metric='minkowski',
    novelty=False  # 训练和检测同时进行
)
predictions = lof.fit_predict(X)
scores = lof.negative_outlier_factor_

print(f"检测到 {np.sum(predictions == -1)} 个异常点")
print(f"LOF 分数范围: [{scores.min():.4f}, {scores.max():.4f}]")
print(f"异常点的 LOF 分数: {scores[predictions == -1]}")

import numpy as np
from sklearn.neighbors import NearestNeighbors

def compute_lof(X, k=20):
    """LOF 算法的简化实现"""
    n = len(X)

    # 1. 计算 k 近邻
    nn = NearestNeighbors(n_neighbors=k + 1)
    nn.fit(X)
    distances, indices = nn.kneighbors(X)

    # 排除自身（第一个邻居是自己）
    k_distances = distances[:, k]  # k 距离
    neighborhoods = indices[:, 1:]  # k 个邻居

    # 2. 计算可达距离
    reach_dist = np.zeros((n, n))
    for i in range(n):
        for j in neighborhoods[i]:
            actual_dist = np.linalg.norm(X[i] - X[j])
            reach_dist[i, j] = max(k_distances[j], actual_dist)

    # 3. 计算局部可达密度
    lrd = np.zeros(n)
    for i in range(n):
        avg_reach_dist = np.mean([reach_dist[i, j] for j in neighborhoods[i]])
        lrd[i] = 1.0 / max(avg_reach_dist, 1e-10)

    # 4. 计算 LOF 分数
    lof_scores = np.zeros(n)
    for i in range(n):
        neighbor_lrd = np.mean([lrd[j] for j in neighborhoods[i]])
        lof_scores[i] = neighbor_lrd / max(lrd[i], 1e-10)

    return lof_scores

# 测试
np.random.seed(42)
X_test = np.vstack([
    np.random.randn(100, 2) * 0.5,
    np.array([[3, 3], [3.5, 3.5]])
])
scores = compute_lof(X_test, k=10)
anomalies = np.where(scores > 2.0)[0]
print(f"异常索引: {anomalies}, LOF 分数: {scores[anomalies]}")

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np

class AnomalyAutoEncoder(nn.Module):
    """用于异常检测的自编码器"""

    def __init__(self, input_dim, hidden_dim=64, latent_dim=16):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, latent_dim)
        )
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, input_dim)
        )

    def forward(self, x):
        z = self.encoder(x)
        x_recon = self.decoder(z)
        return x_recon

    def anomaly_score(self, x):
        """计算重建误差作为异常得分"""
        x_recon = self(x)
        return torch.mean((x - x_recon) ** 2, dim=1)

# 训练示例
model = AnomalyAutoEncoder(input_dim=10)
optimizer = optim.Adam(model.parameters(), lr=0.001)
criterion = nn.MSELoss()

# 仅用正常数据训练
X_normal = torch.randn(500, 10)
for epoch in range(100):
    optimizer.zero_grad()
    recon = model(X_normal)
    loss = criterion(recon, X_normal)
    loss.backward()
    optimizer.step()
    if (epoch + 1) % 20 == 0:
        print(f"Epoch {epoch+1}, Loss: {loss.item():.6f}")

import numpy as np
from sklearn.preprocessing import StandardScaler

class DeepAnomalyDetector:
    """完整的深度异常检测流程"""

    def __init__(self, input_dim, threshold_percentile=95):
        self.input_dim = input_dim
        self.threshold_percentile = threshold_percentile
        self.model = AnomalyAutoEncoder(input_dim)
        self.scaler = StandardScaler()
        self.threshold = None

    def fit(self, X_normal):
        """仅用正常数据训练"""
        X_scaled = self.scaler.fit_transform(X_normal)
        X_tensor = torch.FloatTensor(X_scaled)

        optimizer = optim.Adam(self.model.parameters(), lr=0.001, weight_decay=1e-5)
        criterion = nn.MSELoss()

        # 训练
        for epoch in range(200):
            optimizer.zero_grad()
            recon = self.model(X_tensor)
            loss = criterion(recon, X_tensor)
            loss.backward()
            optimizer.step()

        # 在训练集上设定阈值
        train_errors = self.model.anomaly_score(X_tensor).detach().numpy()
        self.threshold = np.percentile(train_errors, self.threshold_percentile)
        print(f"异常阈值 (P{self.threshold_percentile}): {self.threshold:.6f}")

    def predict(self, X):
        """检测新数据中的异常"""
        X_scaled = self.scaler.transform(X)
        X_tensor = torch.FloatTensor(X_scaled)
        scores = self.model.anomaly_score(X_tensor).detach().numpy()
        return (scores > self.threshold).astype(int), scores

# 使用示例
detector = DeepAnomalyDetector(input_dim=10, threshold_percentile=95)
X_train = np.random.randn(1000, 10)
detector.fit(X_train)

X_test = np.vstack([np.random.randn(50, 10), np.random.uniform(-5, 5, (5, 10))])
preds, scores = detector.predict(X_test)
print(f"检测到 {preds.sum()} 个异常点")

# pip install pyod
from pyod.models.iforest import IForest
from pyod.models.lof import LOF
from pyod.models.ocsvm import OCSVM
from pyod.models.cof import COF
from pyod.utils.data import generate_data
from pyod.utils.metric import evaluate_print
import numpy as np

# 1. 生成带有真实标签的测试数据
X_train, y_train, X_test, y_test = generate_data(
    n_train=1000,
    n_test=500,
    n_features=10,
    contamination=0.05,
    random_state=42,
    behaviour='new'
)

# 2. 定义多个检测器
detectors = {
    "Isolation Forest": IForest(contamination=0.05, random_state=42),
    "LOF": LOF(contamination=0.05, n_neighbors=20),
    "One-Class SVM": OCSVM(contamination=0.05, kernel='rbf'),
    "Connectivity-Based": COF(contamination=0.05, n_neighbors=20),
}

# 3. 训练并评估每个检测器
for name, clf in detectors.items():
    clf.fit(X_train)
    # 预测
    y_pred_train = clf.predict(X_train)
    y_pred_test = clf.predict(X_test)

    # 异常得分
    train_scores = clf.decision_function(X_train)
    test_scores = clf.decision_function(X_test)

    print(f"\n{'='*40}")
    print(f"算法: {name}")
    print(f"训练集异常比例: {np.mean(y_pred_train == 1):.4f}")
    print(f"测试集异常比例: {np.mean(y_pred_test == 1):.4f}")
    print(f"平均异常得分: {np.mean(test_scores):.4f}")

from pyod.models.combination import aom, moa, average, maximization
from pyod.utils.data import generate_data
import numpy as np

# 生成数据
X_train, y_train, X_test, y_test = generate_data(
    n_train=1000, n_test=500, n_features=10,
    contamination=0.05, random_state=42, behaviour='new'
)

# 训练多个基础检测器
from pyod.models.iforest import IForest
from pyod.models.lof import LOF
from pyod.models.ocsvm import OCSVM
from pyod.models.pca import PCA as PCA_OD

clf1 = IForest(contamination=0.05, random_state=42).fit(X_train)
clf2 = LOF(contamination=0.05).fit(X_train)
clf3 = OCSVM(contamination=0.05).fit(X_train)
clf4 = PCA_OD(contamination=0.05).fit(X_train)

# 获取所有检测器的异常得分
scores = np.column_stack([
    clf1.decision_function(X_test),
    clf2.decision_function(X_test),
    clf3.decision_function(X_test),
    clf4.decision_function(X_test),
])

# 组合策略对比
comb_methods = {
    "Average": average(scores),
    "AOM (Average of Max)": aom(scores, n_buckets=5),
    "MOA (Max of Average)": moa(scores, n_buckets=5),
    "Maximization": maximization(scores),
}

from sklearn.metrics import roc_auc_score

for method, combined_scores in comb_methods.items():
    auc = roc_auc_score(y_test, combined_scores)
    print(f"{method:20s} ROC-AUC: {auc:.4f}")

# 最佳组合策略
best_method = max(comb_methods, key=lambda m: roc_auc_score(y_test, comb_methods[m]))
print(f"\n最佳组合策略: {best_method}")