SDE Solvers API Reference<a class="headerlink" href="#sde-solvers-api-reference" title="Link to this heading">

class hpfracc.solvers.sde_solvers.SDESolution(t, y, fractional_order, method, drift_func, diffusion_func, metadata=None)[source]

Bases: object

Solution object for SDE solvers.

Parameters:

t (ndarray)
y (ndarray)
fractional_order (float | FractionalOrder)
method (str)
drift_func (Callable)
diffusion_func (Callable)
metadata (Dict[str, Any])

t: ndarray

y: ndarray

fractional_order: float | FractionalOrder

method: str

drift_func: Callable

diffusion_func: Callable

metadata: Dict[str, Any] = None

__init__(t, y, fractional_order, method, drift_func, diffusion_func, metadata=None)

Parameters:

t (ndarray)
y (ndarray)
fractional_order (float | FractionalOrder)
method (str)
drift_func (Callable)
diffusion_func (Callable)
metadata (Dict[str, Any] | None)

Return type:

None

class hpfracc.solvers.sde_solvers.FractionalSDESolver(fractional_order, definition='caputo', adaptive=False, rtol=1e-05, atol=1e-08)[source]

Bases: ABC

Base class for fractional SDE solvers.

A fractional SDE takes the form:

D^α X(t) = f(t, X(t)) dt + g(t, X(t)) dW(t)

where:

α is the fractional order
f is the drift function
g is the diffusion function
W(t) is a Wiener process

Parameters:

fractional_order (float | FractionalOrder)
definition (str)
adaptive (bool)
rtol (float)
atol (float)

__init__(fractional_order, definition='caputo', adaptive=False, rtol=1e-05, atol=1e-08)[source]

Initialize fractional SDE solver.

Parameters:

fractional_order (float | FractionalOrder) – Fractional order (0 < α < 2)
definition (str) – Type of fractional derivative (“caputo” or “riemann_liouville”)
adaptive (bool) – Use adaptive step size
rtol (float) – Relative tolerance for adaptive stepping
atol (float) – Absolute tolerance for adaptive stepping

abstract solve(drift, diffusion, x0, t_span, **kwargs)[source]

Solve fractional SDE.

Parameters:

drift (Callable[[float, ndarray], ndarray]) – Drift function f(t, x) -> R^d
diffusion (Callable[[float, ndarray], ndarray]) – Diffusion function g(t, x) -> R^(d x m) where m is dimension of noise
x0 (ndarray) – Initial condition
t_span (Tuple[float, float]) – Time interval (t0, tf)
**kwargs – Additional solver parameters

Returns:

SDESolution object containing trajectory

Return type:

class hpfracc.solvers.sde_solvers.FastHistoryConvolution(alpha, num_steps, dim)[source]

Bases: object

Helper class for efficient history convolution in fractional SDE solvers.

Currently uses optimized dot product (O(N^2)), but structured to support FFT-based block convolution (O(N log N)) in future updates.

Parameters:

alpha (float)
num_steps (int)
dim (int)

__init__(alpha, num_steps, dim)[source]

Parameters:

alpha (float)
num_steps (int)
dim (int)

update(value)[source]

Add new value to history.

Parameters:: value (ndarray)

convolve()[source]

Compute convolution of weights with history.

Return type:: ndarray

class hpfracc.solvers.sde_solvers.FractionalEulerMaruyama(*args, **kwargs)[source]

Bases: FractionalSDESolver

Fractional Euler-Maruyama method for solving fractional SDEs.

This is a first-order strong convergence method.

__init__(*args, **kwargs)[source]

Initialize fractional SDE solver.

Parameters:

fractional_order – Fractional order (0 < α < 2)
definition – Type of fractional derivative (“caputo” or “riemann_liouville”)
adaptive – Use adaptive step size
rtol – Relative tolerance for adaptive stepping
atol – Absolute tolerance for adaptive stepping

solve(drift, diffusion, x0, t_span, num_steps=100, seed=None, **kwargs)[source]

Solve fractional SDE using Euler-Maruyama method.

Parameters:

drift (Callable) – Drift function f(t, x)
diffusion (Callable) – Diffusion function g(t, x)
x0 (ndarray) – Initial condition
t_span (Tuple[float, float]) – Time interval (t0, tf)
num_steps (int) – Number of time steps
seed (int | None) – Random seed for Wiener process

Returns:

SDESolution object

Return type:

class hpfracc.solvers.sde_solvers.FractionalMilstein(*args, **kwargs)[source]

Bases: FractionalSDESolver

Fractional Milstein method for solving fractional SDEs.

This is a second-order strong convergence method with improved accuracy.

__init__(*args, **kwargs)[source]

Initialize fractional SDE solver.

Parameters:

fractional_order – Fractional order (0 < α < 2)
definition – Type of fractional derivative (“caputo” or “riemann_liouville”)
adaptive – Use adaptive step size
rtol – Relative tolerance for adaptive stepping
atol – Absolute tolerance for adaptive stepping

solve(drift, diffusion, x0, t_span, num_steps=100, seed=None, **kwargs)[source]

Solve fractional SDE using Milstein method.

Parameters:

drift (Callable) – Drift function f(t, x)
diffusion (Callable) – Diffusion function g(t, x)
x0 (ndarray) – Initial condition
t_span (Tuple[float, float]) – Time interval (t0, tf)
num_steps (int) – Number of time steps
seed (int | None) – Random seed for Wiener process

Returns:

SDESolution object

Return type:

hpfracc.solvers.sde_solvers.solve_fractional_sde(drift, diffusion, x0, t_span, fractional_order=0.5, method='euler_maruyama', num_steps=100, seed=None, **kwargs)[source]

Solve a fractional SDE.

Parameters:

drift (Callable) – Drift function f(t, x) -> R^d
diffusion (Callable) – Diffusion function g(t, x) -> R^(d x m)
x0 (ndarray) – Initial condition
t_span (Tuple[float, float]) – Time interval (t0, tf)
fractional_order (float | FractionalOrder) – Fractional order (0 < α < 2)
method (str) – Solver method (“euler_maruyama”, “milstein”, “predictor_corrector”)
num_steps (int) – Number of time steps
seed (int | None) – Random seed for Wiener process
**kwargs – Additional solver parameters

Returns:

SDESolution object

Return type:

Example

>>> def drift(t, x):
...     return -0.5 * x
>>> def diffusion(t, x):
...     return 0.2 * np.eye(1)
>>> x0 = np.array([1.0])
>>> sol = solve_fractional_sde(drift, diffusion, x0, (0, 1), 0.5, num_steps=100)
>>> print(sol.y[-1])

hpfracc.solvers.sde_solvers.solve_fractional_sde_system(drift, diffusion, x0, t_span, fractional_order, method='euler_maruyama', noise_type='additive', num_steps=100, seed=None, **kwargs)[source]

Solve a system of coupled fractional SDEs.

Parameters:

drift (Callable) – Drift function f(t, x) -> R^d
diffusion (Callable) – Diffusion function g(t, x) -> R^(d x m)
x0 (ndarray) – Initial condition
t_span (Tuple[float, float]) – Time interval (t0, tf)
fractional_order (float | FractionalOrder | List[float]) – Fractional order(s) for system
method (str) – Solver method
noise_type (str) – Type of noise (“additive” or “multiplicative”)
num_steps (int) – Number of time steps
seed (int | None) – Random seed
**kwargs – Additional parameters

Returns:

SDESolution object

Return type:

Noise Models

Stochastic Noise Models for Fractional SDEs

This module provides various noise models including Brownian motion, fractional Brownian motion, Lévy processes, and coloured noise.

class hpfracc.solvers.noise_models.NoiseModel[source]

Bases: ABC

Base class for stochastic noise models.

abstract generate_increment(t, dt, size=(), seed=None)[source]

Generate noise increment.

Parameters:

t (float) – Current time
dt (float) – Time step
size (Tuple[int, ...]) – Output shape
seed (int | None) – Random seed

Returns:

Noise increment array

Return type:

class hpfracc.solvers.noise_models.BrownianMotion(scale=1.0)[source]

Bases: NoiseModel

Standard Brownian motion (Wiener process).

Generates independent Gaussian increments with variance dt.

Parameters:: scale (float)

__init__(scale=1.0)[source]

Initialize Brownian motion.

Parameters:: scale (float) – Scaling factor for noise

generate_increment(t, dt, size=(), seed=None)[source]

Generate Wiener increment.

Parameters:

t (float)
dt (float)
size (Tuple[int, ...])
seed (int | None)

Return type:

variance(dt)[source]

Get variance of increment.

Parameters:: dt (float)
Return type:: float

class hpfracc.solvers.noise_models.FractionalBrownianMotion(hurst=0.5, scale=1.0)[source]

Bases: NoiseModel

Fractional Brownian motion (fBm).

A Gaussian process with long-range dependence characterized by the Hurst exponent H (0 < H < 1).

Parameters:

hurst (float)
scale (float)

__init__(hurst=0.5, scale=1.0)[source]

Initialize fractional Brownian motion.

Parameters:

hurst (float) – Hurst exponent (H=0.5 gives standard Brownian motion)
scale (float) – Scaling factor for noise

generate_increment(t, dt, size=(), seed=None)[source]

Generate fractional Brownian motion increment.

Note: This is a simplified implementation. For full fBm, need to account for covariance structure.

Parameters:

t (float)
dt (float)
size (Tuple[int, ...])
seed (int | None)

Return type:

property is_standard_bm: bool: Check if this is standard Brownian motion.

class hpfracc.solvers.noise_models.LevyNoise(alpha=1.5, beta=0.0, scale=1.0, location=0.0)[source]

Bases: NoiseModel

Lévy noise for jump diffusions.

Uses stable distributions to model heavy-tailed noise.

Parameters:

alpha (float)
beta (float)
scale (float)
location (float)

__init__(alpha=1.5, beta=0.0, scale=1.0, location=0.0)[source]

Initialize Lévy noise.

Parameters:

alpha (float) – Stability parameter (0 < α ≤ 2)
beta (float) – Skewness parameter (-1 ≤ β ≤ 1)
scale (float) – Scale parameter
location (float) – Location parameter

generate_increment(t, dt, size=(), seed=None)[source]

Generate Lévy noise increment.

Uses stable distribution sampling (simplified implementation).

Parameters:

t (float)
dt (float)
size (Tuple[int, ...])
seed (int | None)

Return type:

class hpfracc.solvers.noise_models.ColouredNoise(correlation_time=1.0, amplitude=1.0, seed=None)[source]

Bases: NoiseModel

Coloured noise (Ornstein-Uhlenbeck process).

Gaussian noise with exponential autocorrelation.

Parameters:

correlation_time (float)
amplitude (float)
seed (int | None)

__init__(correlation_time=1.0, amplitude=1.0, seed=None)[source]

Initialize coloured noise.

Parameters:

correlation_time (float) – Correlation time constant
amplitude (float) – Noise amplitude
seed (int | None) – Random seed for state initialization

generate_increment(t, dt, size=(), seed=None)[source]

Generate coloured noise increment.

Parameters:

t (float)
dt (float)
size (Tuple[int, ...])
seed (int | None)

Return type:

reset()[source]: Reset the noise process state.

class hpfracc.solvers.noise_models.NoiseConfig(noise_type='brownian', hurst=0.5, scale=1.0, alpha=1.5, beta=0.0, correlation_time=1.0, amplitude=1.0)[source]

Bases: object

Configuration for noise models.

Parameters:

noise_type (str)
hurst (float)
scale (float)
alpha (float)
beta (float)
correlation_time (float)
amplitude (float)

noise_type: str = 'brownian'

hurst: float = 0.5

scale: float = 1.0

alpha: float = 1.5

beta: float = 0.0

correlation_time: float = 1.0

amplitude: float = 1.0

__init__(noise_type='brownian', hurst=0.5, scale=1.0, alpha=1.5, beta=0.0, correlation_time=1.0, amplitude=1.0)

Parameters:

noise_type (str)
hurst (float)
scale (float)
alpha (float)
beta (float)
correlation_time (float)
amplitude (float)

Return type:

None

hpfracc.solvers.noise_models.create_noise_model(config)[source]

Create a noise model from configuration.

Parameters:: config (NoiseConfig) – Noise configuration
Returns:: NoiseModel instance
Return type:: NoiseModel

hpfracc.solvers.noise_models.generate_noise_trajectory(noise_model, t_span, num_steps, size=(), seed=None)[source]

Generate a complete noise trajectory.

Parameters:

noise_model (NoiseModel) – Noise model to use
t_span (Tuple[float, float]) – Time interval (t0, tf)
num_steps (int) – Number of steps
size (Tuple[int, ...]) – Shape of noise increments
seed (int | None) – Random seed

Returns:

Tuple of (time array, noise increments array)

Return type:

Tuple[ndarray, ndarray]

Neural Fractional SDE

Neural Fractional Stochastic Differential Equations

This module provides neural network-based fractional SDEs with adjoint training methods for efficient gradient-based learning.

class hpfracc.ml.neural_fsde.NeuralFSDEConfig(input_dim=2, hidden_dim=64, output_dim=2, num_layers=3, activation='tanh', use_adjoint=True, solver='dopri5', rtol=1e-05, atol=1e-05, fractional_order=None, device=None, dtype=torch.float32, enable_performance_monitoring=False, memory_optimization=True, use_advanced_solvers=True, diffusion_dim=1, noise_type='additive', drift_net=None, diffusion_net=None, use_sde_adjoint=True, learn_alpha=False)[source]

Bases: NeuralODEConfig

Configuration for neural fractional SDE models.

Parameters:

input_dim (int)
hidden_dim (int)
output_dim (int)
num_layers (int)
activation (str)
use_adjoint (bool)
solver (str)
rtol (float)
atol (float)
fractional_order (float | FractionalOrder | None)
device (torch.device | None)
dtype (torch.dtype)
enable_performance_monitoring (bool)
memory_optimization (bool)
use_advanced_solvers (bool)
diffusion_dim (int)
noise_type (str)
drift_net (torch.nn.Module | None)
diffusion_net (torch.nn.Module | None)
use_sde_adjoint (bool)
learn_alpha (bool)

diffusion_dim: int = 1

noise_type: str = 'additive'

drift_net: torch.nn.Module | None = None

diffusion_net: torch.nn.Module | None = None

use_sde_adjoint: bool = True

learn_alpha: bool = False

__init__(input_dim=2, hidden_dim=64, output_dim=2, num_layers=3, activation='tanh', use_adjoint=True, solver='dopri5', rtol=1e-05, atol=1e-05, fractional_order=None, device=None, dtype=torch.float32, enable_performance_monitoring=False, memory_optimization=True, use_advanced_solvers=True, diffusion_dim=1, noise_type='additive', drift_net=None, diffusion_net=None, use_sde_adjoint=True, learn_alpha=False)

Parameters:

input_dim (int)
hidden_dim (int)
output_dim (int)
num_layers (int)
activation (str)
use_adjoint (bool)
solver (str)
rtol (float)
atol (float)
fractional_order (float | FractionalOrder | None)
device (torch.device | None)
dtype (torch.dtype)
enable_performance_monitoring (bool)
memory_optimization (bool)
use_advanced_solvers (bool)
diffusion_dim (int)
noise_type (str)
drift_net (torch.nn.Module | None)
diffusion_net (torch.nn.Module | None)
use_sde_adjoint (bool)
learn_alpha (bool)

Return type:

None

class hpfracc.ml.neural_fsde.NeuralFractionalSDE(*args, **kwargs)[source]

Bases: BaseNeuralODE

Neural network-based fractional SDE with adjoint training.

Extends neural ODE framework to fractional stochastic differential equations for modeling stochastic dynamics with memory effects.

The model learns: - Drift function f(t, x): deterministic dynamics - Diffusion function g(t, x): stochastic noise magnitude - Fractional order: memory effects in dynamics

Parameters:: config (NeuralFSDEConfig)

__init__(config)[source]

Initialize neural fractional SDE.

Parameters:: config (NeuralFSDEConfig) – Configuration for the neural fSDE

_build_drift_network()[source]: Build neural network for drift function.

_build_diffusion_network()[source]: Build neural network for diffusion function.

drift_function(t, x)[source]

Alias for drift() for compatibility with tests.

Parameters:

t (torch.Tensor)
x (torch.Tensor)

Return type:

torch.Tensor

diffusion_function(t, x)[source]

Alias for diffusion() for compatibility with tests.

Parameters:

t (torch.Tensor)
x (torch.Tensor)

Return type:

torch.Tensor

drift(t, x)[source]

Compute drift function f(t, x).

Parameters:

t (torch.Tensor) – Time tensor
x (torch.Tensor) – State tensor

Returns:

Drift vector

Return type:

torch.Tensor

diffusion(t, x)[source]

Compute diffusion function g(t, x).

Parameters:

t (torch.Tensor) – Time tensor
x (torch.Tensor) – State tensor

Returns:

Diffusion matrix

Return type:

torch.Tensor

fractional_order()[source]

Get current fractional order.

Return type:: float

forward(x0, t, method='euler_maruyama', num_steps=100, seed=None)[source]

Forward pass through neural fractional SDE.

Parameters:

x0 (torch.Tensor) – Initial condition
t (torch.Tensor) – Time points (1D tensor or 2D batch)
method (str) – Solver method
num_steps (int) – Number of integration steps
seed (int | None) – Random seed for reproducibility

Returns:

Trajectory solution

Return type:

torch.Tensor

_solve_fractional_sde_torch(x0, t_start, t_end, num_steps, seed=None)[source]

PyTorch-native fractional SDE solver (Euler-Maruyama).

Parameters:

x0 (torch.Tensor)
t_start (torch.Tensor)
t_end (torch.Tensor)
num_steps (int)
seed (int | None)

Return type:

torch.Tensor

get_fractional_order()[source]

Get the fractional order parameter.

Return type:: float | torch.Tensor

adjoint_forward(x0, t, **kwargs)[source]

Adjoint-compatible forward pass.

Parameters:

x0 (torch.Tensor)
t (torch.Tensor)

Return type:

torch.Tensor

hpfracc.ml.neural_fsde.create_neural_fsde(input_dim=None, output_dim=None, hidden_dim=64, num_layers=3, fractional_order=0.5, diffusion_dim=1, noise_type='additive', learn_alpha=False, use_adjoint=True, drift_net=None, diffusion_net=None, config=None)[source]

Factory function to create a neural fractional SDE.

Parameters:

input_dim (int | None) – Input dimension
output_dim (int | None) – Output dimension
hidden_dim (int) – Hidden layer dimension
num_layers (int) – Number of hidden layers
fractional_order (float) – Initial fractional order
diffusion_dim (int) – Dimension of noise
noise_type (str) – Type of noise (“additive” or “multiplicative”)
learn_alpha (bool) – Whether to learn fractional order
use_adjoint (bool) – Use adjoint method for backpropagation
drift_net (torch.nn.Module | None) – Custom drift network
diffusion_net (torch.nn.Module | None) – Custom diffusion network
config (NeuralFSDEConfig | None)

Returns:

NeuralFractionalSDE instance

Return type:

NeuralFractionalSDE

SDE Adjoint Methods

Advanced SDE Adjoint Optimization Utilities

Provides memory-efficient checkpointing, mixed precision training, and sparse gradient accumulation for neural fractional SDEs.

class hpfracc.ml.sde_adjoint_utils.CheckpointConfig(checkpoint_frequency=10, max_checkpoints=100, checkpoint_strategy='uniform', enable_checkpointing=True)[source]

Bases: object

Configuration for gradient checkpointing.

Parameters:

checkpoint_frequency (int)
max_checkpoints (int)
checkpoint_strategy (str)
enable_checkpointing (bool)

checkpoint_frequency: int = 10

max_checkpoints: int = 100

checkpoint_strategy: str = 'uniform'

enable_checkpointing: bool = True

__init__(checkpoint_frequency=10, max_checkpoints=100, checkpoint_strategy='uniform', enable_checkpointing=True)

Parameters:

checkpoint_frequency (int)
max_checkpoints (int)
checkpoint_strategy (str)
enable_checkpointing (bool)

Return type:

None

class hpfracc.ml.sde_adjoint_utils.MixedPrecisionConfig(enable_amp=False, half_precision=False, dtype_fp16=torch.float16, dtype_fp32=torch.float32, loss_scaling=1.0)[source]

Bases: object

Configuration for mixed precision training.

Parameters:

enable_amp (bool)
half_precision (bool)
dtype_fp16 (torch.dtype)
dtype_fp32 (torch.dtype)
loss_scaling (float)

enable_amp: bool = False

half_precision: bool = False

loss_scaling: float = 1.0

__init__(enable_amp=False, half_precision=False, dtype_fp16=torch.float16, dtype_fp32=torch.float32, loss_scaling=1.0)

Parameters:

enable_amp (bool)
half_precision (bool)
dtype_fp16 (torch.dtype)
dtype_fp32 (torch.dtype)
loss_scaling (float)

Return type:

None

class hpfracc.ml.sde_adjoint_utils.SDEStateCheckpoint(config)[source]

Bases: object

Checkpoint manager for SDE trajectory states. Enables memory-efficient training by saving intermediate states.

Parameters:: config (CheckpointConfig)

__init__(config)[source]

Parameters:: config (CheckpointConfig)

save_checkpoint(step, state, metadata=None)[source]

Save a checkpoint of the current state.

Parameters:

step (int)
state (torch.Tensor)
metadata (Dict[str, Any] | None)

load_checkpoint(step)[source]

Load checkpoint closest to the given step.

Parameters:: step (int)
Return type:: torch.Tensor | None

clear()[source]: Clear all checkpoints.

class hpfracc.ml.sde_adjoint_utils.MixedPrecisionManager(config)[source]

Bases: object

Manager for mixed precision training in SDEs. Handles automatic mixed precision (AMP) and float16 operations.

Parameters:: config (MixedPrecisionConfig)

__init__(config)[source]

Parameters:: config (MixedPrecisionConfig)

autocast()[source]: Create autocast context for mixed precision.

scale_loss(loss)[source]

Scale loss for mixed precision training.

Parameters:: loss (torch.Tensor)
Return type:: torch.Tensor

scale_gradients(loss)[source]

Scale gradients after backward pass.

Parameters:: loss (torch.Tensor)

step_optimizer(optimizer)[source]

Update optimizer with scaled gradients.

Parameters:: optimizer (torch.optim.Optimizer)

unscale_gradients(optimizer)[source]

Unscale gradients for clipping.

Parameters:: optimizer (torch.optim.Optimizer)

class hpfracc.ml.sde_adjoint_utils.SparseGradientAccumulator(sparsity_threshold=1e-06)[source]

Bases: object

Accumulator for sparse gradients in high-dimensional SDE systems. Reduces memory usage by only storing non-zero gradients.

Parameters:: sparsity_threshold (float)

__init__(sparsity_threshold=1e-06)[source]

Initialize sparse gradient accumulator.

Parameters:: sparsity_threshold (float) – Threshold below which gradients are considered zero

accumulate(grad, param_name=None)[source]

Accumulate a gradient with sparsity.

Parameters:

grad (torch.Tensor) – Gradient tensor
param_name (str | None) – Optional parameter name for debugging

get_dense_gradient(shape)[source]

Reconstruct dense gradient from sparse representation.

Parameters:: shape (Tuple[int, ...]) – Original gradient shape
Returns:: Dense gradient tensor
Return type:: torch.Tensor

clear()[source]: Clear accumulated gradients.

get_sparsity_ratio()[source]

Get ratio of non-zero to total elements.

Return type:: float

hpfracc.ml.sde_adjoint_utils.checkpoint_trajectory(func, *args, checkpoint_freq=10, **kwargs)[source]

Execute function with checkpointing to save memory.

Uses gradient checkpointing to trade compute for memory.

Parameters:

func (Callable) – Function to execute
*args – Arguments to pass to function
checkpoint_freq (int) – Frequency of checkpoints
**kwargs – Additional keyword arguments

Returns:

Tuple of (final_output, checkpointed_states)

Return type:

Tuple[torch.Tensor, List[torch.Tensor]]

class hpfracc.ml.sde_adjoint_utils.SDEAdjointOptimizer(model, optimizer, checkpoint_config=None, mixed_precision_config=None, enable_sparse_gradients=False)[source]

Bases: object

Optimizer wrapper for SDE adjoint training with advanced optimizations.

Combines checkpointing, mixed precision, and sparse gradients.

Parameters:

model (torch.nn.Module)
optimizer (torch.optim.Optimizer)
checkpoint_config (CheckpointConfig | None)
mixed_precision_config (MixedPrecisionConfig | None)
enable_sparse_gradients (bool)

__init__(model, optimizer, checkpoint_config=None, mixed_precision_config=None, enable_sparse_gradients=False)[source]

Parameters:

model (torch.nn.Module)
optimizer (torch.optim.Optimizer)
checkpoint_config (CheckpointConfig | None)
mixed_precision_config (MixedPrecisionConfig | None)
enable_sparse_gradients (bool)

step(loss)[source]

Optimization step with advanced features.

Parameters:: loss (torch.Tensor) – Loss tensor to optimize

save_state_checkpoint(step, state, metadata=None)[source]

Save state checkpoint.

Parameters:

step (int)
state (torch.Tensor)
metadata (Dict[str, Any] | None)

load_state_checkpoint(step)[source]

Load state checkpoint.

Parameters:: step (int)
Return type:: torch.Tensor | None

clear_checkpoints()[source]: Clear all checkpoints.

Graph-SDE Coupling

Graph-SDE Coupling for Spatio-Temporal Dynamics

This module provides coupling layers that integrate spatial dynamics (via graph neural networks) with temporal dynamics (via fractional SDEs) for modeling spatio-temporal phenomena.

class hpfracc.ml.graph_sde_coupling.CouplingType[source]

Bases: object

Types of spatial-temporal coupling.

BIDIRECTIONAL = 'bidirectional'

SPATIAL_TO_TEMPORAL = 'spatial_to_temporal'

TEMPORAL_TO_SPATIAL = 'temporal_to_spatial'

GATED = 'gated'

class hpfracc.ml.graph_sde_coupling.SpatialTemporalCoupling(*args, **kwargs)[source]

Bases: Module

Coupling layer between spatial (graph) and temporal (SDE) dynamics.

Computes learned coupling between spatial embeddings and temporal states.

Parameters:

spatial_dim (int)
temporal_dim (int)
coupling_dim (int)
coupling_type (str)
use_attention (bool)

__init__(spatial_dim, temporal_dim, coupling_dim, coupling_type='bidirectional', use_attention=True)[source]

Initialize coupling layer.

Parameters:

spatial_dim (int) – Dimension of spatial (graph) features
temporal_dim (int) – Dimension of temporal (SDE) features
coupling_dim (int) – Dimension of coupling embedding
coupling_type (str) – Type of coupling mechanism
use_attention (bool) – Use attention mechanism for coupling

forward(spatial_features, temporal_features)[source]

Apply spatial-temporal coupling.

Parameters:

spatial_features (torch.Tensor) – Graph node features (batch, num_nodes, spatial_dim)
temporal_features (torch.Tensor) – SDE state features (batch, num_nodes, temporal_dim)

Returns:

Tuple of (coupled_spatial, coupled_temporal)

Return type:

Tuple[torch.Tensor, torch.Tensor]

class hpfracc.ml.graph_sde_coupling.GraphFractionalSDELayer(*args, **kwargs)[source]

Bases: Module

Layer that couples graph-based spatial dynamics with fractional SDE temporal evolution.

Architecture: - Graph convolution for spatial features - Fractional SDE for temporal dynamics at each node - Learned coupling between spatial and temporal embeddings

Parameters:

input_dim (int)
hidden_dim (int)
output_dim (int)
fractional_order (float | FractionalOrder)
coupling_type (str)
num_sde_steps (int)
backend (BackendType)

__init__(input_dim, hidden_dim, output_dim, fractional_order=0.5, coupling_type='bidirectional', num_sde_steps=10, backend=BackendType.AUTO)[source]

Initialize Graph-Fractional SDE layer.

Parameters:

input_dim (int) – Input feature dimension
hidden_dim (int) – Hidden dimension for both spatial and temporal
output_dim (int) – Output feature dimension
fractional_order (float | FractionalOrder) – Fractional order for SDE dynamics
coupling_type (str) – Type of spatial-temporal coupling
num_sde_steps (int) – Number of SDE integration steps
backend (BackendType) – Computation backend

drift(x)[source]

Compute drift term for SDE.

Parameters:: x (torch.Tensor)
Return type:: torch.Tensor

diffusion(x)[source]

Compute diffusion term for SDE.

Parameters:: x (torch.Tensor)
Return type:: torch.Tensor

forward(x, edge_index, edge_weight=None)[source]

Forward pass through graph-SDE layer.

Parameters:

x (torch.Tensor) – Node features (batch, num_nodes, input_dim)
edge_index (torch.Tensor) – Edge connectivity (2, num_edges)
edge_weight (torch.Tensor | None) – Optional edge weights (num_edges,)

Returns:

Updated node features (batch, num_nodes, output_dim)

Return type:

torch.Tensor

class hpfracc.ml.graph_sde_coupling.MultiScaleGraphSDE(*args, **kwargs)[source]

Bases: Module

Multi-scale graph-SDE network with adaptive time stepping.

Handles different time scales for graph updates vs SDE evolution, optimal for stiff coupled systems.

Parameters:

input_dim (int)
hidden_dims (list)
output_dim (int)
fractional_order (float | FractionalOrder)
spatial_time_scale (float)
temporal_time_scale (float)

__init__(input_dim, hidden_dims, output_dim, fractional_order=0.5, spatial_time_scale=1.0, temporal_time_scale=0.1)[source]

Initialize multi-scale graph-SDE.

Parameters:

input_dim (int) – Input dimension
hidden_dims (list) – List of hidden dimensions for each layer
output_dim (int) – Output dimension
fractional_order (float | FractionalOrder) – Fractional order for SDE
spatial_time_scale (float) – Time scale for spatial (graph) dynamics
temporal_time_scale (float) – Time scale for temporal (SDE) dynamics

forward(x, edge_index, edge_weight=None)[source]

Forward pass with multi-scale dynamics.

Parameters:

x (torch.Tensor) – Node features
edge_index (torch.Tensor) – Edge connectivity
edge_weight (torch.Tensor | None) – Optional edge weights

Returns:

Output features

Return type:

torch.Tensor

Coupled System Solvers

Coupled System Solvers for Graph-SDE Dynamics

This module provides numerical solvers for systems of coupled spatial-temporal dynamics, integrating graph-based spatial evolution with fractional SDE temporal evolution.

class hpfracc.solvers.coupled_solvers.CoupledSolution(t, spatial, temporal, coupling, metadata=None)[source]

Bases: object

Solution object for coupled graph-SDE systems.

Parameters:

t (ndarray)
spatial (ndarray)
temporal (ndarray)
coupling (ndarray)
metadata (Dict[str, Any])

t: ndarray

spatial: ndarray

temporal: ndarray

coupling: ndarray

metadata: Dict[str, Any] = None

__init__(t, spatial, temporal, coupling, metadata=None)

Parameters:

t (ndarray)
spatial (ndarray)
temporal (ndarray)
coupling (ndarray)
metadata (Dict[str, Any] | None)

Return type:

None

class hpfracc.solvers.coupled_solvers.CoupledSystemSolver(fractional_orders, coupling_strength=1.0)[source]

Bases: ABC

Base class for coupled system solvers.

Parameters:

fractional_orders (float | FractionalOrder | Dict[str, float])
coupling_strength (float)

__init__(fractional_orders, coupling_strength=1.0)[source]

Initialize coupled system solver.

Parameters:

fractional_orders (float | FractionalOrder | Dict[str, float]) – Fractional order(s) for system
coupling_strength (float) – Strength of spatial-temporal coupling

abstract solve(graph_dynamics, sde_drift, sde_diffusion, adjacency, node_features, t_span, **kwargs)[source]

Solve coupled system.

Parameters:

graph_dynamics (Callable)
sde_drift (Callable)
sde_diffusion (Callable)
adjacency (ndarray)
node_features (ndarray)
t_span (Tuple[float, float])

Return type:

class hpfracc.solvers.coupled_solvers.OperatorSplittingSolver(fractional_orders, coupling_strength=1.0, split_order=2)[source]

Bases: CoupledSystemSolver

Operator splitting solver for graph-SDE dynamics.

Uses Strang splitting for second-order accuracy by splitting spatial and temporal operators.

Parameters:

fractional_orders (float | FractionalOrder | Dict[str, float])
coupling_strength (float)
split_order (int)

__init__(fractional_orders, coupling_strength=1.0, split_order=2)[source]

Initialize operator splitting solver.

Parameters:

fractional_orders (float | FractionalOrder | Dict[str, float]) – Fractional order(s)
coupling_strength (float) – Coupling strength
split_order (int) – Splitting order (1=Lie-Trotter, 2=Strang)

solve(graph_dynamics, sde_drift, sde_diffusion, adjacency, node_features, t_span, num_steps=100, seed=None, **kwargs)[source]

Solve using operator splitting.

For Strang splitting (order 2): - Half step of spatial dynamics - Full step of temporal dynamics - Half step of spatial dynamics

Parameters:

graph_dynamics (Callable)
sde_drift (Callable)
sde_diffusion (Callable)
adjacency (ndarray)
node_features (ndarray)
t_span (Tuple[float, float])
num_steps (int)
seed (int | None)

Return type:

_spatial_step(graph_dynamics, adjacency, state, dt)[source]

Single spatial (graph) evolution step.

Parameters:

graph_dynamics (Callable)
adjacency (ndarray)
state (ndarray)
dt (float)

Return type:

_temporal_step(drift, diffusion, state, t, dt)[source]

Single temporal (SDE) evolution step.

Parameters:

drift (Callable)
diffusion (Callable)
state (ndarray)
t (float)
dt (float)

Return type:

class hpfracc.solvers.coupled_solvers.MonolithicSolver(fractional_orders, coupling_strength=1.0)[source]

Bases: CoupledSystemSolver

Monolithic solver for strongly coupled graph-SDE systems.

Solves the full coupled system simultaneously for better accuracy in strongly coupled regimes, at the cost of higher memory usage.

Parameters:

fractional_orders (float | FractionalOrder | Dict[str, float])
coupling_strength (float)

solve(graph_dynamics, sde_drift, sde_diffusion, adjacency, node_features, t_span, num_steps=100, seed=None, **kwargs)[source]

Solve monolithic coupled system.

Parameters:

graph_dynamics (Callable)
sde_drift (Callable)
sde_diffusion (Callable)
adjacency (ndarray)
node_features (ndarray)
t_span (Tuple[float, float])
num_steps (int)
seed (int | None)

Return type:

hpfracc.solvers.coupled_solvers.solve_coupled_graph_sde(graph_dynamics, sde_drift, sde_diffusion, adjacency, node_features, t_span, fractional_orders=0.5, coupling_type='bidirectional', coupling_strength=1.0, solver='operator_splitting', **kwargs)[source]

Solve coupled graph-SDE system.

Parameters:

graph_dynamics (Callable) – Spatial dynamics function f(spatial, adjacency)
sde_drift (Callable) – Temporal drift function f_spatial(t, temporal)
sde_diffusion (Callable) – Temporal diffusion function g_temporal(t, temporal)
adjacency (ndarray) – Graph adjacency matrix
node_features (ndarray) – Initial node features
t_span (Tuple[float, float]) – Time interval
fractional_orders (float | FractionalOrder | Dict[str, float]) – Fractional order(s)
coupling_type (str) – Coupling type (“bidirectional”, “spatial_to_temporal”, etc.)
coupling_strength (float) – Strength of coupling
solver (str) – Solver type (“operator_splitting”, “monolithic”, “multiscale”)
**kwargs – Additional solver parameters

Returns:

CoupledSolution object

Return type: