"""
fESN -- a :class:`~memory_esn.double.DoubleReservoirESN` whose
second reservoir sees a *fractionally differenced* view of the input.
Reservoir 1 receives the raw series (short-term temporal structure); reservoir 2
receives the pure-memory fractional filter ``f(t) = ((1-B)^d - 1) u(t) =
sum_{k>=1} omega_k u(t-k)`` -- i.e. standard fractional differencing with the
present (zero-lag) term removed, so the memory reservoir is driven only by past
inputs, with polynomially-decaying weights that preserve long memory.
``d`` may be a single order (default, univariate) or a list of orders; supplying
several orders stacks them into a **multivariate** input for the second reservoir
-- the fractional-differencing analogue of wESN's ``wavelet_components``.
"""
from __future__ import annotations
from typing import List, Optional, Tuple, Union
import numpy as np
from ._speedups import fractional_diff as _fractional_diff
from .double import DoubleReservoirESN, _resolve_d, _to_pair
Pair = Union[Tuple, list]
[docs]
class fESN(DoubleReservoirESN):
"""Fractional Echo State Network: dual-reservoir ESN with a
fractional-differencing memory branch.
Parameters
----------
n_reservoir, spectral_radius, input_scaling, input_init, reservoir_init, bias_init, leaky, activation, bias_scaling, noise, sparsity :
Scalar or length-2 pair (reservoir 1 = raw, reservoir 2 = fracdiff).
d : float or list of float, default=0.5
Fractional differencing order(s) (0 = none, 1 = first difference).
A single value (default) gives a univariate fracdiff branch; a list, e.g.
``[0.3, 0.6]``, applies each order and stacks the results, making the
second reservoir's input **multivariate** (one block of channels per order).
K : int, default=100
Number of lags in the fractional-difference filter.
skip_initial : bool, default=True
If True, drop the first ``K`` timesteps (where the filter window is not
yet full) instead of using a partial window.
random_state, alphas, ridge_cv_params, concatenate_inputs, verbose :
See :class:`~memory_esn.multi.MultiESN`.
Notes
-----
When ``skip_initial=True`` and ``continuation=False``, outputs are ``K``
timesteps shorter than the input; align targets accordingly.
Examples
--------
>>> # univariate fracdiff branch (default)
>>> fesn = fESN(n_reservoir=(200, 150), d=0.5, K=100, random_state=0)
>>> fesn.fit(X_train, y_train, washout=100) # single raw series in
>>> y_pred = fesn.predict(X_test)
>>> # multivariate: three differencing orders stacked
>>> fesn = fESN(n_reservoir=(200, 150), d=[0.3, 0.5, 0.8], K=100)
"""
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,
d: Union[float, List[float], Tuple[float, ...]] = 0.5,
K: int = 100,
skip_initial: bool = True,
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_reservoir=n_reservoir,
spectral_radius=spectral_radius,
input_scaling=input_scaling,
input_init=input_init,
reservoir_init=reservoir_init,
bias_init=bias_init,
leaky=leaky,
activation=activation,
bias_scaling=bias_scaling,
noise=noise,
sparsity=sparsity,
random_state=random_state,
alphas=alphas,
ridge_cv_params=ridge_cv_params,
concatenate_inputs=concatenate_inputs,
verbose=verbose,
)
self.d_ = d
self._d_list_ = _resolve_d(d) # normalized list of orders
self.K_ = K
self.skip_initial_ = skip_initial
self._last_K_inputs: Optional[np.ndarray] = None # for continuation
# ---------------------------------------------------------- transforms
@staticmethod
def _as_2d(X: np.ndarray) -> np.ndarray:
# 1-D univariate series -> (N, 1), not (1, N).
X = np.asarray(X)
if X.ndim == 1:
X = X.reshape(-1, 1)
return X
def _apply_fractional_diff(self, X: np.ndarray, continuation: bool) -> np.ndarray:
"""Fractionally difference ``X`` at each order in ``d``; stack the results.
In continuation mode the stored history is prepended (once, shared by all
orders) so every order's output rows align and can be concatenated.
"""
if continuation and self._last_K_inputs is not None:
X_ext = np.vstack([self._last_K_inputs, X])
else:
X_ext = X
blocks = []
for d in self._d_list_:
# Pure-memory filter ((1-B)^d - 1): drop the present (zero-lag) term,
# so f(t) = sum_{k>=1} omega_k u(t-k) (paper's memory input).
X_diff = _fractional_diff(
X_ext, d=d, K=self.K_, skip_initial=self.skip_initial_,
include_zero_lag=False,
)
# When history was prepended and we did NOT skip_initial, the first K
# rows correspond to the prepended context -> drop them. With
# skip_initial=True the filter already dropped exactly those K rows.
if continuation and self._last_K_inputs is not None and not self.skip_initial_:
X_diff = X_diff[self._last_K_inputs.shape[0]:]
blocks.append(X_diff)
return blocks[0] if len(blocks) == 1 else np.hstack(blocks)
def _prepare_inputs(self, X: np.ndarray, continuation: bool):
"""Return aligned (raw, fracdiff) inputs for the two reservoirs."""
X = self._as_2d(X)
X_fracdiff = self._apply_fractional_diff(X, continuation=continuation)
if continuation:
X_direct = X
elif self.skip_initial_:
X_direct = X[self.K_:, :]
else:
X_direct = X
return X_direct, X_fracdiff
# ----------------------------------------------------------------- API
[docs]
def fit(
self,
X: np.ndarray,
y: np.ndarray,
washout: int = 0,
x0_list: Optional[list] = None,
) -> "fESN":
"""Fit on a single raw series ``X`` (targets aligned to the fracdiff branch)."""
X = self._as_2d(X)
y = self._as_2d(y)
X_direct, X_fracdiff = self._prepare_inputs(X, continuation=False)
y_aligned = y[self.K_:, :] if self.skip_initial_ else y
if y_aligned.shape[0] != X_direct.shape[0]:
raise ValueError(
"Target length mismatch after fractional differencing: "
f"X_direct={X_direct.shape[0]}, y={y_aligned.shape[0]}"
)
super().fit([X_direct, X_fracdiff], y_aligned, washout=washout, x0_list=x0_list)
self._last_K_inputs = X[-self.K_:, :].copy() if self.K_ > 0 else None
return self
[docs]
def predict(
self,
X: np.ndarray,
x0_list: Optional[list] = None,
continuation: bool = False,
) -> np.ndarray:
"""Predict from a single raw series ``X``.
With ``skip_initial=True`` and ``continuation=False`` the output is ``K``
steps shorter than ``X``; with ``continuation=True`` it matches ``X``.
"""
X = self._as_2d(X)
X_direct, X_fracdiff = self._prepare_inputs(X, continuation=continuation)
y_pred = super().predict([X_direct, X_fracdiff], x0_list=x0_list, continuation=continuation)
self._last_K_inputs = X[-self.K_:, :].copy() if self.K_ > 0 else None
return y_pred
[docs]
def reset_states(self) -> None:
"""Reset reservoir states and the fractional-difference history buffer."""
super().reset_states()
self._last_K_inputs = None
def __repr__(self) -> str:
d_repr = self._d_list_[0] if len(self._d_list_) == 1 else self._d_list_
return (
f"fESN(sizes={[r.n_reservoir for r in self.reservoirs_]}, "
f"d={d_repr}, K={self.K_})"
)
# Backward-compatible alias (the class is now named ``fESN``)
FractionalESN = fESN
