异常检测

import numpy as np
import matplotlib.pyplot as plt
from scipy.stats import norm
 
# 生成高斯分布数据
mu, sigma = 0, 1
data = np.random.normal(mu, sigma, 1000)
 
# 估计参数
mu_est = np.mean(data)
sigma_est = np.std(data)
 
# 画出数据分布和拟合的高斯曲线
plt.hist(data, bins=50, density=True, alpha=0.6, label='数据')
x = np.linspace(-4, 4, 100)
plt.plot(x, norm.pdf(x, mu_est, sigma_est), 'r-', label='拟合的高斯分布')
plt.xlabel('x')
plt.ylabel('概率密度')
plt.legend()
plt.show()

def fitGaussian(X):
    """
    拟合多元高斯分布（假设特征独立）
    X: 数据矩阵 (m x n)
    """
    mu = np.mean(X, axis=0)
    sigma2 = np.var(X, axis=0)
    return mu, sigma2
 
def multivariateGaussian(X, mu, sigma2):
    """
    计算每个样本的概率密度
    """
    n = len(mu)
    sigma2 = sigma2.reshape(-1, 1) if sigma2.ndim == 1 else sigma2
    
    X_centered = X - mu
    p = (2 * np.pi) ** (-n/2) * np.prod(sigma2 ** (-0.5)) * \
        np.exp(-0.5 * np.sum(X_centered ** 2 / sigma2.T, axis=1))
    
    return p
 
def selectThreshold(y_val, p_val):
    """
    选择最优阈值
    y_val: 验证集标签 (1=异常, 0=正常)
    p_val: 验证集样本的概率
    """
    best_epsilon = 0
    best_F1 = 0
    
    step_size = (max(p_val) - min(p_val)) / 1000
    for epsilon in np.arange(min(p_val), max(p_val), step_size):
        predictions = (p_val < epsilon).astype(int)
        
        tp = np.sum((predictions == 1) & (y_val == 1))
        fp = np.sum((predictions == 1) & (y_val == 0))
        fn = np.sum((predictions == 0) & (y_val == 1))
        
        if tp + fp == 0 or tp + fn == 0:
            continue
            
        precision = tp / (tp + fp)
        recall = tp / (tp + fn)
        
        if precision + recall == 0:
            continue
            
        F1 = 2 * precision * recall / (precision + recall)
        
        if F1 > best_F1:
            best_F1 = F1
            best_epsilon = epsilon
    
    return best_epsilon, best_F1
 
# 使用示例
# 训练集（只包含正常样本）
X_train = np.random.randn(1000, 2)
 
# 拟合模型
mu, sigma2 = fitGaussian(X_train)
 
# 验证集（包含一些异常样本）
X_val = np.vstack([np.random.randn(100, 2), np.random.randn(10, 2) * 3 + 5])
y_val = np.array([0]*100 + [1]*10)  # 1表示异常
 
# 计算概率
p_val = multivariateGaussian(X_val, mu, sigma2)
 
# 选择阈值
epsilon, F1 = selectThreshold(y_val, p_val)
print(f"最优阈值: {epsilon:.6f}, F1分数: {F1:.3f}")
 
# 在测试集上检测异常
X_test = np.random.randn(50, 2)
p_test = multivariateGaussian(X_test, mu, sigma2)
anomalies = p_test < epsilon
print(f"检测到 {np.sum(anomalies)} 个异常")

# 检查特征分布
plt.figure(figsize=(12, 4))
for i in range(X_train.shape[1]):
    plt.subplot(1, X_train.shape[1], i+1)
    plt.hist(X_train[:, i], bins=50)
    plt.title(f'特征 {i+1}')
plt.show()
 
# 如果分布很偏，尝试对数变换
X_train_transformed = np.log(X_train + 1)  # +1避免log(0)

def fitMultivariateGaussian(X):
    """
    拟合多变量高斯分布
    """
    mu = np.mean(X, axis=0)
    Sigma = np.cov(X.T)
    return mu, Sigma
 
def multivariateGaussianFull(X, mu, Sigma):
    """
    计算多变量高斯分布的概率密度
    """
    n = len(mu)
    X_centered = X - mu
    
    det_Sigma = np.linalg.det(Sigma)
    inv_Sigma = np.linalg.inv(Sigma)
    
    norm_const = 1 / ((2 * np.pi) ** (n/2) * det_Sigma ** 0.5)
    exponent = -0.5 * np.sum(X_centered @ inv_Sigma * X_centered, axis=1)
    
    p = norm_const * np.exp(exponent)
    return p

# 特征：网络连接的各项指标
features = [
    'duration',  # 连接持续时间
    'src_bytes',  # 源到目标的字节数
    'dst_bytes',  # 目标到源的字节数
    'wrong_fragment',  # 错误的分片数
    'urgent',  # 紧急包的数量
    'hot',  # 访问热点资源的次数
    # ... 更多特征
]
 
# 训练集：正常网络流量
X_train = loadNormalTraffic()
 
# 拟合异常检测模型
mu, sigma2 = fitGaussian(X_train)
 
# 实时监控
while True:
    new_connection = getNextConnection()
    p = multivariateGaussian(new_connection.reshape(1, -1), mu, sigma2)
    
    if p < epsilon:
        alert("检测到可疑连接！")
        logAnomaly(new_connection)