Skip to content

regression

This module contains the regressors for the deep_river package.

Modules:

Name Description
multioutput
regressor
rolling_regressor
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.
MultiTargetRegressor

Incremental multi-target regression wrapper for PyTorch modules.
RNNRegressor

Rolling RNN regressor for sequential / time-series data.
Regressor

Incremental wrapper for PyTorch regression models.
RollingRegressor

Incremental regressor with a fixed-size rolling window.

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 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
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 
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
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 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
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) -> 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 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def learn_one(self, x: dict, y: base.typing.RegTarget) -> 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 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
@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) -> Series

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

Returns a series of predictions.

Source code in deep_river/regression/rolling_regressor.py 
225
226
227
228
229
230
231
232
def predict_many(self, X: pd.DataFrame) -> pd.Series:
    """Predict targets for multiple samples (appends to a copy of the window).

    Returns a series of predictions.
    """

    y_preds = [self.predict_one(row) for row in X.to_dict(orient="records")]
    return pd.Series(y_preds, index=X.index)

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 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
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.
    """
    y_pred = self._rolling_prediction(x)
    if isinstance(y_pred, torch.Tensor):
        y_pred = y_pred.detach().view(-1)[-1].cpu().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 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
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.
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 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
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 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
@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) -> Series

Predict target values for multiple instances.

Source code in deep_river/regression/regressor.py 
139
140
141
142
143
144
145
146
def predict_many(self, X: pd.DataFrame) -> pd.Series:
    """Predict target values for multiple instances."""
    self._update_observed_features(X)
    x_t = self._df2tensor(X)
    self.module.eval()
    with torch.inference_mode():
        y_preds = self.module(x_t).detach().cpu().view(-1).tolist()
    return pd.Series(y_preds, index=X.index)

predict_one 

predict_one(x: dict) -> RegTarget

Predict target value for a single instance.

Source code in deep_river/regression/regressor.py 
130
131
132
133
134
135
136
137
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 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
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.
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 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
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 
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
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 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
@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) -> Series

Predict target values for multiple instances.

Source code in deep_river/regression/regressor.py 
139
140
141
142
143
144
145
146
def predict_many(self, X: pd.DataFrame) -> pd.Series:
    """Predict target values for multiple instances."""
    self._update_observed_features(X)
    x_t = self._df2tensor(X)
    self.module.eval()
    with torch.inference_mode():
        y_preds = self.module(x_t).detach().cpu().view(-1).tolist()
    return pd.Series(y_preds, index=X.index)

predict_one 

predict_one(x: dict) -> RegTarget

Predict target value for a single instance.

Source code in deep_river/regression/regressor.py 
130
131
132
133
134
135
136
137
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 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
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)

MultiTargetRegressor 

MultiTargetRegressor(
    module: Module,
    loss_fn: Union[str, Callable] = "mse",
    optimizer_fn: Union[str, Callable] = "sgd",
    is_feature_incremental: bool = False,
    is_target_incremental: bool = False,
    lr: float = 0.001,
    device: str = "cpu",
    seed: int = 42,
    **kwargs
)

Bases: MultiTargetRegressor, DeepEstimator

Incremental multi-target regression wrapper for PyTorch modules.

This estimator adapts a torch.nn.Module to the :mod:river streaming API for multi-target (a.k.a. multi-output) regression. It optionally supports feature-incremental learning (dynamic growth of the input layer when new feature names appear) as provided by :class:deep_river.base.DeepEstimator and additionally (optionally) target-incremental learning: if new target names appear during the stream, the output layer can be expanded on-the-fly so the model natively handles the enlarged target vector.

Targets are tracked via an ordered :class:~sortedcontainers.SortedSet to guarantee deterministic ordering between training and prediction. Incoming target dictionaries / frames are converted into dense tensors with columns arranged according to the observed target name order. Missing targets (when the model has been expanded but a prior sample omits some target) are imputed with 0.0.

Parameters:

Name Type Description Default
module Module

PyTorch module producing an output tensor of shape (N, T) where T is the current number of target variables.

 required
loss_fn str | Callable

Loss identifier or custom callable passed through :func:deep_river.utils.get_loss_fn.

 'mse'
optimizer_fn str | Callable

Optimizer identifier (e.g. 'adam', 'sgd') or factory / class.

 'sgd'
is_feature_incremental bool

If True, unseen feature names trigger expansion of the first trainable layer (see :class:DeepEstimator).

 False
is_target_incremental bool

If True, unseen target names trigger expansion of the last trainable layer. Expansion preserves existing weights and initialises new units with small random values.

 False
lr float

Learning rate.

 1e-3
device str

Torch device (e.g. 'cuda').

 'cpu'
seed int

Random seed for reproducibility.

 42
**kwargs

Extra arguments stored for persistence / cloning.

 {}

Examples:

>>> import torch
>>> from torch import nn
>>> from deep_river.regression.multioutput import MultiTargetRegressor
>>> class TinyMultiNet(nn.Module):
...     def __init__(self, n_features, n_outputs):
...         super().__init__()
...         self.net = nn.Sequential(
...             nn.Linear(n_features, 8),
...             nn.ReLU(),
...             nn.Linear(8, n_outputs)
...         )
...     def forward(self, x):
...         return self.net(x)
>>> model = MultiTargetRegressor(
...     module=TinyMultiNet(3, 2),
...     loss_fn='mse',
...     optimizer_fn='sgd',
...     is_feature_incremental=True,
...     is_target_incremental=True,
... )
>>> x = {'a': 1.0, 'b': 2.0, 'c': 3.0}
>>> y = {'y1': 10.0, 'y2': 20.0}
>>> _ = model.learn_one(x, y)
>>> model.predict_one(x)
{'y1': ..., 'y2': ...}
Notes
  • The module's last trainable leaf layer is treated as output layer for
  • If is_target_incremental is disabled, the number of outputs is fixed and encountering a new target name will only register it internally (the tensor conversion will still allocate a slot, but the model's output layer size will not change, possibly causing a mismatch). Therefore, enabling target incrementality is recommended for truly open-world streams.

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

Learn from a batch of multi-target instances.
learn_one

Learn from a single multi-target instance.
load

Load a previously saved estimator.
predict_many

Predict target values for multiple instances.
predict_one

Predict a dictionary of target values for a single instance.
save

Persist the estimator (architecture, weights, optimiser & runtime state).
Source code in deep_river/regression/multioutput.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def __init__(
    self,
    module: torch.nn.Module,
    loss_fn: Union[str, Callable] = "mse",
    optimizer_fn: Union[str, Callable] = "sgd",
    is_feature_incremental: bool = False,
    is_target_incremental: bool = False,
    lr: float = 1e-3,
    device: str = "cpu",
    seed: int = 42,
    **kwargs,
):
    super().__init__(
        module=module,
        loss_fn=loss_fn,
        optimizer_fn=optimizer_fn,
        lr=lr,
        device=device,
        seed=seed,
        is_feature_incremental=is_feature_incremental,
        **kwargs,
    )
    self.is_target_incremental = is_target_incremental
    self.observed_targets: SortedSet[FeatureName] = SortedSet()

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 
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
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: Union[
        DataFrame, Series, Mapping[str, Sequence[RegTarget]]
    ],
) -> None

Learn from a batch of multi-target instances.

Parameters:

Name Type Description Default
X DataFrame

Feature matrix (rows are samples, columns are feature names).

 required
y DataFrame | Series | mapping

Target matrix. Preferred is a DataFrame with one column per target. A Series is interpreted as one target. A mapping of name -> list is converted into a DataFrame first.

 required
Source code in deep_river/regression/multioutput.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def learn_many(
    self,
    X: pd.DataFrame,
    y: Union[pd.DataFrame, pd.Series, Mapping[str, Sequence[RegTarget]]],
) -> None:
    """Learn from a batch of multi-target instances.

    Parameters
    ----------
    X : pandas.DataFrame
        Feature matrix (rows are samples, columns are feature names).
    y : pandas.DataFrame | pandas.Series | mapping
        Target matrix. Preferred is a DataFrame with one column per target.
        A Series is interpreted as *one* target. A mapping of ``name -> list``
        is converted into a DataFrame first.
    """
    self._update_observed_features(X)
    y_df = self._coerce_targets_to_frame(y)
    self._update_observed_targets(y_df)

    x_t = self._df2tensor(X)
    y_t = self._multi_target_frame_to_tensor(y_df)
    self._learn(x_t, y_t)

learn_one 

learn_one(
    x: dict[Hashable, Any],
    y: dict[Hashable, RegTarget],
    **kwargs: Any
) -> None

Learn from a single multi-target instance.

Parameters:

Name Type Description Default
x dict[str, float]

Feature mapping.

 required
y dict[str, float]

Mapping of target name -> target value.

 required
Source code in deep_river/regression/multioutput.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def learn_one(
    self,
    x: dict[Hashable, Any],
    y: dict[Hashable, RegTarget],
    **kwargs: Any,
) -> None:
    """Learn from a single multi-target instance.

    Parameters
    ----------
    x : dict[str, float]
        Feature mapping.
    y : dict[str, float]
        Mapping of target name -> target value.
    """
    self._update_observed_features(x)
    self._update_observed_targets(y)
    x_t = self._dict2tensor(dict(x))
    y_t = self._single_target_dict_to_tensor(y)
    self._learn(x_t, 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 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
@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:

Type Description
DataFrame

DataFrame whose columns follow the ordering of observed_targets.
Source code in deep_river/regression/multioutput.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def predict_many(self, X: pd.DataFrame) -> pd.DataFrame:
    """Predict target values for multiple instances.

    Returns
    -------
    pandas.DataFrame
        DataFrame whose columns follow the ordering of ``observed_targets``.
    """
    self._update_observed_features(X)
    x_t = self._df2tensor(X)
    self.module.eval()
    with torch.inference_mode():
        y_pred = self.module(x_t)
        if y_pred.is_cuda:
            y_pred = y_pred.cpu()
    if y_pred.dim() == 1:
        y_pred = y_pred.view(-1, 1)
    cols = list(self.observed_targets)
    if y_pred.shape[1] < len(cols):
        pad = torch.zeros(
            (y_pred.shape[0], len(cols) - y_pred.shape[1]),
            dtype=y_pred.dtype,
        )
        y_pred = torch.cat([y_pred, pad], dim=1)
    elif y_pred.shape[1] > len(cols):
        extra = [f"__extra_{i}" for i in range(y_pred.shape[1] - len(cols))]
        cols = cols + extra
    return pd.DataFrame(y_pred.tolist(), columns=cols, index=X.index)

predict_one 

predict_one(x: dict) -> dict[FeatureName, RegTarget]

Predict a dictionary of target values for a single instance.

Source code in deep_river/regression/multioutput.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def predict_one(self, x: dict) -> dict[FeatureName, RegTarget]:
    """Predict a dictionary of target values for a single instance."""
    self._update_observed_features(x)
    x_t = self._dict2tensor(dict(x))
    self.module.eval()
    with torch.inference_mode():
        y_pred_t = self.module(x_t).squeeze(0)
        if y_pred_t.dim() == 0:
            y_pred_t = y_pred_t.view(1)
        if y_pred_t.is_cuda:
            y_pred_t = y_pred_t.cpu()
        y_list: list[float] = [float(v) for v in y_pred_t.tolist()]
    return {
        cast(FeatureName, t): cast(
            RegTarget, (y_list[i] if i < len(y_list) else float("nan"))
        )
        for i, t in enumerate(self.observed_targets)
    }

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 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
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 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
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 
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
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 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
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) -> 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 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def learn_one(self, x: dict, y: base.typing.RegTarget) -> 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 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
@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) -> Series

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

Returns a series of predictions.

Source code in deep_river/regression/rolling_regressor.py 
225
226
227
228
229
230
231
232
def predict_many(self, X: pd.DataFrame) -> pd.Series:
    """Predict targets for multiple samples (appends to a copy of the window).

    Returns a series of predictions.
    """

    y_preds = [self.predict_one(row) for row in X.to_dict(orient="records")]
    return pd.Series(y_preds, index=X.index)

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 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
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.
    """
    y_pred = self._rolling_prediction(x)
    if isinstance(y_pred, torch.Tensor):
        y_pred = y_pred.detach().view(-1)[-1].cpu().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 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
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)

Regressor 

Regressor(
    module: Module,
    loss_fn: Union[str, Callable],
    optimizer_fn: Union[str, Type[Optimizer]],
    lr: float = 0.001,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs
)

Bases: DeepEstimator, MiniBatchRegressor

Incremental wrapper for PyTorch regression models.

Provides feature-incremental learning (optional) by expanding the first trainable layer on-the-fly when unseen feature names are encountered. Suitable for streaming / online regression tasks using the :mod:river API.

Parameters:

Name Type Description Default
module Module

PyTorch module that outputs a numeric prediction (shape (N, 1) or (N,)).

 required
loss_fn str | Callable

Loss identifier or callable (e.g. 'mse').

 required
optimizer_fn str | Type[Optimizer]

Optimizer spec ('adam', 'sgd' or optimizer class).

 required
lr float

Learning rate.

 1e-3
is_feature_incremental bool

If True, expands the input layer for new feature names.

 False
device str

Torch device.

 'cpu'
seed int

Random seed for reproducibility.

 42
**kwargs

Extra args stored for cloning/persistence.

 {}

Examples:

Real-world streaming regression on the Bikes dataset from :mod:`river`.
We retain only numeric features (discarding timestamps/strings) to build
dense tensors. We maintain an online MAE; the exact value may vary depending
on library version and hardware.

>>> import random, numpy as np
>>> import torch
>>> from torch import nn, manual_seed
>>> from river import datasets, metrics
>>> from deep_river.regression import Regressor
>>> _ = 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))])
>>> class SmallNet(nn.Module):
...     def __init__(self, n_features):
...         super().__init__()
...         self.net = nn.Sequential(
...             nn.Linear(n_features, 8),
...             nn.ReLU(),
...             nn.Linear(8, 1)
...         )
...     def forward(self, x):
...         return self.net(x)
>>> model = Regressor(module=SmallNet(len(numeric_keys)), loss_fn='mse',
...                     optimizer_fn='sgd', lr=1e-2)
>>> mae = metrics.MAE()
>>> for i, (x, y) in enumerate(datasets.Bikes().take(200)):
...     x_num = {k: x[k] for k in numeric_keys}
...     y_pred = model.predict_one(x_num)
...     model.learn_one(x_num, y)
...     mae.update(y, y_pred)
>>> print(f"MAE: {mae.get():.4f}")
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.
predict_one

Predict target value for a single instance.
save

Persist the estimator (architecture, weights, optimiser & runtime state).
Source code in deep_river/regression/regressor.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def __init__(
    self,
    module: nn.Module,
    loss_fn: Union[str, Callable],
    optimizer_fn: Union[str, Type[optim.Optimizer]],
    lr: float = 0.001,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs,
):
    super().__init__(
        module=module,
        loss_fn=loss_fn,
        optimizer_fn=optimizer_fn,
        device=device,
        lr=lr,
        is_feature_incremental=is_feature_incremental,
        seed=seed,
        **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 
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
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 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
@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) -> Series

Predict target values for multiple instances.

Source code in deep_river/regression/regressor.py 
139
140
141
142
143
144
145
146
def predict_many(self, X: pd.DataFrame) -> pd.Series:
    """Predict target values for multiple instances."""
    self._update_observed_features(X)
    x_t = self._df2tensor(X)
    self.module.eval()
    with torch.inference_mode():
        y_preds = self.module(x_t).detach().cpu().view(-1).tolist()
    return pd.Series(y_preds, index=X.index)

predict_one 

predict_one(x: dict) -> RegTarget

Predict target value for a single instance.

Source code in deep_river/regression/regressor.py 
130
131
132
133
134
135
136
137
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 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
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)

RollingRegressor 

RollingRegressor(
    module: Module,
    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,
    window_size: int = 10,
    append_predict: bool = False,
    **kwargs
)

Bases: RollingDeepEstimator, Regressor

Incremental regressor with a fixed-size rolling window.

Maintains the most recent window_size observations in a deque and feeds them as a (sequence_length, batch=1, n_features) tensor to the wrapped PyTorch module. This enables simple sequence style conditioning for models such as RNN/LSTM/GRU without storing the full historical stream.

Parameters:

Name Type Description Default
module Module

Wrapped regression module (expects rolling tensor input shape).

 required
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 first trainable layer when new feature names appear.

 False
device str

Torch device.

 'cpu'
seed int

Random seed.

 42
window_size int

Number of most recent samples kept in the rolling buffer.

 10
append_predict bool

If True, predicted samples (during prediction) are appended to the window enabling simple autoregressive rollouts.

 False
**kwargs

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

 {}

Examples:

Real-world regression example using the Bikes dataset from river. We keep only
the numeric features so the rolling tensor construction succeeds. A small GRU
is trained online and we track a running MAE. The exact value may vary across
library versions and hardware.

>>> import random, numpy as np
>>> from torch import nn, manual_seed
>>> from river import datasets, metrics
>>> from deep_river.regression.rolling_regressor import RollingRegressor
>>> _ = 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))])
>>> class TinySeq(nn.Module):
...     def __init__(self, n_features):
...         super().__init__()
...         self.rnn = nn.GRU(n_features, 8)
...         self.head = nn.Linear(8, 1)
...     def forward(self, x):
...         out, _ = self.rnn(x)
...         return self.head(out[-1])
>>> model = RollingRegressor(module=TinySeq(len(numeric_keys)), window_size=8)
>>> mae = metrics.MAE()
>>> window_size = 8
>>> for i, (x, y) in enumerate(datasets.Bikes().take(200)):
...     x_num = {k: x[k] for k in numeric_keys}
...     if i >= window_size:
...         y_pred = model.predict_one(x_num)
...         mae.update(y, y_pred)
...     model.learn_one(x_num, y)
>>> assert 0.0 <= mae.get() < 15.0
>>> print(f"MAE: {mae.get():.4f}")
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/rolling_regressor.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def __init__(
    self,
    module: torch.nn.Module,
    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,
    window_size: int = 10,
    append_predict: bool = False,
    **kwargs,
):
    super().__init__(
        module=module,
        loss_fn=loss_fn,
        optimizer_fn=optimizer_fn,
        lr=lr,
        is_feature_incremental=is_feature_incremental,
        device=device,
        seed=seed,
        window_size=window_size,
        append_predict=append_predict,
        **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 
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
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 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
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) -> 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 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def learn_one(self, x: dict, y: base.typing.RegTarget) -> 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 
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
@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) -> Series

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

Returns a series of predictions.

Source code in deep_river/regression/rolling_regressor.py 
225
226
227
228
229
230
231
232
def predict_many(self, X: pd.DataFrame) -> pd.Series:
    """Predict targets for multiple samples (appends to a copy of the window).

    Returns a series of predictions.
    """

    y_preds = [self.predict_one(row) for row in X.to_dict(orient="records")]
    return pd.Series(y_preds, index=X.index)

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 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
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.
    """
    y_pred = self._rolling_prediction(x)
    if isinstance(y_pred, torch.Tensor):
        y_pred = y_pred.detach().view(-1)[-1].cpu().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 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
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)