python 日常笔记 hmmlearn 隐性马尔科夫模型案例分析

Posted 2023-05-04

参考技术A

什么是马尔科夫模型?用来干什么?
大家可以参考这篇

<隐性马尔科夫模型简介,只聊原理, (保证没有数学)>
https://www.jianshu.com/p/3b4fbcea2744

关于HMM有两个主要问题：
已知上述三个参数，和当前观测序列，求解隐藏状态的变化
所有参数未知，只有数据，如何获得三个参数
需要使用hmmlearn 包

隐藏状态:

观察情况

初始状态矩阵

状态转移矩阵

发射矩阵

构建模型

将上述设定的参数传入其中

输入我们观察到的序列

输出的结果是
观察到的取出的色子点数: 1, 2, 3, 2, 4, 4, 5, 1, 6, 4
decode 方法计算： A, B, A, B, A, B, A, B, A, B
predict方法计算： A, B, A, B, A, B, A, B, A, B

取出的色子点数: 3, 2, 2, 1, 3, 4, 1, 2, 2, 3, 3, 2, 1, 1, 4, 4
从哪个盒子取的: A, B, A, B, A, B, A, B, A, B, A, B, A, B, A, B

取出的色子点数: 6, 5, 4, 3, 3, 4, 1, 5, 6, 5, 2, 2, 1, 1, 4, 4
从哪个盒子取的: A, A, B, A, B, A, B, A, A, A, B, A, B, A, B, A

方案：　我们先假定一个初始概率分布，转移矩阵，以及发射矩阵，按照这些参数生成很多很多数据.
然后将数据放到HMM模型里，利用数据学习，检查模型学到的参数和我们假定的参数是不是一致。
按照假象的规则去生成一些列数据

将数据填入模型中,进行学习

输出的结果是
开始学习 2020-01-05 10:03:02.832792 共 10000条数据
结束学习 2020-01-05 13:05:53.612298
因为是无监督学习,所以模型不会把 A B 排定先后顺序,但是 三个参数是相互关联的,所以顺序其实无关系
初始概率
[0.20509604 0.79490396]

状态转移矩阵
[[0.31460223 0.68539777]
[0.6213235 0.3786765 ]]

从隐藏状态到 显示状态的发散矩阵
[[1.67834277e-01 1.74886284e-01 1.69078215e-01 1.68723388e-01
1.61611529e-01 1.57866306e-01]
[2.51185996e-01 2.46793569e-01 2.46239587e-01 2.53539909e-01
1.54840968e-06 2.23939182e-03]]

可见学习的还是很好的, 只是时间有点长(3个小时),
但是结果非常符合预期, 主要原因是,我们的数据非常干净,没有噪音. 如果在数据中混杂这噪音,可能结果就不会这么好了

https://hmmlearn.readthedocs.io/en/latest/api.html

Python中的输入输出隐马尔可夫模型实现

【中文标题】Python中的输入输出隐马尔可夫模型实现

【英文标题】：Input Output Hidden Markov Model Implementation in Python

【发布时间】：2019-10-31 15:14:30

【问题描述】：

我正在尝试使用输入输出架构实现隐马尔可夫模型，但我找不到任何好的 Python 实现。

任何人都可以分享 Python 包并考虑 HMM 的以下实现。

允许连续排放。 允许协变量的功能（即 I/O HMM 中的自变量）。

此时，我正在努力寻找相同的 python 实现。

我在 hmmlearn 中找不到相关示例。

以下是我测试过的几个库：

hmmlearn：hmmlearn 允许将多个特征传递给发射/观察，但不提供包含协变量（即自变量）的支持。

hmms：不支持添加连续发射的功能，也不支持添加自变量。

IOHMM：我能够使用此库训练 HMM 模型，但在训练模型后找不到用于进行预测的文档。

因此，我正在寻找符合目的的包。

from IOHMM import UnSupervisedIOHMM
from IOHMM import OLS, DiscreteMNL, CrossEntropyMNL, forward_backward

SHMM = UnSupervisedIOHMM(num_states=3, max_EM_iter=200, EM_tol=1e-6)

SHMM.set_models(model_emissions = [OLS(est_stderr=True)], 
                model_transition=CrossEntropyMNL(solver='lbfgs'),
                model_initial=CrossEntropyMNL(solver='lbfgs'))

SHMM.set_inputs(covariates_initial = [], covariates_transition = [], covariates_emissions = [['Insulin']])


SHMM.set_outputs([['Glucose']])

SHMM.set_data([data])

SHMM.train() 

经过上述训练，我无法弄清楚如何获得发射概率和隐藏状态序列。

【问题讨论】：

如果你能解释你是如何得到输出的，那将会很有帮助。底层逻辑是什么？

输出只是排放/观察。

hmmlearn 的哪一部分你不明白？这里没有人会为您编写示例，因为 a）我们不为人们编写代码 b）您甚至没有给我们任何迹象表明您自己尝试过

嗨，克里斯，感谢您的意见。我已经编辑了这个问题，以便更好地理解这个疑问。

根据github.com/Mogeng/IOHMM/blob/master/examples/notebooks/…，您只需要SHMM.model_emissions 即可获得排放量

【参考方案1】：

参考“https://web.ece.ucsb.edu/Faculty/Rabiner/ece259/Reprints/tutorial%20on%20hmm%20and%20applications.pdf”和库“https://hmmlearn.readthedocs.io/en/latest/”我找到了这个解决方案：

1-通过log_gamma（后验分布）：

state_sequences = []
for i in range(100):
    for j in range(lengths[i]):
        state_sequences.append(np.argmax(np.exp(SHMM.log_gammas[i])[j]))
pred_state_seq = [state_sequences[df[df['unit'] == i].index[0]:df[df['unit'] == i].index[-1] + 1] for i in
                  range(1, df_A['unit'].max() + 1)]

2- 维特比算法：

from hmmlearn import _hmmc

transmat = np.empty((num_states, num_states))
for i in range(num_states):
    transmat = np.concatenate((transmat, np.exp(SHMM.model_transition[i].predict_log_proba(np.array([[]])))))
transmat = transmat[num_states:]

startprob = np.exp(SHMM.model_initial.predict_log_proba(np.array([[]]))).squeeze()


def log_mask_zero(a):
    """
    Compute the log of input probabilities masking divide by zero in log.

    Notes
    -----
    During the M-step of EM-algorithm, very small intermediate start
    or transition probabilities could be normalized to zero, causing a
    *RuntimeWarning: divide by zero encountered in log*.

    This function masks this unharmful warning.
    """
    a = np.asarray(a)
    with np.errstate(divide="ignore"):
        return np.log(a)


def _do_viterbi_pass(framelogprob):
    n_samples, n_components = framelogprob.shape
    state_sequence, logprob = _hmmc._viterbi(n_samples, n_components, log_mask_zero(startprob),
                                             log_mask_zero(transmat), framelogprob)
    return logprob, state_sequence


def _decode_viterbi(X):
    framelogprob = SHMM.log_Eys[X]
    return _do_viterbi_pass(framelogprob)


def decode():
    decoder = "viterbi": _decode_viterbi["viterbi"]
    logprob = 0
    sub_state_sequences = []
    for sub_X in range(100):
        # XXX decoder works on a single sample at a time!
        sub_logprob, sub_state_sequence = decoder(sub_X)
        logprob += sub_logprob
        sub_state_sequences.append(sub_state_sequence)
    return logprob, np.concatenate(sub_state_sequences)


def predict():
    """
    Find most likely state sequence corresponding to ``X``.

    Parameters
    ----------
    X : array-like, shape (n_samples, n_features)
        Feature matrix of individual samples.
    lengths : array-like of integers, shape (n_sequences, ), optional
        Lengths of the individual sequences in ``X``. The sum of
        these should be ``n_samples``.

    Returns
    -------
    state_sequence : array, shape (n_samples, )
        Labels for each sample from ``X``.
    """
    logprob, state_sequence = decode()
    return logprob, state_sequence


_, state_seq = predict()

pred_state_seq = [state_seq[df[df['unit'] == i].index[0]:df[df['unit'] == i].index[-1] + 1] for i in
                  range(1, df_A['unit'].max() + 1)]