import json import matplotlib.pyplot as plt import numpy as np import pandas as pd import seaborn as sns from numpy.linalg import norm from scipy.stats import entropy, pearsonr, spearmanr from sklearn.metrics import auc, f1_score, mean_squared_error, roc_curve from sklearn.preprocessing import minmax_scale SHELVE_PATHS = { "gpt2": "./data/gpt2_eval.json", "gpt2background": "./data/gpt2background_eval.json", "gpt2odin": "./data/gpt2odin_eval.json", } LABEL2VALUE = { "O": 0.0, # "Not a breakdown" "T": 0.5, # "Possible breakdown" "X": 1.0, # "Breakdown" } VALUE2LABEL = { "0.0": "Not a breakdown", "0.5": "Possible breakdown", "1.0": "Breakdown", } PRED_COLUMN2LABEL = { "gpt2_target_score": "gpt2_target_score", "gpt2_argmax_score": "GPT2 Argmax Score", "llr": "LLR", "h_max": "ODIN-h-max", "g": "ODIN-g", } SUM_COLUMNS = ["gpt2_target_score", "gpt2_argmax_score", "h_max", "g"] PRED_COLUMNS = ["gpt2_target_score", "gpt2_argmax_score", "llr", "h_max", "g"] ANNOTATION_COLUMNS = ["annotations_avg", "annotations_maj"] def JSD(P, Q): _P = P / norm(P, ord=1) _Q = Q / norm(Q, ord=1) _M = 0.5 * (_P + _Q) return 0.5 * (entropy(_P, _M, base=np.e) + entropy(_Q, _M, base=np.e)) def _sum(x): if isinstance(x[0], list): total_sum = 0.0 for item in x: if len(item) == 1: item = item[0] total_sum += item else: raise ValueError("List instead of scalar is provied") return total_sum else: return sum(x) def _convert_values(annotations): return [LABEL2VALUE[annotation.upper()] for annotation in annotations] def _majority_vote(annotations): values, counts = np.unique(annotations, return_counts=True) ind = np.argmax(counts) return values[ind] def _to_labels(pos_probs, threshold): return (pos_probs >= threshold).astype("int") def _to_predict_dist(pos_probs): return [pos_probs, 1 - pos_probs] def _to_anno_dist(annnotations, lbls): total = float(len(annnotations)) if "PB+B" in lbls: positive = float(len([a for a in annnotations if a >= 0.5])) else: positive = float(len([a for a in annnotations if a > 0.5])) positive = positive / total negative = 1 - positive return [positive, negative] def _calculate_llr(x): regular, background = x # we are already in log space return float(sum(regular) - sum(background)) def calculate_metrics(): merged_eval_data = dict() for model_name, eval_shelve_path in SHELVE_PATHS.items(): with open(eval_shelve_path, "rt") as fin: model_eval_data = json.load(fin) for key, value in model_eval_data.items(): if key not in merged_eval_data: merged_eval_data[key] = value else: merged_eval_data[key].update(value) merged_eval_data_df = pd.DataFrame.from_dict(merged_eval_data, orient="index") merged_eval_data_df["annotations"] = merged_eval_data_df["annotations"].apply( _convert_values ) merged_eval_data_df["annotations_avg"] = merged_eval_data_df["annotations"].apply( np.mean ) merged_eval_data_df["annotations_maj"] = merged_eval_data_df["annotations"].apply( _majority_vote ) merged_eval_data_df["llr"] = merged_eval_data_df[ ["gpt2_target_score", "gpt2background_target_score"] ].apply(_calculate_llr, axis=1) for col in SUM_COLUMNS: # sum since we are in log-space merged_eval_data_df[col] = merged_eval_data_df[col].apply(_sum) for col in PRED_COLUMNS: merged_eval_data_df[col] = minmax_scale(merged_eval_data_df[col]) # inverse the probability score, because we predict the probability for not a breakdown merged_eval_data_df[col] = merged_eval_data_df[col].apply(lambda x: 1.0 - x) for pred_col in PRED_COLUMNS: for c in [0.0, 0.5, 1.0]: data = merged_eval_data_df[pred_col][ merged_eval_data_df["annotations_maj"] == c ] ax = sns.kdeplot(data, label=VALUE2LABEL[str(c)]) title = PRED_COLUMN2LABEL[pred_col] ax.set(xlabel="Normalized Score") plt.legend() plt.title(f"{title}") plt.savefig(f"{pred_col}_dist.png") plt.close() fpr, tpr, thresholds = roc_curve( y_true=merged_eval_data_df["annotations_maj"].apply(str), y_score=merged_eval_data_df[pred_col], pos_label=str(1.0), ) auc_val = auc(fpr, tpr) plt.plot(fpr, tpr) plt.plot(np.array(range(2)), linestyle="dotted") plt.title(f"{title} AUC: {auc_val}") plt.savefig(f"{pred_col}_roc.png") plt.close() # calculate F1 scores print(f"{pred_col} >>> F1 Scores") thresholds = np.arange(0.1, 0.9, 0.001) test_ys = [ (merged_eval_data_df["annotations_maj"] >= 0.5).astype("int"), (merged_eval_data_df["annotations_maj"] > 0.5).astype("int"), ] for test_y, lbls in zip( test_ys, [ "PB+B", "B", ], ): probs = merged_eval_data_df[pred_col] f1_scores = [ round(f1_score(test_y, _to_labels(probs, t)), 4) for t in thresholds ] f1_ix = np.argmax(f1_scores) print( f"{pred_col} ({lbls}): " + "Threshold=%.4f, F-Score=%.4f" % (thresholds[f1_ix], f1_scores[f1_ix]) ) # calculate F1 scores print(f"{pred_col} >>> Dist Scores") for lbls in [ "PB+B", "B", ]: for dist_score_func in [JSD, mean_squared_error]: test_y = ( merged_eval_data_df["annotations"] .apply(_to_anno_dist, lbls=lbls) .values.tolist() ) pred_y = ( merged_eval_data_df[pred_col] .apply(_to_predict_dist) .values.tolist() ) dist_scores = [dist_score_func(*pair) for pair in zip(test_y, pred_y)] dist_scores = np.mean(np.ma.masked_invalid(dist_scores)) dist_scores = round(dist_scores, 4) dist_score_name = dist_score_func.__name__ print(f"{pred_col} {dist_score_name} ({lbls}): {dist_scores}") # calculate correlations scores print(f"{pred_col} >>> Correlation Scores") for annotation_col in ANNOTATION_COLUMNS: for corr_func in [pearsonr, spearmanr]: corr = corr_func( merged_eval_data_df[annotation_col], merged_eval_data_df[pred_col], ) corr_name = corr_func.__name__ corr = [round(val, 4) for val in corr] print(f"{pred_col}-{annotation_col}-{corr_name}: {corr}")