Discrete Hidden Markov Models

cuthbert provides exact filtering and smoothing for state-space models with discrete states (aka hidden Markov models):

In this case, we assume the latent states \(x_t\) are discrete, i.e. \(x_t \in \{1, 2, \ldots, K\}\) and therefore distributions over a state can be stored as an array of normalized probabilities.

Observations are handled via a general log likelihood functions that can also be stored as an array of log likelihoods \(b_i = \log p(y_t \mid x_t = i)\) of size \(K\).

The core atomic functions can be found in cuthbertlib.discrete.

cuthbert.discrete.filter

Parallel-in-time Bayesian filter for discrete hidden Markov models.

References

• https://ieeexplore.ieee.org/document/9512397
• https://github.com/EEA-sensors/sequential-parallelization-examples/tree/main/python/temporal-parallelization-inference-in-HMMs
• https://github.com/probml/dynamax/blob/main/dynamax/hidden_markov_model/parallel_inference.py

build_filter(get_init_dist, get_trans_matrix, get_obs_lls)

Builds a filter object for discrete hidden Markov models.

Parameters:

Name	Type	Description	Default
get_init_dist	GetInitDist	Function to get initial state probabilities \(m_i = p(x_0 = i)\).	required
get_trans_matrix	GetTransitionMatrix	Function to get the transition matrix \(A_{ij} = p(x_t = j \mid x_{t-1} = i)\).	required
get_obs_lls	GetObsLogLikelihoods	Function to get observation log likelihoods \(b_i = \log p(y_t | x_t = i)\).	required

Returns:

Type	Description
Filter	Filter object. Suitable for associative scan.

Source code in cuthbert/discrete/filter.py

def build_filter(
    get_init_dist: GetInitDist,
    get_trans_matrix: GetTransitionMatrix,
    get_obs_lls: GetObsLogLikelihoods,
) -> Filter:
    r"""Builds a filter object for discrete hidden Markov models.

    Args:
        get_init_dist: Function to get initial state probabilities $m_i = p(x_0 = i)$.
        get_trans_matrix: Function to get the transition matrix $A_{ij} = p(x_t = j \mid x_{t-1} = i)$.
        get_obs_lls: Function to get observation log likelihoods $b_i = \log p(y_t | x_t = i)$.

    Returns:
        Filter object. Suitable for associative scan.
    """
    return Filter(
        init_prepare=partial(
            init_prepare, get_init_dist=get_init_dist, get_obs_lls=get_obs_lls
        ),
        filter_prepare=partial(
            filter_prepare, get_trans_matrix=get_trans_matrix, get_obs_lls=get_obs_lls
        ),
        filter_combine=filter_combine,
        associative=True,
    )

cuthbert.discrete.smoother

Parallel-in-time Bayesian smoother for discrete hidden Markov models.

References

• https://ieeexplore.ieee.org/document/9512397
• https://github.com/EEA-sensors/sequential-parallelization-examples/tree/main/python/temporal-parallelization-inference-in-HMMs

build_smoother(get_trans_matrix)

Builds a smoother object for discrete hidden Markov models.

Parameters:

Name	Type	Description	Default
get_trans_matrix	GetTransitionMatrix	Function to get the transition matrix \(A_{ij} = p(x_t = j \mid x_{t-1} = i)\).	required

Returns:

Type	Description
Smoother	Smoother object. Suitable for associative scan.

Source code in cuthbert/discrete/smoother.py

def build_smoother(get_trans_matrix: GetTransitionMatrix) -> Smoother:
    r"""Builds a smoother object for discrete hidden Markov models.

    Args:
        get_trans_matrix: Function to get the transition matrix $A_{ij} = p(x_t = j \mid x_{t-1} = i)$.

    Returns:
        Smoother object. Suitable for associative scan.
    """
    return Smoother(
        convert_filter_to_smoother_state=convert_filter_to_smoother_state,
        smoother_prepare=partial(smoother_prepare, get_trans_matrix=get_trans_matrix),
        smoother_combine=smoother_combine,
        associative=True,
    )

cuthbert.discrete.types

Provides types for representing discrete HMMs.

GetInitDist

Bases: Protocol

Protocol for specifying the initial distribution.

__call__(model_inputs)

Get the initial distribution.

Parameters:

Name	Type	Description	Default
model_inputs	ArrayTreeLike	Model inputs.	required

Returns:

Type	Description
Array	An array \(m\) of shape (N,) where N is the number of states, with \(m_i = p(x_0 = i)\).

Source code in cuthbert/discrete/types.py

def __call__(self, model_inputs: ArrayTreeLike) -> Array:
    """Get the initial distribution.

    Args:
        model_inputs: Model inputs.

    Returns:
        An array $m$ of shape (N,) where N is the number of states,
            with $m_i = p(x_0 = i)$.
    """
    ...

GetTransitionMatrix

Bases: Protocol

Protocol for specifying the transition matrix.

__call__(model_inputs)

Get the transition matrix.

Parameters:

Name	Type	Description	Default
model_inputs	ArrayTreeLike	Model inputs.	required

Returns:

Type	Description
Array	An array \(A\) of shape (N, N) where N is the number of states, with \(A_{ij} = p(x_t = j \mid x_{t-1} = i)\).

Source code in cuthbert/discrete/types.py

def __call__(self, model_inputs: ArrayTreeLike) -> Array:
    r"""Get the transition matrix.

    Args:
        model_inputs: Model inputs.

    Returns:
        An array $A$ of shape (N, N) where N is the number of
            states, with $A_{ij} = p(x_t = j \mid x_{t-1} = i)$.
    """
    ...

GetObsLogLikelihoods

Bases: Protocol

Protocol for specifying the observation log likelihoods.

__call__(model_inputs)

Get the observation log likelihoods.

Parameters:

Name	Type	Description	Default
model_inputs	ArrayTreeLike	Model inputs.	required

Returns:

Type	Description
Array	An array \(b\) of shape (N,) where N is the number of states, with \(b_i = \log p(y_t \mid x_t = i)\).

Source code in cuthbert/discrete/types.py

def __call__(self, model_inputs: ArrayTreeLike) -> Array:
    r"""Get the observation log likelihoods.

    Args:
        model_inputs: Model inputs.

    Returns:
        An array $b$ of shape (N,) where N is the number of states,
            with $b_i = \log p(y_t \mid x_t = i)$.
    """
    ...