"""
DoubleReservoirESN -- a :class:`~memory_esn.multi.MultiESN` fixed to two reservoirs.
It takes two explicit inputs, ``fit([X1, X2], y)``, and is the shared base for
:class:`~memory_esn.fractional.fESN` and
:class:`~memory_esn.wavelet.wESN`, which feed reservoir 1 the raw
series and reservoir 2 a transformed view of it.
"""
from __future__ import annotations
from typing import List, Optional, Tuple, Union
import numpy as np
from .multi import MultiESN
Pair = Union[Tuple, list]
def _to_pair(param, name):
"""Broadcast a scalar to a 2-tuple, or validate a length-2 pair."""
if isinstance(param, (list, tuple)):
if len(param) != 2:
raise ValueError(
f"{name} must be a scalar or a length-2 pair, got length {len(param)}"
)
return list(param)
return [param, param]
def _resolve_d(d) -> List[float]:
"""Normalize a fractional-differencing order spec to a non-empty list.
A scalar becomes ``[d]`` (univariate); a list/tuple/array of orders is kept as
a list, so callers can stack one fracdiff channel per order (multivariate).
Shared by :class:`~memory_esn.fractional.fESN` and
:class:`~memory_esn.wavelet.wESN`.
"""
if isinstance(d, (list, tuple, np.ndarray)):
dl = [float(x) for x in d]
if len(dl) == 0:
raise ValueError("d must specify at least one differencing order.")
return dl
return [float(d)]
[docs]
class DoubleReservoirESN(MultiESN):
"""Two-reservoir Echo State Network.
Each per-reservoir hyperparameter is either a scalar (applied to both
reservoirs) or a length-2 pair ``(value_reservoir_1, value_reservoir_2)``.
Parameters
----------
n_reservoir, spectral_radius, input_scaling, input_init, reservoir_init, bias_init, leaky, activation, bias_scaling, noise, sparsity :
Scalar or length-2 pair. See :class:`~memory_esn.base.BaseESN`.
random_state, alphas, ridge_cv_params, concatenate_inputs, verbose :
See :class:`~memory_esn.multi.MultiESN`.
Examples
--------
>>> esn = DoubleReservoirESN(n_reservoir=(150, 100), random_state=0)
>>> esn.fit([X1, X2], y, washout=100)
>>> y_pred = esn.predict([X1_test, X2_test])
"""
def __init__(
self,
n_reservoir: Union[int, Pair] = 100,
spectral_radius: Union[float, Pair] = 0.9,
input_scaling: Union[float, Pair] = 0.5,
input_init: Union[str, Pair] = "uniform",
reservoir_init: Union[str, Pair] = "uniform",
bias_init: Union[str, Pair] = "uniform",
leaky: Union[float, Pair] = 1.0,
activation: Union[str, Pair] = "tanh",
bias_scaling: Union[float, Pair] = 0.0,
noise: Union[float, Pair] = 0.0,
sparsity: Union[float, Pair] = 0.9,
random_state: Union[int, Pair, None] = None,
alphas: Tuple[float, ...] = (0.01, 0.1, 1.0, 10.0),
ridge_cv_params: Optional[dict] = None,
concatenate_inputs: bool = True,
verbose: bool = False,
):
super().__init__(
n_reservoirs=2,
n_reservoir=_to_pair(n_reservoir, "n_reservoir"),
spectral_radius=_to_pair(spectral_radius, "spectral_radius"),
input_scaling=_to_pair(input_scaling, "input_scaling"),
input_init=_to_pair(input_init, "input_init"),
reservoir_init=_to_pair(reservoir_init, "reservoir_init"),
bias_init=_to_pair(bias_init, "bias_init"),
leaky=_to_pair(leaky, "leaky"),
activation=_to_pair(activation, "activation"),
bias_scaling=_to_pair(bias_scaling, "bias_scaling"),
noise=_to_pair(noise, "noise"),
sparsity=_to_pair(sparsity, "sparsity"),
random_state=random_state,
alphas=alphas,
ridge_cv_params=ridge_cv_params,
concatenate_inputs=concatenate_inputs,
verbose=verbose,
)
def __repr__(self) -> str:
return (
f"DoubleReservoirESN(sizes={[r.n_reservoir for r in self.reservoirs_]})"
)
