67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
import numpy as np
|
||
from sklearn.metrics import f1_score, precision_score, recall_score
|
||
|
||
|
||
def adjust_predicts(score, label,
|
||
threshold=None,
|
||
pred=None,
|
||
calc_latency=False):
|
||
"""
|
||
该函数是point-adjust官方源码
|
||
Calculate adjusted predict labels using given `score`, `threshold` (or given `pred`) and `label`.
|
||
Args:
|
||
score =: The anomaly score
|
||
label : The ground-truth label
|
||
threshold (float): The threshold of anomaly score.
|
||
A point is labeled as "anomaly" if its score is lower than the threshold.
|
||
pred : if not None, adjust `pred` and ignore `score` and `threshold`,
|
||
calc_latency (bool):
|
||
Returns:
|
||
np.ndarray: predict labels
|
||
"""
|
||
if len(score) != len(label):
|
||
raise ValueError("score and label must have the same length")
|
||
score = np.asarray(score)
|
||
label = np.asarray(label)
|
||
latency = 0
|
||
if pred is None:
|
||
predict = score < threshold
|
||
else:
|
||
predict = pred
|
||
actual = label > 0.1
|
||
anomaly_state = False
|
||
anomaly_count = 0
|
||
for i in range(len(score)):
|
||
if actual[i] and predict[i] and not anomaly_state:
|
||
anomaly_state = True
|
||
anomaly_count += 1
|
||
for j in range(i, 0, -1):
|
||
if not actual[j]:
|
||
break
|
||
else:
|
||
if not predict[j]:
|
||
predict[j] = True
|
||
latency += 1
|
||
elif not actual[i]:
|
||
anomaly_state = False
|
||
if anomaly_state:
|
||
predict[i] = True
|
||
if calc_latency:
|
||
return predict, latency / (anomaly_count + 1e-4)
|
||
else:
|
||
return predict
|
||
|
||
|
||
def evaluate(y_true: list, y_pred: list) -> float:
|
||
"""
|
||
F1PA评估方法，经过point adjust调整标签后再用F1评分
|
||
:param y_true: 真实标签
|
||
:param y_pred: 检测标签
|
||
:return: 经过pa调整后的f1、recall、precision
|
||
"""
|
||
y_true, y_pred = y_true.copy(), y_pred.copy()
|
||
adjust_y_pred = adjust_predicts(score=np.array([0] * len(y_true)), label=y_true, pred=y_pred)
|
||
f1 = f1_score(y_true, adjust_y_pred)
|
||
return f1
|
||
|