probability_weighted_ae

ProbabilityWeightedAutoencoder(module, loss_fn='mse', optimizer_fn='sgd', lr=0.001, device='cpu', seed=42, skip_threshold=0.9, window_size=250, **kwargs)

Bases: Autoencoder

Wrapper for PyTorch autoencoder models for anomaly detection that reduces the employed learning rate based on an outlier probability estimate of the input example as well as a threshold probability skip_threshold. If the outlier probability is above the threshold, the learning rate is reduced to less than 0. Given the probability estimate \(p_out\), the adjusted learning rate \(lr_adj\) is \(lr * 1 - (rac{p_out}{skip_threshold})\).

PARAMETER	DESCRIPTION
module	Torch Module that builds the autoencoder to be wrapped. The Module should accept parameter n_features so that the returned model's input shape can be determined based on the number of features in the initial training example. TYPE: Type[Module]
loss_fn	Loss function to be used for training the wrapped model. Can be a loss function provided by torch.nn.functional or one of the following: 'mse', 'l1', 'cross_entropy', 'binary_crossentropy', 'smooth_l1', 'kl_div'. TYPE: Union[str, Callable] DEFAULT: 'mse'
optimizer_fn	Optimizer to be used for training the wrapped model. Can be an optimizer class provided by torch.optim or one of the following: "adam", "adam_w", "sgd", "rmsprop", "lbfgs". TYPE: Union[str, Callable] DEFAULT: 'sgd'
lr	Base learning rate of the optimizer. TYPE: float DEFAULT: 0.001
skip_threshold	Threshold probability to use as a reference for the reduction of the base learning rate. TYPE: float DEFAULT: 0.9
device	Device to run the wrapped model on. Can be "cpu" or "cuda". TYPE: str DEFAULT: 'cpu'
seed	Random seed to be used for training the wrapped model. TYPE: int DEFAULT: 42
**kwargs	Parameters to be passed to the module function aside from n_features. Examples DEFAULT: {}

_ = manual_seed(42) dataset = CreditCard().take(5000) metric = metrics.ROCAUC(n_thresholds=50)

class MyAutoEncoder(torch.nn.Module): ... def init(self, n_features, latent_dim=3): ... super(MyAutoEncoder, self).init() ... self.linear1 = nn.Linear(n_features, latent_dim) ... self.nonlin = torch.nn.LeakyReLU() ... self.linear2 = nn.Linear(latent_dim, n_features) ... self.sigmoid = nn.Sigmoid() ... ... def forward(self, X, **kwargs): ... X = self.linear1(X) ... X = self.nonlin(X) ... X = self.linear2(X) ... return self.sigmoid(X)

ae = ProbabilityWeightedAutoencoder(module=MyAutoEncoder, lr=0.005) scaler = MinMaxScaler() model = Pipeline(scaler, ae)

for x, y in dataset: ... score = model.score_one(x) ... model = model.learn_one(x=x) ... metric = metric.update(y, score) ... print(f"ROCAUC: {metric.get():.4f}") ROCAUC: 0.8599

learn_one(x, y=None, **kwargs)

Performs one step of training with a single example, scaling the employed learning rate based on the outlier probability estimate of the input example.

PARAMETER	DESCRIPTION
**kwargs	DEFAULT: {}
x	Input example. TYPE: dict

RETURNS	DESCRIPTION
ProbabilityWeightedAutoencoder	The autoencoder itself.