RNN Classification Models¶

This example shows the application of RNN models in river-torch with and without usage of an incremental class adaption strategy.

In [1]:

from deep_river.classification import RollingClassifier
from river import metrics, compose, preprocessing, datasets
import torch
from tqdm import tqdm

from deep_river.classification import RollingClassifier from river import metrics, compose, preprocessing, datasets import torch from tqdm import tqdm

RNN Model¶

In [2]:

class RnnModule(torch.nn.Module):

    def __init__(self, n_features, hidden_size=1):
        super().__init__()
        self.n_features = n_features
        self.rnn = torch.nn.RNN(
            input_size=n_features, hidden_size=hidden_size, num_layers=1
        )
        self.softmax = torch.nn.Softmax(dim=-1)

    def forward(self, X, **kwargs):
        out, hn = self.rnn(X)  # lstm with input, hidden, and internal state
        hn = hn.view(-1, self.rnn.hidden_size)
        return self.softmax(hn)

class RnnModule(torch.nn.Module): def __init__(self, n_features, hidden_size=1): super().__init__() self.n_features = n_features self.rnn = torch.nn.RNN( input_size=n_features, hidden_size=hidden_size, num_layers=1 ) self.softmax = torch.nn.Softmax(dim=-1) def forward(self, X, **kwargs): out, hn = self.rnn(X) # lstm with input, hidden, and internal state hn = hn.view(-1, self.rnn.hidden_size) return self.softmax(hn)

Classification without incremental class adapation strategy¶

In [3]:

dataset = datasets.Keystroke()
metric = metrics.Accuracy()
optimizer_fn = torch.optim.SGD

model_pipeline = preprocessing.StandardScaler()
model_pipeline |= RollingClassifier(
    module=RnnModule,
    loss_fn="binary_cross_entropy",
    optimizer_fn=torch.optim.SGD,
    window_size=20,
    lr=1e-2,
    append_predict=True,
    is_class_incremental=False,
)
model_pipeline

dataset = datasets.Keystroke() metric = metrics.Accuracy() optimizer_fn = torch.optim.SGD model_pipeline = preprocessing.StandardScaler() model_pipeline |= RollingClassifier( module=RnnModule, loss_fn="binary_cross_entropy", optimizer_fn=torch.optim.SGD, window_size=20, lr=1e-2, append_predict=True, is_class_incremental=False, ) model_pipeline

Out[3]:

StandardScaler

( with_std=True )

RollingClassifier

( module=None loss_fn="binary_cross_entropy" optimizer_fn=<class 'torch.optim.sgd.SGD'> lr=0.01 output_is_logit=True is_class_incremental=False device="cpu" seed=42 window_size=20 append_predict=True )

In [4]:

for x, y in tqdm(dataset):
    y_pred = model_pipeline.predict_one(x)  # make a prediction
    metric = metric.update(y, y_pred)  # update the metric
    model = model_pipeline.learn_one(x, y)  # make the model learn
print(f"Accuracy: {metric.get():.2f}")

for x, y in tqdm(dataset): y_pred = model_pipeline.predict_one(x) # make a prediction metric = metric.update(y, y_pred) # update the metric model = model_pipeline.learn_one(x, y) # make the model learn print(f"Accuracy: {metric.get():.2f}")

20400it [00:30, 666.50it/s]

Accuracy: 0.02

Classification with incremental class adaption strategy¶

In [5]:

dataset = datasets.Keystroke()
metric = metrics.Accuracy()
optimizer_fn = torch.optim.SGD

model_pipeline = preprocessing.StandardScaler()
model_pipeline |= RollingClassifier(
    module=RnnModule,
    loss_fn="binary_cross_entropy",
    optimizer_fn=torch.optim.SGD,
    window_size=20,
    lr=1e-2,
    append_predict=True,
    is_class_incremental=True,
)
model_pipeline

dataset = datasets.Keystroke() metric = metrics.Accuracy() optimizer_fn = torch.optim.SGD model_pipeline = preprocessing.StandardScaler() model_pipeline |= RollingClassifier( module=RnnModule, loss_fn="binary_cross_entropy", optimizer_fn=torch.optim.SGD, window_size=20, lr=1e-2, append_predict=True, is_class_incremental=True, ) model_pipeline

Out[5]:

StandardScaler

( with_std=True )

RollingClassifier

( module=None loss_fn="binary_cross_entropy" optimizer_fn=<class 'torch.optim.sgd.SGD'> lr=0.01 output_is_logit=True is_class_incremental=True device="cpu" seed=42 window_size=20 append_predict=True )

In [6]:

for x, y in tqdm(dataset):
    y_pred = model_pipeline.predict_one(x)  # make a prediction
    metric = metric.update(y, y_pred)  # update the metric
    model = model_pipeline.learn_one(x, y)  # make the model learn
print(f"Accuracy: {metric.get():.2f}")

for x, y in tqdm(dataset): y_pred = model_pipeline.predict_one(x) # make a prediction metric = metric.update(y, y_pred) # update the metric model = model_pipeline.learn_one(x, y) # make the model learn print(f"Accuracy: {metric.get():.2f}")

20400it [00:31, 652.13it/s]

Accuracy: 0.09

LSTM Model¶

In [7]:

class LstmModule(torch.nn.Module):

    def __init__(self, n_features, hidden_size=1):
        super().__init__()
        self.n_features = n_features
        self.lstm = torch.nn.LSTM(
            input_size=n_features, hidden_size=hidden_size, num_layers=1
        )
        self.softmax = torch.nn.Softmax(dim=-1)

    def forward(self, X, **kwargs):
        output, (hn, cn) = self.lstm(
            X
        )  # lstm with input, hidden, and internal state
        hn = hn.view(-1, self.lstm.hidden_size)
        return self.softmax(hn)

class LstmModule(torch.nn.Module): def __init__(self, n_features, hidden_size=1): super().__init__() self.n_features = n_features self.lstm = torch.nn.LSTM( input_size=n_features, hidden_size=hidden_size, num_layers=1 ) self.softmax = torch.nn.Softmax(dim=-1) def forward(self, X, **kwargs): output, (hn, cn) = self.lstm( X ) # lstm with input, hidden, and internal state hn = hn.view(-1, self.lstm.hidden_size) return self.softmax(hn)

Classifcation without incremental class adaption strategy¶

In [8]:

dataset = datasets.Keystroke()
metric = metrics.Accuracy()
optimizer_fn = torch.optim.SGD

model_pipeline = preprocessing.StandardScaler()
model_pipeline |= RollingClassifier(
    module=LstmModule,
    loss_fn="binary_cross_entropy",
    optimizer_fn=torch.optim.SGD,
    window_size=20,
    lr=1e-2,
    append_predict=True,
)
model_pipeline

dataset = datasets.Keystroke() metric = metrics.Accuracy() optimizer_fn = torch.optim.SGD model_pipeline = preprocessing.StandardScaler() model_pipeline |= RollingClassifier( module=LstmModule, loss_fn="binary_cross_entropy", optimizer_fn=torch.optim.SGD, window_size=20, lr=1e-2, append_predict=True, ) model_pipeline

Out[8]:

StandardScaler

( with_std=True )

RollingClassifier

( module=None loss_fn="binary_cross_entropy" optimizer_fn=<class 'torch.optim.sgd.SGD'> lr=0.01 output_is_logit=True is_class_incremental=False device="cpu" seed=42 window_size=20 append_predict=True )

In [9]:

for x, y in tqdm(dataset):
    y_pred = model_pipeline.predict_one(x)  # make a prediction
    metric = metric.update(y, y_pred)  # update the metric
    model = model_pipeline.learn_one(x, y)  # make the model learn
print(f"Accuracy: {metric.get():.2f}")

for x, y in tqdm(dataset): y_pred = model_pipeline.predict_one(x) # make a prediction metric = metric.update(y, y_pred) # update the metric model = model_pipeline.learn_one(x, y) # make the model learn print(f"Accuracy: {metric.get():.2f}")

20400it [00:59, 344.86it/s]

Accuracy: 0.02

Classifcation with incremental class adaption strategy¶

In [10]:

dataset = datasets.Keystroke()
metric = metrics.Accuracy()
optimizer_fn = torch.optim.SGD

model_pipeline = preprocessing.StandardScaler()
model_pipeline |= RollingClassifier(
    module=LstmModule,
    loss_fn="binary_cross_entropy",
    optimizer_fn=torch.optim.SGD,
    window_size=20,
    lr=1e-2,
    append_predict=True,
    is_class_incremental=True,
)
model_pipeline

dataset = datasets.Keystroke() metric = metrics.Accuracy() optimizer_fn = torch.optim.SGD model_pipeline = preprocessing.StandardScaler() model_pipeline |= RollingClassifier( module=LstmModule, loss_fn="binary_cross_entropy", optimizer_fn=torch.optim.SGD, window_size=20, lr=1e-2, append_predict=True, is_class_incremental=True, ) model_pipeline

Out[10]:

StandardScaler

( with_std=True )

RollingClassifier

( module=None loss_fn="binary_cross_entropy" optimizer_fn=<class 'torch.optim.sgd.SGD'> lr=0.01 output_is_logit=True is_class_incremental=True device="cpu" seed=42 window_size=20 append_predict=True )

In [11]:

for x, y in tqdm(dataset):
    y_pred = model_pipeline.predict_one(x)  # make a prediction
    metric = metric.update(y, y_pred)  # update the metric
    model = model_pipeline.learn_one(x, y)  # make the model learn
print(f"Accuracy: {metric.get()}:.2f")

for x, y in tqdm(dataset): y_pred = model_pipeline.predict_one(x) # make a prediction metric = metric.update(y, y_pred) # update the metric model = model_pipeline.learn_one(x, y) # make the model learn print(f"Accuracy: {metric.get()}:.2f")

20400it [01:07, 300.25it/s]

Accuracy: 0.13857843137254902:.2f