forked from JointCloud/JCC-DeepOD
107 lines
3.0 KiB
Python
107 lines
3.0 KiB
Python
import numpy as np
|
|
from sklearn import metrics
|
|
|
|
|
|
def get_sub_seqs(x_arr, seq_len=100, stride=1):
|
|
"""
|
|
|
|
Parameters
|
|
----------
|
|
x_arr: np.array, required
|
|
input original data with shape [time_length, channels]
|
|
|
|
seq_len: int, optional (default=100)
|
|
Size of window used to create subsequences from the data
|
|
|
|
stride: int, optional (default=1)
|
|
number of time points the window will move between two subsequences
|
|
|
|
Returns
|
|
-------
|
|
x_seqs: np.array
|
|
Split sub-sequences of input time-series data
|
|
"""
|
|
|
|
seq_starts = np.arange(0, x_arr.shape[0] - seq_len + 1, stride)
|
|
x_seqs = np.array([x_arr[i:i + seq_len] for i in seq_starts])
|
|
|
|
return x_seqs
|
|
|
|
|
|
def get_sub_seqs_label(y, seq_len=100, stride=1):
|
|
"""
|
|
|
|
Parameters
|
|
----------
|
|
y: np.array, required
|
|
data labels
|
|
|
|
seq_len: int, optional (default=100)
|
|
Size of window used to create subsequences from the data
|
|
|
|
stride: int, optional (default=1)
|
|
number of time points the window will move between two subsequences
|
|
|
|
Returns
|
|
-------
|
|
y_seqs: np.array
|
|
Split label of each sequence
|
|
"""
|
|
|
|
seq_starts = np.arange(0, y.shape[0] - seq_len + 1, stride)
|
|
ys = np.array([y[i:i + seq_len] for i in seq_starts])
|
|
y = np.sum(ys, axis=1) / seq_len
|
|
|
|
y_binary = np.zeros_like(y)
|
|
y_binary[np.where(y!=0)[0]] = 1
|
|
return y_binary
|
|
|
|
|
|
def _get_best_f1(label, score):
|
|
precision, recall, _ = metrics.precision_recall_curve(y_true=label, probas_pred=score)
|
|
f1 = 2 * precision * recall / (precision + recall + 1e-5)
|
|
best_f1 = f1[np.argmax(f1)]
|
|
best_p = precision[np.argmax(f1)]
|
|
best_r = recall[np.argmax(f1)]
|
|
return best_f1, best_p, best_r
|
|
|
|
|
|
def cal_metrics(label, score, pa=False):
|
|
if pa:
|
|
score = _adjust_scores(label, score)
|
|
auroc = metrics.roc_auc_score(label, score)
|
|
ap = metrics.average_precision_score(y_true=label, y_score=score, average=None)
|
|
best_f1, best_p, best_r = _get_best_f1(label, score)
|
|
|
|
return auroc, ap, best_f1, best_p, best_r
|
|
|
|
|
|
def _adjust_scores(label, score):
|
|
"""
|
|
adjust the score for segment detection. i.e., for each ground-truth anomaly segment,
|
|
use the maximum score as the score of all points in that segment. This corresponds to point-adjust f1-score.
|
|
** This function is copied/modified from the source code in [Zhihan Li et al. KDD21]
|
|
|
|
Parameters
|
|
----------
|
|
label: np.array, required
|
|
data label, 0 indicates normal timestamp, and 1 is anomaly
|
|
|
|
score: np.array, required
|
|
anomaly score, higher score indicates higher likelihoods to be anomaly
|
|
"""
|
|
score = score.copy()
|
|
assert len(score) == len(label)
|
|
splits = np.where(label[1:] != label[:-1])[0] + 1
|
|
is_anomaly = label[0] == 1
|
|
pos = 0
|
|
for sp in splits:
|
|
if is_anomaly:
|
|
score[pos:sp] = np.max(score[pos:sp])
|
|
is_anomaly = not is_anomaly
|
|
pos = sp
|
|
sp = len(label)
|
|
if is_anomaly:
|
|
score[pos:sp] = np.max(score[pos:sp])
|
|
return score
|