Linear Gaussian predictive model¶

sc_neurocore.world_model.predictive_model exposes filtering, smoothing, learning, and planning-compatible forecasts for a controlled linear Gaussian state-space model (LGSSM).

Model¶

For latent state (x_t \in \mathbb{R}^d), observation (y_t \in \mathbb{R}^p), and control (u_t \in \mathbb{R}^m):

[ x_{t+1} = A x_t + B u_t + w_t, \qquad w_t \sim \mathcal{N}(0, Q), ]

[ y_t = C x_t + D u_t + v_t, \qquad v_t \sim \mathcal{N}(0, R), ]

with prior (x_0 \sim \mathcal{N}(\mu_0, \Sigma_0)).

LinearGaussianSSM copies every parameter into a finite, C-contiguous float64 array. It validates all dimensions, requires Q to be symmetric positive semidefinite, and requires R and Sigma_0 to be symmetric positive definite. A positive diagonal alone is not accepted as a covariance proof.

Public API and compatibility¶

Python

from sc_neurocore.world_model.predictive_model import (
    EMLearner,
    FilterResult,
    KalmanFilter,
    LinearGaussianSSM,
    PredictiveWorldModel,
    RTSSmoother,
    SmoothResult,
)

The import path, constructor parameter names, public class identities, and pickle references remain sc_neurocore.world_model.predictive_model. The implementation is partitioned by responsibility behind that facade:

Responsibility	Owner
parameter and result contracts	`_lgssm_types.py`
native discovery, selection, and FFI marshalling	`_lgssm_backends.py`
Python and native forward-filter dispatch	`_lgssm_filter.py`
RTS backward recursion	`_lgssm_smoothing.py`
controlled EM learning	`_lgssm_em.py`
planning-compatible state forecasts	`_predictive_world_model.py`

The private import graph is one-way and acyclic. The facade contains no algorithm implementation.

Forward filtering¶

KalmanFilter.filter(observations, controls=None, backend="auto") returns:

filtered state means and covariances;
one-step predicted means and covariances before each observation; and
the sequence log-likelihood.

Observations must have shape (T, p) with T > 0. Controls are required for a model with m > 0 and must have shape (T, m). Every input and returned moment must be finite. Result covariance stacks are checked in one vectorised symmetry/eigenvalue pass before being exposed to the caller.

The Python update factors each innovation covariance once with Cholesky. It uses triangular solves for the innovation quadratic form and Kalman gain; it does not form a matrix inverse. The covariance update uses Joseph form:

[ P_{t|t} = (I-K_t C)P_{t|t-1}(I-K_t C)^T + K_t R K_t^T. ]

A non-positive-definite innovation covariance fails closed with numpy.linalg.LinAlgError.

Native forward-filter chain¶

The maintained forward filter has five execution paths:

Backend	Boundary	Numerical source
Python	NumPy	`_lgssm_filter.py`
Mojo	C ABI via `ctypes`	`accel/mojo/world_model/lgssm.mojo`
Go	C shared library via `ctypes`	`accel/go/lgssm/lgssm.go`
Rust	PyO3	`engine/src/lgssm.rs`
Julia	`juliacall`	`accel/julia/world_model/predictive_model.jl`

An explicit unavailable backend raises RuntimeError; it never silently changes language. backend="auto" follows the stable availability- and initialisation-aware order:

Text Only

Mojo -> Go -> Rust -> Julia -> Python

The source-bound controlled workload described below rejects any unexplained ordering with a material warm-timing inversion greater than 10%. Rust is the one declared exception: it precedes Julia because it is loaded during package import, while probing Julia may initialise a separate runtime. Adjacent timings inside the loaded-host noise band retain their stable order. The artifact records both post-import probe cost and the exact warm median ranking. Python is always the final maintained fallback. All native results pass through the same FilterResult validation boundary as Python results.

The Mojo kernel solves each row of the (d \times p) gain workspace independently; it never indexes that workspace as though it were (p \times p). A dedicated (d = 1, p = 3) test exercises the observation-wider-than-state case through all five backends and verifies complete moment and likelihood parity.

RTS smoothing¶

RTSSmoother.smooth(filter_result) performs the Rauch-Tung-Striebel backward recursion with positive-definite solves. It returns full-sequence posterior moments and lag-one covariance blocks oriented as:

[ \operatorname{Cov}[x_t, x_{t+1} \mid y_{0:T-1}]. ]

That orientation is part of the public result contract. The EM transition update explicitly transposes each lag block when it needs (E[x_{t+1}x_t^T]). A multivariate exact batch-conditioning test verifies the means, covariance blocks, and lag orientation.

RTS smoothing is currently a Python/NumPy responsibility. The native chain accelerates only the forward filter.

Controlled expectation-maximisation¶

EMLearner.fit updates A, C, Q, R, mu_0, and Sigma_0. B and D are treated as known parameters and are preserved exactly.

Controls still participate in the M-step. The transition statistics subtract B @ u_t, and the observation statistics subtract D @ u_t, before solving for A and C. Omitting those terms biases a controlled fit even when B and D themselves are fixed.

All normal-equation right solves use Cholesky factors rather than explicit inverses. Learned covariance matrices are symmetrised and projected only as needed to restore the documented positive-semidefinite or positive-definite contract. Every M-step candidate, including the candidate from the final allowed iteration, is filtered and checked before it becomes the returned model. A material likelihood decrease raises RuntimeError; convergence uses the configured absolute tolerance. log_likelihood_history starts with the initial model and records each evaluated candidate.

The default EM forward-filter backend is Python, making the complete learning workload deterministic and explicit. A caller may select another maintained forward backend for the E-step, but the smoother and M-step remain Python.

Latent-state bases are not uniquely identifiable: an invertible change of basis can alter A and C while preserving the observation distribution. Accordingly, recovery tests judge held-out likelihood and posterior moments, not raw parameter equality alone.

Planning-compatible forecasts¶

PredictiveWorldModel preserves the historical planning surface:

Python

import numpy as np

from sc_neurocore.world_model import PredictiveWorldModel

world_model = PredictiveWorldModel(state_dim=4, action_dim=2, seed=42)
mean, covariance = world_model.predict_next_state_with_cov(
    current_state=np.zeros(4),
    current_cov=np.eye(4),
    action=np.array([0.25, -0.10]),
)

predict_next_state returns (A x_t + B u_t). predict_next_state_with_cov additionally returns (A P_t A^T + Q). forecast and forecast_with_cov return independent arrays for every step, so mutating one returned state cannot alter another.

Benchmark evidence¶

benchmarks/bench_predictive_model.py runs one controlled workload through all five forward backends and fails if array or likelihood parity exceeds the declared tolerances. It warms every backend before timing, then rotates the starting backend across interleaved sampling rounds so changing host load does not systematically favour a later block. The same run separately measures the Python-only RTS and EM workloads; unsupported native RTS/EM rows are not represented as skips.

The committed artifact is benchmarks/results/bench_predictive_model.json. It records:

every timing sample, median, minimum, maximum, post-import probe cost, exact warm measured rank, stable dispatch policy, and interleaving policy;
parity deltas against the Python result;
source SHA-256 hashes for the facade, responsibility modules, native implementations, bridge, engine registration, and benchmark harness;
binary hashes for the selected Rust, Go, and Mojo runtimes;
language versions, CPU model, affinity, frequency governor, and load averages;
the exact invocation and explicit heavy-job/isolation disclosure.

Reproduce the local evidence with:

Bash

taskset -c 2 env PYTHONPATH=src .venv/bin/python \
  benchmarks/bench_predictive_model.py \
  --backends python rust julia go mojo \
  --steps 200 \
  --repeats 25 \
  --em-iterations 10 \
  --other-heavy-jobs-running yes \
  --other-heavy-jobs-note "shared workstation with concurrent repository work" \
  --isolation-note "single-CPU affinity; no exclusive-core reservation" \
  --json benchmarks/results/bench_predictive_model.json

The measurements are loaded-host local-regression evidence. CPU affinity does not imply an exclusive core, so the artifact does not support a promotion-grade cross-host performance claim.

Latest committed local run¶

The 2026-07-14 source-bound run used 200 time steps, 25 interleaved samples per backend, four latent states, three observations, and two controls. It ran on logical CPU 2 of an Intel i5-11600K under the powersave governor while other heavy repository work was active. The core was affinity-pinned but not reserved. These numbers are regression evidence for this host only.

Backend	Probe (ms)	Median (ms)	Min (ms)	Max (ms)	Python / median	Max array delta	Likelihood delta
Python	0.005959	13.847860	12.020334	15.800676	1.000000	0	0
Rust	0.003617	2.806096	2.519172	5.343638	4.934920	6.22e-15	1.71e-13
Julia	6259.491127	2.140998	1.976630	136.903510	6.467946	5.33e-15	1.14e-13
Go	1.802784	1.827930	1.572094	3.094846	7.575706	5.33e-15	2.27e-13
Mojo	67.868924	1.553183	1.382646	2.595167	8.915794	4.00e-15	8.52e-08

The exact warm ranking was Mojo, Go, Julia, Rust, Python. The stable dispatch remains Mojo, Go, Rust, Julia, Python: the artifact records Rust-before-Julia as the sole declared warm-order exception because Rust is already imported and the Julia probe incurred 6.26 seconds of runtime initialisation. All parity deltas remain inside the committed 1e-9 array and 1e-7 likelihood limits.

Python-only RTS smoothing measured 6.222437 ms median (5.728943–9.343452 ms). Ten EM iterations measured 234.127058 ms median (217.413576–265.916984 ms). The artifact records every raw sample, so timing dispersion remains visible rather than being discarded.

Verification surfaces¶

test_linear_gaussian_ssm_parameters.py: parameter and covariance contracts.
test_linear_gaussian_ssm_random.py: random-model dimension and stability contracts.
test_linear_gaussian_filter_result.py: filtering-result contracts.
test_linear_gaussian_smooth_result.py: smoothing-result contracts.
test_kalman_filter.py: analytic updates, controls, covariance stability, validation, and installed-backend parity.
test_rts_smoother.py: exact multivariate batch conditioning and lag orientation.
test_em_learner.py: controlled sufficient statistics, monotonicity, held-out likelihood, convergence, and fail-closed behavior.
test_predictive_world_model.py: planning-facing mean/covariance forecasts.
test_predictive_model_backends.py: loader and FFI boundaries.
test_predictive_model_architecture.py: facade identity, pickle compatibility, responsibility ownership, import DAG, structured solves, and module bounds.
test_predictive_model_benchmark.py: artifact provenance, all-backend parity, loaded-host disclosure, and the reduced real CLI.

Focused tests cover every executable line in the seven Python responsibility modules. The only uncovered branch arcs are the exits of static Protocol declarations; no coverage exclusions or test skips are used.

API reference¶

`sc_neurocore.world_model.predictive_model` ¶

Linear Gaussian filtering, smoothing, learning, and state forecasting.

The model follows Kalman (1960), Rauch, Tung, and Striebel (1965), and the controlled expectation-maximisation formulation of Shumway and Stoffer (1982). Native acceleration applies to the forward Kalman filter; RTS smoothing and the EM M-step remain NumPy implementations.

`EMLearner` ¶

Estimate selected LGSSM parameters by expectation-maximisation.

Parameters¶

max_iter : int, default=50 Positive maximum number of E/M iterations. tol : float, default=1e-4 Non-negative absolute log-likelihood convergence threshold.

Notes¶

The M-step updates A, C, Q, R, mu_0, and Sigma_0. B and D are treated as known, but their control contributions are subtracted from the transition and observation sufficient statistics as required by Shumway and Stoffer (1982).

Raises¶

ValueError If max_iter or tol is outside its documented domain.

Source code in src/sc_neurocore/world_model/_lgssm_em.py

Python
class EMLearner:
    """Estimate selected LGSSM parameters by expectation-maximisation.

    Parameters
    ----------
    max_iter : int, default=50
        Positive maximum number of E/M iterations.
    tol : float, default=1e-4
        Non-negative absolute log-likelihood convergence threshold.

    Notes
    -----
    The M-step updates ``A``, ``C``, ``Q``, ``R``, ``mu_0``, and
    ``Sigma_0``. ``B`` and ``D`` are treated as known, but their control
    contributions are subtracted from the transition and observation
    sufficient statistics as required by Shumway and Stoffer (1982).

    Raises
    ------
    ValueError
        If ``max_iter`` or ``tol`` is outside its documented domain.

    """

    def __init__(self, max_iter: int = 50, tol: float = 1e-4) -> None:
        if isinstance(max_iter, bool) or not isinstance(max_iter, Integral):
            raise ValueError("max_iter must be a positive integer")
        if int(max_iter) <= 0:
            raise ValueError("max_iter must be a positive integer")
        if not isfinite(tol) or tol < 0.0:
            raise ValueError("tol must be finite and non-negative")
        self.max_iter: int = int(max_iter)
        self.tol: float = float(tol)
        self.log_likelihood_history: list[float] = []

    def fit(
        self,
        observations: npt.ArrayLike,
        initial_model: LinearGaussianSSM,
        controls: npt.ArrayLike | None = None,
        backend: str = "python",
    ) -> LinearGaussianSSM:
        """Estimate model parameters from one observation sequence.

        Parameters
        ----------
        observations : array-like, shape (T, p)
            Finite observation sequence with ``T >= 2``.
        initial_model : LinearGaussianSSM
            Starting parameters. The returned model preserves its ``B`` and
            ``D`` values exactly.
        controls : array-like, shape (T, m), optional
            Controls paired with the observations.
        backend : {"auto", "mojo", "go", "rust", "julia", "python"}, default="python"
            Forward-filter backend used in each E-step. The deterministic
            default keeps the entire learning benchmark on the Python path.

        Returns
        -------
        LinearGaussianSSM
            Last accepted M-step model.

        Raises
        ------
        ValueError
            If data shapes or values are invalid or fewer than two samples
            are supplied.
        RuntimeError
            If a likelihood decrease exceeds accumulated float64 round-off.

        """
        observation_array = _normalise_observations(
            observations,
            obs_dim=initial_model.obs_dim,
        )
        if observation_array.shape[0] < 2:
            raise ValueError("EM requires at least two observation time steps")
        control_array = _normalise_controls(
            controls,
            time_steps=observation_array.shape[0],
            control_dim=initial_model.control_dim,
        )

        model = initial_model
        filter_result = KalmanFilter(model).filter(
            observation_array,
            controls=control_array,
            backend=backend,
        )
        previous_likelihood = filter_result.log_likelihood
        self.log_likelihood_history = [previous_likelihood]

        for _ in range(self.max_iter):
            smoothed = RTSSmoother(model).smooth(filter_result)
            candidate = _maximisation_step(
                model,
                observation_array,
                control_array,
                smoothed,
            )
            candidate_result = KalmanFilter(candidate).filter(
                observation_array,
                controls=control_array,
                backend=backend,
            )
            candidate_likelihood = candidate_result.log_likelihood
            self.log_likelihood_history.append(candidate_likelihood)
            decrease = previous_likelihood - candidate_likelihood
            if decrease > _likelihood_tolerance(previous_likelihood):
                raise RuntimeError(
                    "EM log-likelihood decreased beyond float64 round-off: "
                    f"{previous_likelihood} -> {candidate_likelihood}"
                )

            model = candidate
            if abs(candidate_likelihood - previous_likelihood) <= self.tol:
                break
            filter_result = candidate_result
            previous_likelihood = candidate_likelihood

        return model

`fit(observations, initial_model, controls=None, backend='python')` ¶

Estimate model parameters from one observation sequence.

Parameters¶

observations : array-like, shape (T, p) Finite observation sequence with T >= 2. initial_model : LinearGaussianSSM Starting parameters. The returned model preserves its B and D values exactly. controls : array-like, shape (T, m), optional Controls paired with the observations. backend : {"auto", "mojo", "go", "rust", "julia", "python"}, default="python" Forward-filter backend used in each E-step. The deterministic default keeps the entire learning benchmark on the Python path.

Returns¶

LinearGaussianSSM Last accepted M-step model.

Raises¶

ValueError If data shapes or values are invalid or fewer than two samples are supplied. RuntimeError If a likelihood decrease exceeds accumulated float64 round-off.

Source code in src/sc_neurocore/world_model/_lgssm_em.py

Python
def fit(
    self,
    observations: npt.ArrayLike,
    initial_model: LinearGaussianSSM,
    controls: npt.ArrayLike | None = None,
    backend: str = "python",
) -> LinearGaussianSSM:
    """Estimate model parameters from one observation sequence.

    Parameters
    ----------
    observations : array-like, shape (T, p)
        Finite observation sequence with ``T >= 2``.
    initial_model : LinearGaussianSSM
        Starting parameters. The returned model preserves its ``B`` and
        ``D`` values exactly.
    controls : array-like, shape (T, m), optional
        Controls paired with the observations.
    backend : {"auto", "mojo", "go", "rust", "julia", "python"}, default="python"
        Forward-filter backend used in each E-step. The deterministic
        default keeps the entire learning benchmark on the Python path.

    Returns
    -------
    LinearGaussianSSM
        Last accepted M-step model.

    Raises
    ------
    ValueError
        If data shapes or values are invalid or fewer than two samples
        are supplied.
    RuntimeError
        If a likelihood decrease exceeds accumulated float64 round-off.

    """
    observation_array = _normalise_observations(
        observations,
        obs_dim=initial_model.obs_dim,
    )
    if observation_array.shape[0] < 2:
        raise ValueError("EM requires at least two observation time steps")
    control_array = _normalise_controls(
        controls,
        time_steps=observation_array.shape[0],
        control_dim=initial_model.control_dim,
    )

    model = initial_model
    filter_result = KalmanFilter(model).filter(
        observation_array,
        controls=control_array,
        backend=backend,
    )
    previous_likelihood = filter_result.log_likelihood
    self.log_likelihood_history = [previous_likelihood]

    for _ in range(self.max_iter):
        smoothed = RTSSmoother(model).smooth(filter_result)
        candidate = _maximisation_step(
            model,
            observation_array,
            control_array,
            smoothed,
        )
        candidate_result = KalmanFilter(candidate).filter(
            observation_array,
            controls=control_array,
            backend=backend,
        )
        candidate_likelihood = candidate_result.log_likelihood
        self.log_likelihood_history.append(candidate_likelihood)
        decrease = previous_likelihood - candidate_likelihood
        if decrease > _likelihood_tolerance(previous_likelihood):
            raise RuntimeError(
                "EM log-likelihood decreased beyond float64 round-off: "
                f"{previous_likelihood} -> {candidate_likelihood}"
            )

        model = candidate
        if abs(candidate_likelihood - previous_likelihood) <= self.tol:
            break
        filter_result = candidate_result
        previous_likelihood = candidate_likelihood

    return model

`KalmanFilter` ¶

Forward Kalman filter for a linear Gaussian state-space model.

Parameters¶

model : LinearGaussianSSM Validated model parameters shared by the Python and native paths.

Source code in src/sc_neurocore/world_model/_lgssm_filter.py

Python
class KalmanFilter:
    """Forward Kalman filter for a linear Gaussian state-space model.

    Parameters
    ----------
    model : LinearGaussianSSM
        Validated model parameters shared by the Python and native paths.

    """

    def __init__(self, model: LinearGaussianSSM) -> None:
        self.model = model

    def filter(
        self,
        observations: npt.ArrayLike,
        controls: npt.ArrayLike | None = None,
        backend: str = "auto",
    ) -> FilterResult:
        """Filter an observation sequence.

        Parameters
        ----------
        observations : array-like, shape (T, p)
            Finite observations ordered by time.
        controls : array-like, shape (T, m), optional
            Finite controls. They are required when ``m > 0`` and may be
            omitted only for a model with no control input.
        backend : {"auto", "mojo", "go", "rust", "julia", "python"}
            Execution backend. ``auto`` follows the availability- and
            initialisation-aware order Mojo, Go, Rust, Julia, then Python.

        Returns
        -------
        FilterResult
            Filtered and predicted moments plus the sequence log-likelihood.

        Raises
        ------
        ValueError
            If sequence shapes, values, or the backend name are invalid.
        RuntimeError
            If an explicitly requested native backend is unavailable.
        numpy.linalg.LinAlgError
            If an innovation covariance is not positive definite.

        """
        observation_array = _normalise_observations(
            observations,
            obs_dim=self.model.obs_dim,
        )
        control_array = _normalise_controls(
            controls,
            time_steps=observation_array.shape[0],
            control_dim=self.model.control_dim,
        )
        selected_backend = resolve_backend(backend)
        if selected_backend == "python":
            return self._filter_python(observation_array, control_array)
        return filter_native(selected_backend, self.model, observation_array, control_array)

    def _filter_python(
        self,
        observations: FloatArray,
        controls: FloatArray,
    ) -> FilterResult:
        time_steps, obs_dim = observations.shape
        state_dim = self.model.state_dim
        A, B, C, D = self.model.A, self.model.B, self.model.C, self.model.D
        Q, R = self.model.Q, self.model.R

        means = np.zeros((time_steps, state_dim), dtype=np.float64)
        covariances = np.zeros((time_steps, state_dim, state_dim), dtype=np.float64)
        pred_means = np.zeros((time_steps, state_dim), dtype=np.float64)
        pred_covariances = np.zeros((time_steps, state_dim, state_dim), dtype=np.float64)

        predicted_mean = self.model.mu_0.copy()
        predicted_covariance = self.model.Sigma_0.copy()
        identity = np.eye(state_dim, dtype=np.float64)
        gaussian_constant = obs_dim * np.log(2.0 * np.pi)
        log_likelihood = 0.0

        for time_index in range(time_steps):
            pred_means[time_index] = predicted_mean
            pred_covariances[time_index] = predicted_covariance

            control = controls[time_index]
            innovation = observations[time_index] - C @ predicted_mean - D @ control
            innovation_covariance = _symmetrise(C @ predicted_covariance @ C.T + R)
            try:
                lower = np.asarray(
                    np.linalg.cholesky(innovation_covariance),
                    dtype=np.float64,
                )
            except np.linalg.LinAlgError as exc:
                raise np.linalg.LinAlgError(
                    "innovation covariance must be positive definite"
                ) from exc
            solved_innovation = _solve_cholesky(lower, innovation)
            log_determinant = 2.0 * float(np.sum(np.log(np.diag(lower))))
            log_likelihood -= 0.5 * (
                gaussian_constant + log_determinant + float(innovation @ solved_innovation)
            )

            covariance_observation = predicted_covariance @ C.T
            gain = _solve_cholesky(lower, covariance_observation.T).T
            filtered_mean = predicted_mean + gain @ innovation
            residual_operator = identity - gain @ C
            filtered_covariance = _symmetrise(
                residual_operator @ predicted_covariance @ residual_operator.T + gain @ R @ gain.T
            )

            means[time_index] = filtered_mean
            covariances[time_index] = filtered_covariance
            predicted_mean = A @ filtered_mean + B @ control
            predicted_covariance = _symmetrise(A @ filtered_covariance @ A.T + Q)

        return FilterResult(
            means=means,
            covariances=covariances,
            pred_means=pred_means,
            pred_covariances=pred_covariances,
            log_likelihood=log_likelihood,
        )

`filter(observations, controls=None, backend='auto')` ¶

Filter an observation sequence.

Parameters¶

observations : array-like, shape (T, p) Finite observations ordered by time. controls : array-like, shape (T, m), optional Finite controls. They are required when m > 0 and may be omitted only for a model with no control input. backend : {"auto", "mojo", "go", "rust", "julia", "python"} Execution backend. auto follows the availability- and initialisation-aware order Mojo, Go, Rust, Julia, then Python.

Returns¶

FilterResult Filtered and predicted moments plus the sequence log-likelihood.

Raises¶

ValueError If sequence shapes, values, or the backend name are invalid. RuntimeError If an explicitly requested native backend is unavailable. numpy.linalg.LinAlgError If an innovation covariance is not positive definite.

Source code in src/sc_neurocore/world_model/_lgssm_filter.py

Python
def filter(
    self,
    observations: npt.ArrayLike,
    controls: npt.ArrayLike | None = None,
    backend: str = "auto",
) -> FilterResult:
    """Filter an observation sequence.

    Parameters
    ----------
    observations : array-like, shape (T, p)
        Finite observations ordered by time.
    controls : array-like, shape (T, m), optional
        Finite controls. They are required when ``m > 0`` and may be
        omitted only for a model with no control input.
    backend : {"auto", "mojo", "go", "rust", "julia", "python"}
        Execution backend. ``auto`` follows the availability- and
        initialisation-aware order Mojo, Go, Rust, Julia, then Python.

    Returns
    -------
    FilterResult
        Filtered and predicted moments plus the sequence log-likelihood.

    Raises
    ------
    ValueError
        If sequence shapes, values, or the backend name are invalid.
    RuntimeError
        If an explicitly requested native backend is unavailable.
    numpy.linalg.LinAlgError
        If an innovation covariance is not positive definite.

    """
    observation_array = _normalise_observations(
        observations,
        obs_dim=self.model.obs_dim,
    )
    control_array = _normalise_controls(
        controls,
        time_steps=observation_array.shape[0],
        control_dim=self.model.control_dim,
    )
    selected_backend = resolve_backend(backend)
    if selected_backend == "python":
        return self._filter_python(observation_array, control_array)
    return filter_native(selected_backend, self.model, observation_array, control_array)

`RTSSmoother` ¶

Rauch-Tung-Striebel backward smoother.

Parameters¶

model : LinearGaussianSSM Model used to produce the corresponding forward-filter result.

Notes¶

The recursion follows Rauch, Tung, and Striebel (1965). The returned lag-one covariance is oriented as Cov[x_t, x_{t+1} | y].

Source code in src/sc_neurocore/world_model/_lgssm_smoothing.py

Python
class RTSSmoother:
    """Rauch-Tung-Striebel backward smoother.

    Parameters
    ----------
    model : LinearGaussianSSM
        Model used to produce the corresponding forward-filter result.

    Notes
    -----
    The recursion follows Rauch, Tung, and Striebel (1965). The returned
    lag-one covariance is oriented as ``Cov[x_t, x_{t+1} | y]``.

    """

    def __init__(self, model: LinearGaussianSSM) -> None:
        self.model = model

    def smooth(self, filter_result: FilterResult) -> SmoothResult:
        """Smooth every state in a validated forward-filter result.

        Parameters
        ----------
        filter_result : FilterResult
            Forward moments for at least one time step.

        Returns
        -------
        SmoothResult
            Full-sequence posterior means, covariances, and lag-one
            cross-covariances.

        Raises
        ------
        ValueError
            If the filter-result state dimension differs from the model.
        numpy.linalg.LinAlgError
            If a predicted covariance required by the recursion is singular.

        """
        time_steps, state_dim = filter_result.means.shape
        if state_dim != self.model.state_dim:
            raise ValueError(
                "filter_result state dimension "
                f"{state_dim} does not match model state dimension {self.model.state_dim}"
            )

        smoothed_means = filter_result.means.copy()
        smoothed_covariances = filter_result.covariances.copy()
        cross_covariances = np.zeros(
            (time_steps - 1, state_dim, state_dim),
            dtype=np.float64,
        )

        for time_index in range(time_steps - 2, -1, -1):
            predicted_next = filter_result.pred_covariances[time_index + 1]
            covariance_transition = filter_result.covariances[time_index] @ self.model.A.T
            smoothing_gain = _solve_positive_definite(
                predicted_next,
                covariance_transition.T,
            ).T
            smoothed_means[time_index] = filter_result.means[time_index] + smoothing_gain @ (
                smoothed_means[time_index + 1] - filter_result.pred_means[time_index + 1]
            )
            smoothed_covariances[time_index] = _symmetrise(
                filter_result.covariances[time_index]
                + smoothing_gain
                @ (smoothed_covariances[time_index + 1] - predicted_next)
                @ smoothing_gain.T
            )
            cross_covariances[time_index] = smoothing_gain @ smoothed_covariances[time_index + 1]

        return SmoothResult(
            means=smoothed_means,
            covariances=smoothed_covariances,
            cross_covariances=cross_covariances,
        )

`smooth(filter_result)` ¶

Smooth every state in a validated forward-filter result.

Parameters¶

filter_result : FilterResult Forward moments for at least one time step.

Returns¶

SmoothResult Full-sequence posterior means, covariances, and lag-one cross-covariances.

Raises¶

ValueError If the filter-result state dimension differs from the model. numpy.linalg.LinAlgError If a predicted covariance required by the recursion is singular.

Source code in src/sc_neurocore/world_model/_lgssm_smoothing.py

Python
def smooth(self, filter_result: FilterResult) -> SmoothResult:
    """Smooth every state in a validated forward-filter result.

    Parameters
    ----------
    filter_result : FilterResult
        Forward moments for at least one time step.

    Returns
    -------
    SmoothResult
        Full-sequence posterior means, covariances, and lag-one
        cross-covariances.

    Raises
    ------
    ValueError
        If the filter-result state dimension differs from the model.
    numpy.linalg.LinAlgError
        If a predicted covariance required by the recursion is singular.

    """
    time_steps, state_dim = filter_result.means.shape
    if state_dim != self.model.state_dim:
        raise ValueError(
            "filter_result state dimension "
            f"{state_dim} does not match model state dimension {self.model.state_dim}"
        )

    smoothed_means = filter_result.means.copy()
    smoothed_covariances = filter_result.covariances.copy()
    cross_covariances = np.zeros(
        (time_steps - 1, state_dim, state_dim),
        dtype=np.float64,
    )

    for time_index in range(time_steps - 2, -1, -1):
        predicted_next = filter_result.pred_covariances[time_index + 1]
        covariance_transition = filter_result.covariances[time_index] @ self.model.A.T
        smoothing_gain = _solve_positive_definite(
            predicted_next,
            covariance_transition.T,
        ).T
        smoothed_means[time_index] = filter_result.means[time_index] + smoothing_gain @ (
            smoothed_means[time_index + 1] - filter_result.pred_means[time_index + 1]
        )
        smoothed_covariances[time_index] = _symmetrise(
            filter_result.covariances[time_index]
            + smoothing_gain
            @ (smoothed_covariances[time_index + 1] - predicted_next)
            @ smoothing_gain.T
        )
        cross_covariances[time_index] = smoothing_gain @ smoothed_covariances[time_index + 1]

    return SmoothResult(
        means=smoothed_means,
        covariances=smoothed_covariances,
        cross_covariances=cross_covariances,
    )

`FilterResult` `dataclass` ¶

Forward-filter posterior and one-step prediction moments.

Parameters¶

means : numpy.ndarray, shape (T, d) Filtered state means. covariances : numpy.ndarray, shape (T, d, d) Filtered state covariances. pred_means : numpy.ndarray, shape (T, d) One-step predicted state means before observing each sample. pred_covariances : numpy.ndarray, shape (T, d, d) One-step predicted state covariances. log_likelihood : float Sequence log-likelihood under the model.

Source code in src/sc_neurocore/world_model/_lgssm_types.py

Python
@dataclass
class FilterResult:
    """Forward-filter posterior and one-step prediction moments.

    Parameters
    ----------
    means : numpy.ndarray, shape (T, d)
        Filtered state means.
    covariances : numpy.ndarray, shape (T, d, d)
        Filtered state covariances.
    pred_means : numpy.ndarray, shape (T, d)
        One-step predicted state means before observing each sample.
    pred_covariances : numpy.ndarray, shape (T, d, d)
        One-step predicted state covariances.
    log_likelihood : float
        Sequence log-likelihood under the model.

    """

    means: FloatArray
    covariances: FloatArray
    pred_means: FloatArray
    pred_covariances: FloatArray
    log_likelihood: float

    def __post_init__(self) -> None:
        """Validate result shapes, finiteness, symmetry, and covariance signs."""
        self.means = _as_float_array(self.means, name="means", ndim=2)
        self.covariances = _as_float_array(self.covariances, name="covariances", ndim=3)
        self.pred_means = _as_float_array(self.pred_means, name="pred_means", ndim=2)
        self.pred_covariances = _as_float_array(
            self.pred_covariances, name="pred_covariances", ndim=3
        )
        time_steps, state_dim = self.means.shape
        if time_steps == 0 or state_dim == 0:
            raise ValueError("means must have non-zero time and state dimensions")
        expected_vector_shape = (time_steps, state_dim)
        expected_matrix_shape = (time_steps, state_dim, state_dim)
        if self.pred_means.shape != expected_vector_shape:
            raise ValueError(
                f"pred_means must have shape {expected_vector_shape}, got {self.pred_means.shape}"
            )
        for name in ("covariances", "pred_covariances"):
            covariance_stack = getattr(self, name)
            if covariance_stack.shape != expected_matrix_shape:
                raise ValueError(
                    f"{name} must have shape {expected_matrix_shape}, got {covariance_stack.shape}"
                )
            _require_positive_semidefinite_stack(covariance_stack, name=name)
        self.log_likelihood = float(self.log_likelihood)
        if not isfinite(self.log_likelihood):
            raise ValueError("log_likelihood must be finite")

`__post_init__()` ¶

Validate result shapes, finiteness, symmetry, and covariance signs.

Source code in src/sc_neurocore/world_model/_lgssm_types.py

Python
def __post_init__(self) -> None:
    """Validate result shapes, finiteness, symmetry, and covariance signs."""
    self.means = _as_float_array(self.means, name="means", ndim=2)
    self.covariances = _as_float_array(self.covariances, name="covariances", ndim=3)
    self.pred_means = _as_float_array(self.pred_means, name="pred_means", ndim=2)
    self.pred_covariances = _as_float_array(
        self.pred_covariances, name="pred_covariances", ndim=3
    )
    time_steps, state_dim = self.means.shape
    if time_steps == 0 or state_dim == 0:
        raise ValueError("means must have non-zero time and state dimensions")
    expected_vector_shape = (time_steps, state_dim)
    expected_matrix_shape = (time_steps, state_dim, state_dim)
    if self.pred_means.shape != expected_vector_shape:
        raise ValueError(
            f"pred_means must have shape {expected_vector_shape}, got {self.pred_means.shape}"
        )
    for name in ("covariances", "pred_covariances"):
        covariance_stack = getattr(self, name)
        if covariance_stack.shape != expected_matrix_shape:
            raise ValueError(
                f"{name} must have shape {expected_matrix_shape}, got {covariance_stack.shape}"
            )
        _require_positive_semidefinite_stack(covariance_stack, name=name)
    self.log_likelihood = float(self.log_likelihood)
    if not isfinite(self.log_likelihood):
        raise ValueError("log_likelihood must be finite")

`LinearGaussianSSM` `dataclass` ¶

Parameters of a discrete-time linear Gaussian state-space model.

Parameters¶

A : numpy.ndarray, shape (d, d) State-transition matrix. B : numpy.ndarray, shape (d, m) Control-input matrix. Use an empty second dimension when m = 0. C : numpy.ndarray, shape (p, d) Observation matrix. D : numpy.ndarray, shape (p, m) Direct control-to-observation matrix. Q : numpy.ndarray, shape (d, d) Symmetric positive-semidefinite process covariance. R : numpy.ndarray, shape (p, p) Symmetric positive-definite observation covariance. mu_0 : numpy.ndarray, shape (d,) Prior state mean. Sigma_0 : numpy.ndarray, shape (d, d) Symmetric positive-definite prior covariance.

Raises¶

ValueError If a parameter has an incompatible shape, non-finite value, or invalid covariance contract.

Source code in src/sc_neurocore/world_model/_lgssm_types.py

Python
@dataclass
class LinearGaussianSSM:
    """Parameters of a discrete-time linear Gaussian state-space model.

    Parameters
    ----------
    A : numpy.ndarray, shape (d, d)
        State-transition matrix.
    B : numpy.ndarray, shape (d, m)
        Control-input matrix. Use an empty second dimension when ``m = 0``.
    C : numpy.ndarray, shape (p, d)
        Observation matrix.
    D : numpy.ndarray, shape (p, m)
        Direct control-to-observation matrix.
    Q : numpy.ndarray, shape (d, d)
        Symmetric positive-semidefinite process covariance.
    R : numpy.ndarray, shape (p, p)
        Symmetric positive-definite observation covariance.
    mu_0 : numpy.ndarray, shape (d,)
        Prior state mean.
    Sigma_0 : numpy.ndarray, shape (d, d)
        Symmetric positive-definite prior covariance.

    Raises
    ------
    ValueError
        If a parameter has an incompatible shape, non-finite value, or invalid
        covariance contract.

    """

    A: FloatArray
    B: FloatArray
    C: FloatArray
    D: FloatArray
    Q: FloatArray
    R: FloatArray
    mu_0: FloatArray
    Sigma_0: FloatArray

    def __post_init__(self) -> None:
        """Copy parameters into finite, C-contiguous float64 arrays and validate them."""
        self.A = _as_float_array(self.A, name="A", ndim=2)
        self.B = _as_float_array(self.B, name="B", ndim=2)
        self.C = _as_float_array(self.C, name="C", ndim=2)
        self.D = _as_float_array(self.D, name="D", ndim=2)
        self.Q = _as_float_array(self.Q, name="Q", ndim=2)
        self.R = _as_float_array(self.R, name="R", ndim=2)
        self.mu_0 = _as_float_array(self.mu_0, name="mu_0", ndim=1)
        self.Sigma_0 = _as_float_array(self.Sigma_0, name="Sigma_0", ndim=2)

        d = self.A.shape[0]
        if d == 0 or self.A.shape != (d, d):
            raise ValueError(f"A must be a non-empty square matrix, got {self.A.shape}")
        if self.B.shape[0] != d:
            raise ValueError(f"B must have {d} rows, got {self.B.shape}")
        m = self.B.shape[1]
        p = self.C.shape[0]
        if p == 0 or self.C.shape != (p, d):
            raise ValueError(f"C must have shape (p, {d}) with p > 0, got {self.C.shape}")

        expected_shapes = {
            "D": (p, m),
            "Q": (d, d),
            "R": (p, p),
            "mu_0": (d,),
            "Sigma_0": (d, d),
        }
        for name, expected_shape in expected_shapes.items():
            actual_shape = getattr(self, name).shape
            if actual_shape != expected_shape:
                raise ValueError(f"{name} must have shape {expected_shape}, got {actual_shape}")

        _require_positive_semidefinite(self.Q, name="Q")
        _require_positive_definite(self.R, name="R")
        _require_positive_definite(self.Sigma_0, name="Sigma_0")

    @property
    def state_dim(self) -> int:
        """Return the latent-state dimension ``d``."""
        return int(self.A.shape[0])

    @property
    def obs_dim(self) -> int:
        """Return the observation dimension ``p``."""
        return int(self.C.shape[0])

    @property
    def control_dim(self) -> int:
        """Return the control-input dimension ``m``."""
        return int(self.B.shape[1])

    @classmethod
    def random(
        cls,
        state_dim: int,
        obs_dim: int,
        control_dim: int = 0,
        seed: int = 42,
    ) -> LinearGaussianSSM:
        """Construct a stable random model for initialisation and examples.

        Parameters
        ----------
        state_dim : int
            Positive latent-state dimension.
        obs_dim : int
            Positive observation dimension.
        control_dim : int, default=0
            Non-negative control dimension.
        seed : int, default=42
            NumPy random-generator seed.

        Returns
        -------
        LinearGaussianSSM
            Model whose state-transition spectral radius is ``0.95``.

        Raises
        ------
        ValueError
            If a dimension is not an integer in its documented domain.

        """
        d = _require_dimension(state_dim, name="state_dim", allow_zero=False)
        p = _require_dimension(obs_dim, name="obs_dim", allow_zero=False)
        m = _require_dimension(control_dim, name="control_dim", allow_zero=True)
        rng = np.random.default_rng(seed)
        raw = rng.standard_normal((d, d)) * 0.5
        spectral_radius = float(np.max(np.abs(np.linalg.eigvals(raw))))
        minimum_scale = float(np.finfo(np.float64).tiny)
        A = np.asarray(raw * (0.95 / max(spectral_radius, minimum_scale)))
        B = rng.standard_normal((d, m)) if m > 0 else np.zeros((d, 0))
        C = rng.standard_normal((p, d)) * 0.5
        D = rng.standard_normal((p, m)) if m > 0 else np.zeros((p, 0))
        return cls(
            A=A,
            B=B,
            C=C,
            D=D,
            Q=np.eye(d) * 0.1,
            R=np.eye(p) * 0.1,
            mu_0=np.zeros(d),
            Sigma_0=np.eye(d),
        )

`state_dim` `property` ¶

Return the latent-state dimension d.

`obs_dim` `property` ¶

Return the observation dimension p.

`control_dim` `property` ¶

Return the control-input dimension m.

`__post_init__()` ¶

Copy parameters into finite, C-contiguous float64 arrays and validate them.

Source code in src/sc_neurocore/world_model/_lgssm_types.py

Python
def __post_init__(self) -> None:
    """Copy parameters into finite, C-contiguous float64 arrays and validate them."""
    self.A = _as_float_array(self.A, name="A", ndim=2)
    self.B = _as_float_array(self.B, name="B", ndim=2)
    self.C = _as_float_array(self.C, name="C", ndim=2)
    self.D = _as_float_array(self.D, name="D", ndim=2)
    self.Q = _as_float_array(self.Q, name="Q", ndim=2)
    self.R = _as_float_array(self.R, name="R", ndim=2)
    self.mu_0 = _as_float_array(self.mu_0, name="mu_0", ndim=1)
    self.Sigma_0 = _as_float_array(self.Sigma_0, name="Sigma_0", ndim=2)

    d = self.A.shape[0]
    if d == 0 or self.A.shape != (d, d):
        raise ValueError(f"A must be a non-empty square matrix, got {self.A.shape}")
    if self.B.shape[0] != d:
        raise ValueError(f"B must have {d} rows, got {self.B.shape}")
    m = self.B.shape[1]
    p = self.C.shape[0]
    if p == 0 or self.C.shape != (p, d):
        raise ValueError(f"C must have shape (p, {d}) with p > 0, got {self.C.shape}")

    expected_shapes = {
        "D": (p, m),
        "Q": (d, d),
        "R": (p, p),
        "mu_0": (d,),
        "Sigma_0": (d, d),
    }
    for name, expected_shape in expected_shapes.items():
        actual_shape = getattr(self, name).shape
        if actual_shape != expected_shape:
            raise ValueError(f"{name} must have shape {expected_shape}, got {actual_shape}")

    _require_positive_semidefinite(self.Q, name="Q")
    _require_positive_definite(self.R, name="R")
    _require_positive_definite(self.Sigma_0, name="Sigma_0")

`random(state_dim, obs_dim, control_dim=0, seed=42)` `classmethod` ¶

Construct a stable random model for initialisation and examples.

Parameters¶

state_dim : int Positive latent-state dimension. obs_dim : int Positive observation dimension. control_dim : int, default=0 Non-negative control dimension. seed : int, default=42 NumPy random-generator seed.

Returns¶

LinearGaussianSSM Model whose state-transition spectral radius is 0.95.

Raises¶

ValueError If a dimension is not an integer in its documented domain.

Source code in src/sc_neurocore/world_model/_lgssm_types.py

Python
@classmethod
def random(
    cls,
    state_dim: int,
    obs_dim: int,
    control_dim: int = 0,
    seed: int = 42,
) -> LinearGaussianSSM:
    """Construct a stable random model for initialisation and examples.

    Parameters
    ----------
    state_dim : int
        Positive latent-state dimension.
    obs_dim : int
        Positive observation dimension.
    control_dim : int, default=0
        Non-negative control dimension.
    seed : int, default=42
        NumPy random-generator seed.

    Returns
    -------
    LinearGaussianSSM
        Model whose state-transition spectral radius is ``0.95``.

    Raises
    ------
    ValueError
        If a dimension is not an integer in its documented domain.

    """
    d = _require_dimension(state_dim, name="state_dim", allow_zero=False)
    p = _require_dimension(obs_dim, name="obs_dim", allow_zero=False)
    m = _require_dimension(control_dim, name="control_dim", allow_zero=True)
    rng = np.random.default_rng(seed)
    raw = rng.standard_normal((d, d)) * 0.5
    spectral_radius = float(np.max(np.abs(np.linalg.eigvals(raw))))
    minimum_scale = float(np.finfo(np.float64).tiny)
    A = np.asarray(raw * (0.95 / max(spectral_radius, minimum_scale)))
    B = rng.standard_normal((d, m)) if m > 0 else np.zeros((d, 0))
    C = rng.standard_normal((p, d)) * 0.5
    D = rng.standard_normal((p, m)) if m > 0 else np.zeros((p, 0))
    return cls(
        A=A,
        B=B,
        C=C,
        D=D,
        Q=np.eye(d) * 0.1,
        R=np.eye(p) * 0.1,
        mu_0=np.zeros(d),
        Sigma_0=np.eye(d),
    )

`SmoothResult` `dataclass` ¶

Rauch-Tung-Striebel smoothed state moments.

Parameters¶

means : numpy.ndarray, shape (T, d) Smoothed state means. covariances : numpy.ndarray, shape (T, d, d) Smoothed state covariances. cross_covariances : numpy.ndarray, shape (T - 1, d, d) Lag-one covariances Cov[x_t, x_{t+1} | y_{0:T-1}].

Source code in src/sc_neurocore/world_model/_lgssm_types.py

Python
@dataclass
class SmoothResult:
    """Rauch-Tung-Striebel smoothed state moments.

    Parameters
    ----------
    means : numpy.ndarray, shape (T, d)
        Smoothed state means.
    covariances : numpy.ndarray, shape (T, d, d)
        Smoothed state covariances.
    cross_covariances : numpy.ndarray, shape (T - 1, d, d)
        Lag-one covariances ``Cov[x_t, x_{t+1} | y_{0:T-1}]``.

    """

    means: FloatArray
    covariances: FloatArray
    cross_covariances: FloatArray

    def __post_init__(self) -> None:
        """Validate smoothed moment shapes, finiteness, and covariance signs."""
        self.means = _as_float_array(self.means, name="means", ndim=2)
        self.covariances = _as_float_array(self.covariances, name="covariances", ndim=3)
        self.cross_covariances = _as_float_array(
            self.cross_covariances, name="cross_covariances", ndim=3
        )
        time_steps, state_dim = self.means.shape
        if time_steps == 0 or state_dim == 0:
            raise ValueError("means must have non-zero time and state dimensions")
        covariance_shape = (time_steps, state_dim, state_dim)
        cross_shape = (time_steps - 1, state_dim, state_dim)
        if self.covariances.shape != covariance_shape:
            raise ValueError(
                f"covariances must have shape {covariance_shape}, got {self.covariances.shape}"
            )
        if self.cross_covariances.shape != cross_shape:
            raise ValueError(
                "cross_covariances must have shape "
                f"{cross_shape}, got {self.cross_covariances.shape}"
            )
        _require_positive_semidefinite_stack(
            self.covariances,
            name="covariances",
        )

`__post_init__()` ¶

Validate smoothed moment shapes, finiteness, and covariance signs.

Source code in src/sc_neurocore/world_model/_lgssm_types.py

Python
def __post_init__(self) -> None:
    """Validate smoothed moment shapes, finiteness, and covariance signs."""
    self.means = _as_float_array(self.means, name="means", ndim=2)
    self.covariances = _as_float_array(self.covariances, name="covariances", ndim=3)
    self.cross_covariances = _as_float_array(
        self.cross_covariances, name="cross_covariances", ndim=3
    )
    time_steps, state_dim = self.means.shape
    if time_steps == 0 or state_dim == 0:
        raise ValueError("means must have non-zero time and state dimensions")
    covariance_shape = (time_steps, state_dim, state_dim)
    cross_shape = (time_steps - 1, state_dim, state_dim)
    if self.covariances.shape != covariance_shape:
        raise ValueError(
            f"covariances must have shape {covariance_shape}, got {self.covariances.shape}"
        )
    if self.cross_covariances.shape != cross_shape:
        raise ValueError(
            "cross_covariances must have shape "
            f"{cross_shape}, got {self.cross_covariances.shape}"
        )
    _require_positive_semidefinite_stack(
        self.covariances,
        name="covariances",
    )

`PredictiveWorldModel` `dataclass` ¶

Forecast latent-state means and covariances through an LGSSM.

Parameters¶

state_dim : int Positive latent-state dimension. action_dim : int Non-negative action dimension. seed : int, default=42 Seed used to initialise the stable random LGSSM.

Notes¶

This class preserves the historical planning-facing API. Use :class:LinearGaussianSSM, :class:KalmanFilter, and :class:RTSSmoother when observations are available.

Source code in src/sc_neurocore/world_model/_predictive_world_model.py

Python
@dataclass
class PredictiveWorldModel:
    """Forecast latent-state means and covariances through an LGSSM.

    Parameters
    ----------
    state_dim : int
        Positive latent-state dimension.
    action_dim : int
        Non-negative action dimension.
    seed : int, default=42
        Seed used to initialise the stable random LGSSM.

    Notes
    -----
    This class preserves the historical planning-facing API. Use
    :class:`LinearGaussianSSM`, :class:`KalmanFilter`, and
    :class:`RTSSmoother` when observations are available.

    """

    state_dim: int
    action_dim: int
    seed: int = 42

    def __post_init__(self) -> None:
        """Initialise a validated stable state-transition model."""
        self.state_dim = _require_dimension(
            self.state_dim,
            name="state_dim",
            allow_zero=False,
        )
        self.action_dim = _require_dimension(
            self.action_dim,
            name="action_dim",
            allow_zero=True,
        )
        self.model = LinearGaussianSSM.random(
            state_dim=self.state_dim,
            obs_dim=self.state_dim,
            control_dim=self.action_dim,
            seed=self.seed,
        )
        self._mu = self.model.mu_0.copy()
        self._Sigma = self.model.Sigma_0.copy()

    def reset(self) -> None:
        """Reset the stored belief moments to the model prior."""
        self._mu = self.model.mu_0.copy()
        self._Sigma = self.model.Sigma_0.copy()

    def predict_next_state(
        self,
        current_state: npt.ArrayLike,
        action: npt.ArrayLike,
    ) -> FloatArray:
        """Predict the next latent-state mean.

        Parameters
        ----------
        current_state : array-like, shape (state_dim,)
            Current state estimate.
        action : array-like, shape (action_dim,)
            Current control input. A scalar is accepted when ``action_dim=1``.

        Returns
        -------
        numpy.ndarray, shape (state_dim,)
            Conditional mean ``A x_t + B u_t``.

        Raises
        ------
        ValueError
            If an input has an incompatible shape or non-finite value.

        """
        state = _normalise_vector(
            current_state,
            name="current_state",
            length=self.state_dim,
        )
        control = _normalise_vector(
            action,
            name="action",
            length=self.action_dim,
            allow_scalar=True,
        )
        return np.asarray(self.model.A @ state + self.model.B @ control)

    def predict_next_state_with_cov(
        self,
        current_state: npt.ArrayLike,
        current_cov: npt.ArrayLike,
        action: npt.ArrayLike,
    ) -> tuple[FloatArray, FloatArray]:
        """Predict the next latent-state mean and covariance.

        Parameters
        ----------
        current_state : array-like, shape (state_dim,)
            Current state estimate.
        current_cov : array-like, shape (state_dim, state_dim)
            Symmetric positive-semidefinite current covariance.
        action : array-like, shape (action_dim,)
            Current control input.

        Returns
        -------
        tuple of numpy.ndarray
            Mean ``A x_t + B u_t`` and covariance ``A P_t A^T + Q``.

        """
        state_covariance = _normalise_state_covariance(
            current_cov,
            state_dim=self.state_dim,
            name="current_cov",
        )
        mean = self.predict_next_state(current_state, action)
        covariance = _symmetrise(self.model.A @ state_covariance @ self.model.A.T + self.model.Q)
        return mean, covariance

    def forecast(
        self,
        initial_state: npt.ArrayLike,
        actions: list[npt.ArrayLike],
    ) -> list[FloatArray]:
        """Forecast a deterministic mean trajectory.

        Parameters
        ----------
        initial_state : array-like, shape (state_dim,)
            State before the first action.
        actions : list of array-like
            Ordered actions, one per returned state.

        Returns
        -------
        list of numpy.ndarray
            Independent state arrays after each action.

        """
        state = _normalise_vector(
            initial_state,
            name="initial_state",
            length=self.state_dim,
        )
        trajectory: list[FloatArray] = []
        for action in actions:
            state = self.predict_next_state(state, action)
            trajectory.append(state.copy())
        return trajectory

    def forecast_with_cov(
        self,
        initial_state: npt.ArrayLike,
        initial_cov: npt.ArrayLike,
        actions: list[npt.ArrayLike],
    ) -> list[tuple[FloatArray, FloatArray]]:
        """Forecast a mean and covariance trajectory.

        Parameters
        ----------
        initial_state : array-like, shape (state_dim,)
            State before the first action.
        initial_cov : array-like, shape (state_dim, state_dim)
            Initial symmetric positive-semidefinite covariance.
        actions : list of array-like
            Ordered actions, one per returned state.

        Returns
        -------
        list of tuple of numpy.ndarray
            Independent ``(mean, covariance)`` pairs after each action.

        """
        state = _normalise_vector(
            initial_state,
            name="initial_state",
            length=self.state_dim,
        )
        covariance = _normalise_state_covariance(
            initial_cov,
            state_dim=self.state_dim,
            name="initial_cov",
        )
        trajectory: list[tuple[FloatArray, FloatArray]] = []
        for action in actions:
            state, covariance = self.predict_next_state_with_cov(
                state,
                covariance,
                action,
            )
            trajectory.append((state.copy(), covariance.copy()))
        return trajectory

`__post_init__()` ¶

Initialise a validated stable state-transition model.

Source code in src/sc_neurocore/world_model/_predictive_world_model.py

Python
def __post_init__(self) -> None:
    """Initialise a validated stable state-transition model."""
    self.state_dim = _require_dimension(
        self.state_dim,
        name="state_dim",
        allow_zero=False,
    )
    self.action_dim = _require_dimension(
        self.action_dim,
        name="action_dim",
        allow_zero=True,
    )
    self.model = LinearGaussianSSM.random(
        state_dim=self.state_dim,
        obs_dim=self.state_dim,
        control_dim=self.action_dim,
        seed=self.seed,
    )
    self._mu = self.model.mu_0.copy()
    self._Sigma = self.model.Sigma_0.copy()

`reset()` ¶

Reset the stored belief moments to the model prior.

Source code in src/sc_neurocore/world_model/_predictive_world_model.py

Python
def reset(self) -> None:
    """Reset the stored belief moments to the model prior."""
    self._mu = self.model.mu_0.copy()
    self._Sigma = self.model.Sigma_0.copy()

`predict_next_state(current_state, action)` ¶

Predict the next latent-state mean.

Parameters¶

current_state : array-like, shape (state_dim,) Current state estimate. action : array-like, shape (action_dim,) Current control input. A scalar is accepted when action_dim=1.

Returns¶

numpy.ndarray, shape (state_dim,) Conditional mean A x_t + B u_t.

Raises¶

ValueError If an input has an incompatible shape or non-finite value.

Source code in src/sc_neurocore/world_model/_predictive_world_model.py

Python
def predict_next_state(
    self,
    current_state: npt.ArrayLike,
    action: npt.ArrayLike,
) -> FloatArray:
    """Predict the next latent-state mean.

    Parameters
    ----------
    current_state : array-like, shape (state_dim,)
        Current state estimate.
    action : array-like, shape (action_dim,)
        Current control input. A scalar is accepted when ``action_dim=1``.

    Returns
    -------
    numpy.ndarray, shape (state_dim,)
        Conditional mean ``A x_t + B u_t``.

    Raises
    ------
    ValueError
        If an input has an incompatible shape or non-finite value.

    """
    state = _normalise_vector(
        current_state,
        name="current_state",
        length=self.state_dim,
    )
    control = _normalise_vector(
        action,
        name="action",
        length=self.action_dim,
        allow_scalar=True,
    )
    return np.asarray(self.model.A @ state + self.model.B @ control)

`predict_next_state_with_cov(current_state, current_cov, action)` ¶

Predict the next latent-state mean and covariance.

Parameters¶

current_state : array-like, shape (state_dim,) Current state estimate. current_cov : array-like, shape (state_dim, state_dim) Symmetric positive-semidefinite current covariance. action : array-like, shape (action_dim,) Current control input.

Returns¶

tuple of numpy.ndarray Mean A x_t + B u_t and covariance A P_t A^T + Q.

Source code in src/sc_neurocore/world_model/_predictive_world_model.py

Python
def predict_next_state_with_cov(
    self,
    current_state: npt.ArrayLike,
    current_cov: npt.ArrayLike,
    action: npt.ArrayLike,
) -> tuple[FloatArray, FloatArray]:
    """Predict the next latent-state mean and covariance.

    Parameters
    ----------
    current_state : array-like, shape (state_dim,)
        Current state estimate.
    current_cov : array-like, shape (state_dim, state_dim)
        Symmetric positive-semidefinite current covariance.
    action : array-like, shape (action_dim,)
        Current control input.

    Returns
    -------
    tuple of numpy.ndarray
        Mean ``A x_t + B u_t`` and covariance ``A P_t A^T + Q``.

    """
    state_covariance = _normalise_state_covariance(
        current_cov,
        state_dim=self.state_dim,
        name="current_cov",
    )
    mean = self.predict_next_state(current_state, action)
    covariance = _symmetrise(self.model.A @ state_covariance @ self.model.A.T + self.model.Q)
    return mean, covariance

`forecast(initial_state, actions)` ¶

Forecast a deterministic mean trajectory.

Parameters¶

initial_state : array-like, shape (state_dim,) State before the first action. actions : list of array-like Ordered actions, one per returned state.

Returns¶

list of numpy.ndarray Independent state arrays after each action.

Source code in src/sc_neurocore/world_model/_predictive_world_model.py

Python
def forecast(
    self,
    initial_state: npt.ArrayLike,
    actions: list[npt.ArrayLike],
) -> list[FloatArray]:
    """Forecast a deterministic mean trajectory.

    Parameters
    ----------
    initial_state : array-like, shape (state_dim,)
        State before the first action.
    actions : list of array-like
        Ordered actions, one per returned state.

    Returns
    -------
    list of numpy.ndarray
        Independent state arrays after each action.

    """
    state = _normalise_vector(
        initial_state,
        name="initial_state",
        length=self.state_dim,
    )
    trajectory: list[FloatArray] = []
    for action in actions:
        state = self.predict_next_state(state, action)
        trajectory.append(state.copy())
    return trajectory

`forecast_with_cov(initial_state, initial_cov, actions)` ¶

Forecast a mean and covariance trajectory.

Parameters¶

initial_state : array-like, shape (state_dim,) State before the first action. initial_cov : array-like, shape (state_dim, state_dim) Initial symmetric positive-semidefinite covariance. actions : list of array-like Ordered actions, one per returned state.

Returns¶

list of tuple of numpy.ndarray Independent (mean, covariance) pairs after each action.

Source code in src/sc_neurocore/world_model/_predictive_world_model.py

Python
def forecast_with_cov(
    self,
    initial_state: npt.ArrayLike,
    initial_cov: npt.ArrayLike,
    actions: list[npt.ArrayLike],
) -> list[tuple[FloatArray, FloatArray]]:
    """Forecast a mean and covariance trajectory.

    Parameters
    ----------
    initial_state : array-like, shape (state_dim,)
        State before the first action.
    initial_cov : array-like, shape (state_dim, state_dim)
        Initial symmetric positive-semidefinite covariance.
    actions : list of array-like
        Ordered actions, one per returned state.

    Returns
    -------
    list of tuple of numpy.ndarray
        Independent ``(mean, covariance)`` pairs after each action.

    """
    state = _normalise_vector(
        initial_state,
        name="initial_state",
        length=self.state_dim,
    )
    covariance = _normalise_state_covariance(
        initial_cov,
        state_dim=self.state_dim,
        name="initial_cov",
    )
    trajectory: list[tuple[FloatArray, FloatArray]] = []
    for action in actions:
        state, covariance = self.predict_next_state_with_cov(
            state,
            covariance,
            action,
        )
        trajectory.append((state.copy(), covariance.copy()))
    return trajectory

`getattr(name)` ¶

Return the live historical Rust-availability flag on private access.

Source code in src/sc_neurocore/world_model/predictive_model.py

Python
def __getattr__(name: str) -> object:
    """Return the live historical Rust-availability flag on private access."""
    if name == "_HAS_RUST_LGSSM":
        return _backend_runtime._HAS_RUST_LGSSM
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

References¶

Kalman, R. E. (1960), “A New Approach to Linear Filtering and Prediction Problems.”
Rauch, H. E., Tung, F., and Striebel, C. T. (1965), “Maximum Likelihood Estimates of Linear Dynamic Systems.”
Shumway, R. H., and Stoffer, D. S. (1982), “An Approach to Time Series Smoothing and Forecasting Using the EM Algorithm.”
Bishop, C. M. (2006), Pattern Recognition and Machine Learning, section 13.3.

Linear Gaussian predictive model¶

Model¶

Public API and compatibility¶

Forward filtering¶

Native forward-filter chain¶

RTS smoothing¶

Controlled expectation-maximisation¶

Planning-compatible forecasts¶

Benchmark evidence¶

Latest committed local run¶

Verification surfaces¶

API reference¶

sc_neurocore.world_model.predictive_model ¶

EMLearner ¶

Parameters¶

Notes¶

Raises¶

fit(observations, initial_model, controls=None, backend='python') ¶

Parameters¶

Returns¶

Raises¶

KalmanFilter ¶

Parameters¶

filter(observations, controls=None, backend='auto') ¶

Parameters¶

Returns¶

Raises¶

RTSSmoother ¶

Parameters¶

Notes¶

smooth(filter_result) ¶

Parameters¶

Returns¶

Raises¶

FilterResult dataclass ¶

Parameters¶

__post_init__() ¶

LinearGaussianSSM dataclass ¶

Parameters¶

Raises¶

state_dim property ¶

obs_dim property ¶

control_dim property ¶

__post_init__() ¶

random(state_dim, obs_dim, control_dim=0, seed=42) classmethod ¶

Parameters¶

Returns¶

Raises¶

SmoothResult dataclass ¶

Parameters¶

__post_init__() ¶

PredictiveWorldModel dataclass ¶

Parameters¶

Notes¶

__post_init__() ¶

reset() ¶

predict_next_state(current_state, action) ¶

Parameters¶

Returns¶

Raises¶

predict_next_state_with_cov(current_state, current_cov, action) ¶

Parameters¶

Returns¶

forecast(initial_state, actions) ¶

Parameters¶

Returns¶

forecast_with_cov(initial_state, initial_cov, actions) ¶

Parameters¶

Returns¶

__getattr__(name) ¶

References¶

`sc_neurocore.world_model.predictive_model` ¶

`EMLearner` ¶

`fit(observations, initial_model, controls=None, backend='python')` ¶

`KalmanFilter` ¶

`filter(observations, controls=None, backend='auto')` ¶

`RTSSmoother` ¶

`smooth(filter_result)` ¶

`FilterResult` `dataclass` ¶

`__post_init__()` ¶

`LinearGaussianSSM` `dataclass` ¶

`state_dim` `property` ¶

`obs_dim` `property` ¶

`control_dim` `property` ¶

`__post_init__()` ¶

`random(state_dim, obs_dim, control_dim=0, seed=42)` `classmethod` ¶

`SmoothResult` `dataclass` ¶

`__post_init__()` ¶

`PredictiveWorldModel` `dataclass` ¶

`__post_init__()` ¶

`reset()` ¶

`predict_next_state(current_state, action)` ¶

`predict_next_state_with_cov(current_state, current_cov, action)` ¶

`forecast(initial_state, actions)` ¶

`forecast_with_cov(initial_state, initial_cov, actions)` ¶

`getattr(name)` ¶