paddlets.models.anomaly.dl.utils

percentile(anomaly_score: ndarray, q: float = 100)[source]

The percentile function to get anomaly threshold.

Parameters

anomaly_score (np.ndarray) – Anomaly score.
q (float) – The parameter used to calculate the quantile, range is [0, 100].

Returns

Anomaly threshold.

Return type

threshold(float)

my_kl_loss(p: Tensor, q: Tensor)[source]

The Kullback–Leibler divergence.

Parameters

p (paddle.Tensor) – Tensor of arbitrary shape in log-probabilities.
q (paddle.Tensor) – Tensor of the same shape as input.

Returns

Got loss.

Return type

loss(paddle.Tensor)

adjust_learning_rate(optimizer: Callable[[...], Optimizer], epoch: int, lr_: float)[source]

Dynamic Learning Rate Adjustment.

Parameters

optimizer (Callable[..., Optimizer]) – Optimizer algorithm.
epoch (int) – Max epochs during training.
lr (float) – Learning rate.

series_prior_loss(output_list: List[Tensor], input: Tensor, criterion: Callable[[...], Tensor] = MSELoss(), win_size: int = 100, k: int = 3)[source]

Calculate Association discrepancy in train.

Parameters

output_list (List[paddle.Tensor]) – Model ouput tensor list.
input (paddle.Tensor) – Target tensor.
criterion (Callable[..., paddle.Tensor]) – Loss function.
win_size (int) – The size of the loopback window, i.e. the number of time steps feed to the model.
k (int) – The optimization is to enlarge the association discrepancy.

Returns

Series_loss and rec_loss. loss2(paddle.Tensor): Prior_loss and rec_loss. for_loss_one(paddle.Tensor): Rec_loss and Association discrepancy.

Return type

loss1(paddle.Tensor)

series_prios_energy(output_list, loss, temperature=50, win_size=100)[source]

calculate Association discrepancy in test.

Parameters

output_list (List[paddle.Tensor]) – Model ouput tensor list.
loss (paddle.Tensor) – Got loss.
temperature (int|float) – A parameter to adjust series loss and prior loss.
win_size (int) – The size of the loopback window, i.e. the number of time steps feed to the model.

Returns

Anomaly score for predict.

Return type

cri(np.ndarray)

anomaly_get_threshold(model: ~typing.Callable[[...], ~paddle.Tensor], train_dataloader: ~paddle.fluid.reader.DataLoader, thre_dataloader: ~paddle.fluid.reader.DataLoader, temperature: float = 50, anormly_ratio: float = 4, criterion: ~typing.Callable[[...], ~paddle.Tensor] = MSELoss(), my_kl_loss: ~typing.Callable[[...], ~paddle.Tensor] = <function my_kl_loss>, win_size: int = 100)[source]

Threshold is calculated based on Association-based Anomaly Criterion.

Parameters

model (Callable[..., paddle.Tensor]) – Anomaly transformer model.
train_dataloader (paddle.io.DataLoader) – Train set.
thre_dataloader (List[paddle.io.DataLoader]|None) – Test set.
temperature (int|float) – A parameter to adjust series loss and prior loss.
anormly_ratio (int|float) – The Proportion of Anomaly data in train set and test set.
criterion (Callable[..., paddle.Tensor]|None) – Loss function for the reconstruction loss.
my_kl_loss (Callable[..., paddle.Tensor]|None) – Loss function for association discrepancy.
win_size (int) – The size of the loopback window, i.e. the number of time steps feed to the model.

Returns

The threshold to judge anomaly.

Return type

threshold(float|None)

result_adjust(pred: ndarray, real: ndarray)[source]

The adjustment is a widely-used convention in time series anomaly detection.

Parameters

pred (List[float]|np.ndarray) – The model prediction results.
real (List[float]|np.ndarray) – Ground truth target values.

Returns

Adjusted prediction results.

Return type

pred(np.ndarray)

smooth_l1_loss_vae(output_tensor_list: List[Tensor], kld_beta: float = 0.2)[source]

smooth l1 loss.

Parameters

output_tensor_list (list[paddle.Tensor]) – Model ouput.
kld_beta (float) – Kld beta in 0~1.

Returns

Got loss.

Return type

loss(float)

to_tsdataset(scenario: str = 'anomaly_label') → Callable[[...], Callable[[...], TSDataset]][source]

A decorator, used for converting ndarray to tsdataset in anomaly dl models.

Parameters: scenario (str) – The task type. [“anomaly_label”, “anomaly_score”] is optional.
Returns: Wrapped core function.
Return type: Callable[…, Callable[…, TSDataset]]

epsilon_th(anomaly_score: ndarray, reg_level: int = 1)[source]

Threshold method proposed by Hundman et. al. (https://arxiv.org/abs/1802.04431) Code from TelemAnom (https://github.com/khundman/telemanom)

Parameters

anomaly_score (np.ndarray) – Anomaly score.
reg_level (int) – The parameter used to calculate the threshold.

Returns

Anomaly threshold.

Return type

threshold(float)