Python中如何实现基于HMM的异常检测？隐马尔可夫模型

hmmlearn

hmmlearn

score_samples

import numpy as np
from hmmlearn import hmm
import matplotlib.pyplot as plt
from scipy.stats import norm

# 1. 生成一些“正常”的序列数据
# 假设我们有两个隐藏状态：状态0（低值波动）和状态1（高值波动）
# 每个状态下的观测数据服从不同的高斯分布
np.random.seed(42)

# 状态转移概率矩阵
# 从状态0到状态0的概率是0.9，到状态1是0.1
# 从状态1到状态1的概率是0.9，到状态0是0.1
transmat = np.array([[0.9, 0.1],
                     [0.1, 0.9]])

# 观测均值和协方差（对于GaussianHMM）
# 状态0的观测均值是0，协方差是1
# 状态1的观测均值是5，协方差是1
means = np.array([[0.], [5.]])
covars = np.array([[1.], [1.]])

# 初始状态概率
startprob = np.array([0.5, 0.5])

# 创建一个高斯HMM模型
model = hmm.GaussianHMM(n_components=2, covariance_type="full", n_iter=100)
model.startprob_ = startprob
model.transmat_ = transmat
model.means_ = means
model.covars_ = covars

# 生成一些正常序列数据
# 模拟100个序列，每个序列长度为50
normal_data = []
lengths = []
for _ in range(100):
    X, Z = model.sample(n_samples=50)
    normal_data.append(X)
    lengths.append(50)

normal_data_flat = np.concatenate(normal_data)

# 2. 训练HMM模型（使用真实数据时，通常只用正常数据训练）
# 这里我们假设我们不知道真实的参数，需要从数据中学习
# 重新初始化一个模型进行训练
trained_model = hmm.GaussianHMM(n_components=2, covariance_type="full", n_iter=100, random_state=0)
trained_model.fit(normal_data_flat, lengths)

print("训练后的模型参数:")
print("初始状态概率:", trained_model.startprob_)
print("状态转移概率:", trained_model.transmat_)
print("观测均值:", trained_model.means_)
print("观测协方差:", trained_model.covars_)

# 3. 异常检测
# 计算正常数据在训练模型下的对数似然
normal_scores = [trained_model.score(seq) for seq in normal_data]

# 生成一些“异常”的序列数据
# 异常数据可能表现为：
# 1. 持续处于某个不常出现的状态
# 2. 观测值与正常状态的分布显著不符
# 3. 状态转换模式异常
abnormal_data = []
# 异常类型1：观测值异常高
abnormal_data.append(np.random.normal(loc=10, scale=1, size=(50, 1)))
# 异常类型2：观测值异常低
abnormal_data.append(np.random.normal(loc=-5, scale=1, size=(50, 1)))
# 异常类型3：混合了正常状态，但有突然的偏离
abnormal_seq_mixed = np.concatenate([np.random.normal(loc=0, scale=1, size=(25, 1)),
                                     np.random.normal(loc=10, scale=1, size=(25, 1))])
abnormal_data.append(abnormal_seq_mixed)

abnormal_scores = [trained_model.score(seq) for seq in abnormal_data]

print("\n正常序列的对数似然:", np.mean(normal_scores), "±", np.std(normal_scores))
print("异常序列的对数似然:", abnormal_scores)

# 设定阈值（例如，低于正常均值减去3倍标准差）
threshold = np.mean(normal_scores) - 3 * np.std(normal_scores)
print(f"\n设定的异常检测阈值: {threshold:.2f}")

# 可视化分数分布
plt.figure(figsize=(10, 6))
plt.hist(normal_scores, bins=20, alpha=0.7, label='正常序列分数', color='skyblue')
for i, score in enumerate(abnormal_scores):
    plt.axvline(x=score, color='red', linestyle='--', label=f'异常序列{i+1}分数' if i == 0 else "")
plt.axvline(x=threshold, color='green', linestyle='-', linewidth=2, label='异常阈值')
plt.title('HMM对数似然分数分布')
plt.xlabel('对数似然分数')
plt.ylabel('序列数量')
plt.legend()
plt.grid(axis='y', alpha=0.75)
plt.show()

# 判断新序列是否异常
new_unseen_sequence = np.random.normal(loc=8, scale=1, size=(50, 1)) # 一个可能是异常的序列
new_score = trained_model.score(new_unseen_sequence)
print(f"\n新序列的分数: {new_score:.2f}")
if new_score < threshold:
    print("该新序列被检测为异常。")
else:
    print("该新序列被检测为正常。")

DiscreteHMM

GaussianHMM

GMMHMM

GaussianHMM

GMMHMM

GaussianHMM

GMMHMM