timesead.models.other.ncad ========================== .. py:module:: timesead.models.other.ncad Classes ------- .. autoapisummary:: timesead.models.other.ncad.TCNEncoder timesead.models.other.ncad.ContrastiveClassifier timesead.models.other.ncad.NCAD timesead.models.other.ncad.NCADTrainer timesead.models.other.ncad.NCADAnomalyDetector timesead.models.other.ncad.LocalOutlierInjectionTransform Functions --------- .. autoapisummary:: timesead.models.other.ncad.l2_distance timesead.models.other.ncad.coe_batch timesead.models.other.ncad.mixup_batch Module Contents --------------- .. py:class:: TCNEncoder(in_channels: int, out_channels: int, kernel_size: int, tcn_channels: int, tcn_layers: int, tcn_out_channels: int, maxpool_out_channels: int = 1, normalize_embedding: bool = True) Bases: :py:obj:`torch.nn.Module` Encoder of a time series using a Temporal Convolution Network (TCN). The computed representation is the output of a fully connected layer applied to the output of an adaptive max pooling layer applied on top of the TCN, which reduces the length of the time series to a fixed size. Takes as input a three-dimensional tensor (`B`, `C_in`, `L`) where `B` is the batch size, `C_in` is the number of input channels, and `L` is the length of the input. Outputs a two-dimensional tensor (`B`, `C_out`), `C_in` is the number of input channels C_in=tcn_channels* :param in_channels: Number of input channels. :param out_channels: Dimension of the output representation vector. :param kernel_size: Kernel size of the applied non-residual convolutions. :param tcn_channels: Number of channels manipulated in the causal CNN. :param tcn_layers: Depth of the causal CNN. :param tcn_out_channels: Number of channels produced by the TCN. The TCN outputs a tensor of shape (B, tcn_out_channels, T) :param maxpool_out_channels: Fixed length to which each channel of the TCN is reduced. :param normalize_embedding: Normalize size of the embeddings Initialize internal Module state, shared by both nn.Module and ScriptModule. .. py:attribute:: network .. py:attribute:: normalize_embedding :value: True .. py:method:: forward(x) .. py:class:: ContrastiveClassifier(distance: Union[Callable[[torch.Tensor, torch.Tensor], str], torch.Tensor]) Bases: :py:obj:`torch.nn.Module` Contrastive Classifier. Calculates the distance between two random vectors, and returns an exponential transformation of it, which can be interpreted as the logits for the two vectors being different. p : Probability of x1 and x2 being different p = 1 - exp( -dist(x1,x2) ) :param distance: A Function which takes two (batches of) vectors and returns a (batch of) positive number. .. py:attribute:: distance .. py:attribute:: eps :value: 1e-10 .. py:method:: forward(x1: torch.Tensor, x2: torch.Tensor) -> torch.Tensor .. py:function:: l2_distance(x1: torch.Tensor, x2: torch.Tensor) -> torch.Tensor .. py:class:: NCAD(ts_channels: int, suspect_window_length: int = 1, tcn_kernel_size: int = 7, tcn_layers: int = 8, tcn_out_channels: int = 20, tcn_maxpool_out_channels: int = 8, embedding_rep_dim: int = 120, normalize_embedding: bool = True, distance: Union[Callable, str] = l2_distance) Bases: :py:obj:`timesead.models.BaseModel` Neural Contrastive Detection in Time Series Initialize internal Module state, shared by both nn.Module and ScriptModule. .. py:attribute:: suspect_window_length :value: 1 .. py:attribute:: encoder .. py:attribute:: classifier .. py:method:: forward(inputs: Tuple[torch.Tensor, Ellipsis]) -> torch.Tensor .. py:function:: coe_batch(x: torch.Tensor, y: torch.Tensor, coe_rate: float, suspect_window_length: int) -> Tuple[torch.Tensor, torch.Tensor] Contextual Outlier Exposure. :param x: Tensor of shape (batch, time, D) :param y: Tensor of shape (batch, ) :param coe_rate: Number of generated anomalies as proportion of the batch size. .. py:function:: mixup_batch(x: torch.Tensor, y: torch.Tensor, mixup_rate: float) -> Tuple[torch.Tensor, torch.Tensor] :param x: Tensor of shape (batch, time, D) :param y: Tensor of shape (batch, ) :param mixup_rate: Number of generated anomalies as proportion of the batch size. .. py:class:: NCADTrainer(*args, coe_rate: float = 1.116, mixup_rate: float = 1.96, **kwargs) Bases: :py:obj:`timesead.optim.trainer.Trainer` .. py:attribute:: coe_rate :value: 1.116 .. py:attribute:: mixup_rate :value: 1.96 .. py:method:: validate_batch(network: torch.nn.Module, val_metrics: Dict[str, Callable], b_inputs: Tuple[torch.Tensor, Ellipsis], b_targets: Tuple[torch.Tensor, Ellipsis], *args, **kwargs) -> Dict[str, float] .. py:method:: train_batch(network: NCAD, losses: List[timesead.optim.loss.Loss], optimizers: List[torch.optim.Optimizer], epoch: int, num_epochs: int, b_inputs: Tuple[torch.Tensor, Ellipsis], b_targets: Tuple[torch.Tensor, Ellipsis]) -> List[float] .. py:class:: NCADAnomalyDetector(model: NCAD) Bases: :py:obj:`timesead.models.common.AnomalyDetector` Base class for all neural network modules. Your models should also subclass this class. Modules can also contain other Modules, allowing them to be nested in a tree structure. You can assign the submodules as regular attributes:: import torch.nn as nn import torch.nn.functional as F class Model(nn.Module): def __init__(self) -> None: super().__init__() self.conv1 = nn.Conv2d(1, 20, 5) self.conv2 = nn.Conv2d(20, 20, 5) def forward(self, x): x = F.relu(self.conv1(x)) return F.relu(self.conv2(x)) Submodules assigned in this way will be registered, and will also have their parameters converted when you call :meth:`to`, etc. .. note:: As per the example above, an ``__init__()`` call to the parent class must be made before assignment on the child. :ivar training: Boolean represents whether this module is in training or evaluation mode. :vartype training: bool Initialize internal Module state, shared by both nn.Module and ScriptModule. .. py:attribute:: model .. py:method:: compute_online_anomaly_score(inputs: Tuple[torch.Tensor, Ellipsis]) -> torch.Tensor Compute the online anomaly score for a batch of inputs. The output tensor must have the same shape as the output of `format_targets` when called with the corresponding targets for this batch. This method expects a window (or a batch of windows) as its input and should return a score for the last point in the window. :param inputs: tuple of input tensors :return: Tensor of shape (B,) that contains the anomaly scores for this batch .. py:method:: compute_offline_anomaly_score(inputs: Tuple[torch.Tensor, Ellipsis]) -> torch.Tensor Compute the offline anomaly score for a batch of inputs. The output tensor must have the same shape as the output of `format_targets` when called with the corresponding targets for this batch. This method expects a window (or a batch of windows) as its input and should return a score for the last point in the window. :param inputs: tuple of input tensors :return: Tensor of shape (N,) that contains the anomaly scores for this batch .. py:method:: fit(dataset: torch.utils.data.DataLoader) -> None Fit this anomaly detector on a dataset. Note that we assume only normal data here. :param dataset: A dataset .. py:method:: format_online_targets(targets: Tuple[torch.Tensor, Ellipsis]) -> torch.Tensor Format the labels for a batch of targets. The output tensor must have the same shape as the output of `compute_online_anomaly_score` when called with the corresponding inputs for this batch. :param targets: tuple of target tensors :return: Tensor of shape (B,) that contains the ground truth labels for this batch .. py:class:: LocalOutlierInjectionTransform(parent: timesead.data.transforms.Transform, max_duration_spike: int = 2, spike_multiplier_range: Tuple[float, float] = (0.5, 2.0), spike_value_range: Tuple[float, float] = (-np.inf, np.inf), area_radius: int = 100, num_spikes: int = 10) Bases: :py:obj:`timesead.data.transforms.Transform` Inject spikes based on local noise .. py:attribute:: max_duration_spike :value: 2 .. py:attribute:: spike_multiplier_range :value: (0.5, 2.0) .. py:attribute:: spike_value_range .. py:attribute:: area_radius :value: 100 .. py:attribute:: num_spikes :value: 10