Schemes

schemes

SDE Integration schemes.

This module facilitates the implementation of SDE integration schemes according to an ABC contract. Similar to the BaseRandomIncrement class, the BaseScheme class assumes basicially nothing about the properties of the integration algorithm, ensuring flexibility for a broad range of options. It enforces a simple interface via the step method. The actual implementation of the user schemes determines their user-friendliness. The recommended workflow can be seen in the ExplicitEulerMaruyamaScheme class. Here, the user has to provide drift and diffusion functions only for a single trajectory. Broadcasting over trajectories is not necessary. This is taken care of by a jit-compiled, parallel for loop using numba. Other algorithms should follow this pattern, which can be realized by a simple closure around the step function.

Classes:

Name	Description
BaseScheme	ABC for SDE integration schemes.
ExplicitEulerMaruyamaScheme	Implementation of the Explicit Euler-Maruyama scheme.

pysde.schemes.nobeartype module-attribute

nobeartype = beartype(conf=BeartypeConf(strategy=BeartypeStrategy.O0))

Decorator to deactivate type checking.

pysde.schemes.BaseScheme

Bases: ABC

ABC for SDE integration schemes.

The base class enforces a uniform interface through the step method.

Methods:

Name	Description
step	Perform a single step of the integration scheme

step abstractmethod

step(current_state: npt.NDArray[np.floating], current_time: Real, step_size: Real) -> npt.NDArray[np.floating]

Implement a single step in the integration scheme.

Parameters:

Name	Type	Description	Default
current_state	npt.NDArray[np.floating]	The current state of the system, in vectorized form \(d_X \times N\)	required
current_time	Real	The current time \(t\) of the stochastic process.	required
step_size	Real	Discrete step size \(\Delta t\).	required

Returns:

Type	Description
npt.NDArray[np.floating]	npt.NDArray[np.floating]: The updated state of the system, in vectorized form \(d_X \times N\).

Raises:

Type	Description
NotImplementedError	If the method is not implemented in a subclass.

pysde.schemes.ExplicitEulerMaruyamaScheme

Bases: BaseScheme

Implementation of the Explicit Euler-Maruyama scheme.

The explicit Euler-Maruyama scheme is a simple and efficient method for integrating SDEs. For a vector SDE of the form \(d\mathbf{X}_t = \mathbf{b}(\mathbf{X}_t,t)dt + \mathbf{\Sigma}(X_t,t)d\mathbf{W}_t\), the scheme is given by

\[ X_{k+1} = X_k + \mathbf{b}(\mathbf{X}_k,t_k)\Delta t + \mathbf{\Sigma}(X_k,t_k)\Delta W_k. \]

Methods:

Name	Description
step	Perform a single step of the integration scheme

__init__

__init__(drift_function: Callable[[npt.NDArray[np.floating], Real], npt.NDArray[np.floating]], diffusion_function: Callable[[npt.NDArray[np.floating], Real], npt.NDArray[np.floating]], random_increment: increments.BaseRandomIncrement) -> None

Initialize scheme with drift, diffusion, and Brownian increment.

Drift and diffusion need to be implemented as functions of the current state and time. They can employ standard Python and numpy functionality, to be jit-compilable with numba. These functions have to be provided for single trajectories only, broadcasting is taken care of internally.

Warning

Drift and diffusion have to be defined as proper functions with the def keyword, not as lambda functions.

Parameters:

Name	Type	Description	Default
drift_function	Callable[[npt.NDArray[np.floating], Real], npt.NDArray[np.floating]]	Callable representing the drift function \(\mathbf{b}(\mathbf{X}_t,t)\).	required
diffusion_function	Callable[[npt.NDArray[np.floating], Real], npt.NDArray[np.floating]]	Callable representing the diffusion function \(\mathbf{\Sigma}(\mathbf{X}_t,t)\).	required
random_increment	increments.BaseRandomIncrement	BaseRandomIncrement to approximate the Wiener process increment \(\Delta W_t\).	required

step

step(current_state: npt.NDArray[np.floating], current_time: Real, step_size: Real) -> npt.NDArray[np.floating]

Perform a single step of the integration scheme.

This method is a closure to the static _step function, which implements the actual loop over trajectories. in numba.

Parameters:

Name	Type	Description	Default
current_state	npt.NDArray[np.floating]	The current state of the system, in vectorized form \(d_X \times N\).	required
current_time	Real	The current time \(t\) of the stochastic process.	required
step_size	Real	Discrete step size \(\Delta t\).	required

Returns:

Type	Description
npt.NDArray[np.floating]	npt.NDArray[np.floating]: The updated state of the system, in vectorized form \(d_X \times N\).

_step staticmethod

_step(current_state: npt.NDArray[np.floating], current_time: Real, step_size: Real, drift_function: Callable[[npt.NDArray[np.floating], Real], npt.NDArray[np.floating]], diffusion_function: Callable[[npt.NDArray[np.floating], Real], npt.NDArray[np.floating]], vectorized_increment: npt.NDArray[np.floating]) -> npt.NDArray[np.floating]

Numba jittable step function for thread-parallel loop over trajectories.

Warning

Jitted functions with numba cannot be typ checked with beartype. The @nobeartype decorator is used to suppress the type checking for this function.