import numpy as np
import pandas as pd
from CommonServerPython import *
from sklearn.metrics import precision_recall_curve, precision_score, recall_score
from tabulate import tabulate
# pylint: disable=no-member
METRICS = {}
METRICS["Precision"] = (
"The precision of the class in the evaluation set that were classified as this class by the "
"model. Precision is calculated by dividing the TPs of the class by the number of incidents that "
"the model predicted as this class."
)
METRICS["TP (true positive)"] = "The number of incidents from the class in the evaluation set that were predicted correctly. "
METRICS["FP (false positive)"] = "The number of incidents from other classes that were predicted incorrectly as this class."
METRICS["Coverage"] = (
"The number of incidents from the class in the evaluation set for which the confidence level of "
"the model exceeded the threshold in the prediction."
)
METRICS["Total"] = "The total number of incidents from the class in the evaluation set."
def bold_hr(s):
return f"**{s}:**"
def binarize(arr, threshold):
return np.where(arr >= threshold, 1.0, 0)
def calculate_confusion_matrix(y_true, y_pred, y_pred_per_class, threshold):
indices_higher_than_threshold = set()
for i, y in enumerate(y_pred):
if y_pred_per_class[y][i] >= threshold:
indices_higher_than_threshold.add(i)
y_true_at_threshold = [y for i, y in enumerate(y_true) if i in indices_higher_than_threshold]
y_pred_at_threshold = [y for i, y in enumerate(y_pred) if i in indices_higher_than_threshold]
test_tag = pd.Series(y_true_at_threshold)
ft_test_predictions_labels = pd.Series(y_pred_at_threshold)
csr_matrix = pd.crosstab(test_tag, ft_test_predictions_labels, rownames=["True"], colnames=["Predicted"], margins=True)
return csr_matrix
def generate_metrics_df(y_true, y_true_per_class, y_pred, y_pred_per_class, threshold):
df = pd.DataFrame(columns=["Class", "Precision", "Recall", "TP", "FP", "Coverage", "Total"])
for class_ in sorted(y_pred_per_class):
row = calculate_df_row(class_, threshold, y_true_per_class, y_pred_per_class)
df = pd.concat([df, pd.DataFrame([row])], ignore_index=True)
df = pd.concat(
[
df,
pd.DataFrame(
[
{
"Class": "All",
"Precision": df["Precision"].mean(),
"Recall": df["Recall"].mean(),
"TP": df["TP"].sum(),
"FP": df["FP"].sum(),
"Coverage": df["Coverage"].sum(),
"Total": df["Total"].sum(),
}
]
),
],
ignore_index=True,
)
df = df[["Class", "Precision", "TP", "FP", "Coverage", "Total"]]
explained_metrics = ["Precision", "TP (true positive)", "FP (false positive)", "Coverage", "Total"]
explanation = [f"{bold_hr(metric)} {METRICS[metric]}" for metric in explained_metrics]
df = df.set_index("Class")
return df, explanation
def calculate_df_row(class_, threshold, y_true_per_class, y_pred_per_class):
y_pred_class = y_pred_per_class[class_]
y_true_class = y_true_per_class[class_]
y_pred_class_binary = binarize(y_pred_class, threshold)
precision = precision_score(y_true=y_true_class, y_pred=y_pred_class_binary)
recall = recall_score(y_true=y_true_class, y_pred=y_pred_class_binary)
classified_correctly = sum(
1 for y_true_i, y_pred_i in zip(y_true_class, y_pred_class_binary) if y_true_i == 1 and y_pred_i == 1
)
above_thresh = sum(
1
for i, y_true_i in enumerate(y_true_class)
if y_true_i == 1 and any(y_pred_per_class[c][i] >= threshold for c in y_pred_per_class)
)
fp = sum(1 for i, y_true_i in enumerate(y_true_class) if y_true_i == 0 and y_pred_class_binary[i] == 1.0)
total = int(sum(y_true_class))
row = {
"Class": class_,
"Precision": precision,
"Recall": recall,
"TP": classified_correctly,
"FP": fp,
"Coverage": int(above_thresh),
"Total": total,
}
return row
def reformat_df_fractions_to_percentage(metrics_df):
hr_df = metrics_df.copy()
hr_df["Precision"] = hr_df["Precision"].apply(lambda p: f"{p * 100:.1f}%")
hr_df["TP"] = hr_df.apply(
lambda row: "{}/{} ({:.1f}%)".format(int(row["TP"]), int(row["Coverage"]), float(row["TP"]) * 100 / row["Coverage"]),
axis=1,
)
hr_df["Coverage"] = hr_df.apply(
lambda row: "{}/{} ({:.1f}%)".format(int(row["Coverage"]), row["Total"], float(row["Coverage"]) * 100 / row["Total"]),
axis=1,
)
return hr_df
def output_report(
y_true,
y_true_per_class,
y_pred,
y_pred_per_class,
found_threshold,
target_precision,
actual_threshold_precision,
detailed_output=True,
):
csr_matrix_at_threshold = calculate_confusion_matrix(y_true, y_pred, y_pred_per_class, found_threshold)
csr_matrix_no_threshold = calculate_confusion_matrix(y_true, y_pred, y_pred_per_class, 0)
metrics_df, metrics_explanation = generate_metrics_df(y_true, y_true_per_class, y_pred, y_pred_per_class, found_threshold)
coverage = metrics_df.loc[["All"]]["Coverage"][0]
test_set_size = metrics_df.loc[["All"]]["Total"][0]
human_readable_threshold = ["## Summary"]
# in case the found threshold meets the target accuracy
if actual_threshold_precision >= target_precision or abs(found_threshold - target_precision) < 10**-2:
human_readable_threshold += [
f"- A confidence threshold of {found_threshold:.2f} meets the conditions of required precision."
]
else:
human_readable_threshold += [
"- Could not find a threshold which meets the conditions of required precision. "
f"The confidence threshold of {found_threshold:.2f} achieved highest "
"possible precision"
]
human_readable_threshold += [
f"- {int(coverage)}/{int(test_set_size)} incidents of the evaluation set were predicted with higher confidence"
f" than this threshold.",
f"- The remainder, {int(test_set_size - coverage)}/{int(test_set_size)} incidents of the evaluation set, were"
f" predicted with lower confidence than this threshold "
"(these predictions were ignored).",
f"- Expected coverage ratio: The model will attempt to provide a prediction for"
f" {float(coverage) / test_set_size * 100:.2f}% of incidents. "
f"({int(coverage)}/{int(test_set_size)})",
"- Evaluation of the model performance using this probability threshold can be found below:",
]
pd.set_option("display.max_columns", None)
tablualted_csr = tabulate(reformat_df_fractions_to_percentage(metrics_df), tablefmt="pipe", headers="keys")
class_metrics_human_readable = ["## Metrics per Class", tablualted_csr]
class_metrics_explanation_human_readable = ["### Metrics Explanation"] + ["- " + row for row in metrics_explanation]
csr_matrix_readable = [
"## Confusion Matrix",
"This table displays the predictions of the model on the evaluation set per each " + "class:",
tabulate(csr_matrix_at_threshold, tablefmt="pipe", headers="keys").replace("True", "True \\ Predicted"),
"\n",
]
csr_matrix_no_thresh_readable = [
"## Confusion Matrix - No Threshold",
"This table displays the predictions of the model on the evaluation set per each " + "class when no threshold is used:",
tabulate(csr_matrix_no_threshold, tablefmt="pipe", headers="keys").replace("True", "True \\ Predicted"),
"\n",
]
human_readable = [] # type: ignore
if detailed_output:
human_readable += human_readable_threshold + ["\n"]
else:
human_readable += [f"## Results for confidence threshold = {found_threshold:.2f}"] + ["\n"]
human_readable += class_metrics_human_readable + ["\n"]
human_readable += class_metrics_explanation_human_readable
human_readable += csr_matrix_readable
human_readable += csr_matrix_no_thresh_readable
human_readable = "\n".join(human_readable)
contents = {
"threshold": found_threshold,
"csr_matrix_at_threshold": csr_matrix_at_threshold.to_json(orient="index"),
"csr_matrix_no_threshold": csr_matrix_no_threshold.to_json(orient="index"),
"metrics_df": metrics_df.to_json(),
}
entry = {
"Type": entryTypes["note"],
"Contents": contents,
"ContentsFormat": formats["json"],
"HumanReadable": human_readable,
"HumanReadableFormat": formats["markdown"],
"EntryContext": {
"GetMLModelEvaluation": {
"Threshold": found_threshold,
"ConfusionMatrixAtThreshold": csr_matrix_at_threshold.to_json(orient="index"),
"ConfusionMatrixNoThreshold": csr_matrix_no_threshold.to_json(orient="index"),
"Metrics": metrics_df.to_json(),
}
},
}
return entry
def merge_entries(entry, per_class_entry):
entry = {
"Type": entryTypes["note"],
"Contents": entry["Contents"],
"ContentsFormat": formats["json"],
"HumanReadable": entry["HumanReadable"] + "\n" + per_class_entry["HumanReadable"],
"HumanReadableFormat": formats["markdown"],
"EntryContext": {**entry["EntryContext"], **per_class_entry["EntryContext"]},
}
return entry
def find_threshold(y_true, y_pred_all_classes, customer_target_precision, target_recall, detailed_output=True):
labels = sorted(set(y_true + list(y_pred_all_classes[0].keys())))
n_instances = len(y_true)
y_true_per_class = {class_: np.zeros(n_instances) for class_ in labels}
for i, y in enumerate(y_true):
y_true_per_class[y][i] = 1.0
y_pred_per_class = {class_: np.zeros(n_instances) for class_ in labels}
y_pred = []
for i, y in enumerate(y_pred_all_classes):
predicted_class = sorted(y.items(), key=lambda x: x[1], reverse=True)[0][0]
y_pred_per_class[predicted_class][i] = y[predicted_class]
y_pred.append(predicted_class)
class_to_arrs = {class_: {} for class_ in labels} # type: Dict[str, Dict[str, Any]]
for class_ in labels:
precision_arr, recall_arr, thresholds_arr = precision_recall_curve(y_true_per_class[class_], y_pred_per_class[class_])
class_to_arrs[class_]["precisions"] = precision_arr
class_to_arrs[class_]["recalls"] = recall_arr
class_to_arrs[class_]["thresholds"] = thresholds_arr
# find threshold for all classes such as precision of all classes are higher than target precision:
unified_threshold, unified_threshold_precision, target_unified_precision = find_best_threshold_for_target_precision(
class_to_arrs, customer_target_precision, labels
)
if unified_threshold is None or unified_threshold_precision is None:
error_message = f"Could not find any threshold at ranges {target_unified_precision} - {customer_target_precision:.2f}."
return_error(error_message)
entry = output_report(
np.array(y_true),
y_true_per_class,
np.array(y_pred),
y_pred_per_class,
unified_threshold,
customer_target_precision,
unified_threshold_precision,
detailed_output,
)
per_class_entry = calculate_per_class_report_entry(class_to_arrs, labels, y_pred_per_class, y_true_per_class)
res = merge_entries(entry, per_class_entry)
return res
def find_best_threshold_for_target_precision(class_to_arrs, customer_target_precision, labels):
target_unified_precision = round(customer_target_precision, 2)
unified_threshold_found = False
threshold = None
threshold_precision = None
while not unified_threshold_found:
threshold_per_class = {}
precision_per_class = {}
for class_ in labels:
# indexing is done by purpose - the ith precision corresponds with threshold i-1. Last precision is 1
for i, precision in enumerate(class_to_arrs[class_]["precisions"][:-1]):
if class_to_arrs[class_]["thresholds"][i] == 0:
continue
if precision > target_unified_precision:
threshold_per_class[class_] = class_to_arrs[class_]["thresholds"][i]
precision_per_class[class_] = precision
break
if len(threshold_per_class) == len(labels):
threshold_candidates = sorted(threshold_per_class.values())
for threshold in threshold_candidates:
legal_threshold_for_all_classes = True
threshold_precision = sys.maxsize
for class_ in labels:
i = np.argmax(class_to_arrs[class_]["thresholds"] >= threshold) # type: ignore
threshold_precision_for_class = class_to_arrs[class_]["precisions"][i]
threshold_precision = min(threshold_precision, threshold_precision_for_class) # type: ignore
if threshold_precision_for_class >= target_unified_precision:
legal_threshold_for_all_classes = True
else:
legal_threshold_for_all_classes = False
break
if legal_threshold_for_all_classes:
unified_threshold_found = True
break
elif target_unified_precision < 0:
break
target_unified_precision -= 0.01
return threshold, threshold_precision, target_unified_precision
def calculate_per_class_report_entry(class_to_arrs, labels, y_pred_per_class, y_true_per_class):
per_class_hr = ["## Per-Class Report"]
per_class_hr += ["The following tables present evlauation of the model per class at different confidence thresholds:"]
class_to_thresholds = {}
for class_ in labels:
class_to_thresholds[class_] = {0.001} # using no threshold
for target_precision in np.arange(0.95, 0.5, -0.05):
# indexing is done by purpose - the ith precision corresponds with threshold i-1. Last precision is 1
for i, precision in enumerate(class_to_arrs[class_]["precisions"][:-1]):
if class_to_arrs[class_]["thresholds"][i] == 0:
continue
if precision > target_precision and class_to_arrs[class_]["recalls"][i] > 0:
threshold = class_to_arrs[class_]["thresholds"][i]
class_to_thresholds[class_].add(threshold)
break
if len(class_to_thresholds[class_]) >= 4:
break
per_class_context = {}
for class_ in labels:
class_threshold_df = pd.DataFrame(columns=["Threshold", "Precision", "Recall", "TP", "FP", "Coverage", "Total"])
for threshold in sorted(class_to_thresholds[class_]):
row = calculate_df_row(class_, threshold, y_true_per_class, y_pred_per_class)
row["Threshold"] = threshold
class_threshold_df = pd.concat([class_threshold_df, pd.DataFrame([row])], ignore_index=True)
class_threshold_df = reformat_df_fractions_to_percentage(class_threshold_df)
class_threshold_df["Threshold"] = class_threshold_df["Threshold"].apply(lambda p: f"{p:.2f}")
class_threshold_df = class_threshold_df[["Threshold", "Precision", "TP", "FP", "Coverage", "Total"]]
class_threshold_df = class_threshold_df.sort_values(by="Coverage", ascending=False)
class_threshold_df = class_threshold_df.drop_duplicates(subset="Threshold", keep="first")
class_threshold_df = class_threshold_df.drop_duplicates(subset="Precision", keep="first")
class_threshold_df = class_threshold_df.set_index("Threshold")
per_class_context[class_] = class_threshold_df.to_json()
tabulated_class_df = tabulate(class_threshold_df, tablefmt="pipe", headers="keys")
per_class_hr += [f"### {class_}", tabulated_class_df]
per_class_entry = {
"Type": entryTypes["note"],
"ContentsFormat": formats["json"],
"Contents": [],
"HumanReadable": "\n".join(per_class_hr),
"HumanReadableFormat": formats["markdown"],
"EntryContext": {"GetMLModelEvaluation": {"PerClassReport": per_class_context}},
}
return per_class_entry
def convert_str_to_json(str_json, var_name):
try:
y_true = json.loads(str_json)
return y_true
except Exception as e:
return_error(f"Exception while reading {var_name} :{e}")
def main():
try:
y_pred_all_classes = demisto.args()["yPred"]
y_true = demisto.args()["yTrue"]
target_precision = calculate_and_validate_float_parameter("targetPrecision")
target_recall = calculate_and_validate_float_parameter("targetRecall")
detailed_output = "detailedOutput" in demisto.args() and demisto.args()["detailedOutput"] == "true"
y_true = convert_str_to_json(y_true, "yTrue")
y_pred_all_classes = convert_str_to_json(y_pred_all_classes, "yPred")
if not (y_true and y_pred_all_classes):
raise DemistoException('Either "yPred" or "yTrue" are empty.')
entries = find_threshold(
y_true=y_true,
y_pred_all_classes=y_pred_all_classes,
customer_target_precision=target_precision,
target_recall=target_recall,
detailed_output=detailed_output,
)
demisto.results(entries)
except Exception as e:
return_error(f"Error in GetMLModelEvaluation:\n{e}")
def calculate_and_validate_float_parameter(var_name):
try:
res = float(demisto.args()[var_name]) if var_name in demisto.args() else 0
except Exception:
return_error(f"{var_name} must be a float between 0-1 or left empty")
if res < 0 or res > 1:
return_error(f"{var_name} must be a float between 0-1 or left empty")
return res
if __name__ in ["__main__", "__builtin__", "builtins"]:
main()