Skip to content

forecasting

This module contains forecasters for the deep_river package.

Modules:

Name Description
forecaster
zoo

Classes:

Name Description
DeepForecaster

Incremental PyTorch forecaster compatible with River's time-series API.
GRUForecaster

Autoregressive forecaster backed by torch.nn.GRU.
LSTMForecaster

Autoregressive forecaster backed by torch.nn.LSTM.
LinearForecaster

Autoregressive linear forecaster with optional exogenous features.
LiquidForecaster

Autoregressive forecaster backed by closed-form liquid recurrent cells.
MLPForecaster

Autoregressive multi-layer perceptron forecaster.
NBEATSForecaster

Compact N-BEATS-style residual MLP forecaster for online point forecasts.
RNNForecaster

Autoregressive forecaster backed by torch.nn.RNN.

DeepForecaster 

DeepForecaster(
    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,
    is_sequence_model: bool = False,
    gradient_clip_value: float | None = 1.0,
    **kwargs
)

Bases: DeepEstimator, Forecaster

Incremental PyTorch forecaster compatible with River's time-series API.

DeepForecaster learns a one-step-ahead model from the recent endogenous target history and optional exogenous features. Multi-step forecasts are produced autoregressively by feeding each predicted value back into a copied history buffer.

Parameters:

Name Type Description Default
module Module

PyTorch module. In flat mode it receives a tensor of shape (1, window_size + n_exogenous_features). In sequence mode it receives (window_size, 1, 1 + n_exogenous_features).

 required
window_size int

Number of past target values used as autoregressive context.

 10
is_sequence_model bool

Whether module expects sequence-shaped input.

 False
loss_fn Union[str, Callable]
'mse'
optimizer_fn Union[str, Callable]
'mse'
lr Union[str, Callable]
'mse'
is_feature_incremental Union[str, Callable]
'mse'
device Union[str, Callable]
'mse'
seed Union[str, Callable]
'mse'
gradient_clip_value float | None

Standard deep-river estimator configuration.

 1.0
**kwargs float | None

Standard deep-river estimator configuration.

 1.0

Methods:

Name Description
clone

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

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

Forecast horizon steps ahead using autoregressive rollouts.
learn_one

Update the model with one target and optional exogenous features.
load

Load a previously saved estimator.
save

Persist the estimator (architecture, weights, optimiser & runtime state).
Source code in deep_river/forecasting/forecaster.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    module: 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,
    is_sequence_model: bool = False,
    gradient_clip_value: float | None = 1.0,
    **kwargs,
):
    if window_size < 1:
        raise ValueError("window_size must be at least 1")

    self.window_size = window_size
    self.is_sequence_model = is_sequence_model
    self._y_window: Deque[float] = collections.deque(maxlen=window_size)

    super().__init__(
        module=module,
        loss_fn=loss_fn,
        optimizer_fn=optimizer_fn,
        lr=lr,
        is_feature_incremental=is_feature_incremental,
        device=device,
        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()))

forecast 

forecast(
    horizon: int, xs: list[dict] | None = None
) -> list

Forecast horizon steps ahead using autoregressive rollouts.

Source code in deep_river/forecasting/forecaster.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def forecast(self, horizon: int, xs: list[dict] | None = None) -> list:
    """Forecast ``horizon`` steps ahead using autoregressive rollouts."""
    if horizon < 0:
        raise ValueError("horizon must be non-negative")
    if xs is None:
        xs = [{} for _ in range(horizon)]
    if len(xs) != horizon:
        raise ValueError(
            "the length of xs should be equal to the specified horizon"
        )

    y_window = collections.deque(self._y_window, maxlen=self.window_size)
    forecasts = []

    self.module.eval()
    with torch.inference_mode():
        for x in xs:
            x = x or {}
            self._update_observed_features(x)
            x_t = self._make_input_tensor(y_window, x)
            y_pred = self.module(x_t)
            if isinstance(y_pred, torch.Tensor):
                y_pred = y_pred.detach().view(-1)[-1].cpu().item()
            else:
                y_pred = float(y_pred)
            forecasts.append(y_pred)
            y_window.append(float(y_pred))

    return forecasts

learn_one 

learn_one(y: float, x: dict | None = None) -> None

Update the model with one target and optional exogenous features.

Source code in deep_river/forecasting/forecaster.py 
79
80
81
82
83
84
85
86
def learn_one(self, y: float, x: dict | None = None) -> None:
    """Update the model with one target and optional exogenous features."""
    x = x or {}
    self._update_observed_features(x)
    x_t = self._make_input_tensor(self._y_window, x)
    y_t = torch.tensor([[y]], dtype=torch.float32, device=self.device)
    self._learn(x_t, y_t)
    self._y_window.append(float(y))

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

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)

GRUForecaster 

GRUForecaster(
    n_features: int = 0,
    window_size: int = 10,
    hidden_size: int = 16,
    num_layers: int = 1,
    dropout: float = 0.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,
    gradient_clip_value: float | None = 1.0,
    **kwargs
)

Bases: DeepForecaster

Autoregressive forecaster backed by torch.nn.GRU.

Methods:

Name Description
clone

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

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

Forecast horizon steps ahead using autoregressive rollouts.
learn_one

Update the model with one target and optional exogenous features.
load

Load a previously saved estimator.
save

Persist the estimator (architecture, weights, optimiser & runtime state).
Source code in deep_river/forecasting/zoo.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def __init__(
    self,
    n_features: int = 0,
    window_size: int = 10,
    hidden_size: int = 16,
    num_layers: int = 1,
    dropout: float = 0.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,
    gradient_clip_value: float | None = 1.0,
    **kwargs,
):
    self.n_features = n_features
    self.hidden_size = hidden_size
    self.num_layers = num_layers
    self.dropout = dropout
    torch.manual_seed(seed)
    module = GRUForecaster.GRUModule(
        input_size=1 + n_features,
        hidden_size=hidden_size,
        num_layers=num_layers,
        dropout=dropout,
    )
    kwargs.pop("module", None)
    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,
        is_sequence_model=True,
        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()))

forecast 

forecast(
    horizon: int, xs: list[dict] | None = None
) -> list

Forecast horizon steps ahead using autoregressive rollouts.

Source code in deep_river/forecasting/forecaster.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def forecast(self, horizon: int, xs: list[dict] | None = None) -> list:
    """Forecast ``horizon`` steps ahead using autoregressive rollouts."""
    if horizon < 0:
        raise ValueError("horizon must be non-negative")
    if xs is None:
        xs = [{} for _ in range(horizon)]
    if len(xs) != horizon:
        raise ValueError(
            "the length of xs should be equal to the specified horizon"
        )

    y_window = collections.deque(self._y_window, maxlen=self.window_size)
    forecasts = []

    self.module.eval()
    with torch.inference_mode():
        for x in xs:
            x = x or {}
            self._update_observed_features(x)
            x_t = self._make_input_tensor(y_window, x)
            y_pred = self.module(x_t)
            if isinstance(y_pred, torch.Tensor):
                y_pred = y_pred.detach().view(-1)[-1].cpu().item()
            else:
                y_pred = float(y_pred)
            forecasts.append(y_pred)
            y_window.append(float(y_pred))

    return forecasts

learn_one 

learn_one(y: float, x: dict | None = None) -> None

Update the model with one target and optional exogenous features.

Source code in deep_river/forecasting/forecaster.py 
79
80
81
82
83
84
85
86
def learn_one(self, y: float, x: dict | None = None) -> None:
    """Update the model with one target and optional exogenous features."""
    x = x or {}
    self._update_observed_features(x)
    x_t = self._make_input_tensor(self._y_window, x)
    y_t = torch.tensor([[y]], dtype=torch.float32, device=self.device)
    self._learn(x_t, y_t)
    self._y_window.append(float(y))

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

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)

LSTMForecaster 

LSTMForecaster(
    n_features: int = 0,
    window_size: int = 10,
    hidden_size: int = 16,
    num_layers: int = 1,
    dropout: float = 0.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,
    gradient_clip_value: float | None = 1.0,
    **kwargs
)

Bases: DeepForecaster

Autoregressive forecaster backed by torch.nn.LSTM.

Methods:

Name Description
clone

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

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

Forecast horizon steps ahead using autoregressive rollouts.
learn_one

Update the model with one target and optional exogenous features.
load

Load a previously saved estimator.
save

Persist the estimator (architecture, weights, optimiser & runtime state).
Source code in deep_river/forecasting/zoo.py 
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
343
344
345
346
347
348
349
350
351
352
353
def __init__(
    self,
    n_features: int = 0,
    window_size: int = 10,
    hidden_size: int = 16,
    num_layers: int = 1,
    dropout: float = 0.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,
    gradient_clip_value: float | None = 1.0,
    **kwargs,
):
    self.n_features = n_features
    self.hidden_size = hidden_size
    self.num_layers = num_layers
    self.dropout = dropout
    torch.manual_seed(seed)
    module = LSTMForecaster.LSTMModule(
        input_size=1 + n_features,
        hidden_size=hidden_size,
        num_layers=num_layers,
        dropout=dropout,
    )
    kwargs.pop("module", None)
    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,
        is_sequence_model=True,
        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()))

forecast 

forecast(
    horizon: int, xs: list[dict] | None = None
) -> list

Forecast horizon steps ahead using autoregressive rollouts.

Source code in deep_river/forecasting/forecaster.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def forecast(self, horizon: int, xs: list[dict] | None = None) -> list:
    """Forecast ``horizon`` steps ahead using autoregressive rollouts."""
    if horizon < 0:
        raise ValueError("horizon must be non-negative")
    if xs is None:
        xs = [{} for _ in range(horizon)]
    if len(xs) != horizon:
        raise ValueError(
            "the length of xs should be equal to the specified horizon"
        )

    y_window = collections.deque(self._y_window, maxlen=self.window_size)
    forecasts = []

    self.module.eval()
    with torch.inference_mode():
        for x in xs:
            x = x or {}
            self._update_observed_features(x)
            x_t = self._make_input_tensor(y_window, x)
            y_pred = self.module(x_t)
            if isinstance(y_pred, torch.Tensor):
                y_pred = y_pred.detach().view(-1)[-1].cpu().item()
            else:
                y_pred = float(y_pred)
            forecasts.append(y_pred)
            y_window.append(float(y_pred))

    return forecasts

learn_one 

learn_one(y: float, x: dict | None = None) -> None

Update the model with one target and optional exogenous features.

Source code in deep_river/forecasting/forecaster.py 
79
80
81
82
83
84
85
86
def learn_one(self, y: float, x: dict | None = None) -> None:
    """Update the model with one target and optional exogenous features."""
    x = x or {}
    self._update_observed_features(x)
    x_t = self._make_input_tensor(self._y_window, x)
    y_t = torch.tensor([[y]], dtype=torch.float32, device=self.device)
    self._learn(x_t, y_t)
    self._y_window.append(float(y))

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

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)

LinearForecaster 

LinearForecaster(
    n_features: int = 0,
    window_size: 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: DeepForecaster

Autoregressive linear forecaster with optional exogenous features.

Methods:

Name Description
clone

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

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

Forecast horizon steps ahead using autoregressive rollouts.
learn_one

Update the model with one target and optional exogenous features.
load

Load a previously saved estimator.
save

Persist the estimator (architecture, weights, optimiser & runtime state).
Source code in deep_river/forecasting/zoo.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(
    self,
    n_features: int = 0,
    window_size: 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
    torch.manual_seed(seed)
    module = LinearForecaster.LinearModule(input_size=window_size + n_features)
    kwargs.pop("module", None)
    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,
        is_sequence_model=False,
        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()))

forecast 

forecast(
    horizon: int, xs: list[dict] | None = None
) -> list

Forecast horizon steps ahead using autoregressive rollouts.

Source code in deep_river/forecasting/forecaster.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def forecast(self, horizon: int, xs: list[dict] | None = None) -> list:
    """Forecast ``horizon`` steps ahead using autoregressive rollouts."""
    if horizon < 0:
        raise ValueError("horizon must be non-negative")
    if xs is None:
        xs = [{} for _ in range(horizon)]
    if len(xs) != horizon:
        raise ValueError(
            "the length of xs should be equal to the specified horizon"
        )

    y_window = collections.deque(self._y_window, maxlen=self.window_size)
    forecasts = []

    self.module.eval()
    with torch.inference_mode():
        for x in xs:
            x = x or {}
            self._update_observed_features(x)
            x_t = self._make_input_tensor(y_window, x)
            y_pred = self.module(x_t)
            if isinstance(y_pred, torch.Tensor):
                y_pred = y_pred.detach().view(-1)[-1].cpu().item()
            else:
                y_pred = float(y_pred)
            forecasts.append(y_pred)
            y_window.append(float(y_pred))

    return forecasts

learn_one 

learn_one(y: float, x: dict | None = None) -> None

Update the model with one target and optional exogenous features.

Source code in deep_river/forecasting/forecaster.py 
79
80
81
82
83
84
85
86
def learn_one(self, y: float, x: dict | None = None) -> None:
    """Update the model with one target and optional exogenous features."""
    x = x or {}
    self._update_observed_features(x)
    x_t = self._make_input_tensor(self._y_window, x)
    y_t = torch.tensor([[y]], dtype=torch.float32, device=self.device)
    self._learn(x_t, y_t)
    self._y_window.append(float(y))

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

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)

LiquidForecaster 

LiquidForecaster(
    n_features: int = 0,
    window_size: int = 10,
    hidden_size: int = 16,
    num_layers: int = 1,
    dropout: float = 0.0,
    time_delta: float = 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,
    gradient_clip_value: float | None = 1.0,
    **kwargs
)

Bases: DeepForecaster

Autoregressive forecaster backed by closed-form liquid recurrent cells.

The recurrent state follows a fixed-step liquid update where each hidden unit learns a positive time constant. time_delta is constant across all steps, matching regularly sampled time series.

Methods:

Name Description
clone

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

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

Forecast horizon steps ahead using autoregressive rollouts.
learn_one

Update the model with one target and optional exogenous features.
load

Load a previously saved estimator.
save

Persist the estimator (architecture, weights, optimiser & runtime state).
Source code in deep_river/forecasting/zoo.py 
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
479
480
481
482
483
484
485
486
487
488
489
def __init__(
    self,
    n_features: int = 0,
    window_size: int = 10,
    hidden_size: int = 16,
    num_layers: int = 1,
    dropout: float = 0.0,
    time_delta: float = 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,
    gradient_clip_value: float | None = 1.0,
    **kwargs,
):
    if is_feature_incremental:
        raise ValueError("LiquidForecaster does not support feature incrementality")
    if num_layers < 1:
        raise ValueError("num_layers must be at least 1")
    if time_delta <= 0:
        raise ValueError("time_delta must be positive")

    self.n_features = n_features
    self.hidden_size = hidden_size
    self.num_layers = num_layers
    self.dropout = dropout
    self.time_delta = time_delta
    torch.manual_seed(seed)
    module = LiquidForecaster.LiquidModule(
        input_size=1 + n_features,
        hidden_size=hidden_size,
        num_layers=num_layers,
        dropout=dropout,
        time_delta=time_delta,
    )
    kwargs.pop("module", None)
    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,
        is_sequence_model=True,
        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()))

forecast 

forecast(
    horizon: int, xs: list[dict] | None = None
) -> list

Forecast horizon steps ahead using autoregressive rollouts.

Source code in deep_river/forecasting/forecaster.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def forecast(self, horizon: int, xs: list[dict] | None = None) -> list:
    """Forecast ``horizon`` steps ahead using autoregressive rollouts."""
    if horizon < 0:
        raise ValueError("horizon must be non-negative")
    if xs is None:
        xs = [{} for _ in range(horizon)]
    if len(xs) != horizon:
        raise ValueError(
            "the length of xs should be equal to the specified horizon"
        )

    y_window = collections.deque(self._y_window, maxlen=self.window_size)
    forecasts = []

    self.module.eval()
    with torch.inference_mode():
        for x in xs:
            x = x or {}
            self._update_observed_features(x)
            x_t = self._make_input_tensor(y_window, x)
            y_pred = self.module(x_t)
            if isinstance(y_pred, torch.Tensor):
                y_pred = y_pred.detach().view(-1)[-1].cpu().item()
            else:
                y_pred = float(y_pred)
            forecasts.append(y_pred)
            y_window.append(float(y_pred))

    return forecasts

learn_one 

learn_one(y: float, x: dict | None = None) -> None

Update the model with one target and optional exogenous features.

Source code in deep_river/forecasting/forecaster.py 
79
80
81
82
83
84
85
86
def learn_one(self, y: float, x: dict | None = None) -> None:
    """Update the model with one target and optional exogenous features."""
    x = x or {}
    self._update_observed_features(x)
    x_t = self._make_input_tensor(self._y_window, x)
    y_t = torch.tensor([[y]], dtype=torch.float32, device=self.device)
    self._learn(x_t, y_t)
    self._y_window.append(float(y))

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

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)

MLPForecaster 

MLPForecaster(
    n_features: int = 0,
    window_size: int = 10,
    n_width: int = 16,
    n_layers: int = 2,
    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,
    gradient_clip_value: float | None = 1.0,
    **kwargs
)

Bases: DeepForecaster

Autoregressive multi-layer perceptron forecaster.

Methods:

Name Description
clone

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

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

Forecast horizon steps ahead using autoregressive rollouts.
learn_one

Update the model with one target and optional exogenous features.
load

Load a previously saved estimator.
save

Persist the estimator (architecture, weights, optimiser & runtime state).
Source code in deep_river/forecasting/zoo.py 
 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
101
102
103
104
105
106
107
108
109
110
111
112
113
def __init__(
    self,
    n_features: int = 0,
    window_size: int = 10,
    n_width: int = 16,
    n_layers: int = 2,
    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,
    gradient_clip_value: float | None = 1.0,
    **kwargs,
):
    self.n_features = n_features
    self.n_width = n_width
    self.n_layers = n_layers
    torch.manual_seed(seed)
    module = MLPForecaster.MLPModule(
        input_size=window_size + n_features,
        n_width=n_width,
        n_layers=n_layers,
    )
    kwargs.pop("module", None)
    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,
        is_sequence_model=False,
        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()))

forecast 

forecast(
    horizon: int, xs: list[dict] | None = None
) -> list

Forecast horizon steps ahead using autoregressive rollouts.

Source code in deep_river/forecasting/forecaster.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def forecast(self, horizon: int, xs: list[dict] | None = None) -> list:
    """Forecast ``horizon`` steps ahead using autoregressive rollouts."""
    if horizon < 0:
        raise ValueError("horizon must be non-negative")
    if xs is None:
        xs = [{} for _ in range(horizon)]
    if len(xs) != horizon:
        raise ValueError(
            "the length of xs should be equal to the specified horizon"
        )

    y_window = collections.deque(self._y_window, maxlen=self.window_size)
    forecasts = []

    self.module.eval()
    with torch.inference_mode():
        for x in xs:
            x = x or {}
            self._update_observed_features(x)
            x_t = self._make_input_tensor(y_window, x)
            y_pred = self.module(x_t)
            if isinstance(y_pred, torch.Tensor):
                y_pred = y_pred.detach().view(-1)[-1].cpu().item()
            else:
                y_pred = float(y_pred)
            forecasts.append(y_pred)
            y_window.append(float(y_pred))

    return forecasts

learn_one 

learn_one(y: float, x: dict | None = None) -> None

Update the model with one target and optional exogenous features.

Source code in deep_river/forecasting/forecaster.py 
79
80
81
82
83
84
85
86
def learn_one(self, y: float, x: dict | None = None) -> None:
    """Update the model with one target and optional exogenous features."""
    x = x or {}
    self._update_observed_features(x)
    x_t = self._make_input_tensor(self._y_window, x)
    y_t = torch.tensor([[y]], dtype=torch.float32, device=self.device)
    self._learn(x_t, y_t)
    self._y_window.append(float(y))

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

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)

NBEATSForecaster 

NBEATSForecaster(
    n_features: int = 0,
    window_size: int = 10,
    n_width: int = 32,
    n_layers: int = 2,
    n_blocks: int = 2,
    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,
    gradient_clip_value: float | None = 1.0,
    **kwargs
)

Bases: DeepForecaster

Compact N-BEATS-style residual MLP forecaster for online point forecasts.

Methods:

Name Description
clone

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

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

Forecast horizon steps ahead using autoregressive rollouts.
learn_one

Update the model with one target and optional exogenous features.
load

Load a previously saved estimator.
save

Persist the estimator (architecture, weights, optimiser & runtime state).
Source code in deep_river/forecasting/zoo.py 
555
556
557
558
559
560
561
562
563
564
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 __init__(
    self,
    n_features: int = 0,
    window_size: int = 10,
    n_width: int = 32,
    n_layers: int = 2,
    n_blocks: int = 2,
    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,
    gradient_clip_value: float | None = 1.0,
    **kwargs,
):
    if is_feature_incremental:
        raise ValueError("NBEATSForecaster does not support feature incrementality")

    self.n_features = n_features
    self.n_width = n_width
    self.n_layers = n_layers
    self.n_blocks = n_blocks
    torch.manual_seed(seed)
    module = NBEATSForecaster.NBEATSModule(
        input_size=window_size + n_features,
        n_width=n_width,
        n_layers=n_layers,
        n_blocks=n_blocks,
    )
    kwargs.pop("module", None)
    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,
        is_sequence_model=False,
        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()))

forecast 

forecast(
    horizon: int, xs: list[dict] | None = None
) -> list

Forecast horizon steps ahead using autoregressive rollouts.

Source code in deep_river/forecasting/forecaster.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def forecast(self, horizon: int, xs: list[dict] | None = None) -> list:
    """Forecast ``horizon`` steps ahead using autoregressive rollouts."""
    if horizon < 0:
        raise ValueError("horizon must be non-negative")
    if xs is None:
        xs = [{} for _ in range(horizon)]
    if len(xs) != horizon:
        raise ValueError(
            "the length of xs should be equal to the specified horizon"
        )

    y_window = collections.deque(self._y_window, maxlen=self.window_size)
    forecasts = []

    self.module.eval()
    with torch.inference_mode():
        for x in xs:
            x = x or {}
            self._update_observed_features(x)
            x_t = self._make_input_tensor(y_window, x)
            y_pred = self.module(x_t)
            if isinstance(y_pred, torch.Tensor):
                y_pred = y_pred.detach().view(-1)[-1].cpu().item()
            else:
                y_pred = float(y_pred)
            forecasts.append(y_pred)
            y_window.append(float(y_pred))

    return forecasts

learn_one 

learn_one(y: float, x: dict | None = None) -> None

Update the model with one target and optional exogenous features.

Source code in deep_river/forecasting/forecaster.py 
79
80
81
82
83
84
85
86
def learn_one(self, y: float, x: dict | None = None) -> None:
    """Update the model with one target and optional exogenous features."""
    x = x or {}
    self._update_observed_features(x)
    x_t = self._make_input_tensor(self._y_window, x)
    y_t = torch.tensor([[y]], dtype=torch.float32, device=self.device)
    self._learn(x_t, y_t)
    self._y_window.append(float(y))

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

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)

RNNForecaster 

RNNForecaster(
    n_features: int = 0,
    window_size: int = 10,
    hidden_size: int = 16,
    num_layers: int = 1,
    nonlinearity: str = "tanh",
    dropout: float = 0.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,
    gradient_clip_value: float | None = 1.0,
    **kwargs
)

Bases: DeepForecaster

Autoregressive forecaster backed by torch.nn.RNN.

Methods:

Name Description
clone

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

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

Forecast horizon steps ahead using autoregressive rollouts.
learn_one

Update the model with one target and optional exogenous features.
load

Load a previously saved estimator.
save

Persist the estimator (architecture, weights, optimiser & runtime state).
Source code in deep_river/forecasting/zoo.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def __init__(
    self,
    n_features: int = 0,
    window_size: int = 10,
    hidden_size: int = 16,
    num_layers: int = 1,
    nonlinearity: str = "tanh",
    dropout: float = 0.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,
    gradient_clip_value: float | None = 1.0,
    **kwargs,
):
    self.n_features = n_features
    self.hidden_size = hidden_size
    self.num_layers = num_layers
    self.nonlinearity = nonlinearity
    self.dropout = dropout
    torch.manual_seed(seed)
    module = RNNForecaster.RNNModule(
        input_size=1 + n_features,
        hidden_size=hidden_size,
        num_layers=num_layers,
        nonlinearity=nonlinearity,
        dropout=dropout,
    )
    kwargs.pop("module", None)
    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,
        is_sequence_model=True,
        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()))

forecast 

forecast(
    horizon: int, xs: list[dict] | None = None
) -> list

Forecast horizon steps ahead using autoregressive rollouts.

Source code in deep_river/forecasting/forecaster.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def forecast(self, horizon: int, xs: list[dict] | None = None) -> list:
    """Forecast ``horizon`` steps ahead using autoregressive rollouts."""
    if horizon < 0:
        raise ValueError("horizon must be non-negative")
    if xs is None:
        xs = [{} for _ in range(horizon)]
    if len(xs) != horizon:
        raise ValueError(
            "the length of xs should be equal to the specified horizon"
        )

    y_window = collections.deque(self._y_window, maxlen=self.window_size)
    forecasts = []

    self.module.eval()
    with torch.inference_mode():
        for x in xs:
            x = x or {}
            self._update_observed_features(x)
            x_t = self._make_input_tensor(y_window, x)
            y_pred = self.module(x_t)
            if isinstance(y_pred, torch.Tensor):
                y_pred = y_pred.detach().view(-1)[-1].cpu().item()
            else:
                y_pred = float(y_pred)
            forecasts.append(y_pred)
            y_window.append(float(y_pred))

    return forecasts

learn_one 

learn_one(y: float, x: dict | None = None) -> None

Update the model with one target and optional exogenous features.

Source code in deep_river/forecasting/forecaster.py 
79
80
81
82
83
84
85
86
def learn_one(self, y: float, x: dict | None = None) -> None:
    """Update the model with one target and optional exogenous features."""
    x = x or {}
    self._update_observed_features(x)
    x_t = self._make_input_tensor(self._y_window, x)
    y_t = torch.tensor([[y]], dtype=torch.float32, device=self.device)
    self._learn(x_t, y_t)
    self._y_window.append(float(y))

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

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)