systems.base.core.DiscreteStochasticSystem
systems.base.core.DiscreteStochasticSystem(*args, **kwargs)Concrete symbolic discrete-time stochastic dynamical system.
Extends DiscreteSymbolicSystem to handle stochastic difference equations. Users subclass this and implement define_system() to specify both deterministic and stochastic terms.
Represents systems of the form: x[k+1] = f(x[k], u[k]) + g(x[k], u[k]) * w[k] y[k] = h(x[k])
where: f: Deterministic next state (inherited from parent) g: Noise gain matrix (diffusion) w[k]: IID standard normal noise ~ N(0, I)
The noise enters additively through the gain matrix g, which can be: - Constant (additive noise): g(x, u) = G - State-dependent (multiplicative): g(x, u) = G(x, u) - Control-dependent: g(x, u) = G(x, u)
Users must define both dynamics (_f_sym) and diffusion (diffusion_expr) in define_system().
Attributes (Set by User in define_system)
diffusion_expr : sp.Matrix Symbolic diffusion matrix g(x, u), shape (nx, nw) REQUIRED - must be set in define_system() sde_type : SDEType or str SDE interpretation (‘ito’ or ‘stratonovich’) Optional - defaults to Itô (convention for discrete time) Note: In discrete time, both interpretations are equivalent
Attributes (Created Automatically)
diffusion_handler : DiffusionHandler Generates and caches diffusion functions noise_characteristics : NoiseCharacteristics Automatic noise structure analysis results nw : int Number of independent noise sources is_stochastic : bool Always True for this class
Examples
Discrete-time Ornstein-Uhlenbeck process:
>>> class DiscreteOU(DiscreteStochasticSystem):
... '''AR(1) process with mean reversion.'''
...
... def define_system(self, alpha=1.0, sigma=0.5, dt=0.1):
... x = sp.symbols('x')
... u = sp.symbols('u')
... alpha_sym = sp.symbols('alpha', positive=True)
... sigma_sym = sp.symbols('sigma', positive=True)
... dt_sym = sp.symbols('dt', positive=True)
...
... # Deterministic part (Euler discretization)
... self.state_vars = [x]
... self.control_vars = [u]
... self._f_sym = sp.Matrix([(1 - alpha_sym*dt_sym) * x + u])
... self.parameters = {alpha_sym: alpha, sigma_sym: sigma, dt_sym: dt}
... self._dt = dt # REQUIRED!
... self.order = 1
...
... # Stochastic part (additive noise)
... self.diffusion_expr = sp.Matrix([[sigma_sym]])
... self.sde_type = 'ito'
>>>
>>> # Instantiate system
>>> system = DiscreteOU(alpha=2.0, sigma=0.3, dt=0.1)
>>>
>>> # Automatic noise analysis
>>> print(system.noise_characteristics.noise_type)
NoiseType.ADDITIVE
>>>
>>> # Evaluate deterministic and stochastic parts
>>> x_k = np.array([1.0])
>>> u_k = np.array([0.0])
>>> f = system(x_k, u_k) # Deterministic next state mean
>>> g = system.diffusion(x_k, u_k) # Noise gain
>>>
>>> # Full stochastic step
>>> w_k = np.random.randn(1)
>>> x_next = f + g @ w_kAttributes
| Name | Description |
|---|---|
| diffusion_expr | Symbolic diffusion matrix g(x, u) - MUST be set in define_system() |
| is_stochastic | Return True (this is a stochastic system). |
| sde_type | SDE interpretation - ‘ito’ or ‘stratonovich’ (equivalent in discrete time) |
Methods
| Name | Description |
|---|---|
| can_optimize_for_additive | Check if additive-noise optimizations are applicable. |
| compile_all | Compile both deterministic and diffusion for all backends. |
| compile_diffusion | Pre-compile diffusion functions for specified backends. |
| depends_on_control | Check if diffusion depends on control inputs. |
| depends_on_state | Check if diffusion depends on state variables. |
| depends_on_time | Check if diffusion depends on time (always False for time-invariant). |
| diffusion | Evaluate diffusion term g(x[k], u[k]). |
| get_constant_noise | Get constant noise matrix for additive noise. |
| get_info | Get comprehensive system information. |
| get_noise_type | Get classified noise type. |
| get_optimization_opportunities | Get optimization opportunities based on noise structure. |
| get_sde_type | Get SDE interpretation type (Itô convention for discrete). |
| is_additive_noise | Check if noise is additive (constant, state-independent). |
| is_diagonal_noise | Check if noise sources are independent (diagonal diffusion). |
| is_multiplicative_noise | Check if noise is multiplicative (state-dependent). |
| is_pure_diffusion | Check if system is pure diffusion (zero deterministic part). |
| is_scalar_noise | Check if system has single noise source. |
| linearize | Compute linearization including diffusion: Ad = ∂f/∂x, Bd = ∂f/∂u, Gd = g(x_eq). |
| print_equations | Print symbolic equations using discrete-time stochastic notation. |
| print_stochastic_info | Print formatted stochastic system information. |
| reset_all_caches | Clear both deterministic and diffusion caches. |
| reset_diffusion_cache | Clear cached diffusion functions. |
| simulate_stochastic | Simulate stochastic discrete system with optional Monte Carlo. |
| step_stochastic | Compute full stochastic step: x[k+1] = f(x[k], u[k]) + g(x[k], u[k]) * w[k]. |
can_optimize_for_additive
systems.base.core.DiscreteStochasticSystem.can_optimize_for_additive()Check if additive-noise optimizations are applicable.
compile_all
systems.base.core.DiscreteStochasticSystem.compile_all(
backends=None,
verbose=False,
**kwargs,
)Compile both deterministic and diffusion for all backends.
Returns
| Name | Type | Description |
|---|---|---|
| Dict[Backend, Dict[str, float]] | Nested dict: backend → {‘deterministic’: time, ‘diffusion’: time} |
compile_diffusion
systems.base.core.DiscreteStochasticSystem.compile_diffusion(
backends=None,
verbose=False,
**kwargs,
)Pre-compile diffusion functions for specified backends.
depends_on_control
systems.base.core.DiscreteStochasticSystem.depends_on_control()Check if diffusion depends on control inputs.
depends_on_state
systems.base.core.DiscreteStochasticSystem.depends_on_state()Check if diffusion depends on state variables.
depends_on_time
systems.base.core.DiscreteStochasticSystem.depends_on_time()Check if diffusion depends on time (always False for time-invariant).
diffusion
systems.base.core.DiscreteStochasticSystem.diffusion(
x,
u=None,
k=0,
backend=None,
)Evaluate diffusion term g(x[k], u[k]).
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| x | StateVector | State vector (nx,) or batched (batch, nx) | required |
| u | Optional[ControlVector] | Control vector (nu,) or batched (batch, nu) None for autonomous systems | None |
| k | int | Time step (currently ignored for time-invariant systems) | 0 |
| backend | Optional[Backend] | Backend selection (None = auto-detect) | None |
Returns
| Name | Type | Description |
|---|---|---|
| ArrayLike | Diffusion matrix g(x, u), shape (nx, nw) or (batch, nx, nw) |
Examples
Controlled system:
>>> g = system.diffusion(np.array([2.0]), np.array([0.0]))
>>> print(g.shape)
(1, 1)Autonomous system:
>>> g = system.diffusion(np.array([2.0])) # u=None
>>> print(g.shape)
(1, 1)For additive noise (precompute once):
>>> if system.is_additive_noise():
... G = system.get_constant_noise() # Precompute once
... # Use G directly in simulation - huge speedup!get_constant_noise
systems.base.core.DiscreteStochasticSystem.get_constant_noise(backend='numpy')Get constant noise matrix for additive noise.
For additive noise, diffusion is constant and can be precomputed once for significant performance gains.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| backend | Backend | Backend for array type | 'numpy' |
Returns
| Name | Type | Description |
|---|---|---|
| ArrayLike | Constant diffusion matrix (nx, nw) |
Raises
| Name | Type | Description |
|---|---|---|
| ValueError | If noise is not additive |
Examples
>>> if system.is_additive_noise():
... G = system.get_constant_noise('numpy')
... print(G)
... [[0.3]]
...
... # Simulation loop with precomputed noise
... for k in range(1000):
... w_k = np.random.randn(nw)
... x_next = system(x, u) + G @ w_k
... x = x_nextget_info
systems.base.core.DiscreteStochasticSystem.get_info()Get comprehensive system information.
get_noise_type
systems.base.core.DiscreteStochasticSystem.get_noise_type()Get classified noise type.
get_optimization_opportunities
systems.base.core.DiscreteStochasticSystem.get_optimization_opportunities()Get optimization opportunities based on noise structure.
get_sde_type
systems.base.core.DiscreteStochasticSystem.get_sde_type()Get SDE interpretation type (Itô convention for discrete).
is_additive_noise
systems.base.core.DiscreteStochasticSystem.is_additive_noise()Check if noise is additive (constant, state-independent).
is_diagonal_noise
systems.base.core.DiscreteStochasticSystem.is_diagonal_noise()Check if noise sources are independent (diagonal diffusion).
is_multiplicative_noise
systems.base.core.DiscreteStochasticSystem.is_multiplicative_noise()Check if noise is multiplicative (state-dependent).
is_pure_diffusion
systems.base.core.DiscreteStochasticSystem.is_pure_diffusion()Check if system is pure diffusion (zero deterministic part).
is_scalar_noise
systems.base.core.DiscreteStochasticSystem.is_scalar_noise()Check if system has single noise source.
linearize
systems.base.core.DiscreteStochasticSystem.linearize(x_eq, u_eq=None)Compute linearization including diffusion: Ad = ∂f/∂x, Bd = ∂f/∂u, Gd = g(x_eq).
For stochastic systems, linearization returns three matrices: - Ad: State Jacobian ∂f/∂x - Bd: Control Jacobian ∂f/∂u - Gd: Diffusion matrix evaluated at equilibrium g(x_eq, u_eq)
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| x_eq | StateVector | Equilibrium state (nx,) | required |
| u_eq | Optional[ControlVector] | Equilibrium control (nu,) | None |
Returns
| Name | Type | Description |
|---|---|---|
| Tuple[ArrayLike, ArrayLike, ArrayLike] | (Ad, Bd, Gd) where: - Ad: State Jacobian (nx, nx) - Bd: Control Jacobian (nx, nu) - Gd: Diffusion matrix (nx, nw) |
Examples
>>> x_eq = np.zeros(2)
>>> u_eq = np.zeros(1)
>>> Ad, Bd, Gd = system.linearize(x_eq, u_eq)
>>>
>>> # Check discrete stability: |λ| < 1
>>> eigenvalues = np.linalg.eigvals(Ad)
>>> is_stable = np.all(np.abs(eigenvalues) < 1)
>>>
>>> # Stochastic covariance propagation
>>> # P[k+1] = Ad @ P[k] @ Ad.T + Gd @ Gd.Tprint_equations
systems.base.core.DiscreteStochasticSystem.print_equations(simplify=True)Print symbolic equations using discrete-time stochastic notation.
Overrides parent to show both deterministic and stochastic parts.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| simplify | bool | If True, simplify expressions before printing | True |
Examples
>>> system.print_equations()
======================================================================
DiscreteOU (Discrete-Time Stochastic, dt=0.1)
======================================================================
State Variables: [x]
Control Variables: [u]
Dimensions: nx=1, nu=1, nw=1
Noise Type: additiveDeterministic Part: f(x[k], u[k]) f_0 = 0.9*x + u
Stochastic Part: g(x[k], u[k]) g_0 = [0.3]
Full Dynamics: x[k+1] = f(x[k], u[k]) + g(x[k], u[k]) * w[k] where w[k] ~ N(0, I) ======================================================================
print_stochastic_info
systems.base.core.DiscreteStochasticSystem.print_stochastic_info()Print formatted stochastic system information.
reset_all_caches
systems.base.core.DiscreteStochasticSystem.reset_all_caches(backends=None)Clear both deterministic and diffusion caches.
reset_diffusion_cache
systems.base.core.DiscreteStochasticSystem.reset_diffusion_cache(backends=None)Clear cached diffusion functions.
simulate_stochastic
systems.base.core.DiscreteStochasticSystem.simulate_stochastic(
x0,
u_sequence=None,
n_steps=100,
n_paths=1,
seed=None,
**kwargs,
)Simulate stochastic discrete system with optional Monte Carlo.
Performs either single-path or Monte Carlo simulation of the stochastic difference equation.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| x0 | StateVector | Initial state (nx,) | required |
| u_sequence | Optional[Union[ControlVector, DiscreteControlInput]] | Control sequence (same format as parent simulate()) | None |
| n_steps | int | Number of simulation steps | 100 |
| n_paths | int | Number of Monte Carlo paths (default: 1) | 1 |
| seed | Optional[int] | Random seed for reproducibility | None |
| **kwargs | Additional simulation options | {} |
Returns
| Name | Type | Description |
|---|---|---|
| DiscreteSimulationResult | TypedDict containing: - states: State trajectories - Single path: (n_steps+1, nx) - Multiple paths: (n_paths, n_steps+1, nx) - controls: Control sequence (n_steps, nu) - time_steps: [0, 1, …, n_steps] - dt: Sampling period - metadata: Additional info including: - n_paths: Number of paths - noise_type: Detected noise type - seed: Random seed used |
Examples
Single trajectory:
>>> result = system.simulate_stochastic(
... x0=np.array([1.0]),
... u_sequence=None,
... n_steps=1000,
... seed=42
... )
>>> plt.plot(result['time_steps'], result['states'][:, 0])Monte Carlo simulation:
>>> result = system.simulate_stochastic(
... x0=np.array([1.0]),
... u_sequence=None,
... n_steps=1000,
... n_paths=100,
... seed=42
... )
>>> # result['states'] has shape (100, 1001, 1)
>>> mean_traj = result['states'].mean(axis=0)
>>> std_traj = result['states'].std(axis=0)
>>>
>>> plt.plot(result['time_steps'], mean_traj[:, 0], label='Mean')
>>> plt.fill_between(
... result['time_steps'],
... mean_traj[:, 0] - std_traj[:, 0],
... mean_traj[:, 0] + std_traj[:, 0],
... alpha=0.3
... )State feedback with stochastic dynamics:
>>> def policy(x, k):
... return -0.5 * x
>>> result = system.simulate_stochastic(
... x0=np.array([1.0]),
... u_sequence=policy,
... n_steps=1000
... )step_stochastic
systems.base.core.DiscreteStochasticSystem.step_stochastic(
x,
u=None,
w=None,
k=0,
backend=None,
)Compute full stochastic step: x[k+1] = f(x[k], u[k]) + g(x[k], u[k]) * w[k].
This is the primary method for stochastic simulation, computing the complete state update including both deterministic and stochastic components.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| x | StateVector | Current state (nx,) or batched (batch, nx) | required |
| u | Optional[ControlVector] | Control (nu,) or batched (batch, nu), None for autonomous | None |
| w | Optional[ArrayLike] | Standard normal noise (nw,) or batched (batch, nw) If None, generated automatically | None |
| k | int | Time step (currently ignored for time-invariant systems) | 0 |
| backend | Optional[Backend] | Backend selection (None = auto-detect) | None |
Returns
| Name | Type | Description |
|---|---|---|
| StateVector | Next state x[k+1], same shape and backend as input |
Examples
Automatic noise generation:
>>> x_next = system.step_stochastic(x_k, u_k)Custom noise (for reproducibility):
>>> w_k = np.random.randn(system.nw)
>>> x_next = system.step_stochastic(x_k, u_k, w=w_k)Deterministic (w=0):
>>> x_next = system.step_stochastic(x_k, u_k, w=np.zeros(system.nw))Batched inputs:
>>> x_batch = np.random.randn(100, 2) # 100 states
>>> u_batch = np.random.randn(100, 1) # 100 controls
>>> w_batch = np.random.randn(100, 1) # 100 noise samples
>>> x_next_batch = system.step_stochastic(x_batch, u_batch, w_batch)Notes
For batched inputs: x: (batch, nx) u: (batch, nu) or None w: (batch, nw) or None → x_next: (batch, nx)
If w is None, noise is generated per sample in batch.
With DiffusionHandler batching support: - Additive noise: g returns (nx, nw) - constant - Multiplicative noise: g returns (batch, nx, nw) - state-dependent
The function automatically handles both cases efficiently.