Source code for alibi_detect.od.pytorch.mahalanobis

import torch

from alibi_detect.od.pytorch.base import TorchOutlierDetector
from alibi_detect.utils._types import TorchDeviceType

[docs]class MahalanobisTorch(TorchOutlierDetector): ensemble = False
[docs] def __init__( self, min_eigenvalue: float = 1e-6, device: TorchDeviceType = None, ): """PyTorch backend for Mahalanobis detector. Parameters ---------- min_eigenvalue Eigenvectors with eigenvalues below this value will be discarded. device Device type used. The default tries to use the GPU and falls back on CPU if needed. Can be specified by passing either ``'cuda'``, ``'gpu'``, ``'cpu'`` or an instance of ``torch.device``. """ super().__init__(device=device) self.min_eigenvalue = min_eigenvalue
[docs] def forward(self, x: torch.Tensor) -> torch.Tensor: """Detect if `x` is an outlier. Parameters ---------- x `torch.Tensor` with leading batch dimension. Returns ------- `torch.Tensor` of ``bool`` values with leading batch dimension. Raises ------ ThresholdNotInferredException If called before detector has had `infer_threshold` method called. """ scores = self.score(x) if not torch.jit.is_scripting(): self.check_threshold_inferred() preds = scores > self.threshold return preds
[docs] def score(self, x: torch.Tensor) -> torch.Tensor: """Computes the score of `x` Parameters ---------- x The tensor of instances. First dimension corresponds to batch. Returns ------- Tensor of scores for each element in `x`. Raises ------ NotFitException If called before detector has been fit. """ self.check_fitted() x_pcs = self._compute_linear_proj(x) return (x_pcs**2).sum(-1)
[docs] def fit(self, x_ref: torch.Tensor): """Fits the detector Parameters ---------- x_ref The Dataset tensor. """ self.x_ref = x_ref self._compute_linear_pcs(self.x_ref) self._set_fitted()
def _compute_linear_pcs(self, x: torch.Tensor): """Computes the principal components of the data. Parameters ---------- x The reference dataset. """ self.means = x.mean(0) x = x - self.means cov_mat = (x.t() @ x)/(len(x)-1) D, V = torch.linalg.eigh(cov_mat) non_zero_inds = D > self.min_eigenvalue self.pcs = V[:, non_zero_inds] / D[None, non_zero_inds].sqrt() def _compute_linear_proj(self, x: torch.Tensor) -> torch.Tensor: """Projects the data point being tested onto the principal components. Parameters ---------- x The data point being tested. """ x_cen = x - self.means x_proj = x_cen @ self.pcs return x_proj