import itertools import re from collections import Counter import pandas as pd import src.config as config import src.utils.string_utils as string_utils from constants.investment_columns import COLUMN_ORDER from rapidfuzz import fuzz # Imports for the testbed postprocessing functions from src.investment import postprocessing_funcs from src.prompts.fieldset import FieldSet def filter_file_names(testbed, results): # Ensure FILE_NAME values match between testbed and results common_file_names = set(testbed["FILE_NAME"]).intersection( set(results["FILE_NAME"]) ) files_to_remove = [ "86-0898663-Founder Health, LLC dba Preferred Homecare-ICMProviderAgreement_120661" ] for filename in files_to_remove: if filename in common_file_names: common_file_names.remove(filename.upper()) testbed = testbed[testbed["FILE_NAME"].isin(common_file_names)] results = results[results["FILE_NAME"].isin(common_file_names)] return testbed, results 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): # Fix column names with leading/trailing spaces df.columns = df.columns.str.strip() # 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 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 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) """ # 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"]) ) FN_list = [ truth for pred, truth in zip(merged[f"{field}_pred"], merged[f"{field}_truth"]) if is_negative(pred) and is_positive(truth) ] # 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, FN_list def evaluate_provider_fields_separately(testbed_df, results_df, verbose=False): """Evaluates TIN, NPI, and NAME fields separately for each provider.""" print("\n*************** Order-Independent Provider Analysis ****************") # 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") # Display the results provider_metrics_df = pd.DataFrame( [ { "Field": f"PROV_OTHER_{field_type}", "Precision": metrics["precision"], "Recall": metrics["recall"], "F1": metrics["f1"], "TP": metrics["tp"], "FP": metrics["fp"], "FN": metrics["fn"], "average_accuracy": metrics["average_accuracy"], } for field_type, metrics in results.items() ] ) print( provider_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": 15, "Precision": 12, "Recall": 12, "F1": 12, "TP": 8, "FP": 8, "FN": 8, }, ) ) return results, provider_metrics_df 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(" ", "").replace(",", "").upper() prediction = str(prediction).replace(" ", "").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, "unmatched_labels": []} 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 confusion_matrix[field]["unmatched_labels"] = [] # 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"] if best_matches[field]["tp"] == 0: confusion_matrix[field]["unmatched_labels"].append( (label_idx, label_row[field]) ) # 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"] ) # add unmatched labels to the confusion matrix for cases where whole rows were unmatched if list(set(labels_df.index) - set(matched_label_indices)): confusion_matrix[field]["unmatched_labels"].append(("|", "|")) for idx in list(set(labels_df.index) - set(matched_label_indices)): confusion_matrix[field]["unmatched_labels"].append( (idx, labels_df.loc[idx, field]) ) 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} recall_dict_value = {} 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 recall_dict_value[field + "_missed"] = confusion_matrix[field][ "unmatched_labels" ] precision_dicts.append(precision_dict) recall_dicts.append(recall_dict | recall_dict_value) # 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( postprocessing_funcs.format_rate_fields_with_commas ) if "REIMB_PCT_RATE" in df.columns: df["REIMB_PCT_RATE"] = df["REIMB_PCT_RATE"].apply( postprocessing_funcs.format_rate_fields_with_commas ) for col in df.columns: if "_IND" in col: df[col] = df[col].apply(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(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(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(postprocessing_funcs.validate_and_reformat_date) # Normalize CPT fields if "CPT" in col: df[col] = df[col].apply(postprocessing_funcs.normalize_cpt_fields) # Add reimb_id df = postprocessing_funcs.generate_reimb_ids(df) # Standardize output column order - this should ALWAYS be the final postprocessing step df = postprocessing_funcs.reorder_columns(df, COLUMN_ORDER) return df def get_df_stats(df): print("*************** Test Bed Stats ****************") print(f"{len(df['FILE_NAME'].unique())} contracts") print(f"{len(df.columns)} fields") print( f"{(df == '').all().sum()} fields not populated:" ) # Count of completely empty fields print(f"{df.columns[(df == '').all()].tolist()}") # List of completely empty fields def get_row_comparison(testbed, results): row_comparison = pd.DataFrame(index=testbed["FILE_NAME"].unique()) row_comparison["testbed"] = testbed.groupby("FILE_NAME").size() row_comparison["results"] = results.groupby("FILE_NAME").size() print("\n*************** Row Count Comparison ****************") print( f"Files with different row counts: {(row_comparison['testbed'] != row_comparison['results']).sum()}" ) return row_comparison def get_consecutive_page_groups(pages) -> list: """Group consecutive page numbers together. e.g., [4, 5, 6, 8, 9] -> [[4, 5, 6], [8, 9]] """ if not pages: return [] # Convert to sorted list of integers page_nums = sorted([int(p) for p in pages if str(p).isdigit()]) if not page_nums: return [] groups = [] current_group = [page_nums[0]] for i in range(1, len(page_nums)): if page_nums[i] == page_nums[i - 1] + 1: # consecutive pages current_group.append(page_nums[i]) else: groups.append(current_group) current_group = [page_nums[i]] groups.append(current_group) # Append the last group return groups def group_name(group): """Convert a group of consecutive pages to a readable name. E.g., [4, 5, 6] -> "4-6", [8, 9] -> "8-9", [13] -> "13" """ if len(group) == 1: return str(group[0]) else: return f"{group[0]}-{group[-1]}" def normalize_page(page_str): """Normalize page numbers to integer strings for comparison (e.g., "4.0" -> "4", "4" -> "4", "13.0.0" -> "13")""" if pd.isna(page_str) or page_str == "": return None try: # Split on first period return str(page_str).strip().split(".")[0] except (ValueError, TypeError): return str(page_str).strip() def get_one_to_one_analysis(testbed, results): 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"] ) one_to_one_fields = [ f for f in one_to_one_fields if f not in {"TIN", "NPI", "NAME", "CLIENT_NAME", "GLOBAL_LESSER_OF_STATEMENT"} ] # 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 and f in results.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", "FN_list"] ) comparison_df = pd.DataFrame(index=testbed["FILE_NAME"].unique()) for field in one_to_one_fields: 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"), ) # Normalize strings by stripping whitespace and converting to same case merged[f"{field}_truth_norm"] = ( merged[f"{field}_truth"].astype(str).str.strip().str.upper() ) merged[f"{field}_pred_norm"] = ( merged[f"{field}_pred"].astype(str).str.strip().str.upper() ) # Compare values using normalized strings merged[field] = ( merged[f"{field}_truth_norm"] == merged[f"{field}_pred_norm"] ) & (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 Exception as e: print( f"\n❌ ERROR: Field '{field}' has inconsistent values for the same FILE_NAME", f"This indicates a data quality issue in your testbed - one-to-one fields should have", f"the same value for all rows of the same file.\n", ) # Check for problematic files in the original data before any deduplication problem_files = [] for file_name in testbed_one_to_one["FILE_NAME"].unique(): file_data = testbed_one_to_one[ testbed_one_to_one["FILE_NAME"] == file_name ] unique_values = file_data[field].unique() if len(unique_values) > 1: problem_files.append((file_name, unique_values.tolist())) if problem_files: print("Problematic file(s) and ALL their conflicting values:") for file_name, values in problem_files: print(f"\nFile: {file_name}") print(f" All '{field}' values found: {values}") print(f" (Should all be the same value)") print( f"\n💡 Fix: Ensure all rows for the same file have identical '{field}' values in your testbed." ) print(f"Original pandas error: {e}") raise ValueError( f"Data quality error: Field '{field}' has inconsistent values for FILE_NAME. " f"Please fix the testbed data (or the results data) before proceeding." ) # Calculate metrics precision, recall, accuracy, FN_list = calculate_field_metrics( merged, field ) metrics_df.loc[len(metrics_df)] = [ field, precision, recall, accuracy, FN_list, ] else: metrics_df.loc[len(metrics_df)] = [field, None, None, None, None] comparison_df[field] = None # Display results with clean formatting print( metrics_df[["Field", "Precision", "Recall", "Accuracy"]].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}, ) ) return metrics_df, comparison_df def get_one_to_n_analysis(testbed, results): print("\n*************** One-to-N Stats ****************") one_to_n_fields = { "methodology_breakout": [ field for field in FieldSet(file_path=config.FIELD_JSON_PATH) .filter(field_type="methodology_breakout") .list_fields() if field != "GREATER_OF_IND" ], "fee_schedule_breakout": [ "AARETE_DERIVED_FEE_SCHEDULE", "AARETE_DERIVED_FEE_SCHEDULE_VERSION", ], "trigger_cap": ["TRIGGER_CAP_THRESHOLD_AMT", "TRIGGER_BASE_THRESHOLD"], "rate_escalator": [ "RATE_ESCALATOR_IND", "RATE_ESCALATOR_TERM", "RATE_ESCALATOR_MAX_RATE_INC_PCT", "RATE_ESCALATOR_RATE_CHANGE_TIMELINE", ], "addition": [ "ADDITION_DESC", "ADDITION_MAX_PCT_RATE_INC", "ADDITION_MAX_FEE_RATE_INC", "ADDITION_RATE_CHANGE_TIMELINE", "AARETE_DERIVED_ADDITION_RATE_CHANGE_TIMELINE", ], "dynamic_primary": [ "AARETE_DERIVED_LOB", "AARETE_DERIVED_PROGRAM", "AARETE_DERIVED_NETWORK", "AARETE_DERIVED_PRODUCT", ], "reimb_prov_info": ["REIMB_PROV_TIN", "REIMB_PROV_NPI", "REIMB_PROV_NAME"], } one_to_n_metrics = {} for field_type, fields in one_to_n_fields.items(): N = len(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 = match_rows(fields, testbed_file, results_file, N) confusion_matrix["Filename"] = file accuracies.append(confusion_matrix) precision_df, recall_df, overall = calculate_precision_recall(accuracies) one_to_n_metrics[field_type] = { "precision": precision_df, "recall": recall_df, "overall": overall, } print(f"\nOverall Metrics: {field_type}") 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, }, ) ) return one_to_n_metrics def get_reimb_primary(testbed, results): common_files = set(testbed["FILE_NAME"]).intersection(set(results["FILE_NAME"])) reimb_primary_metric_rows = [] for file in common_files: result = compare_term_extraction_results( file, predictions_df=results, gt_df=testbed ) reimb_primary_metric_rows.append(result.copy()) reimb_primary_df = pd.DataFrame(reimb_primary_metric_rows) return reimb_primary_df def export_to_excel( output_file, testbed, row_comparison, metrics_df, comparison_df, one_to_n_metrics, reimb_primary_df, provider_metrics_df, ): 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) row_comparison.to_excel(writer, sheet_name="Row Counts") 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 for field_type, metrics in one_to_n_metrics.items(): precision_df = metrics["precision"] recall_df = metrics["recall"] overall = metrics["overall"] # Add the field type to the sheet names precision_df.to_excel( writer, sheet_name=f"{field_type} Precision", index=False ) recall_df.to_excel(writer, sheet_name=f"{field_type} Recall", index=False) overall.to_excel(writer, sheet_name=f"{field_type} Overall") # Reimbursement Primary Results reimb_primary_df.to_excel( writer, sheet_name="Reimbursement Primary", index=False ) # Order-independent provider analysis provider_metrics_df.to_excel(writer, sheet_name="Provider Fields", index=False) # 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}")