2.6. drrc.reservoircomputer module

class ReservoirComputer(nodes: int, degree: int, leakage: float, spectral_radius: float, input_scaling: float, input_length: int, input_bias: float, output_bias: float, regularization: float, training_includeinput: bool = True, identical_inputmatrix: bool = False, identical_adjacencymatrix: bool = False, identical_outputmatrix: bool | tuple | str = False, reset_matrices: bool = True)

Bases: object

Class to model dynamics using a single reservoir. The class provides a function to train the reservoirs redout, propagate the reservoir state or perform a prediction. Further, private methods exist to collect the reservoir state or construct the input and adjacency matrix.

Initializes a single reservoir based on an echo state network.

Parameters:

• nodes (int) – amount of nodes within the reservoir

• degree (int) – degree of the adjacency matrix of the echo-state network

• leakage (float) – describes how fast the reservoir forgets old data (0 fixed reservoir state, 1 no memory only driven by inputs)

• spectral_radius (float) – spectral radius of the adjacency matrix

• input_scaling (float) – scaling of the input data or weights

• input_length (int) – length of the input data

• input_bias (float) – bias added to the input data

• output_bias (float) – bias added to the output data

• regularization (float) – regularization parameter for the ridge regression

• training_includeinput (bool) – if True the input data is included in the extended state vector, i.e. used in the ridge regression

• identical_inputmatrix (bool) – if True the input matrix is the same for all ReservoirComputer instances

• identical_adjacencymatrix (bool) – if True the adjacency matrix is the same for all ReservoirComputer instances

• identical_outputmatrix (bool | tuple | str) – if False the output matrix is different for all ReservoirComputer instances, else it is the same.

• reset_matrices (bool) – if True the matrices are reset, else they are kept if they are identical for all parallel reservoirs.

train(input: ndarray, output: ndarray, transient_steps: int) → tuple[ndarray, ndarray]

Train a reservoir with given training data input, output of shape (training_steps+transient_steps, variables_in), (training_steps, variables_out), respectively.

For a training input time series \(\vec{u}(t)\) the reservoir is updated and its states \(\vec{r}(t)\) are collected in the extended interior state vector \(\vec{x}(t)=[\vec{r}(t), \vec{u}(t), b_{\mathrm{in}}]\).

The output matrix \(\boldsymbol{W}_{\mathrm{out}}\) is trained to preform \(\vec{y}(t)=\boldsymbol{W}_{\mathrm{out}}\vec{x}(t)\), where \(\vec{y}(t)\) is the desired output time series. The training uses Tikhonov regularisation:

\[\boldsymbol{W}_{\mathrm{out}}=\boldsymbol{Y}\boldsymbol{X}^T\left(\boldsymbol{X}\boldsymbol{X}^T+\beta\boldsymbol{I}\right)^{-1}\]

, which analytically minimizes

\[\sum_m\|\vec{y}^{\mathrm{true}}(m)-\boldsymbol{W}_{\mathrm{out}}\vec{x}_m\|^2+\beta\|\boldsymbol{W}_{\mathrm{out}}\|_2^2\quad.\]

The output matrix \(\boldsymbol{W}_{\mathrm{out}}\) is saved as a ridge regression model sklearn.linear_model.Ridge and the data on which the fit is performed (training data and reservoir states) is returned for testing purposes.

Parameters:

• np.ndarray[float] – Input data following shape as (time_steps, variables_in), where time_steps is the number of timesteps (including transient steps) and variables_in is the number of input variables

• np.ndarray[float] – Output data following shape as (time_steps, variables_out), where time_steps is the number of timesteps (excluding transient steps) and variables_out is the number of output variables

• transient_steps (int) – Number of transient steps to adjust the reservoir state to the input data. These reservoir states are not used in the fit.

Returns:

Time series of the extended interior state vector \(\vec{x}(t)\) and corresponding output data \(\vec{y}(t)\) used for training.

Return type:

tuple[np.ndarray[float]]

train_on_multiple_datasets(inputs: list[ndarray], outputs: list[ndarray], transient_steps) → tuple[ndarray, ndarray]

Train a reservoir on lists of with given training data inputs, outputs. Where the i-th list entry is of shape (training_steps_i+transient_steps, variables_in), (training_steps_i, variables_out), respectively.

For each element in inputs, i.e. training time series \(\vec{u}(t)\) the reservoir is updated and its states \(\vec{r}(t)\) are collected in the extended interior state vector \(\vec{x}(t)=[\vec{r}(t), \vec{u}(t), b_{\mathrm{in}}]\).

The output matrix \(\boldsymbol{W}_{\mathrm{out}}\) is trained to preform \(\vec{y}(t)=\boldsymbol{W}_{\mathrm{out}}\vec{x}(t)\) for all elements in the lists inputs, outputs, where \(\vec{y}(t)\) is the desired output time series. The training uses Tikhonov regularisation:

\[\boldsymbol{W}_{\mathrm{out}}=\boldsymbol{Y}\boldsymbol{X}^T\left(\boldsymbol{X}\boldsymbol{X}^T+\beta\boldsymbol{I}\right)^{-1}\]

, which analytically minimizes

\[\sum_m\|\vec{y}^{\mathrm{true}}(m)-\boldsymbol{W}_{\mathrm{out}}\vec{x}_m\|^2+\beta\|\boldsymbol{W}_{\mathrm{out}}\|_2^2\quad.\]

The output matrix \(\boldsymbol{W}_{\mathrm{out}}\) is saved as a ridge regression model sklearn.linear_model.Ridge and the data on which the fit is performed (training data and reservoir states) is returned for testing purposes.

Parameters:

• inputs (list[np.ndarray[float]]) – List of input training data, each following the shape (time_steps, variables_in), where time_steps is the number of timesteps (including transient steps) and variables_in is the number of input variables

• outputs (list[np.ndarray[float]]) – List of training output data, each following the shape (time_steps, variables_out), where time_steps is the number of timesteps (excluding transient steps) and variables_out is the number of output variables

• transient_steps (int) – Number of transient steps to adjust the reservoir state to the input data. These reservoir states are not used in the fit. Returns:

Returns:

Over all list elements inputs, outputs appended time series of the extended interior state vector \(\vec{x}(t)\) and corresponding output data \(\vec{y}(t)\) used for training.

Return type:

tuple[np.ndarray[float]]

predict_single_step(input: ndarray) → ndarray

Predict a single time step.

Uses the input data to updata the reservoir state following

\[\vec{s}(t+1) = (1-\alpha) \vec{s}(t) + \alpha \tanh(\nu \boldsymbol{W}_{\mathrm{in}} [\vec{u}(t), b_{\mathrm{in}}] + \rho \boldsymbol{W} s(t))\]

and predict the following time step with

\[\vec{u}(t + \Delta t) = \boldsymbol{W}_{\mathrm{out}} [\vec{s}(t), \vec{u}(t), b_{\mathrm{out}}] \,.\]

Parameters:

np.ndarray[float] – Input data containing one timestep, of shape: (1, ReservoirComputer.input_length).

Returns:

Predicted output data for the next timestep, of shape (1, ReservoirComputer.input_length).

Return type:

np.ndarray[float]

Warning

This function uses an hstack, which is slow, if there is not a preallocated array for the output. Allocating an arrow for the esv might be a good idea.

propagate_reservoir_state(input: ndarray) → None

Propagates a reservoir state to the next timestep using an input time series \(\vec{u}(t)\) following

\[\vec{s}(t+1) = (1-\alpha) \vec{s}(t) + \alpha \tanh(\nu \boldsymbol{W}_{\mathrm{in}} [\vec{u}(t), b_{\mathrm{in}}] + \rho \boldsymbol{W} s(t))\]

Parameters:

np.ndarray[float] – Input (transient) time series \(\vec{u}(t)\) of shape (variables,).

Notes

This function returns a reference, hence modifying the return value directly modifies the reseroir state.

_collect_reservoir_state(input_data: ndarray) → ndarray

Collects the reservoir states, which are updated following

\[\vec{s}(t+1) = (1-\alpha) \vec{s}(t) + \alpha \tanh(\nu \boldsymbol{W}_{\mathrm{in}} [\vec{u}(t), b_{\mathrm{in}}] + \rho \boldsymbol{W} s(t))\]

for a given input input_data, i.e. a time series \(\vec{u}(t)\).

Parameters:

input_data (np.ndarray[float]) – Input time series \(\vec{u}(t)\) of shape (time_steps, variables).

Returns:

Reservoir time series \(\vec{s}(t)\) of shape (time_steps, self.nodes).

Return type:

np.ndarray[float]

_get_input_matrix() → tuple[ndarray, ndarray]

Return \(\boldsymbol{W}_{\mathrm{in}}\), which maps the input to reservoir nodes.

Use the shape of input data and nnumber of nodes to determine the matrix.

Returns the input matrix for the input data and separately the input contribution of the input bias

_get_adjacency_matrix() → ndarray | coo_matrix

Returns the adjacency matrix \(\boldsymbol{W}\) of the reservoir. This is the cuppeling matrix of the inner nodes. The spectral radius of the adjacency matrix is set to one and is multiplied in the update step propagate_reservoir_state.