孤立森林【python，机器学习，算法】

作用与特征

孤立森林主要用于样本的异常点检测，异常点检测又被称为离群点检测(outlier detection)，那么什么样的数据才能算作异常数据呢，一般情况异常点具有以下两个特性：

• 异常数据跟样本中大多数数据不太一样。
• 异常数据在整体数据样本中占比比较小。

直观理解

先简单解释一下什么是孤立森林： 「假设我们用一个随机超平面来切割（split）数据空间（data space）, 切一次可以生成两个子空间（想象拿刀切蛋糕一分为二）。

之后我们再继续用一个随机超平面来切割每个子空间，循环下去，直到每子空间里面只有一个数据点为止。

直观上来讲，我们可以发现那些密度很高的簇是可以被切很多次才会停止切割，但是那些密度很低的点很容易很早的就停到一个子空间里了」。

哪些很容易被切分出去的点就会被定义为异常点。

孤立森林构建流程

1. 构建森林
   那么和随机森林一样，孤立森林由 iTree（isolation tree）组成，iTree树和随机森林的决策树不太一样，构建过程只是一个完全随机的过程。构建步骤如下：
   • 随机选择一个特征tree_feature作为建树的节点。
   • 如果样本只剩一个或者树的路径深度已经超过最大深度，那么可以将当前节点作为叶子节点直接返回。
     • 叶子节点返回值为【0，1】，其中 0 表示叶子节点，1 表示叶子节点的路径长度为 1。
   • 从所选样本中，找出tree_feature
     的最大值和最小值，然后在最大值和最小值之间随机选择一个值作为分割点split_val。
   • 构建树的左右节点。
     • 样本中小于split_val的划分到左边节点。
     • 样本中大于等于split_val的划分到右边节点。
   • 返回当前节点信息：【1,left_child,right_child,tree_feature,split_val】。
2. 使用森林进行评估。
   使用训练好的孤立森林进行数据评估，检测异常数据。具体步骤如下：
   • 遍历每一个样本数据。
   • 计算样本数据的异常分数。计算公式如下：
     • 样本的异常分数$s(i)=2^{-\frac{E(h(i))}{c(n)}}$。
     • 其中$E(h(i))$表示样本i的期望路径长度，计算方法如下：
       1. 将样本i带入每课树中，计算其路径长度。
       2. 将计算得到的所有长度相加再除以树的棵树，就得到了样本的期望。
     • 其中二叉搜索树的平均路径长度$c(n)=2H(n-1)-\frac{2(n-1)}{n}$，用来对结果进行归一化处理。这里的n表示树的数量。
     • 而调和数$H(n-1)=\ln n-1-\zeta$，欧拉常数$\zeta \approx 0.5772156649$。
   • 根据异常分数判断样本是否为异常点。异常分数的取值范围为0-1,分数越接近 1，表示该点越有可能是异常孤立的点。

代码实现

import numpy as np
import torch
from matplotlib import pyplot as plt


def iTree(X: torch.Tensor, current_path_len, max_tree_height):
	"""孤立森林中的树

	:param X: 数据集
	:param current_path_len: 当前路径长
	:param max_tree_height: 树高最大值
	:return: 决策树信息
	"""
	# 当前路径长度大于等于树的最大高度或者样本数量小于等于 1，返回叶子节点信息：0 表示叶子节点，以及样本的数量
	if current_path_len >= max_tree_height or len(X) <= 1:
		return [0, len(X)]
	# 随机选取一个样本特征
	random_select_feature = np.random.randint(0, len(X[0]))
	# 找到特征下的最大值和最小值
	feature_max_val = X[:, random_select_feature].max()
	feature_min_val = X[:, random_select_feature].min()
	# 在最大值和最小值之间随机选一个值作为分割点
	separate_val = (np.random.rand() * (feature_max_val - feature_min_val)
	                + feature_min_val)
	lchild = iTree(X[X[:, random_select_feature] < separate_val, :],
	               current_path_len + 1, max_tree_height)
	rchild = iTree(X[X[:, random_select_feature] >= separate_val, :],
	               current_path_len + 1, max_tree_height)
	# 返回当前节点信息
	return [1, lchild, rchild, random_select_feature, separate_val]


def c(n):
	"""计算二叉搜索树的平均路径长度，用来对结果进行归一化处理

	公式：c(n)= 2H( n − 1 ) − 2 ( n − 1 )/n
		 H(i) 表示调和数，近似值为：ln(i)+ ζ，
		 其中 ζ 表示欧拉常数，约等于 0.5772156649，
		 n 表示样本的数量。
		 平均路径长度的期望是一个常数，该公式提供了一个标准化的基准，用于将路径长度标准化。
	:param n: 表示单棵树中的样本数量
	:return: 平均路径长度
	"""
	return 0 if n == 1 else 2 * (np.log(n - 1) + 0.5772156649) - (
			2 * (n - 1) / n)


def PathLength(x, iTree, current_path_len):
	"""计算样本在树中的路径长度

	:param x: 样本
	:param iTree: 孤立树
	:param current_path_len: 当前长度
	:return: 叶子节点的路径长度。
	"""
	# 到达叶子节点或者达到最大路长度时，结束计算。
	if iTree[0] == 0:
		return current_path_len + c(iTree[1])
	# 样本中的特征值小于分叉点的值时，搜索左子树
	if x[iTree[3]] < iTree[4]:
		return PathLength(x, iTree[1], current_path_len + 1)
	# 搜索右子树
	return PathLength(x, iTree[2], current_path_len + 1)


def myIForest(X, n_trees, tree_size):
	"""孤立森林，即构建多颗树

	:param X:样本集
	:param n_trees: 树的数量
	:param tree_size: 每棵树有多少个样本，即采样大小
	:return: 树的集合
	"""
	Ts = []
	# 树高的最大值
	max_tree_height = np.ceil(np.log(tree_size))
	for i in range(n_trees):
		x_i = np.random.choice(range(len(X)), [tree_size], replace=False)
		Ts.append(iTree(X[x_i], 0, max_tree_height))
	return Ts


def anomalyScore(x, Ts, tree_size):
	"""计算样本的样本的异常分数，异常分数的取值为 0-1,值越大越可能是异常点

	:param x: 样本
	:param Ts: 树的集合
	:param tree_size: 树中包含的样本数量
	:return: 异常分数值
	"""
	# 样本在所有树中的路径长度期望
	E_x_len = 0
	for T in Ts:
		E_x_len += PathLength(x, T, 0)
	E_x_len /= len(Ts)
	s = 2 ** (-E_x_len / c(tree_size))
	return s


# %% 定义正常分布、超参数、绘图矩阵
torch.manual_seed(0)
np.random.seed(0)
points = torch.randn([512, 2])
# 将 80 个样本形成一个簇
points[-80:] = torch.randn([80, 2]) / 3 + 4
# 定义树的数量和树的大小
n_tree, tree_size = 100, 256

x, y = np.arange(-4.5, 5.5, 0.1), np.arange(-4.5, 5.5, 0.1)
# 网格化
X, Y = np.meshgrid(x, y)

XY = np.stack([X, Y], -1)
# 生成和 X 一样大小的 0 矩阵 Z
Z = np.zeros_like(X)
# %% 自定义孤立森林、异常值可视化、决策边界
# 训练树
myTs = myIForest(points, n_tree, tree_size)
# 评分
for i in range(XY.shape[0]):
	for j in range(XY.shape[1]):
		Z[i, j] = anomalyScore(XY[i, j], myTs, tree_size)
plt.plot(points[:, 0], points[:, 1], '.', c="purple", alpha=0.3)
# 绘制登高线
plt.contourf(X, Y, Z)
cont = plt.contour(X, Y, Z, levels=[0.55])
plt.clabel(cont, inline=True, fontsize=10)
plt.show()
# %% pyOD孤立森林、异常值可视化、决策边界
from pyod.models.iforest import IForest

ifor = IForest(n_tree, tree_size, 0.1, random_state=0)
ifor.fit(points)
h, w = XY.shape[0], XY.shape[1]
XY = XY.reshape(-1, 2)
Z = Z.reshape(-1)
Z = ifor.decision_function(XY)
Z = Z.reshape(h, w)
XY = XY.reshape(h, w, 2)

plt.plot(points[:, 0], points[:, 1], '.', c="purple", alpha=0.3)
plt.contourf(X, Y, Z)
cont = plt.contour(X, Y, Z, levels=[0])  # 决策边界为0
plt.clabel(cont, inline=True, fontsize=10)
plt.show()

这个示例实现了孤立森林算法，并将实现的算法与第三方库实现的算法进行可视化的比较展示，从结果可以看出，该手撕代码实现与生产结果差异并不大。