Merged in feature/full-test-script (pull request #497)
Initial commit - single testbed comparison script * Initial commit - single testbed comparison script Approved-by: Alex Galarce
This commit is contained in:
Katon Minhas
committed by
Alex Galarce
Alex Galarce
parent
0939399f0e
commit
9984d046d0
@@ -0,0 +1,216 @@
|
||||
|
||||
# Imports
|
||||
import pandas as pd
|
||||
|
||||
from src.prompts.investment_prompts import FieldSet
|
||||
import warnings
|
||||
import src.config as config
|
||||
import src.utils.string_utils as string_utils
|
||||
import src.testbed.testbed_utils as testbed_utils
|
||||
|
||||
warnings.filterwarnings("ignore")
|
||||
|
||||
########################## READ AND PREPROCESS ##########################
|
||||
# Read files
|
||||
testbed = pd.read_excel("Doczy-Testbed.xlsx") # This is the testbed
|
||||
results = pd.read_csv("inv-test-0425-RESULTS.csv") # This is the script results
|
||||
|
||||
# Remove suffixes from the FILE_NAME values (like .txt, .pdf ONLY)
|
||||
testbed['FILE_NAME'] = testbed['FILE_NAME'].apply(lambda x: '.'.join(x.split('.')[:-1]))
|
||||
results['FILE_NAME'] = results['FILE_NAME'].apply(lambda x: '.'.join(x.split('.')[:-1]))
|
||||
|
||||
# Ensure FILE_NAME values match between testbed and results
|
||||
common_file_names = set(testbed['FILE_NAME']).intersection(set(results['FILE_NAME']))
|
||||
testbed = testbed[testbed['FILE_NAME'].isin(common_file_names)]
|
||||
results = results[results['FILE_NAME'].isin(common_file_names)]
|
||||
|
||||
# Preprocess testbed and results data
|
||||
# Convert numeric cells to float with 2 decimal places
|
||||
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
|
||||
|
||||
testbed = testbed.applymap(convert_to_float)
|
||||
results = results.applymap(convert_to_float)
|
||||
|
||||
# Replace "UNKNOWN" with an empty string and convert all Null values to empty string
|
||||
testbed = testbed.replace("UNKNOWN", "").fillna("")
|
||||
results = results.replace("UNKNOWN", "").fillna("")
|
||||
|
||||
testbed = testbed.applymap(lambda x: str(x).strip().upper())
|
||||
results = results.applymap(lambda x: str(x).strip().upper())
|
||||
|
||||
# For columns with TIN in the column name, remove any "-" from the values
|
||||
tin_columns = [col for col in testbed.columns if 'TIN' in col]
|
||||
for col in tin_columns:
|
||||
testbed[col] = testbed[col].str.replace("-", "", regex=False)
|
||||
results[col] = results[col].str.replace("-", "", regex=False)
|
||||
|
||||
########################## Analyze Testbed ##########################
|
||||
print("*************** Test Bed Stats ****************")
|
||||
print(f"{len(testbed['FILE_NAME'].unique())} contracts")
|
||||
print(f"{len(testbed.columns)} fields")
|
||||
|
||||
print(f"{(testbed == '').all().sum()} fields not populated:") # Count of completely empty fields
|
||||
print(f"{testbed.columns[(testbed == '').all()].tolist()}") # List of completely empty fields
|
||||
|
||||
########################## One-to-One Analysis ##########################
|
||||
print("\n*************** One-to-One Stats ****************")
|
||||
one_to_one_fields = FieldSet(file_path=config.FIELD_JSON_PATH).filter(relationship="one_to_one").list_fields()
|
||||
one_to_one_fields.extend(['PROV_GROUP_TIN',
|
||||
'PROV_GROUP_NPI',
|
||||
'PROV_GROUP_NAME_FULL',
|
||||
'PROV_OTHER_TIN',
|
||||
'PROV_OTHER_NPI',
|
||||
'PROV_OTHER_NAME_FULL'])
|
||||
for field in ['TIN', 'NPI', 'NAME']:
|
||||
one_to_one_fields.remove(field)
|
||||
|
||||
# Create filtered dataframes with only one-to-one fields
|
||||
one_to_one_columns = ['FILE_NAME'] + [f for f in one_to_one_fields if f in testbed.columns]
|
||||
testbed_one_to_one = testbed[one_to_one_columns].drop_duplicates()
|
||||
results_one_to_one = results[one_to_one_columns].drop_duplicates()
|
||||
|
||||
metrics_df = pd.DataFrame(columns=['Field', 'Precision', 'Recall', 'Accuracy'])
|
||||
comparison_df = pd.DataFrame(index=testbed_one_to_one['FILE_NAME'].unique())
|
||||
|
||||
for field in one_to_one_fields:
|
||||
print(field)
|
||||
if field in testbed.columns and field in results.columns and field != 'FILE_NAME':
|
||||
|
||||
# Merge and compare values (no need to drop_duplicates here since we already did)
|
||||
merged = pd.merge(testbed_one_to_one[['FILE_NAME', field]],
|
||||
results_one_to_one[['FILE_NAME', field]],
|
||||
on='FILE_NAME',
|
||||
suffixes=('_truth', '_pred'))
|
||||
|
||||
# Compare values
|
||||
merged[field] = (merged[f"{field}_truth"] == merged[f"{field}_pred"]) & (merged[f"{field}_truth"] != '')
|
||||
# Drop duplicate rows
|
||||
merged = merged.drop_duplicates(subset=['FILE_NAME', field])
|
||||
|
||||
|
||||
# Add to comparison DataFrame
|
||||
try:
|
||||
comparison_df[field] = merged.set_index('FILE_NAME')[field]
|
||||
except:
|
||||
print(merged['FILE_NAME'].value_counts())
|
||||
raise
|
||||
|
||||
# Calculate metrics
|
||||
precision, recall, accuracy = testbed_utils.calculate_field_metrics(testbed_one_to_one,
|
||||
results_one_to_one,
|
||||
field)
|
||||
metrics_df.loc[len(metrics_df)] = [field, precision, recall, accuracy]
|
||||
else:
|
||||
metrics_df.loc[len(metrics_df)] = [field, None, None, None]
|
||||
comparison_df[field] = None
|
||||
|
||||
# Display results with clean formatting
|
||||
print(metrics_df.to_string(
|
||||
index=False,
|
||||
float_format=lambda x: '{:.2f}'.format(x) if pd.notnull(x) else 'Not found',
|
||||
justify='left',
|
||||
col_space={'Field': 40, 'Precision': 15, 'Recall': 15, 'Accuracy': 15}
|
||||
))
|
||||
|
||||
########################## One-to-N Analysis ##########################
|
||||
|
||||
print("\n*************** One-to-N Stats ****************")
|
||||
one_to_n_fields = ["AARETE_DERIVED_PROV_TYPE",
|
||||
"AARETE_DERIVED_PRODUCT",
|
||||
"AARETE_DERIVED_LOB",
|
||||
"AARETE_DERIVED_PROGRAM",
|
||||
"AARETE_DERIVED_NETWORK",
|
||||
"LESSER_OF_IND",
|
||||
"GREATER_OF_IND",
|
||||
"AARETE_DERIVED_REIMB_METHOD",
|
||||
"UNIT_OF_MEASURE",
|
||||
"REIMB_PCT_RATE",
|
||||
"REIMB_FEE_RATE",
|
||||
"NOT_TO_EXCEED_IND",
|
||||
"DEFAULT_IND",
|
||||
"AARETE_DERIVED_FEE_SCHEDULE",
|
||||
"AARETE_DERIVED_FEE_SCHEDULE_VERSION",
|
||||
"REIMB_PROV_TIN",
|
||||
"REIMB_PROV_NPI",
|
||||
"REIMB_PROV_NAME"] # comparison fields
|
||||
|
||||
N = len(one_to_n_fields)
|
||||
accuracies = []
|
||||
|
||||
for file in testbed['FILE_NAME'].unique():
|
||||
testbed_file = testbed[testbed['FILE_NAME'] == file]
|
||||
results_file = results[results['FILE_NAME'] == file]
|
||||
|
||||
# Ensure there are rows in both dataframes
|
||||
if testbed_file.shape[0] == 0 or results_file.shape[0] == 0:
|
||||
print(f"No rows in either labels or predictions for file: {file}")
|
||||
continue
|
||||
|
||||
confusion_matrix = testbed_utils.match_rows(one_to_n_fields, testbed_file, results_file, N)
|
||||
|
||||
confusion_matrix["Filename"] = file
|
||||
accuracies.append(confusion_matrix)
|
||||
|
||||
precision_df, recall_df, overall = testbed_utils.calculate_precision_recall(accuracies)
|
||||
|
||||
print("\nOverall Metrics:")
|
||||
print(overall.to_string(
|
||||
index=True,
|
||||
float_format=lambda x: '{:.2f}'.format(x) if pd.notnull(x) else 'Not found',
|
||||
justify='left',
|
||||
col_space={'tp': 15, 'fp': 15, 'fn': 15, 'prec': 15, 'rec': 15, 'f1': 15}
|
||||
))
|
||||
|
||||
|
||||
########################## Export to Excel ##########################
|
||||
print("\nExporting results to Excel...")
|
||||
|
||||
# Create Excel writer object
|
||||
output_file = "20250428-testbed-metrics.xlsx"
|
||||
with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
|
||||
# Test Bed Stats
|
||||
stats_df = pd.DataFrame({
|
||||
'Metric': ['Total Contracts', 'Total Fields', 'Empty Fields'],
|
||||
'Value': [
|
||||
len(testbed['FILE_NAME'].unique()),
|
||||
len(testbed.columns),
|
||||
(testbed == '').all().sum()
|
||||
]
|
||||
})
|
||||
empty_fields_df = pd.DataFrame({'Empty Fields': testbed.columns[(testbed == '').all()].tolist()})
|
||||
|
||||
stats_df.to_excel(writer, sheet_name='Testbed Stats', index=False)
|
||||
empty_fields_df.to_excel(writer, sheet_name='Empty Fields', index=False)
|
||||
|
||||
# One-to-One Results
|
||||
metrics_df.to_excel(writer, sheet_name='One-to-One Metrics', index=False)
|
||||
comparison_df.to_excel(writer, sheet_name='One-to-One Details') # Add new sheet
|
||||
|
||||
# One-to-N Results
|
||||
precision_df.to_excel(writer, sheet_name='One-to-N Precision', index=False)
|
||||
recall_df.to_excel(writer, sheet_name='One-to-N Recall', index=False)
|
||||
overall.to_excel(writer, sheet_name='One-to-N Overall')
|
||||
|
||||
# Auto-adjust column widths
|
||||
for sheet_name in writer.sheets:
|
||||
worksheet = writer.sheets[sheet_name]
|
||||
for column in worksheet.columns:
|
||||
max_length = 0
|
||||
column = [cell for cell in column]
|
||||
for cell in column:
|
||||
try:
|
||||
if len(str(cell.value)) > max_length:
|
||||
max_length = len(str(cell.value))
|
||||
except:
|
||||
pass
|
||||
adjusted_width = (max_length + 2)
|
||||
worksheet.column_dimensions[column[0].column_letter].width = adjusted_width
|
||||
|
||||
print(f"Results exported to {output_file}")
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
|
||||
import pandas as pd
|
||||
import src.utils.string_utils as string_utils
|
||||
|
||||
def calculate_field_metrics(testbed_df, results_df, field):
|
||||
"""Calculate precision, recall, and accuracy for a single field.
|
||||
|
||||
Args:
|
||||
testbed_df: DataFrame containing ground truth
|
||||
results_df: DataFrame containing predictions
|
||||
field: Name of field to evaluate
|
||||
"""
|
||||
# Get unique combinations of FILE_NAME and field value
|
||||
results_unique = results_df[['FILE_NAME', field]].drop_duplicates()
|
||||
testbed_unique = testbed_df[['FILE_NAME', field]].drop_duplicates()
|
||||
|
||||
# Merge on FILE_NAME to compare values
|
||||
comparison = pd.merge(
|
||||
testbed_unique,
|
||||
results_unique,
|
||||
on='FILE_NAME',
|
||||
suffixes=('_truth', '_pred')
|
||||
)
|
||||
|
||||
# Calculate metrics
|
||||
true_positives = sum((comparison[f'{field}_truth'] == comparison[f'{field}_pred']) &
|
||||
(comparison[f'{field}_truth'] != ''))
|
||||
truth_populated = sum(comparison[f'{field}_truth'] != '')
|
||||
pred_populated = sum(comparison[f'{field}_pred'] != '')
|
||||
total_files = len(testbed_unique)
|
||||
|
||||
precision = true_positives / pred_populated if pred_populated > 0 else 0
|
||||
recall = true_positives / truth_populated if truth_populated > 0 else 0
|
||||
accuracy = true_positives / total_files if total_files > 0 else 0
|
||||
|
||||
return precision, recall, accuracy
|
||||
|
||||
|
||||
|
||||
|
||||
def match_rows(fields, labels_df, predictions_df, N):
|
||||
confusion_matrix = {field_name: {"tp": 0, "fp": 0} for field_name in fields}
|
||||
|
||||
# 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 label_row[field] == prediction_row[field] or (string_utils.is_empty(label_row[field]) and string_utils.is_empty(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 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
|
||||
Reference in New Issue
Block a user