Skip to content

zoo

Classes:

Name Description
LSTMRegressor

Rolling LSTM regressor for sequential / time-series data.

LinearRegression

Incremental linear regression with optional feature growth and gradient clipping.

MultiLayerPerceptron

Multi-layer perceptron regressor with optional feature growth.

RNNRegressor

Rolling RNN regressor for sequential / time-series data.

LSTMRegressor

LSTMRegressor(
    n_features: int = 10,
    hidden_size: int = 32,
    num_layers: int = 1,
    dropout: float = 0.0,
    gradient_clip_value: float | None = 1.0,
    loss_fn: Union[str, Callable] = "mse",
    optimizer_fn: Union[str, Type[Optimizer]] = "adam",
    lr: float = 0.001,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs
)

Bases: RollingRegressor

Rolling LSTM regressor for sequential / time-series data.

Improves over a naïve single-unit LSTM by separating the hidden representation (hidden_size) from the 1D regression output head. Supports optional dropout and multiple LSTM layers. Designed to work with a rolling window maintained by :class:~deep_river.base.RollingDeepEstimator.

Parameters:

Name Type Description Default
n_features int

Number of input features per timestep (may grow if feature-incremental).

10
hidden_size int

Dimensionality of the LSTM hidden state.

32
num_layers int

Number of stacked LSTM layers.

1
dropout float

Dropout probability applied after the LSTM (and internally by PyTorch if num_layers > 1). Capped internally for safety.

0.0
gradient_clip_value float | None

Gradient norm clipping threshold (helps stability). None disables it.

1.0
loss_fn Union[str, Callable]

Standard configuration.

'mse'
optimizer_fn Union[str, Callable]

Standard configuration.

'mse'
lr Union[str, Callable]

Standard configuration.

'mse'
is_feature_incremental Union[str, Callable]

Standard configuration.

'mse'
device Union[str, Callable]

Standard configuration.

'mse'
seed Union[str, Callable]

Standard configuration.

'mse'
**kwargs Union[str, Callable]

Standard configuration.

'mse'

Examples:

Streaming regression on the Bikes dataset (only numeric features kept). The exact MAE value may vary depending on library version and hardware::

>>> import random, numpy as np, torch
>>> from torch import manual_seed
>>> from river import datasets, metrics
>>> from deep_river.regression.zoo import LSTMRegressor
>>> _ = manual_seed(42); random.seed(42); np.random.seed(42)
>>> first_x, _ = next(iter(datasets.Bikes()))
>>> numeric_keys = sorted([k for k,v in first_x.items() if isinstance(v,(int,float))])
>>> reg = LSTMRegressor(
...     n_features=len(numeric_keys), hidden_size=8, num_layers=1,
...     optimizer_fn='sgd', lr=1e-2, is_feature_incremental=True,
... )
>>> mae = metrics.MAE()
>>> for i, (x, y) in enumerate(datasets.Bikes().take(200)):
...     x_num = {k: x[k] for k in numeric_keys}
...     if i > 0:
...         y_pred = reg.predict_one(x_num)
...         mae.update(y, y_pred)
...     reg.learn_one(x_num, y)
>>> assert 0.0 <= mae.get() < 20.0
>>> print(f"MAE: {mae.get():.4f}")  # doctest: +ELLIPSIS
MAE: ...

Methods:

Name Description
clone

Return a fresh estimator instance with (optionally) copied state.

draw

Render a (partial) computational graph of the wrapped model.

learn_many

Batch update with multiple samples using the rolling window.

learn_one

Update model using a single (x, y) and current rolling window.

load

Load a previously saved estimator.

predict_many

Predict targets for multiple samples (appends to a copy of the window).

predict_one

Predict a single regression target using rolling context.

save

Persist the estimator (architecture, weights, optimiser & runtime state).

Source code in deep_river/regression/zoo.py
def __init__(
    self,
    n_features: int = 10,
    hidden_size: int = 32,
    num_layers: int = 1,
    dropout: float = 0.0,
    gradient_clip_value: float | None = 1.0,
    loss_fn: Union[str, Callable] = "mse",
    optimizer_fn: Union[str, Type[optim.Optimizer]] = "adam",
    lr: float = 1e-3,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs,
):
    self.n_features = n_features
    self.hidden_size = hidden_size
    self.num_layers = num_layers
    self.dropout = dropout
    self.gradient_clip_value = gradient_clip_value
    module = LSTMRegressor.LSTMModule(
        n_features=n_features,
        hidden_size=hidden_size,
        num_layers=num_layers,
        dropout=dropout,
    )
    if "module" in kwargs:
        del kwargs["module"]
    super().__init__(
        module=module,
        loss_fn=loss_fn,
        optimizer_fn=optimizer_fn,
        is_feature_incremental=is_feature_incremental,
        device=device,
        lr=lr,
        seed=seed,
        gradient_clip_value=gradient_clip_value,
        **kwargs,
    )

clone

clone(
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
)

Return a fresh estimator instance with (optionally) copied state.

Parameters:

Name Type Description Default
new_params dict | None

Parameter overrides for the cloned instance.

None
include_attributes bool

If True, runtime state (observed features, buffers) is also copied.

False
copy_weights bool

If True, model weights are copied (otherwise the module is re‑initialised).

False
Source code in deep_river/base.py
def clone(
    self,
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
):
    """Return a fresh estimator instance with (optionally) copied state.

    Parameters
    ----------
    new_params : dict | None
        Parameter overrides for the cloned instance.
    include_attributes : bool, default=False
        If True, runtime state (observed features, buffers) is also copied.
    copy_weights : bool, default=False
        If True, model weights are copied (otherwise the module is re‑initialised).
    """
    new_params = new_params or {}
    copy_weights = new_params.pop("copy_weights", copy_weights)

    params = {**self._get_all_init_params(), **new_params}

    if "module" not in new_params:
        params["module"] = self._rebuild_module()

    new_est = self.__class__(**self._filter_kwargs(self.__class__.__init__, params))

    if copy_weights and hasattr(self.module, "state_dict"):
        new_est.module.load_state_dict(self.module.state_dict())

    if include_attributes:
        new_est._restore_runtime_state(self._get_runtime_state())

    return new_est

draw

draw()

Render a (partial) computational graph of the wrapped model.

Imports graphviz and torchviz lazily. Raises an informative ImportError if the optional dependencies are not installed.

Source code in deep_river/base.py
def draw(self):  # type: ignore[override]
    """Render a (partial) computational graph of the wrapped model.

    Imports ``graphviz`` and ``torchviz`` lazily. Raises an informative
    ImportError if the optional dependencies are not installed.
    """
    try:  # pragma: no cover
        from torchviz import make_dot  # type: ignore
    except Exception as err:  # noqa: BLE001
        raise ImportError(
            "graphviz and torchviz must be installed to draw the model."
        ) from err

    first_parameter = next(self.module.parameters())
    input_shape = first_parameter.size()
    y_pred = self.module(torch.rand(input_shape))
    return make_dot(y_pred.mean(), params=dict(self.module.named_parameters()))

learn_many

learn_many(X: DataFrame, y: Series) -> None

Batch update with multiple samples using the rolling window.

Only performs an optimisation step once the internal window has reached window_size length to ensure a full sequence is available.

Source code in deep_river/regression/rolling_regressor.py
def learn_many(self, X: pd.DataFrame, y: pd.Series) -> None:
    """Batch update with multiple samples using the rolling window.

    Only performs an optimisation step once the internal window has reached
    ``window_size`` length to ensure a full sequence is available.
    """
    self._update_observed_features(X)

    X = X[list(self.observed_features)]
    self._x_window.extend(X.values.tolist())

    if len(self._x_window) == self.window_size:
        X_t = self._deque2rolling_tensor(self._x_window)

        # Convert y to tensor (ensuring proper shape for regression)
        y_t = torch.tensor(y.values, dtype=torch.float32, device=self.device).view(
            -1, 1
        )

        self._learn(x=X_t, y=y_t)

learn_one

learn_one(x: dict, y: RegTarget, **kwargs) -> None

Update model using a single (x, y) and current rolling window.

Parameters:

Name Type Description Default
x dict

Feature mapping.

required
y float

Target value.

required
Source code in deep_river/regression/rolling_regressor.py
def learn_one(self, x: dict, y: base.typing.RegTarget, **kwargs) -> None:
    """Update model using a single (x, y) and current rolling window.

    Parameters
    ----------
    x : dict
        Feature mapping.
    y : float
        Target value.
    """
    self._update_observed_features(x)

    self._x_window.append([x.get(feature, 0) for feature in self.observed_features])

    x_t = self._deque2rolling_tensor(self._x_window)

    # Convert y to tensor (ensuring proper shape for regression)
    y_t = torch.tensor([y], dtype=torch.float32, device=self.device).view(-1, 1)

    self._learn(x=x_t, y=y_t)

load classmethod

load(filepath: Union[str, Path])

Load a previously saved estimator.

The method reconstructs the estimator class, its wrapped module, optimiser state and runtime information (feature names, buffers, etc.).

Source code in deep_river/base.py
@classmethod
def load(cls, filepath: Union[str, Path]):
    """Load a previously saved estimator.

    The method reconstructs the estimator class, its wrapped module, optimiser
    state and runtime information (feature names, buffers, etc.).
    """
    with open(filepath, "rb") as f:
        state = pickle.load(f)

    estimator_cls = cls._import_from_path(state["estimator_class"])
    init_params = state["init_params"]

    # Rebuild module if needed
    if "module" in init_params and isinstance(init_params["module"], dict):
        module_info = init_params.pop("module")
        module_cls = cls._import_from_path(module_info["class"])
        module = module_cls(
            **cls._filter_kwargs(module_cls.__init__, module_info["kwargs"])
        )
        if state.get("model_state_dict"):
            module.load_state_dict(state["model_state_dict"])
        init_params["module"] = module

    estimator = estimator_cls(
        **cls._filter_kwargs(estimator_cls.__init__, init_params)
    )

    if state.get("optimizer_state_dict") and hasattr(estimator, "optimizer"):
        try:
            estimator.optimizer.load_state_dict(
                state["optimizer_state_dict"]  # type: ignore[arg-type]
            )
        except Exception:  # noqa: E722
            pass

    estimator._restore_runtime_state(state.get("runtime_state", {}))
    return estimator

predict_many

predict_many(X: DataFrame) -> DataFrame

Predict targets for multiple samples (appends to a copy of the window).

Returns a single-column DataFrame named 'y_pred'.

Source code in deep_river/regression/rolling_regressor.py
def predict_many(self, X: pd.DataFrame) -> pd.DataFrame:
    """Predict targets for multiple samples (appends to a copy of the window).

    Returns a single-column DataFrame named ``'y_pred'``.
    """

    self._update_observed_features(X)
    X = X[list(self.observed_features)]
    x_win = self._x_window.copy()
    x_win.extend(X.values.tolist())
    if self.append_predict:
        self._x_window = x_win

    self.module.eval()
    with torch.inference_mode():
        x_t = self._deque2rolling_tensor(x_win)
        y_preds = self.module(x_t)
        if isinstance(y_preds, torch.Tensor):
            y_preds = y_preds.detach().cpu().view(-1).numpy().tolist()

    return pd.DataFrame({"y_pred": y_preds})

predict_one

predict_one(x: dict) -> RegTarget

Predict a single regression target using rolling context.

Parameters:

Name Type Description Default
x dict

Feature mapping.

required

Returns:

Type Description
float

Predicted target value.

Source code in deep_river/regression/rolling_regressor.py
def predict_one(self, x: dict) -> base.typing.RegTarget:
    """Predict a single regression target using rolling context.

    Parameters
    ----------
    x : dict
        Feature mapping.

    Returns
    -------
    float
        Predicted target value.
    """
    self._update_observed_features(x)

    x_win = self._x_window.copy()
    x_win.append([x.get(feature, 0) for feature in self.observed_features])
    if self.append_predict:
        self._x_window = x_win

    self.module.eval()
    with torch.inference_mode():
        x_t = self._deque2rolling_tensor(x_win)
        y_pred = self.module(x_t)
        if isinstance(y_pred, torch.Tensor):
            y_pred = y_pred.detach().view(-1)[-1].cpu().numpy().item()
        else:
            y_pred = float(y_pred)

    return y_pred

save

save(filepath: Union[str, Path]) -> None

Persist the estimator (architecture, weights, optimiser & runtime state).

Parameters:

Name Type Description Default
filepath str | Path

Destination file. Parent directories are created automatically.

required
Source code in deep_river/base.py
def save(self, filepath: Union[str, Path]) -> None:
    """Persist the estimator (architecture, weights, optimiser & runtime state).

    Parameters
    ----------
    filepath : str | Path
        Destination file. Parent directories are created automatically.
    """
    filepath = Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)

    state = {
        "estimator_class": f"{type(self).__module__}.{type(self).__name__}",
        "init_params": self._get_all_init_params(),
        "model_state_dict": getattr(self.module, "state_dict", lambda: {})(),
        "optimizer_state_dict": getattr(self.optimizer, "state_dict", lambda: {})(),
        "runtime_state": self._get_runtime_state(),
    }

    with open(filepath, "wb") as f:
        pickle.dump(state, f)

LinearRegression

LinearRegression(
    n_features: int = 10,
    loss_fn: Union[str, Callable] = "mse",
    optimizer_fn: Union[str, Type[Optimizer]] = "sgd",
    lr: float = 0.001,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    gradient_clip_value: float | None = 1.0,
    **kwargs
)

Bases: Regressor

Incremental linear regression with optional feature growth and gradient clipping.

A thin wrapper that instantiates a single linear layer and enables dynamic feature expansion when is_feature_incremental=True. The model outputs a single continuous target value.

Parameters:

Name Type Description Default
n_features int

Initial number of input features (columns). The input layer can expand if feature incrementality is enabled and new feature names appear.

10
loss_fn str | Callable

Loss used for optimisation.

'mse'
optimizer_fn str | type

Optimizer specification.

'sgd'
lr float

Learning rate.

1e-3
is_feature_incremental bool

Whether to expand the input layer when new features appear.

False
device str

Torch device.

'cpu'
seed int

Random seed.

42
gradient_clip_value float | None

Gradient norm clipping threshold. Disabled if None.

None
**kwargs

Forwarded to :class:~deep_river.base.DeepEstimator.

{}

Examples:

Streaming regression on the Bikes dataset (only numeric features kept).
The exact MAE value may vary depending on library version and hardware::

>>> import random, numpy as np, torch
>>> from torch import manual_seed
>>> from river import datasets, metrics
>>> from deep_river.regression.zoo import LinearRegression
>>> _ = manual_seed(42); random.seed(42); np.random.seed(42)
>>> first_x, _ = next(iter(datasets.Bikes()))
>>> numeric_keys = sorted([k for k,v in first_x.items() if isinstance(v,(int,float))])
>>> reg = LinearRegression(n_features=len(numeric_keys),
...                        loss_fn='mse', lr=1e-2,
...                        is_feature_incremental=True)
>>> mae = metrics.MAE()
>>> for i, (x, y) in enumerate(datasets.Bikes().take(200)):
...     x_num = {k: x[k] for k in numeric_keys}
...     if i > 0:
...         y_pred = reg.predict_one(x_num)
...         mae.update(y, y_pred)
...     reg.learn_one(x_num, y)
>>> assert 0.0 <= mae.get() < 20.0
>>> print(f"MAE: {mae.get():.4f}")  # doctest: +ELLIPSIS
MAE: ...

Methods:

Name Description
clone

Return a fresh estimator instance with (optionally) copied state.

draw

Render a (partial) computational graph of the wrapped model.

load

Load a previously saved estimator.

predict_many

Predict target values for multiple instances (returns single-column DataFrame).

predict_one

Predict target value for a single instance.

save

Persist the estimator (architecture, weights, optimiser & runtime state).

Source code in deep_river/regression/zoo.py
def __init__(
    self,
    n_features: int = 10,
    loss_fn: Union[str, Callable] = "mse",
    optimizer_fn: Union[str, Type[optim.Optimizer]] = "sgd",
    lr: float = 1e-3,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    gradient_clip_value: float | None = 1.0,
    **kwargs,
):
    self.n_features = n_features
    module = LinearRegression.LRModule(n_features=n_features)
    if "module" in kwargs:
        del kwargs["module"]
    super().__init__(
        module=module,
        loss_fn=loss_fn,
        optimizer_fn=optimizer_fn,
        is_feature_incremental=is_feature_incremental,
        device=device,
        lr=lr,
        seed=seed,
        gradient_clip_value=gradient_clip_value,
        **kwargs,
    )

clone

clone(
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
)

Return a fresh estimator instance with (optionally) copied state.

Parameters:

Name Type Description Default
new_params dict | None

Parameter overrides for the cloned instance.

None
include_attributes bool

If True, runtime state (observed features, buffers) is also copied.

False
copy_weights bool

If True, model weights are copied (otherwise the module is re‑initialised).

False
Source code in deep_river/base.py
def clone(
    self,
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
):
    """Return a fresh estimator instance with (optionally) copied state.

    Parameters
    ----------
    new_params : dict | None
        Parameter overrides for the cloned instance.
    include_attributes : bool, default=False
        If True, runtime state (observed features, buffers) is also copied.
    copy_weights : bool, default=False
        If True, model weights are copied (otherwise the module is re‑initialised).
    """
    new_params = new_params or {}
    copy_weights = new_params.pop("copy_weights", copy_weights)

    params = {**self._get_all_init_params(), **new_params}

    if "module" not in new_params:
        params["module"] = self._rebuild_module()

    new_est = self.__class__(**self._filter_kwargs(self.__class__.__init__, params))

    if copy_weights and hasattr(self.module, "state_dict"):
        new_est.module.load_state_dict(self.module.state_dict())

    if include_attributes:
        new_est._restore_runtime_state(self._get_runtime_state())

    return new_est

draw

draw()

Render a (partial) computational graph of the wrapped model.

Imports graphviz and torchviz lazily. Raises an informative ImportError if the optional dependencies are not installed.

Source code in deep_river/base.py
def draw(self):  # type: ignore[override]
    """Render a (partial) computational graph of the wrapped model.

    Imports ``graphviz`` and ``torchviz`` lazily. Raises an informative
    ImportError if the optional dependencies are not installed.
    """
    try:  # pragma: no cover
        from torchviz import make_dot  # type: ignore
    except Exception as err:  # noqa: BLE001
        raise ImportError(
            "graphviz and torchviz must be installed to draw the model."
        ) from err

    first_parameter = next(self.module.parameters())
    input_shape = first_parameter.size()
    y_pred = self.module(torch.rand(input_shape))
    return make_dot(y_pred.mean(), params=dict(self.module.named_parameters()))

load classmethod

load(filepath: Union[str, Path])

Load a previously saved estimator.

The method reconstructs the estimator class, its wrapped module, optimiser state and runtime information (feature names, buffers, etc.).

Source code in deep_river/base.py
@classmethod
def load(cls, filepath: Union[str, Path]):
    """Load a previously saved estimator.

    The method reconstructs the estimator class, its wrapped module, optimiser
    state and runtime information (feature names, buffers, etc.).
    """
    with open(filepath, "rb") as f:
        state = pickle.load(f)

    estimator_cls = cls._import_from_path(state["estimator_class"])
    init_params = state["init_params"]

    # Rebuild module if needed
    if "module" in init_params and isinstance(init_params["module"], dict):
        module_info = init_params.pop("module")
        module_cls = cls._import_from_path(module_info["class"])
        module = module_cls(
            **cls._filter_kwargs(module_cls.__init__, module_info["kwargs"])
        )
        if state.get("model_state_dict"):
            module.load_state_dict(state["model_state_dict"])
        init_params["module"] = module

    estimator = estimator_cls(
        **cls._filter_kwargs(estimator_cls.__init__, init_params)
    )

    if state.get("optimizer_state_dict") and hasattr(estimator, "optimizer"):
        try:
            estimator.optimizer.load_state_dict(
                state["optimizer_state_dict"]  # type: ignore[arg-type]
            )
        except Exception:  # noqa: E722
            pass

    estimator._restore_runtime_state(state.get("runtime_state", {}))
    return estimator

predict_many

predict_many(X: DataFrame) -> DataFrame

Predict target values for multiple instances (returns single-column DataFrame).

Source code in deep_river/regression/regressor.py
def predict_many(self, X: pd.DataFrame) -> pd.DataFrame:
    """Predict target values for multiple instances (returns single-column DataFrame)."""
    self._update_observed_features(X)
    x_t = self._df2tensor(X)
    self.module.eval()
    with torch.inference_mode():
        y_preds = self.module(x_t)
    return pd.DataFrame(y_preds if not y_preds.is_cuda else y_preds.cpu().numpy())

predict_one

predict_one(x: dict) -> RegTarget

Predict target value for a single instance.

Source code in deep_river/regression/regressor.py
def predict_one(self, x: dict) -> RegTarget:
    """Predict target value for a single instance."""
    self._update_observed_features(x)
    x_t = self._dict2tensor(x)
    self.module.eval()
    with torch.inference_mode():
        y_pred = self.module(x_t).item()
    return y_pred

save

save(filepath: Union[str, Path]) -> None

Persist the estimator (architecture, weights, optimiser & runtime state).

Parameters:

Name Type Description Default
filepath str | Path

Destination file. Parent directories are created automatically.

required
Source code in deep_river/base.py
def save(self, filepath: Union[str, Path]) -> None:
    """Persist the estimator (architecture, weights, optimiser & runtime state).

    Parameters
    ----------
    filepath : str | Path
        Destination file. Parent directories are created automatically.
    """
    filepath = Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)

    state = {
        "estimator_class": f"{type(self).__module__}.{type(self).__name__}",
        "init_params": self._get_all_init_params(),
        "model_state_dict": getattr(self.module, "state_dict", lambda: {})(),
        "optimizer_state_dict": getattr(self.optimizer, "state_dict", lambda: {})(),
        "runtime_state": self._get_runtime_state(),
    }

    with open(filepath, "wb") as f:
        pickle.dump(state, f)

MultiLayerPerceptron

MultiLayerPerceptron(
    n_features: int = 10,
    n_width: int = 5,
    n_layers: int = 5,
    loss_fn: Union[str, Callable] = "mse",
    optimizer_fn: Union[str, Type[Optimizer]] = "sgd",
    lr: float = 0.001,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    gradient_clip_value: float | None = None,
    **kwargs
)

Bases: Regressor

Multi-layer perceptron regressor with optional feature growth.

Stacks n_layers fully connected layers of width n_width with a sigmoid non-linearity (kept for backward compatibility) followed by a single output unit. Can expand its input layer when new feature names appear.

Parameters:

Name Type Description Default
n_features int

Initial number of input features.

10
n_width int

Hidden layer width.

5
n_layers int

Number of hidden layers. Must be >=1.

5
loss_fn Union[str, Callable]

Standard estimator configuration.

'mse'
optimizer_fn Union[str, Callable]

Standard estimator configuration.

'mse'
lr Union[str, Callable]

Standard estimator configuration.

'mse'
is_feature_incremental Union[str, Callable]

Standard estimator configuration.

'mse'
device Union[str, Callable]

Standard estimator configuration.

'mse'
seed Union[str, Callable]

Standard estimator configuration.

'mse'
gradient_clip_value Union[str, Callable]

Standard estimator configuration.

'mse'
**kwargs Union[str, Callable]

Standard estimator configuration.

'mse'
Notes

The use of sigmoid after each hidden layer can cause saturation; for deeper networks consider replacing with ReLU or GELU in a custom module.

Examples:

Streaming regression on the Bikes dataset (only numeric features kept). The exact MAE value may vary depending on library version and hardware::

>>> import random, numpy as np, torch
>>> from torch import manual_seed
>>> from river import datasets, metrics
>>> from deep_river.regression.zoo import MultiLayerPerceptron
>>> _ = manual_seed(42); random.seed(42); np.random.seed(42)
>>> first_x, _ = next(iter(datasets.Bikes()))
>>> numeric_keys = sorted([k for k,v in first_x.items() if isinstance(v,(int,float))])
>>> reg = MultiLayerPerceptron(
...     n_features=len(numeric_keys), n_width=8, n_layers=2,
...     optimizer_fn='sgd', lr=1e-2, is_feature_incremental=True,
... )
>>> mae = metrics.MAE()
>>> for i, (x, y) in enumerate(datasets.Bikes().take(200)):
...     x_num = {k: x[k] for k in numeric_keys}
...     if i > 0:
...         y_pred = reg.predict_one(x_num)
...         mae.update(y, y_pred)
...     reg.learn_one(x_num, y)
>>> assert 0.0 <= mae.get() < 20.0
>>> print(f"MAE: {mae.get():.4f}")  # doctest: +ELLIPSIS
MAE: ...

Methods:

Name Description
clone

Return a fresh estimator instance with (optionally) copied state.

draw

Render a (partial) computational graph of the wrapped model.

load

Load a previously saved estimator.

predict_many

Predict target values for multiple instances (returns single-column DataFrame).

predict_one

Predict target value for a single instance.

save

Persist the estimator (architecture, weights, optimiser & runtime state).

Source code in deep_river/regression/zoo.py
def __init__(
    self,
    n_features: int = 10,
    n_width: int = 5,
    n_layers: int = 5,
    loss_fn: Union[str, Callable] = "mse",
    optimizer_fn: Union[str, Type[optim.Optimizer]] = "sgd",
    lr: float = 1e-3,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    gradient_clip_value: float | None = None,
    **kwargs,
):
    self.n_features = n_features
    self.n_width = n_width
    self.n_layers = n_layers
    module = MultiLayerPerceptron.MLPModule(
        n_features=n_features, n_layers=n_layers, n_width=n_width
    )
    if "module" in kwargs:
        del kwargs["module"]
    super().__init__(
        module=module,
        loss_fn=loss_fn,
        optimizer_fn=optimizer_fn,
        is_feature_incremental=is_feature_incremental,
        device=device,
        lr=lr,
        seed=seed,
        gradient_clip_value=gradient_clip_value,
        **kwargs,
    )

clone

clone(
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
)

Return a fresh estimator instance with (optionally) copied state.

Parameters:

Name Type Description Default
new_params dict | None

Parameter overrides for the cloned instance.

None
include_attributes bool

If True, runtime state (observed features, buffers) is also copied.

False
copy_weights bool

If True, model weights are copied (otherwise the module is re‑initialised).

False
Source code in deep_river/base.py
def clone(
    self,
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
):
    """Return a fresh estimator instance with (optionally) copied state.

    Parameters
    ----------
    new_params : dict | None
        Parameter overrides for the cloned instance.
    include_attributes : bool, default=False
        If True, runtime state (observed features, buffers) is also copied.
    copy_weights : bool, default=False
        If True, model weights are copied (otherwise the module is re‑initialised).
    """
    new_params = new_params or {}
    copy_weights = new_params.pop("copy_weights", copy_weights)

    params = {**self._get_all_init_params(), **new_params}

    if "module" not in new_params:
        params["module"] = self._rebuild_module()

    new_est = self.__class__(**self._filter_kwargs(self.__class__.__init__, params))

    if copy_weights and hasattr(self.module, "state_dict"):
        new_est.module.load_state_dict(self.module.state_dict())

    if include_attributes:
        new_est._restore_runtime_state(self._get_runtime_state())

    return new_est

draw

draw()

Render a (partial) computational graph of the wrapped model.

Imports graphviz and torchviz lazily. Raises an informative ImportError if the optional dependencies are not installed.

Source code in deep_river/base.py
def draw(self):  # type: ignore[override]
    """Render a (partial) computational graph of the wrapped model.

    Imports ``graphviz`` and ``torchviz`` lazily. Raises an informative
    ImportError if the optional dependencies are not installed.
    """
    try:  # pragma: no cover
        from torchviz import make_dot  # type: ignore
    except Exception as err:  # noqa: BLE001
        raise ImportError(
            "graphviz and torchviz must be installed to draw the model."
        ) from err

    first_parameter = next(self.module.parameters())
    input_shape = first_parameter.size()
    y_pred = self.module(torch.rand(input_shape))
    return make_dot(y_pred.mean(), params=dict(self.module.named_parameters()))

load classmethod

load(filepath: Union[str, Path])

Load a previously saved estimator.

The method reconstructs the estimator class, its wrapped module, optimiser state and runtime information (feature names, buffers, etc.).

Source code in deep_river/base.py
@classmethod
def load(cls, filepath: Union[str, Path]):
    """Load a previously saved estimator.

    The method reconstructs the estimator class, its wrapped module, optimiser
    state and runtime information (feature names, buffers, etc.).
    """
    with open(filepath, "rb") as f:
        state = pickle.load(f)

    estimator_cls = cls._import_from_path(state["estimator_class"])
    init_params = state["init_params"]

    # Rebuild module if needed
    if "module" in init_params and isinstance(init_params["module"], dict):
        module_info = init_params.pop("module")
        module_cls = cls._import_from_path(module_info["class"])
        module = module_cls(
            **cls._filter_kwargs(module_cls.__init__, module_info["kwargs"])
        )
        if state.get("model_state_dict"):
            module.load_state_dict(state["model_state_dict"])
        init_params["module"] = module

    estimator = estimator_cls(
        **cls._filter_kwargs(estimator_cls.__init__, init_params)
    )

    if state.get("optimizer_state_dict") and hasattr(estimator, "optimizer"):
        try:
            estimator.optimizer.load_state_dict(
                state["optimizer_state_dict"]  # type: ignore[arg-type]
            )
        except Exception:  # noqa: E722
            pass

    estimator._restore_runtime_state(state.get("runtime_state", {}))
    return estimator

predict_many

predict_many(X: DataFrame) -> DataFrame

Predict target values for multiple instances (returns single-column DataFrame).

Source code in deep_river/regression/regressor.py
def predict_many(self, X: pd.DataFrame) -> pd.DataFrame:
    """Predict target values for multiple instances (returns single-column DataFrame)."""
    self._update_observed_features(X)
    x_t = self._df2tensor(X)
    self.module.eval()
    with torch.inference_mode():
        y_preds = self.module(x_t)
    return pd.DataFrame(y_preds if not y_preds.is_cuda else y_preds.cpu().numpy())

predict_one

predict_one(x: dict) -> RegTarget

Predict target value for a single instance.

Source code in deep_river/regression/regressor.py
def predict_one(self, x: dict) -> RegTarget:
    """Predict target value for a single instance."""
    self._update_observed_features(x)
    x_t = self._dict2tensor(x)
    self.module.eval()
    with torch.inference_mode():
        y_pred = self.module(x_t).item()
    return y_pred

save

save(filepath: Union[str, Path]) -> None

Persist the estimator (architecture, weights, optimiser & runtime state).

Parameters:

Name Type Description Default
filepath str | Path

Destination file. Parent directories are created automatically.

required
Source code in deep_river/base.py
def save(self, filepath: Union[str, Path]) -> None:
    """Persist the estimator (architecture, weights, optimiser & runtime state).

    Parameters
    ----------
    filepath : str | Path
        Destination file. Parent directories are created automatically.
    """
    filepath = Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)

    state = {
        "estimator_class": f"{type(self).__module__}.{type(self).__name__}",
        "init_params": self._get_all_init_params(),
        "model_state_dict": getattr(self.module, "state_dict", lambda: {})(),
        "optimizer_state_dict": getattr(self.optimizer, "state_dict", lambda: {})(),
        "runtime_state": self._get_runtime_state(),
    }

    with open(filepath, "wb") as f:
        pickle.dump(state, f)

RNNRegressor

RNNRegressor(
    n_features: int = 10,
    hidden_size: int = 32,
    num_layers: int = 1,
    nonlinearity: str = "tanh",
    dropout: float = 0.0,
    gradient_clip_value: float | None = 1.0,
    loss_fn: Union[str, Callable] = "mse",
    optimizer_fn: Union[str, Type[Optimizer]] = "adam",
    lr: float = 0.001,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs
)

Bases: RollingRegressor

Rolling RNN regressor for sequential / time-series data.

Uses a nn.RNN backbone and a linear head to output a single continuous target. Leverages the rolling window maintained by :class:RollingRegressor to feed the last window_size observations as a sequence.

Parameters:

Name Type Description Default
n_features int

Number of input features per timestep.

10
hidden_size int

Hidden state dimensionality of the RNN.

32
num_layers int

Number of stacked RNN layers.

1
nonlinearity str

Non-linearity used inside the RNN ('tanh' or 'relu').

'tanh'
dropout float

Dropout applied after extracting the last hidden state (no internal RNN dropout).

0.0
gradient_clip_value float | None

Gradient norm clipping threshold. None disables clipping.

1.0
loss_fn Union[str, Callable]

Standard configuration as in other regressors.

'mse'
optimizer_fn Union[str, Callable]

Standard configuration as in other regressors.

'mse'
lr Union[str, Callable]

Standard configuration as in other regressors.

'mse'
is_feature_incremental Union[str, Callable]

Standard configuration as in other regressors.

'mse'
device Union[str, Callable]

Standard configuration as in other regressors.

'mse'
seed Union[str, Callable]

Standard configuration as in other regressors.

'mse'
**kwargs Union[str, Callable]

Standard configuration as in other regressors.

'mse'

Examples:

Streaming regression on the Bikes dataset (only numeric features kept). The exact MAE value may vary depending on library version and hardware::

>>> import random, numpy as np, torch
>>> from torch import manual_seed
>>> from river import datasets, metrics
>>> from deep_river.regression.zoo import RNNRegressor
>>> _ = manual_seed(42); random.seed(42); np.random.seed(42)
>>> first_x, _ = next(iter(datasets.Bikes()))
>>> numeric_keys = sorted([k for k,v in first_x.items() if isinstance(v,(int,float))])
>>> reg = RNNRegressor(
...     n_features=len(numeric_keys), hidden_size=8, num_layers=1,
...     optimizer_fn='sgd', lr=1e-2, is_feature_incremental=True,
... )
>>> mae = metrics.MAE()
>>> for i, (x, y) in enumerate(datasets.Bikes().take(200)):
...     x_num = {k: x[k] for k in numeric_keys}
...     if i > 0:
...         y_pred = reg.predict_one(x_num)
...         mae.update(y, y_pred)
...     reg.learn_one(x_num, y)
>>> assert 0.0 <= mae.get() < 20.0
>>> print(f"MAE: {mae.get():.4f}")  # doctest: +ELLIPSIS
MAE: ...

Methods:

Name Description
clone

Return a fresh estimator instance with (optionally) copied state.

draw

Render a (partial) computational graph of the wrapped model.

learn_many

Batch update with multiple samples using the rolling window.

learn_one

Update model using a single (x, y) and current rolling window.

load

Load a previously saved estimator.

predict_many

Predict targets for multiple samples (appends to a copy of the window).

predict_one

Predict a single regression target using rolling context.

save

Persist the estimator (architecture, weights, optimiser & runtime state).

Source code in deep_river/regression/zoo.py
def __init__(
    self,
    n_features: int = 10,
    hidden_size: int = 32,
    num_layers: int = 1,
    nonlinearity: str = "tanh",
    dropout: float = 0.0,
    gradient_clip_value: float | None = 1.0,
    loss_fn: Union[str, Callable] = "mse",
    optimizer_fn: Union[str, Type[optim.Optimizer]] = "adam",
    lr: float = 1e-3,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs,
):
    self.n_features = n_features
    self.hidden_size = hidden_size
    self.num_layers = num_layers
    self.nonlinearity = nonlinearity
    self.dropout = dropout
    module = RNNRegressor.RNNModule(
        n_features=n_features,
        hidden_size=hidden_size,
        num_layers=num_layers,
        nonlinearity=nonlinearity,
        dropout=dropout,
    )
    if "module" in kwargs:
        del kwargs["module"]
    super().__init__(
        module=module,
        loss_fn=loss_fn,
        optimizer_fn=optimizer_fn,
        is_feature_incremental=is_feature_incremental,
        device=device,
        lr=lr,
        seed=seed,
        gradient_clip_value=gradient_clip_value,
        **kwargs,
    )

clone

clone(
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
)

Return a fresh estimator instance with (optionally) copied state.

Parameters:

Name Type Description Default
new_params dict | None

Parameter overrides for the cloned instance.

None
include_attributes bool

If True, runtime state (observed features, buffers) is also copied.

False
copy_weights bool

If True, model weights are copied (otherwise the module is re‑initialised).

False
Source code in deep_river/base.py
def clone(
    self,
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
):
    """Return a fresh estimator instance with (optionally) copied state.

    Parameters
    ----------
    new_params : dict | None
        Parameter overrides for the cloned instance.
    include_attributes : bool, default=False
        If True, runtime state (observed features, buffers) is also copied.
    copy_weights : bool, default=False
        If True, model weights are copied (otherwise the module is re‑initialised).
    """
    new_params = new_params or {}
    copy_weights = new_params.pop("copy_weights", copy_weights)

    params = {**self._get_all_init_params(), **new_params}

    if "module" not in new_params:
        params["module"] = self._rebuild_module()

    new_est = self.__class__(**self._filter_kwargs(self.__class__.__init__, params))

    if copy_weights and hasattr(self.module, "state_dict"):
        new_est.module.load_state_dict(self.module.state_dict())

    if include_attributes:
        new_est._restore_runtime_state(self._get_runtime_state())

    return new_est

draw

draw()

Render a (partial) computational graph of the wrapped model.

Imports graphviz and torchviz lazily. Raises an informative ImportError if the optional dependencies are not installed.

Source code in deep_river/base.py
def draw(self):  # type: ignore[override]
    """Render a (partial) computational graph of the wrapped model.

    Imports ``graphviz`` and ``torchviz`` lazily. Raises an informative
    ImportError if the optional dependencies are not installed.
    """
    try:  # pragma: no cover
        from torchviz import make_dot  # type: ignore
    except Exception as err:  # noqa: BLE001
        raise ImportError(
            "graphviz and torchviz must be installed to draw the model."
        ) from err

    first_parameter = next(self.module.parameters())
    input_shape = first_parameter.size()
    y_pred = self.module(torch.rand(input_shape))
    return make_dot(y_pred.mean(), params=dict(self.module.named_parameters()))

learn_many

learn_many(X: DataFrame, y: Series) -> None

Batch update with multiple samples using the rolling window.

Only performs an optimisation step once the internal window has reached window_size length to ensure a full sequence is available.

Source code in deep_river/regression/rolling_regressor.py
def learn_many(self, X: pd.DataFrame, y: pd.Series) -> None:
    """Batch update with multiple samples using the rolling window.

    Only performs an optimisation step once the internal window has reached
    ``window_size`` length to ensure a full sequence is available.
    """
    self._update_observed_features(X)

    X = X[list(self.observed_features)]
    self._x_window.extend(X.values.tolist())

    if len(self._x_window) == self.window_size:
        X_t = self._deque2rolling_tensor(self._x_window)

        # Convert y to tensor (ensuring proper shape for regression)
        y_t = torch.tensor(y.values, dtype=torch.float32, device=self.device).view(
            -1, 1
        )

        self._learn(x=X_t, y=y_t)

learn_one

learn_one(x: dict, y: RegTarget, **kwargs) -> None

Update model using a single (x, y) and current rolling window.

Parameters:

Name Type Description Default
x dict

Feature mapping.

required
y float

Target value.

required
Source code in deep_river/regression/rolling_regressor.py
def learn_one(self, x: dict, y: base.typing.RegTarget, **kwargs) -> None:
    """Update model using a single (x, y) and current rolling window.

    Parameters
    ----------
    x : dict
        Feature mapping.
    y : float
        Target value.
    """
    self._update_observed_features(x)

    self._x_window.append([x.get(feature, 0) for feature in self.observed_features])

    x_t = self._deque2rolling_tensor(self._x_window)

    # Convert y to tensor (ensuring proper shape for regression)
    y_t = torch.tensor([y], dtype=torch.float32, device=self.device).view(-1, 1)

    self._learn(x=x_t, y=y_t)

load classmethod

load(filepath: Union[str, Path])

Load a previously saved estimator.

The method reconstructs the estimator class, its wrapped module, optimiser state and runtime information (feature names, buffers, etc.).

Source code in deep_river/base.py
@classmethod
def load(cls, filepath: Union[str, Path]):
    """Load a previously saved estimator.

    The method reconstructs the estimator class, its wrapped module, optimiser
    state and runtime information (feature names, buffers, etc.).
    """
    with open(filepath, "rb") as f:
        state = pickle.load(f)

    estimator_cls = cls._import_from_path(state["estimator_class"])
    init_params = state["init_params"]

    # Rebuild module if needed
    if "module" in init_params and isinstance(init_params["module"], dict):
        module_info = init_params.pop("module")
        module_cls = cls._import_from_path(module_info["class"])
        module = module_cls(
            **cls._filter_kwargs(module_cls.__init__, module_info["kwargs"])
        )
        if state.get("model_state_dict"):
            module.load_state_dict(state["model_state_dict"])
        init_params["module"] = module

    estimator = estimator_cls(
        **cls._filter_kwargs(estimator_cls.__init__, init_params)
    )

    if state.get("optimizer_state_dict") and hasattr(estimator, "optimizer"):
        try:
            estimator.optimizer.load_state_dict(
                state["optimizer_state_dict"]  # type: ignore[arg-type]
            )
        except Exception:  # noqa: E722
            pass

    estimator._restore_runtime_state(state.get("runtime_state", {}))
    return estimator

predict_many

predict_many(X: DataFrame) -> DataFrame

Predict targets for multiple samples (appends to a copy of the window).

Returns a single-column DataFrame named 'y_pred'.

Source code in deep_river/regression/rolling_regressor.py
def predict_many(self, X: pd.DataFrame) -> pd.DataFrame:
    """Predict targets for multiple samples (appends to a copy of the window).

    Returns a single-column DataFrame named ``'y_pred'``.
    """

    self._update_observed_features(X)
    X = X[list(self.observed_features)]
    x_win = self._x_window.copy()
    x_win.extend(X.values.tolist())
    if self.append_predict:
        self._x_window = x_win

    self.module.eval()
    with torch.inference_mode():
        x_t = self._deque2rolling_tensor(x_win)
        y_preds = self.module(x_t)
        if isinstance(y_preds, torch.Tensor):
            y_preds = y_preds.detach().cpu().view(-1).numpy().tolist()

    return pd.DataFrame({"y_pred": y_preds})

predict_one

predict_one(x: dict) -> RegTarget

Predict a single regression target using rolling context.

Parameters:

Name Type Description Default
x dict

Feature mapping.

required

Returns:

Type Description
float

Predicted target value.

Source code in deep_river/regression/rolling_regressor.py
def predict_one(self, x: dict) -> base.typing.RegTarget:
    """Predict a single regression target using rolling context.

    Parameters
    ----------
    x : dict
        Feature mapping.

    Returns
    -------
    float
        Predicted target value.
    """
    self._update_observed_features(x)

    x_win = self._x_window.copy()
    x_win.append([x.get(feature, 0) for feature in self.observed_features])
    if self.append_predict:
        self._x_window = x_win

    self.module.eval()
    with torch.inference_mode():
        x_t = self._deque2rolling_tensor(x_win)
        y_pred = self.module(x_t)
        if isinstance(y_pred, torch.Tensor):
            y_pred = y_pred.detach().view(-1)[-1].cpu().numpy().item()
        else:
            y_pred = float(y_pred)

    return y_pred

save

save(filepath: Union[str, Path]) -> None

Persist the estimator (architecture, weights, optimiser & runtime state).

Parameters:

Name Type Description Default
filepath str | Path

Destination file. Parent directories are created automatically.

required
Source code in deep_river/base.py
def save(self, filepath: Union[str, Path]) -> None:
    """Persist the estimator (architecture, weights, optimiser & runtime state).

    Parameters
    ----------
    filepath : str | Path
        Destination file. Parent directories are created automatically.
    """
    filepath = Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)

    state = {
        "estimator_class": f"{type(self).__module__}.{type(self).__name__}",
        "init_params": self._get_all_init_params(),
        "model_state_dict": getattr(self.module, "state_dict", lambda: {})(),
        "optimizer_state_dict": getattr(self.optimizer, "state_dict", lambda: {})(),
        "runtime_state": self._get_runtime_state(),
    }

    with open(filepath, "wb") as f:
        pickle.dump(state, f)