异常检测---孤立森林 | one-class SVM

作者：我家小花儿 | 2024-05-30 20:36:41

踩

孤立森林异常检测适用于什么

novelty detection：当训练数据中没有离群点，我们的目标是用训练好的模型去检测另外发现的新样本
outlier dection：当训练数据中包含离群点，模型训练时要匹配训练数据的中心样本，忽视训练样本中的其他异常点。

一、outlier dection

1.孤立森林（Isolation Forest）

　　iForest适用于连续数据（Continuous numerical data）的异常检测，将异常定义为“容易被孤立的离群点（more likely to be separated）可以理解为分布稀疏且离密度高的群体较远的点。用统计学来解释，在数据空间里面，分布稀疏的区域表示数据发生在此区域的概率很低，因此可以认为落在这些区域里的数据是异常的。通常用于网络安全中的攻击检测和流量异常等分析，金融机构则用于挖掘出欺诈行为。对于找出的异常数据，然后要么直接清除异常数据，如数据清理中的去噪数据，要么深入分析异常数据，比如分析攻击，欺诈的行为特征。

（1）算法

论文链接
非参数、无监督
大体思想：类似切蛋糕，切一次生成两个子空间，之后再继续用随机超平面来切割每个子空间，循环，直到每个子空间里只有一个数据点为止。可以发现，密度很高的簇需要切很多次才能结束，而那些密度很低的点很容易很早停到一个子空间里。（参考这里）

算法的流程：参考
- 输入：子采样的样本大小Ψ，树的个数t，默认Ψ=256，t=100，height=8
- 步骤1：从训练数据中随机选择Ψ个点样本点作为subsample，放入树的根节点。
- 步骤2：随机指定一个维度（attribute），在当前节点数据中随机产生一个切割点p——切割点产生于当前节点数据中指定维度的最大值和最小值之间。
- 步骤3：以此切割点生成了一个超平面，然后将当前节点数据空间划分为2个子空间：把指定维度里小于p的数据放在当前节点的左孩子，把大于等于p的数据放在当前节点的右孩子。
- 步骤4：在孩子节点中递归步骤2和3，不断构造新的孩子节点，直到孩子节点中只有一个数据（无法再继续切割）或孩子节点已到达限定高度
- 获得t个iTree之后，iForest 训练就结束，然后我们可以用生成的iForest来评估测试数据了。对于一个训练数据x，我们令其遍历每一棵iTree，然后计算x最终落在每个树第几层（x在树的高度）。然后我们可以得出x在每棵树的高度平均值，即 the average path length over t iTrees。最后，将h(x)带入，计算每条待测数据的异常分数(Anomaly Score)，其计算公式为：　，其中_{$c(n) = 2H(n − 1) − (2(n − 1)/n)$} 是二叉搜索树的平均路径长度，用来对结果进行归一化处理, 其中的H(k)可以通过公式 _{$H(k) = ln(k) + \xi$}来估计，_$\xi$是欧拉常数，其值为0.5772156649。$h(x)$ 为路径长度，$E(h(x))$ 为森林中所有iTree树的平均路径长度。获得每个测试数据的average path length后，我们可以设置一个阈值（边界值），average path length 低于此阈值的测试数据即为异常。论文中对树的高度做了归一化，并得出一个0到1的数值：
  - 如果分数越接近1，其是异常点的可能性越高；
  - 如果分数都比0.5要小，那么基本可以确定为正常数据；
  - 如果所有分数都在0.5附近，那么数据不包含明显的异常样本。
步骤总结：
- 训练：从训练集中进行采样，并构建iTree树；
- 测试：对iForest森林中的每颗iTree树进行测试，记录path length，然后根据异常分数计算公式，计算每条测试数据的anomaly score。
适用场景：适用于连续特征，线性时间复杂度，适用于在线异常检测

（2）实现

源码文件
参数：
- n_estimators：默认=100，配置iTree树的数量
- max_samples：默认=256，配置采样大小
- max_features：默认=全部特征，对高维数据，可以只选取部分特征
参考示例代码

举例：


from sklearn.ensemble import IsolationForest
iForest = IsolationForest(n_estimators=500,random_state=75,behaviour='new')
iForest.fit(source_test[to_columns][:2000],)
 
source_test['is_abnormal'] = iForest.predict(source_test[to_columns])
 
# 记录异常样本在原数据集中的索引
drop_rows = []
for row in source_test.itertuples(index=True, name='Pandas'):
    if getattr(row,'is_abnormal') == -1:
        drop_rows.append(row._asdict())
print(len(drop_rows))
 
# 查看要删除的行的具体信息
drop_rows_dataframe = pd.DataFrame(drop_rows)
drop_rows_dataframe[['signalStrength','signalQuality','is_abnormal','label']]
 
# 查看异常与正常样本的数量
source_test['is_abnormal'].value_counts()

（3）总结　　

2.Local Outlier Factor

Local Outlier Factor（LOF）是基于密度的经典算法（Breuning et. al. 2000）。在 LOF 之前的异常检测算法大多是基于统计方法的，或者是借用了一些聚类算法用于异常点的识别（比如，DBSCAN，OPTICS）。但是，基于统计的异常检测算法通常需要假设数据服从特定的概率分布，这个假设往往是不成立的。而聚类的方法通常只能给出 0/1 的判断（即：是不是异常点），不能量化每个数据点的异常程度。相比较而言，基于密度的LOF算法要更简单、直观。它不需要对数据的分布做太多要求，还能量化每个数据点的异常程度（outlierness）。

具体参考

二、novelty detection

1.One-Class SVM

　　它的训练集不应该掺杂异常点，因为模型可能会去匹配这些异常点。但在数据维度很高，或者对相关数据分布没有任何假设的情况下，OneClassSVM也可以作为一种很好的outlier detection方法。在one-class classification中，仅仅只有一类的信息是可以用于训练，其他类别的（总称outlier）信息是缺失的，也就是区分两个类别的边界线是通过仅有的一类数据的信息学习得到的。

（1）算法

无监督学习
思想：SVDD，期望最小化超球体的体积，从而最小化异常点数据的影响。
适用于小样本、高纬度、非线性问题

　　假设产生的超球体参数为中心$o$和对应的超球体半径$r >0$，超球体体积$V(r)$被最小化；跟传统SVM方法相似，可以要求所有训练数据点$x_i$到中心的距离严格小于$r$。但是同时构造一个惩罚系数为$C$的松弛变量$ζ_i$，优化问题如下所示：

　　采用拉格朗日对偶求解之后，可以判断新的数据点$z$是否在内，如果$z$到中心的距离小于或者等于半径$r$，则不是异常点，如果在超球体以外，则是异常点。在Sklearn中，我们可以采用SVM包里面的OneClassSVM来做异常点检测。OneClassSVM也支持核函数，所以普通SVM里面的调参思路在这里也使用。

SVDD的优化目标：求一个中心为a，半径为R的最小球面

约束条件：

满足这个条件就是说要把training set中的数据点都包在球面里。

具体细节

（2）实现


class sklearn.svm.OneClassSVM(kernel=’rbf’, degree=3, gamma=’auto’,
coef0=0.0, tol=0.001, nu=0.5, shrinking=True, cache_size=200, verbose=False,
 max_iter=-1, random_state=None)

sklearn链接

参数：

kernel：核函数（一般使用高斯核）
nu：设定训练误差(0, 1]，表示异常点比例，默认值为0.5

举例：


import numpy as np
import matplotlib.pyplot as plt
import matplotlib.font_manager
from sklearn import svm
 
xx, yy = np.meshgrid(np.linspace(-5, 5, 500), np.linspace(-5, 5, 500))
# Generate train data
X = 0.3 * np.random.randn(100, 2)
X_train = np.r_[X + 2.1, X - 2.1]
X_test = np.r_[X + 2, X - 2]
# Generate some abnormal novel observations
X_outliers = np.random.uniform(low=0.1, high=4, size=(20, 2))
# fit the model
clf = svm.OneClassSVM(nu=0.1, kernel='rbf', gamma=0.1)
clf.fit(X_train)
y_pred_train = clf.predict(X_train)
y_pred_test = clf.predict(X_test)
y_pred_outliers = clf.predict(X_outliers)
n_error_train = y_pred_train[y_pred_train == -1].size
n_error_test = y_pred_test[y_pred_test == -1].size
n_error_outlier = y_pred_outliers[y_pred_outliers == 1].size
 
# plot the line , the points, and the nearest vectors to the plane
Z = clf.decision_function(np.c_[xx.ravel(), yy.ravel()])
Z = Z.reshape(xx.shape)
 
plt.title("Novelty Detection")
# 填充等高线图
plt.contourf(xx, yy, Z, levels=np.linspace(Z.min(), 0, 7), cmap=plt.cm.PuBu)
# 绘制等高线
a = plt.contour(xx, yy, Z, levels=[0, Z.max()], colors='palevioletred')
plt.clabel(a, inline_spacing=3, fmt='%.2f', fontsize=10)
 
s = 40
b1 = plt.scatter(X_train[:, 0], X_train[:, 1], c='green', s=s, edgecolors='k')
b2 = plt.scatter(X_test[:, 0], X_test[:, 1], c='blueviolet', s=s, edgecolors='k')
c = plt.scatter(X_outliers[:, 0], X_outliers[:, 1], c='gold', s=s, edgecolors='k')
 
plt.axis('tight')
plt.xlim((-5, 5))
plt.ylim((-5, 5))
plt.legend([a.collections[0], b1, b2, c],
           ["learned frontier", 'training observations',
            "new regular observations", "new abnormal observations"],
           loc="upper left",
           prop=matplotlib.font_manager.FontProperties(size=11))
plt.xlabel("error train: %d/200; errors novel regular: %d/40; errors novel abnormal:%d/40" %
           (n_error_train, n_error_test, n_error_outlier))
plt.show()