zoo

Classes:

Name	Description
LSTMRegressorInitialized
LinearRegressionInitialized	Linear Regression model for regression.
MultiLayerPerceptronInitialized	Linear Regression model for regression.

LSTMRegressorInitialized

LSTMRegressorInitialized(
    n_features: int = 10,
    loss_fn: Union[
        str, Callable
    ] = "binary_cross_entropy_with_logits",
    optimizer_fn: Union[str, Type[Optimizer]] = "sgd",
    lr: float = 0.001,
    output_is_logit: bool = True,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs
)

Bases: RollingRegressorInitialized

Methods:

Name	Description
learn_many	Performs one step of training with the most recent training examples
learn_one	Performs one step of training with the most recent training examples
predict_many	Predict the probability of each label given the most recent examples
predict_one	Predict the probability of each label given the most recent examples

Source code in deep_river/regression/zoo.py

def __init__(
    self,
    n_features: int = 10,
    loss_fn: Union[str, Callable] = "binary_cross_entropy_with_logits",
    optimizer_fn: Union[str, Type[optim.Optimizer]] = "sgd",
    lr: float = 1e-3,
    output_is_logit: bool = True,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs,
):
    self.n_features = n_features
    module = LSTMRegressorInitialized.LSTMModule(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,
        **kwargs,
    )

learn_many

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

Performs one step of training with the most recent training examples stored in the sliding window.

Parameters:

Name	Type	Description	Default
X	DataFrame	Input examples.	required
y	Series	Target values.	required

Returns:

Type	Description
Self	The regressor itself.

Source code in deep_river/regression/rolling_regressor.py

def learn_many(self, X: pd.DataFrame, y: pd.Series) -> None:
    """
    Performs one step of training with the most recent training examples
    stored in the sliding window.

    Parameters
    ----------
    X
        Input examples.
    y
        Target values.

    Returns
    -------
    Self
        The regressor itself.
    """
    self._update_observed_features(X)

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

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

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

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

learn_one

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

Performs one step of training with the most recent training examples stored in the sliding window.

Parameters:

Name	Type	Description	Default
x	dict	Input example.	required
y	RegTarget	Target value.	required

Returns:

Type	Description
Self	The regressor itself.

Source code in deep_river/regression/rolling_regressor.py

def learn_one(self, x: dict, y: base.typing.RegTarget, **kwargs) -> None:
    """
    Performs one step of training with the most recent training examples
    stored in the sliding window.

    Parameters
    ----------
    x
        Input example.
    y
        Target value.

    Returns
    -------
    Self
        The regressor itself.
    """
    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)

predict_many

predict_many(X: DataFrame) -> DataFrame

Predict the probability of each label given the most recent examples

Parameters:

Name	Type	Description	Default
X	DataFrame		required

Returns:

Type	Description
DataFrame	DataFrame of probabilities for each label.

Source code in deep_river/regression/rolling_regressor.py

def predict_many(self, X: pd.DataFrame) -> pd.DataFrame:
    """
    Predict the probability of each label given the most recent examples

    Parameters
    ----------
    X

    Returns
    -------
    pd.DataFrame
        DataFrame of probabilities for each label.
    """

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

    self.module.eval()
    with torch.inference_mode():
        x_t = self._deque2rolling_tensor(x_win)
        y_preds = self.module(x_t).detach().tolist()
    return pd.DataFrame(y_preds if not y_preds.is_cuda else y_preds.cpu().numpy())

predict_one

predict_one(x: dict) -> RegTarget

Predict the probability of each label given the most recent examples stored in the sliding window.

Parameters:

Name	Type	Description	Default
x	dict	Input example.	required

Returns:

Type	Description
Dict[ClfTarget, float]	Dictionary of probabilities for each label.

Source code in deep_river/regression/rolling_regressor.py

def predict_one(self, x: dict) -> base.typing.RegTarget:
    """
    Predict the probability of each label given the most recent examples
    stored in the sliding window.

    Parameters
    ----------
    x
        Input example.

    Returns
    -------
    Dict[ClfTarget, float]
        Dictionary of probabilities for each label.
    """
    self._update_observed_features(x)

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

    self.module.eval()
    with torch.inference_mode():
        x_t = self._deque2rolling_tensor(x_win)
        res = self.module(x_t).numpy(force=True).item()

    return res

LinearRegressionInitialized

LinearRegressionInitialized(
    n_features: int = 10,
    loss_fn: Union[
        str, Callable
    ] = "binary_cross_entropy_with_logits",
    optimizer_fn: Union[str, Type[Optimizer]] = "sgd",
    lr: float = 0.001,
    output_is_logit: bool = True,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs
)

Bases: RegressorInitialized

Linear Regression model for regression.

Parameters:

Name	Type	Description	Default
loss_fn	str or Callable	Loss function to be used for training the wrapped model.	'binary_cross_entropy_with_logits'
optimizer_fn	str or Callable	Optimizer to be used for training the wrapped model.	'sgd'
lr	float	Learning rate of the optimizer.	0.001
output_is_logit	bool	Whether the module produces logits as output. If true, either softmax or sigmoid is applied to the outputs when predicting.	True
is_class_incremental	bool	Whether the classifier should adapt to the appearance of previously unobserved classes by adding an unit to the output layer of the network.	required
is_feature_incremental	bool	Whether the model should adapt to the appearance of previously features by adding units to the input layer of the network.	False
device	str	Device to run the wrapped model on. Can be "cpu" or "cuda".	'cpu'
seed	int	Random seed to be used for training the wrapped model.	42
**kwargs		Parameters to be passed to the build_fn function aside from n_features.	{}

Methods:

Name	Description
predict_many	Predicts probabilities for multiple examples.
predict_one	Predicts the target value for a single example.

Source code in deep_river/regression/zoo.py

def __init__(
    self,
    n_features: int = 10,
    loss_fn: Union[str, Callable] = "binary_cross_entropy_with_logits",
    optimizer_fn: Union[str, Type[optim.Optimizer]] = "sgd",
    lr: float = 1e-3,
    output_is_logit: bool = True,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs,
):
    self.n_features = n_features
    module = LinearRegressionInitialized.LRModule(n_features=n_features)
    if "module" in kwargs:
        del kwargs["module"]
    super().__init__(
        module=module,
        loss_fn=loss_fn,
        optimizer_fn=optimizer_fn,
        output_is_logit=output_is_logit,
        is_feature_incremental=is_feature_incremental,
        device=device,
        lr=lr,
        seed=seed,
        **kwargs,
    )

predict_many

predict_many(X: DataFrame) -> DataFrame

Predicts probabilities for multiple examples.

Source code in deep_river/regression/regressor.py

def predict_many(self, X: pd.DataFrame) -> pd.DataFrame:
    """Predicts probabilities for multiple examples."""
    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

Predicts the target value for a single example.

Parameters:

Name	Type	Description	Default
x	dict	Input example.	required

Returns:

Type	Description
RegTarget	Predicted target value.

Source code in deep_river/regression/regressor.py

def predict_one(self, x: dict) -> RegTarget:
    """
    Predicts the target value for a single example.

    Parameters
    ----------
    x
        Input example.

    Returns
    -------
    RegTarget
        Predicted target value.
    """
    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

MultiLayerPerceptronInitialized

MultiLayerPerceptronInitialized(
    n_features: int = 10,
    n_width: int = 5,
    n_layers: int = 5,
    loss_fn: Union[
        str, Callable
    ] = "binary_cross_entropy_with_logits",
    optimizer_fn: Union[str, Type[Optimizer]] = "sgd",
    lr: float = 0.001,
    output_is_logit: bool = True,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs
)

Bases: RegressorInitialized

Linear Regression model for regression.

Parameters:

Name	Type	Description	Default
loss_fn	str or Callable	Loss function to be used for training the wrapped model.	'binary_cross_entropy_with_logits'
optimizer_fn	str or Callable	Optimizer to be used for training the wrapped model.	'sgd'
lr	float	Learning rate of the optimizer.	0.001
output_is_logit	bool	Whether the module produces logits as output. If true, either softmax or sigmoid is applied to the outputs when predicting.	True
is_class_incremental	bool	Whether the classifier should adapt to the appearance of previously unobserved classes by adding an unit to the output layer of the network.	required
is_feature_incremental	bool	Whether the model should adapt to the appearance of previously features by adding units to the input layer of the network.	False
device	str	Device to run the wrapped model on. Can be "cpu" or "cuda".	'cpu'
seed	int	Random seed to be used for training the wrapped model.	42
**kwargs		Parameters to be passed to the build_fn function aside from n_features.	{}

Methods:

Name	Description
predict_many	Predicts probabilities for multiple examples.
predict_one	Predicts the target value for a single example.

Source code in deep_river/regression/zoo.py

def __init__(
    self,
    n_features: int = 10,
    n_width: int = 5,
    n_layers: int = 5,
    loss_fn: Union[str, Callable] = "binary_cross_entropy_with_logits",
    optimizer_fn: Union[str, Type[optim.Optimizer]] = "sgd",
    lr: float = 1e-3,
    output_is_logit: bool = True,
    is_feature_incremental: bool = False,
    device: str = "cpu",
    seed: int = 42,
    **kwargs,
):
    self.n_features = n_features
    self.n_width = n_width
    self.n_layers = n_layers
    module = MultiLayerPerceptronInitialized.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,
        output_is_logit=output_is_logit,
        is_feature_incremental=is_feature_incremental,
        device=device,
        lr=lr,
        seed=seed,
        **kwargs,
    )

predict_many

predict_many(X: DataFrame) -> DataFrame

Predicts probabilities for multiple examples.

Source code in deep_river/regression/regressor.py

def predict_many(self, X: pd.DataFrame) -> pd.DataFrame:
    """Predicts probabilities for multiple examples."""
    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

Predicts the target value for a single example.

Parameters:

Name	Type	Description	Default
x	dict	Input example.	required

Returns:

Type	Description
RegTarget	Predicted target value.

Source code in deep_river/regression/regressor.py

def predict_one(self, x: dict) -> RegTarget:
    """
    Predicts the target value for a single example.

    Parameters
    ----------
    x
        Input example.

    Returns
    -------
    RegTarget
        Predicted target value.
    """
    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