regressor

Classes:

Name	Description
Regressor	Incremental wrapper for PyTorch regression models.

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 (returns single-column DataFrame).
predict_one	Predict target value for a single instance.
save	Persist the estimator (architecture, weights, optimiser & runtime state).

Source code in deep_river/regression/regressor.py

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

def clone(
    self,
    new_params=None,
    include_attributes: bool = False,
    copy_weights: bool = False,
):
    """Return a fresh estimator instance with (optionally) copied state.

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

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

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

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

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

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

    return new_est

draw

draw()

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

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

Source code in deep_river/base.py

def draw(self):  # type: ignore[override]
    """Render a (partial) computational graph of the wrapped model.

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

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

load classmethod

load(filepath: Union[str, Path])

Load a previously saved estimator.

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

Source code in deep_river/base.py

@classmethod
def load(cls, filepath: Union[str, Path]):
    """Load a previously saved estimator.

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

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

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

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

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

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

predict_many

predict_many(X: DataFrame) -> DataFrame

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

Source code in deep_river/regression/regressor.py

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

predict_one

predict_one(x: dict) -> RegTarget

Predict target value for a single instance.

Source code in deep_river/regression/regressor.py

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

save

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

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

Parameters:

Name	Type	Description	Default
filepath	str | Path	Destination file. Parent directories are created automatically.	required

Source code in deep_river/base.py

def save(self, filepath: Union[str, Path]) -> None:
    """Persist the estimator (architecture, weights, optimiser & runtime state).

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

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

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