Alex Galarce
780dac2ec1
Feature/fuzzy match no numbers * Add fuzzy matching function for string comparison and enhance term extraction result evaluation * Enhance term extraction results evaluation by adding counts and readable strings for matched, false negatives, and false positives * Adjust text similarity threshold to 65 and improve readability of comparison results in term extraction evaluation * Merged main into feature/fuzzy-match-no-numbers Approved-by: Katon Minhas
888 lines
38 KiB
Python
888 lines
38 KiB
Python
|
|
import pandas as pd
|
|
import src.utils.string_utils as string_utils
|
|
import itertools
|
|
import re
|
|
|
|
from collections import Counter
|
|
from rapidfuzz import fuzz
|
|
|
|
# Imports for the testbed postprocessing functions
|
|
from src.investment import investment_postprocessing_funcs
|
|
from src.constants.investment_columns import COLUMN_ORDER
|
|
|
|
|
|
def convert_to_float(value):
|
|
try:
|
|
return f"{float(value):.2f}" if isinstance(value, (int, float)) or value.replace('.', '', 1).isdigit() else value
|
|
except:
|
|
return value
|
|
|
|
def normalize_tin_npi(value):
|
|
"""Normalize TIN/NPI values by removing decimal points and trailing zeros."""
|
|
if string_utils.is_empty(value):
|
|
return ""
|
|
|
|
# Skip "UNKNOWN" and non-numeric values
|
|
if value == "UNKNOWN" or not value.replace('.', '', 1).isdigit():
|
|
return value
|
|
|
|
# Convert to integer-like string by removing decimal part
|
|
try:
|
|
# Remove decimal and any trailing zeros
|
|
return str(int(float(value)))
|
|
except:
|
|
return value
|
|
|
|
def testbed_preprocess(df):
|
|
|
|
# Capitalize and remove ".TXT" suffixes from FILE_NAME
|
|
df['FILE_NAME'] = df['FILE_NAME'].str.upper().str.replace(".TXT", "", regex=False)
|
|
|
|
# Apply numeric conversion to non-CPT/diag/rev fields only
|
|
for col in df.columns:
|
|
if "_CD" not in col and "CPT" not in col and "DIAG" not in col and "REV" not in col:
|
|
df[col] = df[col].apply(convert_to_float)
|
|
else:
|
|
# For CPT/diag/rev fields, apply string conversion AND convert | delimiters
|
|
# to commas
|
|
df[col] = df[col].astype(str).str.replace("|", ",", regex=False)
|
|
|
|
df = df.applymap(lambda x: str(x).strip().upper())
|
|
df = df.replace("UNKNOWN", "").fillna("")
|
|
df = df.replace("NAN", "")
|
|
df = df.replace("N/A", "")
|
|
|
|
# Normalize TIN and NPI columns
|
|
tin_columns = [col for col in df.columns if 'TIN' in col]
|
|
npi_columns = [col for col in df.columns if 'NPI' in col]
|
|
for col in tin_columns + npi_columns:
|
|
df[col] = df[col].str.replace("-", "", regex=False)
|
|
df[col] = df[col].apply(normalize_tin_npi)
|
|
df[col] = df[col].str.replace(" ", "", regex=False)
|
|
|
|
df["AUTO_RENEWAL_TERM"] = df['AUTO_RENEWAL_TERM'].str.replace("ONE YEAR", "1 YEAR", regex=False)
|
|
|
|
# Ensure that all empty cells are filled with empty strings
|
|
df = df.fillna("")
|
|
|
|
return df
|
|
|
|
def calculate_field_metrics(merged, field):
|
|
"""Calculate precision, recall, and accuracy for a single field using fuzzy matching.
|
|
|
|
Args:
|
|
merged: DataFrame containing predicted and truth values
|
|
field: Name of field to evaluate
|
|
|
|
Returns:
|
|
tuple: (precision, recall, accuracy)
|
|
"""
|
|
def is_positive(val):
|
|
return not string_utils.is_empty(val)
|
|
|
|
def is_negative(val):
|
|
return string_utils.is_empty(val)
|
|
|
|
def normalize_term(text: str) -> str:
|
|
"""Normalize term text for comparison"""
|
|
replacements = {
|
|
"ONE (1)": "1",
|
|
"ONE": "1",
|
|
"TWO": "2",
|
|
"THREE (3)": "3",
|
|
"THREE": "3",
|
|
"FOUR": "4",
|
|
"FIVE": "5",
|
|
"YEAR TO YEAR": "ANNUAL",
|
|
"FROM THE CONTRACT EFFECTIVE DATE": "",
|
|
"FROM CONTRACT EFFECTIVE DATE": "",
|
|
"TERM": "",
|
|
"YEAR(S)": "YEAR",
|
|
"YEARS": "YEAR",
|
|
"-YEAR": " YEAR",
|
|
" ": " "
|
|
}
|
|
|
|
text = text.upper()
|
|
for old, new in replacements.items():
|
|
text = text.replace(old, new)
|
|
|
|
return text.strip()
|
|
|
|
def is_fuzzy_match(pred, truth):
|
|
"""Compare strings using fuzzy matching with preprocessing
|
|
|
|
Args:
|
|
pred: Predicted value
|
|
truth: Ground truth value
|
|
|
|
Returns:
|
|
bool: True if strings match within threshold
|
|
"""
|
|
if is_negative(pred) or is_negative(truth):
|
|
return False
|
|
|
|
# Normalize both strings
|
|
pred_norm = normalize_term(str(pred))
|
|
truth_norm = normalize_term(str(truth))
|
|
|
|
# First try exact match on normalized strings
|
|
if pred_norm == truth_norm or pred_norm in truth_norm or truth_norm in pred_norm:
|
|
return True
|
|
|
|
# Then try fuzzy match
|
|
match_ratio = fuzz.ratio(pred_norm, truth_norm)
|
|
|
|
# # Debugging line
|
|
# if match_ratio < 100 and field == "field of interest":
|
|
# print(f"Fuzzy match ratio for '{pred_norm}' | '{truth_norm}': {match_ratio}")
|
|
|
|
return match_ratio >= 80
|
|
|
|
# True Positive: both pred and truth are non-empty and fuzzy match >= 90%
|
|
TP = sum(
|
|
is_positive(pred) and
|
|
is_positive(truth) and
|
|
is_fuzzy_match(pred, truth)
|
|
for pred, truth in zip(merged[f'{field}_pred'], merged[f'{field}_truth'])
|
|
)
|
|
|
|
# False Positive: pred is non-empty, truth is empty or no fuzzy match
|
|
FP = sum(
|
|
is_positive(pred) and
|
|
(is_negative(truth) or not is_fuzzy_match(pred, truth))
|
|
for pred, truth in zip(merged[f'{field}_pred'], merged[f'{field}_truth'])
|
|
)
|
|
|
|
# True Negative: both pred and truth are empty
|
|
TN = sum(
|
|
is_negative(pred) and is_negative(truth)
|
|
for pred, truth in zip(merged[f'{field}_pred'], merged[f'{field}_truth'])
|
|
)
|
|
|
|
# False Negative: pred is empty, truth is non-empty
|
|
FN = sum(
|
|
is_negative(pred) and is_positive(truth)
|
|
for pred, truth in zip(merged[f'{field}_pred'], merged[f'{field}_truth'])
|
|
)
|
|
|
|
# print(field)
|
|
# print(f"TP: {TP}, FP: {FP}, TN: {TN}, FN: {FN}")
|
|
|
|
# Calculate metrics
|
|
precision = TP / (TP + FP) if (TP + FP) > 0 else 0
|
|
recall = TP / (TP + FN) if (TP + FN) > 0 else 0
|
|
accuracy = (TP + TN) / (TP + FP + TN + FN)
|
|
|
|
return precision, recall, accuracy
|
|
|
|
|
|
def evaluate_provider_fields_separately(testbed_df, results_df, verbose=False):
|
|
"""Evaluates TIN, NPI, and NAME fields separately for each provider."""
|
|
# Initialize metrics for each field type
|
|
metrics = {
|
|
"TIN": {'tp' : 0, 'fp': 0, 'fn': 0, 'accuracy_file': []},
|
|
"NPI": {'tp' : 0, 'fp': 0, 'fn': 0, 'accuracy_file': []},
|
|
"NAME": {'tp' : 0, 'fp': 0, 'fn': 0, 'accuracy_file': []}
|
|
}
|
|
|
|
if verbose: # instantiate provider_metrics.txt if verbose
|
|
csv_output_dicts = []
|
|
with open("provider_metrics.txt", "w") as f:
|
|
f.write("Provider Metrics\n")
|
|
|
|
|
|
for file in testbed_df['FILE_NAME'].unique():
|
|
# Get rows for this file
|
|
testbed_file = testbed_df[testbed_df['FILE_NAME'] == file]
|
|
results_file = results_df[results_df['FILE_NAME'] == file]
|
|
|
|
# Extract all TINs, NPIs, and NAMEs for ground truth
|
|
truth_tins = []
|
|
truth_npis = []
|
|
truth_names = []
|
|
|
|
# Add group provider info (with splitting)
|
|
for col_type, truth_list in [
|
|
("PROV_GROUP_TIN", truth_tins),
|
|
("PROV_GROUP_NPI", truth_npis),
|
|
("PROV_GROUP_NAME_FULL", truth_names)
|
|
]:
|
|
|
|
value = testbed_file[col_type].iloc[0]
|
|
values = []
|
|
pipe_split = value.split("|")
|
|
for item in pipe_split:
|
|
if col_type in ["PROV_GROUP_NPI", "PROV_GROUP_TIN"] and "," in item: # further split by comma, only if col_type is PROV_GROUP_NPI or PROV_GROUP_TIN
|
|
for comma_item in item.split(","):
|
|
comma_item = comma_item.strip()
|
|
if comma_item:
|
|
values.append(comma_item)
|
|
elif item.strip():
|
|
values.append(item.strip())
|
|
|
|
# Now append all values to the list
|
|
for v in values:
|
|
if v and v != "UNKNOWN":
|
|
truth_list.append(v)
|
|
|
|
# Add other provider info
|
|
for col_type, truth_list in [
|
|
("PROV_OTHER_TIN", truth_tins),
|
|
("PROV_OTHER_NPI", truth_npis),
|
|
("PROV_OTHER_NAME_FULL", truth_names)
|
|
]:
|
|
if not string_utils.is_empty(testbed_file[col_type].iloc[0]):
|
|
values = testbed_file[col_type].iloc[0].split("|")
|
|
if len(values) == 1: # possibly delimited by comma
|
|
values = testbed_file[col_type].iloc[0].split(",")
|
|
for value in values:
|
|
value = value.strip()
|
|
if value:
|
|
truth_list.append(value)
|
|
|
|
# Similarly, extract all TINs, NPIs, and NAMEs for predictions
|
|
pred_tins = []
|
|
pred_npis = []
|
|
pred_names = []
|
|
|
|
# Add group provider info
|
|
value = results_file['PROV_GROUP_TIN'].iloc[0]
|
|
pred_tins.append(value)
|
|
value = results_file['PROV_GROUP_NPI'].iloc[0]
|
|
pred_npis.append(value)
|
|
value = results_file['PROV_GROUP_NAME_FULL'].iloc[0]
|
|
pred_names.append(value)
|
|
|
|
# Add other provider info
|
|
for col_type, pred_list in [
|
|
("PROV_OTHER_TIN", pred_tins),
|
|
("PROV_OTHER_NPI", pred_npis),
|
|
("PROV_OTHER_NAME_FULL", pred_names)
|
|
]:
|
|
values = results_file[col_type].iloc[0].split("|")
|
|
if len(values) == 1: # possibly delimited by comma
|
|
values = results_file[col_type].iloc[0].split(",")
|
|
for value in values:
|
|
value = value.strip()
|
|
pred_list.append(value)
|
|
|
|
file_metrics = {} # Store file-specific metrics for verbose output
|
|
# Calculate metrics for each field type and print verbose output if requested
|
|
for field_type, truth_list, pred_list in [
|
|
("TIN", truth_tins, pred_tins),
|
|
("NPI", truth_npis, pred_npis),
|
|
("NAME", truth_names, pred_names)
|
|
]:
|
|
# Count occurrences in lists, excluding empty strings and "UNKNOWN"
|
|
truth_counter = Counter([x for x in truth_list if x and x != "UNKNOWN"])
|
|
pred_counter = Counter([x for x in pred_list if x and x != "UNKNOWN"])
|
|
|
|
# Filter lists to remove empty strings and "UNKNOWN"
|
|
filtered_truth_list = [x for x in truth_list if x and x != "UNKNOWN"]
|
|
filtered_pred_list = [x for x in pred_list if x and x != "UNKNOWN"]
|
|
|
|
# Keep counters for display/debugging purposes
|
|
truth_counter = Counter(filtered_truth_list)
|
|
pred_counter = Counter(filtered_pred_list)
|
|
|
|
# Convert to sets for set operations
|
|
truth_set = set(filtered_truth_list)
|
|
pred_set = set(filtered_pred_list)
|
|
|
|
# Compute metrics using set operations
|
|
file_tp = len(truth_set.intersection(pred_set)) # True Positives
|
|
file_fp = len(pred_set - truth_set) # False Positives
|
|
file_fn = len(truth_set - pred_set) # False Negatives
|
|
# Calculate accuracy for this file
|
|
file_accuracy = file_tp / (file_tp + file_fp + file_fn) if (file_tp + file_fp + file_fn) > 0 else 1
|
|
|
|
# Update overall metrics
|
|
metrics[field_type]['tp'] += file_tp
|
|
metrics[field_type]['fp'] += file_fp
|
|
metrics[field_type]['fn'] += file_fn
|
|
metrics[field_type]['accuracy_file'].append(file_accuracy)
|
|
|
|
# Store file-specific metrics for verbose output
|
|
file_metrics[field_type] = {
|
|
"truth_counter": dict(truth_counter),
|
|
"pred_counter": dict(pred_counter),
|
|
"file_tp": file_tp,
|
|
"file_fp": file_fp,
|
|
"file_fn": file_fn,
|
|
"file_accuracy": file_accuracy
|
|
}
|
|
|
|
if verbose:
|
|
# Write verbose output to file
|
|
# Initialize file if it doesn't exist
|
|
|
|
with open("provider_metrics.txt", "a") as f:
|
|
f.write(f"File: {file}\n")
|
|
f.write(f"Total TINs: {len(set([tin for tin in truth_tins if tin and tin != 'UNKNOWN']))}, Predicted TINs: {len(set([tin for tin in pred_tins if tin and tin != 'UNKNOWN']))}\n")
|
|
f.write(f"Total NPIs: {len(set([npi for npi in truth_npis if npi and npi != 'UNKNOWN']))}, Predicted NPIs: {len(set([npi for npi in pred_npis if npi and npi != 'UNKNOWN']))}\n")
|
|
f.write(f"Total Names: {len(set([name for name in truth_names if name and name != 'UNKNOWN']))}, Predicted Names: {len(set([name for name in pred_names if name and name != 'UNKNOWN']))}\n")
|
|
f.write("-" * 50 + "\n")
|
|
|
|
for field_type in metrics:
|
|
csv_output_dict = {}
|
|
csv_output_dict["Filename"] = file
|
|
csv_output_dict["Test Bed TINs"] = len(set([tin for tin in truth_tins if tin and tin != 'UNKNOWN']))
|
|
csv_output_dict["Test Bed NPIs"] = len(set([npi for npi in truth_npis if npi and npi != 'UNKNOWN']))
|
|
csv_output_dict["Test Bed Names"] = len(set([name for name in truth_names if name and name != 'UNKNOWN']))
|
|
csv_output_dict["Predicted TINs"] = len(set([tin for tin in pred_tins if tin and tin != 'UNKNOWN']))
|
|
csv_output_dict["Predicted NPIs"] = len(set([npi for npi in pred_npis if npi and npi != 'UNKNOWN']))
|
|
csv_output_dict["Predicted Names"] = len(set([name for name in pred_names if name and name != 'UNKNOWN']))
|
|
print_lines = [
|
|
f"Field Type: {field_type}",
|
|
f"Ground Truth: {dict(file_metrics[field_type]['truth_counter'])}",
|
|
f"Predictions: {dict(file_metrics[field_type]['pred_counter'])}",
|
|
f"True Positives: {file_metrics[field_type]['file_tp']}",
|
|
f"False Positives: {file_metrics[field_type]['file_fp']}",
|
|
f"False Negatives: {file_metrics[field_type]['file_fn']}",
|
|
f"File Accuracy: {file_metrics[field_type]['file_accuracy']}",
|
|
f"Precision: {file_metrics[field_type]['file_tp'] / (file_metrics[field_type]['file_tp'] + file_metrics[field_type]['file_fp']) if (file_metrics[field_type]['file_tp'] + file_metrics[field_type]['file_fp']) > 0 else 0}",
|
|
f"Recall: {file_metrics[field_type]['file_tp'] / (file_metrics[field_type]['file_tp'] + file_metrics[field_type]['file_fn']) if (file_metrics[field_type]['file_tp'] + file_metrics[field_type]['file_fn']) > 0 else 0}",
|
|
"-" * 50
|
|
]
|
|
csv_output_dict["field_type"] = field_type
|
|
csv_output_dict["Ground Truth"] = dict(file_metrics[field_type]['truth_counter']).copy()
|
|
csv_output_dict["Predictions"] = dict(file_metrics[field_type]['pred_counter']).copy()
|
|
csv_output_dict["True Positives"] = file_metrics[field_type]['file_tp']
|
|
csv_output_dict["False Positives"] = file_metrics[field_type]['file_fp']
|
|
csv_output_dict["False Negatives"] = file_metrics[field_type]['file_fn']
|
|
csv_output_dict["File Accuracy"] = file_metrics[field_type]['file_accuracy']
|
|
csv_output_dict["Precision"] = file_metrics[field_type]['file_tp'] / (file_metrics[field_type]['file_tp'] + file_metrics[field_type]['file_fp']) if (file_metrics[field_type]['file_tp'] + file_metrics[field_type]['file_fp']) > 0 else 0
|
|
csv_output_dict["Recall"] = file_metrics[field_type]['file_tp'] / (file_metrics[field_type]['file_tp'] + file_metrics[field_type]['file_fn']) if (file_metrics[field_type]['file_tp'] + file_metrics[field_type]['file_fn']) > 0 else 0
|
|
csv_output_dicts.append(csv_output_dict)
|
|
with open("provider_metrics.txt", "a") as f:
|
|
f.write("\n".join(print_lines) + "\n")
|
|
|
|
# Convert the list of dictionaries to a DataFrame
|
|
csv_output_df = pd.DataFrame(csv_output_dicts)
|
|
# Save the DataFrame to a CSV file
|
|
csv_output_df.to_csv("provider_metrics.csv", index=False)
|
|
|
|
|
|
|
|
# Calculate accuracy, precision, recall, and F1 for each field type
|
|
results = {}
|
|
for field_type, field_metrics in metrics.items():
|
|
precision = field_metrics['tp'] / (field_metrics['tp'] + field_metrics['fp']) if (field_metrics['tp'] + field_metrics['fp']) > 0 else 0
|
|
recall = field_metrics['tp'] / (field_metrics['tp'] + field_metrics['fn']) if (field_metrics['tp'] + field_metrics['fn']) > 0 else 0
|
|
f1 = (2 * precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
|
|
|
|
|
|
|
|
|
|
results[field_type] = {
|
|
"precision": precision,
|
|
"recall": recall,
|
|
"f1": f1,
|
|
"tp": field_metrics['tp'],
|
|
"fp": field_metrics['fp'],
|
|
"fn": field_metrics['fn'],
|
|
"average_accuracy": sum(field_metrics['accuracy_file']) / len(field_metrics['accuracy_file']) if field_metrics['accuracy_file'] else 0
|
|
}
|
|
|
|
if verbose:
|
|
print("Verbose output written to provider_metrics.txt")
|
|
return results
|
|
|
|
|
|
def match_rows(fields, labels_df, predictions_df, N):
|
|
|
|
def is_equivalent(label, prediction):
|
|
"""Check if two strings are equivalent, ignoring whitespace and case."""
|
|
# Check both empty
|
|
if string_utils.is_empty(label) and string_utils.is_empty(prediction):
|
|
return True
|
|
|
|
# Clean
|
|
label = str(label).replace(" ", "").upper()
|
|
prediction = str(prediction).replace(" ", "").upper()
|
|
|
|
# Check equivalent after cleaning
|
|
if label == prediction:
|
|
return True
|
|
|
|
# Check if lists match
|
|
label_list = []
|
|
for part in label.split("|"):
|
|
label_list.extend(part.split(","))
|
|
label_list = [term.strip() for term in label_list if not string_utils.is_empty(term.strip())]
|
|
prediction_list = []
|
|
for part in prediction.split("|"):
|
|
prediction_list.extend(part.split(","))
|
|
prediction_list = [term.strip() for term in prediction_list if not string_utils.is_empty(term.strip())]
|
|
|
|
# If all items in the lists match, regardless of order, return True
|
|
if len(label_list) == len(prediction_list):
|
|
label_counter = Counter(label_list)
|
|
prediction_counter = Counter(prediction_list)
|
|
if all(label_counter[term] == prediction_counter[term] for term in label_counter):
|
|
return True
|
|
|
|
return False
|
|
|
|
confusion_matrix = {field_name: {"tp": 0, "fp": 0} for field_name in fields}
|
|
|
|
# Check for completely blank fields and handle them separately
|
|
for field in fields:
|
|
labels_all_blank = string_utils.is_empty(labels_df[field], pd_mask=True).all()
|
|
preds_all_blank = string_utils.is_empty(predictions_df[field], pd_mask=True).all()
|
|
|
|
if labels_all_blank and preds_all_blank:
|
|
# Perfect agreement on blank field
|
|
confusion_matrix[field]["tp"] = min(len(labels_df), len(predictions_df))
|
|
confusion_matrix[field]["fp"] = 0
|
|
confusion_matrix[field]["fn"] = 0
|
|
# Skip this field in the regular matching process
|
|
fields = [f for f in fields if f != field]
|
|
|
|
# Only proceed with matching if there are fields less to process
|
|
if not fields:
|
|
return confusion_matrix
|
|
|
|
# Track which label rows have been matched
|
|
matched_label_indices = set()
|
|
available_predictions = set(predictions_df.index)
|
|
|
|
for i in range(N):
|
|
num_columns_to_match = N - i
|
|
|
|
# Only consider unmatched label rows
|
|
for label_idx, label_row in labels_df.iterrows():
|
|
if label_idx in matched_label_indices:
|
|
continue # Skip already matched label rows
|
|
|
|
best_match_idx = None
|
|
max_matches = 0
|
|
best_matches = {}
|
|
|
|
# Find the best match among available predictions
|
|
for prediction_idx in available_predictions:
|
|
prediction_row = predictions_df.loc[prediction_idx]
|
|
|
|
# Count matching fields for this pair
|
|
field_counter_dict = {field_name: {"tp": 0} for field_name in fields}
|
|
num_matches = 0
|
|
|
|
for field in fields:
|
|
if is_equivalent(label_row[field], prediction_row[field]):
|
|
num_matches += 1
|
|
field_counter_dict[field]["tp"] = 1
|
|
|
|
if num_matches > max_matches:
|
|
max_matches = num_matches
|
|
best_match_idx = prediction_idx
|
|
best_matches = field_counter_dict
|
|
|
|
# If we found a good enough match
|
|
if max_matches >= num_columns_to_match and best_match_idx is not None:
|
|
# Mark this label row as matched
|
|
matched_label_indices.add(label_idx)
|
|
# Remove the matched prediction row from available ones
|
|
available_predictions.remove(best_match_idx)
|
|
# Update the confusion matrix with the best match
|
|
for field in fields:
|
|
confusion_matrix[field]["tp"] += best_matches[field]["tp"]
|
|
|
|
# Calculate FP and FN based on the number of matched fields
|
|
for field in fields:
|
|
# FN = number of label rows where this field wasn't matched
|
|
confusion_matrix[field]['fn'] = len(labels_df) - confusion_matrix[field]['tp']
|
|
# FP = number of prediction rows where this field wasn't matched
|
|
confusion_matrix[field]['fp'] = len(predictions_df) - confusion_matrix[field]['tp']
|
|
|
|
return confusion_matrix
|
|
|
|
def compare_term_extraction_results(file_name: str, predictions_df: pd.DataFrame, gt_df: pd.DataFrame, comparison_columns: list[str] = ['SERVICE_TERM', 'REIMB_TERM'], text_threshold=65):
|
|
"""Compare term extraction results between predictions and ground truth. This function
|
|
was originally written to evaluate the results of the Methodology Primary process,
|
|
which extracts `SERVICE_TERM` and `REIMB_TERM` from the text.
|
|
This is a 1:n comparison, meaning that the ground truth and predictions can have different
|
|
numbers of rows for the same file.
|
|
|
|
Args:
|
|
file_name (str): The name of the file being processed. This is to filter each
|
|
individual dataframe to the rows that are relevant to the file.
|
|
predictions_df (pd.DataFrame): The dataframe containing the predictions.
|
|
gt_df (pd.DataFrame): The dataframe containing the ground truth labels.
|
|
comparison_columns (list, optional): Columns to compare. Defaults to ['SERVICE_TERM', 'REIMB_TERM'].
|
|
text_threshold (int, optional): The minimum similarity score (0-100) for text parts to be considered matching. Defaults to 65 (especially for service term which can be short).
|
|
|
|
Returns:
|
|
dict: A dictionary containing comparison metrics and sets including:
|
|
- file_name: Name of the file being analyzed
|
|
- pred_count: Number of unique prediction pairs
|
|
- gt_count: Number of unique ground truth pairs
|
|
- common_pairs: Set of pairs found in both predictions and ground truth
|
|
- false_negatives: Set of pairs in ground truth but not in predictions
|
|
- false_positives: Set of pairs in predictions but not in ground truth
|
|
- error: Error message if columns are missing (only included if error occurs)
|
|
"""
|
|
# Step 1: Filter dataframes to only rows for the current file
|
|
pred_rows = predictions_df[predictions_df['FILE_NAME'] == file_name].copy()
|
|
gt_rows = gt_df[gt_df['FILE_NAME'] == file_name].copy()
|
|
|
|
# Step 2: Validate that the comparison columns exist in both dataframes
|
|
for col in comparison_columns:
|
|
if col not in pred_rows.columns or col not in gt_rows.columns:
|
|
return {"error": f"Column {col} not found in one of the dataframes."}
|
|
|
|
# Step 3: normalize text by converting to lowercase for case-insensitive comparison
|
|
pred_rows[comparison_columns] = pred_rows[comparison_columns].apply(lambda x: x.str.lower())
|
|
gt_rows[comparison_columns] = gt_rows[comparison_columns].apply(lambda x: x.str.lower())
|
|
|
|
# Step 4: extract unique tuples of the comparison columns from both dataframes
|
|
# This removes duplicates and converts to lists for indexed access
|
|
pred_tuples = list(set(map(tuple, pred_rows[comparison_columns].drop_duplicates().values)))
|
|
gt_tuples = list(set(map(tuple, gt_rows[comparison_columns].drop_duplicates().values)))
|
|
|
|
# Step 5: Initialize tracking sets for the matching algorithm
|
|
matched_pred = set() # Indices of prediction tuples that have been matched
|
|
matched_gt = set() # Indices of ground truth tuples that have been matched
|
|
common_pairs = set() # Indices that were successfully matched
|
|
|
|
# Step 6: Main matching loop - find best fuzzy match for each prediction
|
|
for i, pred_tuple in enumerate(pred_tuples):
|
|
# Initialize variables to track best match for this prediction
|
|
best_match= None # The GT tuple that matches best
|
|
best_score = 0 # The similarity score of the best match
|
|
best_gt_idx = None # The index of the best matching GT tuple
|
|
|
|
# Step 7: Compare this prediction against all unmatched ground truth tuples
|
|
for j, gt_tuple in enumerate(gt_tuples):
|
|
if j in matched_gt: # Skip already matched ground truth tuples
|
|
continue
|
|
|
|
# Step 8: Check if all columns in the tuple pair match using fuzzy logic
|
|
total_score = 0 # Sum of similarity scores across all columns in the tuple
|
|
all_match = True # Flag to track if all columns pass the threshold
|
|
|
|
# Compare each column value in the tuple pair
|
|
for pred_val, gt_val in zip(pred_tuple, gt_tuple):
|
|
# Use fuzzy matching that treats numbers exactly but text fuzzily
|
|
is_match, score = fuzzy_match_except_numbers(pred_val, gt_val, text_threshold=text_threshold)
|
|
if not is_match:
|
|
all_match = False # If any column fails, the whole tuple fails
|
|
break
|
|
total_score += score # Accumulates scores for successful matches
|
|
|
|
# Step 9: If all columns match, check if this is the best match so far
|
|
if all_match:
|
|
avg_score = total_score / len(pred_tuple) # Calculate average similarity
|
|
if avg_score > best_score:
|
|
best_score = avg_score
|
|
best_match = gt_tuple
|
|
best_gt_idx = j
|
|
|
|
# Step 10: if we found a valid match, record it and prevent re-matching
|
|
if best_match is not None:
|
|
matched_pred.add(i)
|
|
matched_gt.add(best_gt_idx)
|
|
common_pairs.add(pred_tuple)
|
|
|
|
# Step 11: calculate false negatives (ground truth tuples not matched) and false positives (predicted tuples not matched)
|
|
false_negatives = set()
|
|
false_positives = set()
|
|
|
|
for j, gt_tuple in enumerate(gt_tuples):
|
|
if j not in matched_gt:
|
|
false_negatives.add(gt_tuple)
|
|
|
|
for i, pred_tuple in enumerate(pred_tuples):
|
|
if i not in matched_pred:
|
|
false_positives.add(pred_tuple)
|
|
|
|
# Step 12: Return the results as a dictionary
|
|
return {
|
|
"file_name": file_name,
|
|
"pred_count": len(pred_tuples),
|
|
"testbed_count": len(gt_tuples),
|
|
"matched_count": len(common_pairs),
|
|
"false_negatives_count": len(false_negatives),
|
|
"false_positives_count": len(false_positives),
|
|
|
|
# Convert tuples to readable strings for Excel
|
|
"common_pairs_readable": "\n".join([" | ".join(map(str, pair)) for pair in common_pairs]),
|
|
"false_negatives_readable": "\n".join([" | ".join(map(str, pair)) for pair in false_negatives]),
|
|
"false_positives_readable": "\n".join([" | ".join(map(str, pair)) for pair in false_positives]),
|
|
|
|
# Keep original tuples for programmatic use if needed
|
|
"common_pairs": common_pairs,
|
|
"false_negatives": false_negatives,
|
|
"false_positives": false_positives,
|
|
"all_predicted_pairs": set(pred_tuples),
|
|
"all_testbed_pairs": set(gt_tuples),
|
|
}
|
|
|
|
def fuzzy_match_except_numbers(str1, str2, text_threshold=80):
|
|
"""
|
|
Compare two strings with fuzzy matching for text but exact matching for numbers.
|
|
|
|
Parameters:
|
|
- str1, str2: Strings to compare
|
|
- text_threshold: Minimum similarity score (0-100) for text parts to be considered matching
|
|
|
|
Returns:
|
|
- Boolean: True if strings match according to criteria, False otherwise
|
|
- Float: Overall similarity score
|
|
"""
|
|
if not str1 or not str2:
|
|
return str1 == str2, 0 if str1 != str2 else 100
|
|
|
|
# Extract numbers separately
|
|
numbers1 = re.findall(r'-?\d+\.?\d*', str1) # Handles decimals and negatives
|
|
numbers2 = re.findall(r'-?\d+\.?\d*', str2)
|
|
|
|
# If numbers don't match exactly, return False
|
|
if numbers1 != numbers2:
|
|
return False, 0
|
|
|
|
# Remove numbers and clean text for fuzzy comparison
|
|
text1 = re.sub(r'-?\d+\.?\d*', '', str1).strip()
|
|
text2 = re.sub(r'-?\d+\.?\d*', '', str2).strip()
|
|
|
|
# Remove extra punctuation and normalize whitespace
|
|
text1 = re.sub(r'[^\w\s]', ' ', text1)
|
|
text2 = re.sub(r'[^\w\s]', ' ', text2)
|
|
text1 = ' '.join(text1.split()) # Normalize whitespace
|
|
text2 = ' '.join(text2.split())
|
|
|
|
# If no text remains after cleaning, check if both are empty
|
|
if not text1 and not text2:
|
|
return True, 100.0
|
|
|
|
# Use fuzzy matching for text parts
|
|
similarity = fuzz.ratio(text1.lower(), text2.lower())
|
|
return similarity >= text_threshold, similarity
|
|
|
|
def calculate_precision_recall(accuracies):
|
|
""" FOR ONE-TO-N FIELDS
|
|
Calculate precision and recall for each file and overall metrics for all fields.
|
|
|
|
Args:
|
|
accuracies (list): A list of confusion matrices for each file.
|
|
|
|
Returns:
|
|
tuple: (precision_df, recall_df, overall_metrics_df)
|
|
- precision_df: Precision for each field per file.
|
|
- recall_df: Recall for each field per file.
|
|
- overall_metrics_df: Overall precision, recall, and F1 score for all fields.
|
|
"""
|
|
precision_dicts, recall_dicts = [], []
|
|
overall_metrics = {field: {"tp": 0, "fp": 0, "fn": 0} for field in accuracies[0] if field != "Filename"}
|
|
|
|
for confusion_matrix in accuracies:
|
|
filename = confusion_matrix["Filename"]
|
|
precision_dict, recall_dict = {"Filename": filename}, {"Filename": filename}
|
|
|
|
for field in confusion_matrix:
|
|
if field == "Filename":
|
|
continue
|
|
|
|
tp = confusion_matrix[field]["tp"]
|
|
fp = confusion_matrix[field]["fp"]
|
|
fn = confusion_matrix[field]["fn"]
|
|
|
|
# Update overall metrics
|
|
overall_metrics[field]["tp"] += tp
|
|
overall_metrics[field]["fp"] += fp
|
|
overall_metrics[field]["fn"] += fn
|
|
|
|
# Calculate precision and recall
|
|
precision = tp / (tp + fp) if (tp + fp) > 0 else 0
|
|
recall = tp / (tp + fn) if (tp + fn) > 0 else 0
|
|
|
|
precision_dict[field] = precision
|
|
recall_dict[field] = recall
|
|
|
|
precision_dicts.append(precision_dict)
|
|
recall_dicts.append(recall_dict)
|
|
|
|
# Calculate overall precision, recall, and F1 score for all fields
|
|
overall_metrics_list = []
|
|
for field, metrics in overall_metrics.items():
|
|
tp = metrics["tp"]
|
|
fp = metrics["fp"]
|
|
fn = metrics["fn"]
|
|
|
|
precision = tp / (tp + fp) if (tp + fp) > 0 else 0
|
|
recall = tp / (tp + fn) if (tp + fn) > 0 else 0
|
|
f1 = (2 * precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
|
|
|
|
overall_metrics_list.append({
|
|
"Field": field,
|
|
"tp": tp,
|
|
"fp": fp,
|
|
"fn": fn,
|
|
"prec": precision,
|
|
"rec": recall,
|
|
"f1": f1
|
|
})
|
|
|
|
# Convert overall metrics to a DataFrame
|
|
overall_metrics_df = pd.DataFrame(overall_metrics_list).set_index("Field")
|
|
|
|
# Convert precision and recall dictionaries to DataFrames
|
|
precision_df = pd.DataFrame(precision_dicts)
|
|
recall_df = pd.DataFrame(recall_dicts)
|
|
|
|
return precision_df, recall_df, overall_metrics_df
|
|
|
|
|
|
|
|
########## TESTBED POSTPROCESSING CODE ##########
|
|
# These functions are to take MCS's test bed that they provided and apply
|
|
# (most of) the same postprocessing steps that we apply to the results of the model.
|
|
|
|
|
|
|
|
def group_providers_into_json_array(df):
|
|
"""Groups provider information from a DataFrame into a JSON array format and appends it as a new column.
|
|
|
|
This function processes a DataFrame containing provider information, groups the data by the "FILE_NAME" column,
|
|
and creates a JSON array for each group. The JSON array includes details about the group provider and other
|
|
associated providers. The resulting JSON array is added as a new column, "PROV_INFO_JSON", to the DataFrame.
|
|
|
|
Args:
|
|
df (pandas.DataFrame): The input DataFrame containing the following columns:
|
|
- "FILE_NAME": Identifier for grouping rows.
|
|
- "PROV_GROUP_TIN": Tax Identification Number (TIN) of the group provider.
|
|
- "PROV_GROUP_NPI": National Provider Identifier (NPI) of the group provider.
|
|
- "PROV_GROUP_NAME_FULL": Full name of the group provider.
|
|
- "PROV_OTHER_TIN": Pipe-separated TINs of other providers.
|
|
- "PROV_OTHER_NPI": Pipe-separated NPIs of other providers.
|
|
- "PROV_OTHER_NAME_FULL": Pipe-separated full names of other providers.
|
|
|
|
Returns:
|
|
pandas.DataFrame: A modified DataFrame with an additional column "PROV_INFO_JSON",
|
|
which contains a JSON array for each "FILE_NAME" group. Each JSON array includes:
|
|
- "TIN": TIN of the provider (or "N/A" if not available).
|
|
- "NPI": NPI of the provider (or "N/A" if not available).
|
|
- "NAME": Full name of the provider (or "N/A" if not available).
|
|
- "IS_GROUP": Boolean indicating whether the provider is the group provider (True)
|
|
or an associated provider (False).
|
|
|
|
Notes:
|
|
- Missing or NaN values in the input DataFrame are replaced with "N/A" in the JSON output.
|
|
- If the lengths of "PROV_OTHER_TIN", "PROV_OTHER_NPI", and "PROV_OTHER_NAME_FULL" are not equal,
|
|
missing values are filled with "N/A" using itertools.zip_longest.
|
|
- If the OTHER columns are misaligned with one another, the function will NOT
|
|
be able to correct them. It will simply fill in the gaps with "N/A".
|
|
"""
|
|
|
|
df = df.copy()
|
|
# Create a dictionary to hold provider data in JSON array format for each file
|
|
group_providers_dict = {}
|
|
|
|
for group, grouped_df in df.groupby("FILE_NAME"):
|
|
json_dict_list = []
|
|
|
|
# Get the first row of the grouped DataFrame
|
|
first_row = grouped_df.iloc[0]
|
|
|
|
|
|
# Extract TIN/NPI/NAME for group provider.
|
|
group_tin = str(first_row["PROV_GROUP_TIN"]).replace("nan", "N/A")
|
|
group_npi = str(first_row["PROV_GROUP_NPI"]).replace("nan", "N/A")
|
|
group_name_full = str(first_row["PROV_GROUP_NAME_FULL"])
|
|
|
|
# Append the group provider's TIN/NPI/NAME to the list.
|
|
json_dict_list.append({
|
|
"TIN": group_tin,
|
|
"NPI": group_npi,
|
|
"NAME": group_name_full, "IS_GROUP": True})
|
|
|
|
# Create lists of TIN/NPI/NAME for other providers.
|
|
# Split on pipe, then strip whitespace.
|
|
other_tins = str(first_row["PROV_OTHER_TIN"]).split("|")
|
|
other_npis = str(first_row["PROV_OTHER_NPI"]).split("|")
|
|
other_names = str(first_row["PROV_OTHER_NAME_FULL"]).split("|")
|
|
other_tins = [tin.strip().replace("nan", "N/A") for tin in other_tins]
|
|
other_npis = [npi.strip().replace("nan", "N/A") for npi in other_npis]
|
|
other_names = [name.strip().replace("nan", "N/A") for name in other_names]
|
|
|
|
# Append each other provider's TIN/NPI/NAME to the list.
|
|
# Use zip to iterate over the three lists in parallel.
|
|
# IF the lengths of the lists are not equal, use itertools.zip_longest to fill in the gaps with N/A.
|
|
for tin, npi, name in itertools.zip_longest(other_tins, other_npis, other_names, fillvalue="N/A"):
|
|
json_dict_list.append({
|
|
"TIN": tin,
|
|
"NPI": npi,
|
|
"NAME": name,
|
|
"IS_GROUP": False
|
|
})
|
|
|
|
group_providers_dict[group] = json_dict_list
|
|
|
|
# Add JSON data back to original DataFrame
|
|
df["PROV_INFO_JSON"] = df["FILE_NAME"].map(group_providers_dict)
|
|
|
|
|
|
## Reorder columns to place PROV_INFO_JSON right after PROV_OTHER_NAME_FULL
|
|
cols = list(df.columns)
|
|
# Find the position of PROV_OTHER_NAME_FULL
|
|
prov_other_name_full_pos = cols.index("PROV_OTHER_NAME_FULL")
|
|
# Remove PROV_INFO_JSON from its current position
|
|
cols.remove("PROV_INFO_JSON")
|
|
# Insert it after PROV_OTHER_NAME_FULL
|
|
cols.insert(prov_other_name_full_pos + 1, "PROV_INFO_JSON")
|
|
# Reorder the DataFrame
|
|
df = df[cols]
|
|
|
|
return df
|
|
|
|
def testbed_postprocess(df):
|
|
"""Postprocess the testbed DataFrame.
|
|
|
|
Args:
|
|
df (pandas.DataFrame): The input DataFrame containing the testbed data.
|
|
|
|
Returns:
|
|
pandas.DataFrame: The postprocessed DataFrame.
|
|
"""
|
|
# Apply preprocessing steps
|
|
df = testbed_preprocess(df)
|
|
|
|
# Group providers into JSON array format
|
|
df = group_providers_into_json_array(df)
|
|
|
|
# Add CLIENT_NAME column
|
|
df['CLIENT_NAME'] = "AARETE_TESTBED"
|
|
|
|
# Reformat rate fields
|
|
if "REIMB_FEE_RATE" in df.columns:
|
|
# Format rate fields
|
|
df['REIMB_FEE_RATE'] = df['REIMB_FEE_RATE'].apply(investment_postprocessing_funcs.format_rate_fields_with_commas)
|
|
if "REIMB_PCT_RATE" in df.columns:
|
|
df['REIMB_PCT_RATE'] = df['REIMB_PCT_RATE'].apply(investment_postprocessing_funcs.format_rate_fields_with_commas)
|
|
|
|
for col in df.columns:
|
|
if "_IND" in col:
|
|
df[col] = df[col].apply(investment_postprocessing_funcs.normalize_indicator_field)
|
|
# Normalize _IND fields and clean up TIN/NPI fields
|
|
if "TIN" in col or "NPI" in col:
|
|
df[col] = df[col].apply(investment_postprocessing_funcs.remove_hyphens)
|
|
# Apply the flatten_singleton_string_list function
|
|
if "_CD" in col and "CPT" not in col:
|
|
df[col] = df[col].apply(investment_postprocessing_funcs.flatten_singleton_string_list)
|
|
# Convert DATE to YYYY/MM/DD format
|
|
if "_DT" in col or "DATE" in col:
|
|
df[col] = df[col].apply(investment_postprocessing_funcs.validate_and_reformat_date)
|
|
# Normalize CPT fields
|
|
if "CPT" in col:
|
|
df[col] = df[col].apply(investment_postprocessing_funcs.normalize_cpt_fields)
|
|
|
|
df = investment_postprocessing_funcs.update_reimb_prov_name(df)
|
|
|
|
# Add reimb_id
|
|
df = investment_postprocessing_funcs.generate_reimb_ids(df)
|
|
|
|
# Standardize output column order - this should ALWAYS be the final postprocessing step
|
|
df = investment_postprocessing_funcs.reorder_columns(df, COLUMN_ORDER)
|
|
|
|
return df |