zoo

Classes:

Name	Description
LSTMClassifierInitialized	A specialized LSTM-based classifier designed for handling rolling or
LogisticRegressionInitialized	Logistic Regression model for classification.
MultiLayerPerceptronInitialized	Logistic Regression model for classification.

LSTMClassifierInitialized

LSTMClassifierInitialized(
    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: RollingClassifierInitialized

A specialized LSTM-based classifier designed for handling rolling or incremental data classification tasks.

This class leverages LSTM (Long Short-Term Memory) modules to process and classify sequential data. It is built on top of the base RollingClassifierInitialized class, inheriting its functionality for handling incremental learning tasks. Customization options include the definition of the loss function, optimizer, learning rate, and other hyperparameters to suit various use cases.

Attributes:

Name	Type	Description
n_features	int	Number of features in the input data. It defines the input dimension for the LSTM module.
loss_fn	Union[str, Callable]	Specifies the loss function to be used for model training. Can either be a predefined string or a callable function.
optimizer_fn	Union[str, Type[Optimizer]]	Defines the optimizer to be utilized in training. Accepts either a string representing the optimizer name or the optimizer class itself.
lr	float	Learning rate for the chosen optimizer.
output_is_logit	bool	Indicates whether the model output is a raw logit (pre-sigmoid/softmax output).
is_feature_incremental	bool	Specifies if the model supports adding new features incrementally.
device	str	Designates the device for computation, e.g., 'cpu' or 'cuda'.
seed	int	Random seed for reproducibility of results.
kwargs	dict	Additional arguments passed during the initialization.

Methods:

Name	Description
learn_many	Learns from multiple examples using the rolling window.
learn_one	Learns from one example using the rolling window.
predict_proba_many	Predicts probabilities for many examples.
predict_proba_one	Predicts class probabilities using the rolling window.

Source code in deep_river/classification/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 = LSTMClassifierInitialized.LSTMModule(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,
    )

learn_many

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

Learns from multiple examples using the rolling window.

Source code in deep_river/classification/rolling_classifier.py

def learn_many(self, X: pd.DataFrame, y: pd.Series) -> None:
    """Learns from multiple examples using the rolling window."""
    self._update_observed_targets(y)
    self._update_observed_features(X)
    X = X[list(self.observed_features)]
    self._x_window.extend(X.values.tolist())
    X_t = self._deque2rolling_tensor(self._x_window)
    self._learn(x=X_t, y=y)

learn_one

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

Learns from one example using the rolling window.

Source code in deep_river/classification/rolling_classifier.py

def learn_one(self, x: dict, y: ClfTarget, **kwargs) -> None:
    """Learns from one example using the rolling window."""
    self._update_observed_features(x)
    self._update_observed_targets(y)
    self._x_window.append([x.get(feature, 0) for feature in self.observed_features])
    x_t = self._deque2rolling_tensor(self._x_window)
    self._learn(x=x_t, y=y)

predict_proba_many

predict_proba_many(X: DataFrame) -> DataFrame

Predicts probabilities for many examples.

Source code in deep_river/classification/rolling_classifier.py

def predict_proba_many(self, X: pd.DataFrame) -> pd.DataFrame:
    """Predicts probabilities for many examples."""
    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)
        probas = self.module(x_t).detach().tolist()
    return pd.DataFrame(probas)

predict_proba_one

predict_proba_one(x: dict) -> Dict[ClfTarget, float]

Predicts class probabilities using the rolling window.

Source code in deep_river/classification/rolling_classifier.py

def predict_proba_one(self, x: dict) -> Dict[ClfTarget, float]:
    """Predicts class probabilities using the rolling window."""
    self._update_observed_features(x)
    x_win = self._x_window.copy()
    x_win.append([x.get(feature, 0) for feature in self.observed_features])
    if self.append_predict:
        self._x_window = x_win
    self.module.eval()
    with torch.inference_mode():
        x_t = self._deque2rolling_tensor(x_win)
        y_pred = self.module(x_t)
        proba = output2proba(y_pred, self.observed_classes, self.output_is_logit)
    return proba[0]

LogisticRegressionInitialized

LogisticRegressionInitialized(
    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: ClassifierInitialized

Logistic Regression model for classification.

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
learn_many	Updates the model with multiple instances for supervised learning.
learn_one	Learns from a single example.
predict_proba_many	Predicts probabilities for multiple examples.
predict_proba_one	Predicts probabilities for a single example.

Source code in deep_river/classification/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 = LogisticRegressionInitialized.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,
    )

learn_many

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

Updates the model with multiple instances for supervised learning.

The function updates the observed features and targets based on the input data. It converts the data from a pandas DataFrame to a tensor format before learning occurs. The updates to the model are executed through an internal learning mechanism.

Parameters:

Name	Type	Description	Default
X	DataFrame	The data-frame containing instances to be learned by the model. Each row represents a single instance, and each column represents a feature.	required
y	Series	The target values corresponding to the instances in X. Each entry in the series represents the target associated with a row in X.	required

Returns:

Type	Description
None

Source code in deep_river/classification/classifier.py

def learn_many(self, X: pd.DataFrame, y: pd.Series) -> None:
    """
    Updates the model with multiple instances for supervised learning.

    The function updates the observed features and targets based on the input
    data. It converts the data from a pandas DataFrame to a tensor format before
    learning occurs. The updates to the model are executed through an internal
    learning mechanism.

    Parameters
    ----------
    X : pd.DataFrame
        The data-frame containing instances to be learned by the model. Each
        row represents a single instance, and each column represents a feature.
    y : pd.Series
        The target values corresponding to the instances in `X`. Each entry in
        the series represents the target associated with a row in `X`.

    Returns
    -------
    None
    """
    self._update_observed_features(X)
    self._update_observed_targets(y)
    x_t = self._df2tensor(X)
    self._learn(x_t, y)

learn_one

learn_one(x: dict, y: ClfTarget) -> None

Learns from a single example.

Source code in deep_river/classification/classifier.py

def learn_one(self, x: dict, y: base.typing.ClfTarget) -> None:
    """Learns from a single example."""
    self._update_observed_features(x)
    self._update_observed_targets(y)
    x_t = self._dict2tensor(x)
    self._learn(x_t, y)

predict_proba_many

predict_proba_many(X: DataFrame) -> DataFrame

Predicts probabilities for multiple examples.

Source code in deep_river/classification/classifier.py

def predict_proba_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(
        output2proba(y_preds, self.observed_classes, self.output_is_logit)
    )

predict_proba_one

predict_proba_one(x: dict) -> dict[ClfTarget, float]

Predicts probabilities for a single example.

Source code in deep_river/classification/classifier.py

def predict_proba_one(self, x: dict) -> dict[base.typing.ClfTarget, float]:
    """Predicts probabilities for a single example."""
    self._update_observed_features(x)
    x_t = self._dict2tensor(x)
    self.module.eval()
    with torch.inference_mode():
        y_pred = self.module(x_t)
    return output2proba(y_pred, self.observed_classes, self.output_is_logit)[0]

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: ClassifierInitialized

Logistic Regression model for classification.

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
learn_many	Updates the model with multiple instances for supervised learning.
learn_one	Learns from a single example.
predict_proba_many	Predicts probabilities for multiple examples.
predict_proba_one	Predicts probabilities for a single example.

Source code in deep_river/classification/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_width=n_width, n_layers=n_layers, 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,
    )

learn_many

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

Updates the model with multiple instances for supervised learning.

The function updates the observed features and targets based on the input data. It converts the data from a pandas DataFrame to a tensor format before learning occurs. The updates to the model are executed through an internal learning mechanism.

Parameters:

Name	Type	Description	Default
X	DataFrame	The data-frame containing instances to be learned by the model. Each row represents a single instance, and each column represents a feature.	required
y	Series	The target values corresponding to the instances in X. Each entry in the series represents the target associated with a row in X.	required

Returns:

Type	Description
None

Source code in deep_river/classification/classifier.py

def learn_many(self, X: pd.DataFrame, y: pd.Series) -> None:
    """
    Updates the model with multiple instances for supervised learning.

    The function updates the observed features and targets based on the input
    data. It converts the data from a pandas DataFrame to a tensor format before
    learning occurs. The updates to the model are executed through an internal
    learning mechanism.

    Parameters
    ----------
    X : pd.DataFrame
        The data-frame containing instances to be learned by the model. Each
        row represents a single instance, and each column represents a feature.
    y : pd.Series
        The target values corresponding to the instances in `X`. Each entry in
        the series represents the target associated with a row in `X`.

    Returns
    -------
    None
    """
    self._update_observed_features(X)
    self._update_observed_targets(y)
    x_t = self._df2tensor(X)
    self._learn(x_t, y)

learn_one

learn_one(x: dict, y: ClfTarget) -> None

Learns from a single example.

Source code in deep_river/classification/classifier.py

def learn_one(self, x: dict, y: base.typing.ClfTarget) -> None:
    """Learns from a single example."""
    self._update_observed_features(x)
    self._update_observed_targets(y)
    x_t = self._dict2tensor(x)
    self._learn(x_t, y)

predict_proba_many

predict_proba_many(X: DataFrame) -> DataFrame

Predicts probabilities for multiple examples.

Source code in deep_river/classification/classifier.py

def predict_proba_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(
        output2proba(y_preds, self.observed_classes, self.output_is_logit)
    )

predict_proba_one

predict_proba_one(x: dict) -> dict[ClfTarget, float]

Predicts probabilities for a single example.

Source code in deep_river/classification/classifier.py

def predict_proba_one(self, x: dict) -> dict[base.typing.ClfTarget, float]:
    """Predicts probabilities for a single example."""
    self._update_observed_features(x)
    x_t = self._dict2tensor(x)
    self.module.eval()
    with torch.inference_mode():
        y_pred = self.module(x_t)
    return output2proba(y_pred, self.observed_classes, self.output_is_logit)[0]