Source code for pysatl_cpd.algorithms.online.bayesian.algorithm.abstract_bayesian

# -*- coding: ascii -*-
"""Abstract Bayesian base implementation."""

from __future__ import annotations

from abc import ABC, abstractmethod
from copy import deepcopy
from dataclasses import dataclass, field
from typing import Any, Self

import numpy as np
from scipy.special import logsumexp

from pysatl_cpd.algorithms.online.bayesian._enum import BayesianCPFType
from pysatl_cpd.algorithms.online.bayesian.protocol.cpf import IBayesianCPF
from pysatl_cpd.algorithms.online.bayesian.protocol.hazard import IHazard
from pysatl_cpd.algorithms.online.bayesian.protocol.likelihood import ILikelihood
from pysatl_cpd.core.online.ionline_algorithm import (
    OnlineAlgorithm,
    OnlineAlgorithmConfiguration,
    OnlineAlgorithmState,
)
from pysatl_cpd.typedefs import Number, stable_hash

__author__ = "Alexey Tatyanenko"
__copyright__ = "Copyright (c) 2026 PySATL project"
__license__ = "SPDX-License-Identifier: MIT"


[docs] @dataclass(kw_only=True, frozen=True) class BayesianOnlineCPDConfiguration(OnlineAlgorithmConfiguration): """Base configuration for Bayesian online CPD algorithms.""" window: int | None = None cpf_type: BayesianCPFType = BayesianCPFType.MAX_RUN_LENGTH
[docs] def __post_init__(self) -> None: """Validate configuration fields after initialisation. Raises ------ ValueError If *learning_period_size* is negative, or *window* is non-positive when provided. """ if self.learning_period_size < 0: raise ValueError("learning_period_size must be non-negative") if self.window is not None and self.window <= 0: raise ValueError("window must be positive")
def __hash__(self) -> int: return stable_hash( ( type(self).__module__, type(self).__qualname__, self.learning_period_size, self.window, self.cpf_type, ) )
[docs] @dataclass(kw_only=True, frozen=True) class BayesianOnlineCPDState(OnlineAlgorithmState): """Base state for Bayesian online CPD algorithms. Attributes ---------- t Current time step. run_length_log_posterior Log-posterior distribution over run lengths. """ t: int = 0 run_length_log_posterior: Any = field(default_factory=list) def __hash__(self) -> int: posterior = tuple(np.asarray(self.run_length_log_posterior, dtype=np.float64).tolist()) return stable_hash( (type(self).__module__, type(self).__qualname__, self.is_in_learning_period, self.t, posterior) )
[docs] class AbstractBayesian(OnlineAlgorithm[float, BayesianOnlineCPDConfiguration, BayesianOnlineCPDState], ABC): """Base class for configurable online Bayesian detectors. Implements the Bayesian online change-point detection (BOCPD) message-passing loop. Subclasses must provide the *name* property and wire up the concrete *hazard*, *likelihood*, and *cpf* components. Parameters ---------- configuration Algorithm configuration. hazard Hazard function defining the prior over run lengths. likelihood Predictive likelihood model. cpf Change-point score function. """
[docs] def __init__( self, configuration: BayesianOnlineCPDConfiguration, hazard: IHazard, likelihood: ILikelihood, cpf: IBayesianCPF, ) -> None: self._config = configuration self._hazard = hazard self._likelihood = likelihood self._cpf = cpf self._run_length_log_posterior = np.array([0.0], dtype=np.float64) self.t = 0
@property @abstractmethod def name(self) -> str: """Human-readable algorithm name.""" @property def configuration(self) -> BayesianOnlineCPDConfiguration: """Return the current algorithm configuration. Returns ------- BayesianOnlineCPDConfiguration """ return self._config @property def state(self) -> BayesianOnlineCPDState: """Materialise an immutable snapshot of the current state. Returns ------- BayesianOnlineCPDState """ return BayesianOnlineCPDState( is_in_learning_period=self.t < self._config.learning_period_size, t=self.t, run_length_log_posterior=self._run_length_log_posterior.copy(), )
[docs] def process(self, observation: float) -> Number: """Ingest one observation and return the change-point score. Implements the BOCPD message-passing update: 1. Predictive probabilities from the likelihood model. 2. Hazard-based growth / change-point probabilities. 3. Normalise and optionally truncate to *window*. 4. Return CPF score (or 0 during learning period). Parameters ---------- observation New data point. Returns ------- Number Change-point score (0.0 during learning period). """ obs = np.float64(observation) pred_log_probs = self._likelihood.predict(obs, window=self._config.window) self._likelihood.update(obs) run_lengths = np.arange(len(self._run_length_log_posterior), dtype=np.intp) log_h, log_neg_h = self._hazard.hazard(run_lengths) log_growth_probs = self._run_length_log_posterior + log_neg_h + pred_log_probs[1:] log_cp_prob = pred_log_probs[0] + logsumexp(self._run_length_log_posterior + log_h) new_log_posterior_unnorm = np.append(log_cp_prob, log_growth_probs) if self._config.window is not None and len(new_log_posterior_unnorm) > self._config.window: new_log_posterior_unnorm = new_log_posterior_unnorm[-self._config.window :] self._run_length_log_posterior = new_log_posterior_unnorm - logsumexp(new_log_posterior_unnorm) self.t += 1 if self.t <= self._config.learning_period_size: return np.float64(0.0) return np.float64(self._cpf.calculate(self._run_length_log_posterior))
[docs] def reset(self) -> None: """Reset the algorithm to its initial state. Clears step counter, run-length posterior, likelihood, and CPF. Returns ------- None """ self.t = 0 self._run_length_log_posterior = np.array([0.0], dtype=np.float64) self._likelihood.clear() self._cpf.clear()
[docs] def recreate(self) -> Self: """Create a fresh independent copy with identical configuration. Returns ------- Self """ algorithm = deepcopy(self) algorithm.reset() return algorithm