jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in feature/update-testbed-metrics (pull request #871)

Browse Source 
Feature/update testbed metrics

* Combine fee schedule with methodology breakout; remove Reimbursement primary; do not run trigger cap or additions

* Deprecate reimb primary functions

* combine multiple sheets to one

* Clean 1:N metrics

* Improve dynamic primary evaluation

* Resolve dynamic primary

* Update row counts

* black format

* Merged DEV into feature/update-testbed-metrics


Approved-by: Praneel Panchigar
This commit is contained in:
Katon Minhas
2026-02-05 22:11:21 +00:00
parent c62bd1cd6a
commit 0c3546cea9
3 changed files with 167 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/testbed/dynamic_primary_metrics.py
+100 -26
View File
@@ -8,16 +8,100 @@ This module provides tally-based, order-independent analysis for dynamic primary
"""
from collections import Counter
import json
import re
import pandas as pd
import src.config as config
import src.utils.string_utils as string_utils
import src.testbed.testbed_utils as testbed_utils
def parse_value_format(value):
"""Parse a value that may be in multiple formats and return a list of individual values.
Handles:
- JSON format: ["Value1", "Value2"] or string representation with single quotes
- Pipe-separated: "Value1|Value2"
- Comma-separated: "Value1, Value2"
- Single value: "Value1"
- Python list representation: ['Value1', 'Value2']
- Mixed: ['Value1|Value2'] should expand to ['Value1', 'Value2']
Args:
value: The value to parse (may be None, empty, or any of the supported formats)
Returns:
list: List of individual values, normalized and uppercased, or empty list
"""
if pd.isna(value) or value is None:
return []
value_str = str(value).strip()
if not value_str or string_utils.is_empty(value_str):
return []
parsed = []
# Try JSON/Python list format first (with both single and double quotes)
if value_str.startswith("[") and value_str.endswith("]"):
try:
# First try standard JSON
json_list = json.loads(value_str)
if isinstance(json_list, list):
for item in json_list:
# Each item might itself be delimited, so recursively parse
sub_parsed = parse_value_format(str(item))
parsed.extend(sub_parsed)
return parsed if parsed else []
except (json.JSONDecodeError, ValueError):
# Try replacing single quotes with double quotes for Python list format
try:
# Replace single quotes with double quotes carefully
json_str = value_str.replace("'", '"')
json_list = json.loads(json_str)
if isinstance(json_list, list):
for item in json_list:
# Each item might itself be delimited, so recursively parse
sub_parsed = parse_value_format(str(item))
parsed.extend(sub_parsed)
return parsed if parsed else []
except (json.JSONDecodeError, ValueError):
# If both JSON attempts fail, treat as string and extract values
pass
# Try pipe-separated format
if "|" in value_str:
parts = value_str.split("|")
for part in parts:
normalized = part.strip().upper()
if normalized and normalized != "EMPTY":
parsed.append(normalized)
return parsed
# Try comma-separated format
if "," in value_str:
parts = value_str.split(",")
for part in parts:
normalized = part.strip().upper()
if normalized and normalized != "EMPTY":
parsed.append(normalized)
return parsed
# Single value
normalized = value_str.upper()
if normalized and normalized != "EMPTY":
parsed.append(normalized)
return parsed
def analyze_dynamic_primary_fields_tally_based(
testbed_file: pd.DataFrame,
results_file: pd.DataFrame,
fields: list[str],
debug_field: str = None,
):
"""Analyze dynamic primary fields using a tally/count-based approach.
@@ -27,42 +111,32 @@ def analyze_dynamic_primary_fields_tally_based(
2. Missing values: Testbed has more occurrences of a value
3. Extra values: Results have more occurrences of a value
Handles multiple value formats:
- JSON: ["Value1", "Value2"]
- Pipe-separated: "Value1|Value2"
- Comma-separated: "Value1, Value2"
- Single value: "Value1"
Args:
testbed_file: DataFrame containing ground truth for a single file
results_file: DataFrame containing predictions for a single file
fields: List of dynamic primary field names to analyze
debug_field: Specific field to debug (print raw and parsed values)
Returns:
dict: Dictionary containing detailed metrics for each field
"""
analysis_results = {}
def expand_pipe_delimited_values(values):
"""Expand pipe-delimited values into individual values for counting."""
expanded = []
for v in values:
if pd.notna(v) and not string_utils.is_empty(v):
# Split by pipe and add each value
for part in str(v).strip().upper().split("|"):
part = part.strip()
if part and part != "EMPTY":
expanded.append(part)
return expanded
def normalize_value(value):
"""Normalize a value for comparison."""
if pd.isna(value) or string_utils.is_empty(value):
return None
return str(value).strip().upper()
for field in fields:
# Get all values from testbed and results
testbed_values = [normalize_value(v) for v in testbed_file[field]]
results_values = [normalize_value(v) for v in results_file[field]]
# Parse all values from testbed and results using the multi-format parser
testbed_expanded = []
for v in testbed_file[field]:
testbed_expanded.extend(parse_value_format(v))
# Expand pipe-delimited values (e.g., "Duals|Medicare" -> ["Duals", "Medicare"])
testbed_expanded = expand_pipe_delimited_values(testbed_values)
results_expanded = expand_pipe_delimited_values(results_values)
results_expanded = []
for v in results_file[field]:
results_expanded.extend(parse_value_format(v))
# Count occurrences
testbed_counts = Counter(testbed_expanded)
@@ -141,8 +215,9 @@ def get_dynamic_primary_analysis(testbed, results, fields):
continue
# Use tally-based analysis for dynamic primary (order-independent)
# Debug the NETWORK field
file_analysis = analyze_dynamic_primary_fields_tally_based(
testbed_file, results_file, fields
testbed_file, results_file, fields, debug_field="AARETE_DERIVED_NETWORK"
)
# Create detailed results for this file
@@ -359,7 +434,6 @@ def run_dynamic_primary_analysis_only(testbed_path, results_path, output_path=No
Returns:
dict: Dictionary with "summary", "details", and "overall" DataFrames
"""
import src.testbed.testbed_utils as testbed_utils
# Read files
print(f"Loading testbed from: {testbed_path}")
src/testbed/testbed_metrics.py
+3 -7
View File
@@ -9,8 +9,8 @@ warnings.filterwarnings("ignore")
########################## READ AND PREPROCESS ##########################
# Read files
testbed_path = "data/testbed/Doczy-Testbed-Official.csv"
results_path = "data/testbed/inv-test-20260122-generic-RESULTS-CLEAN.csv"
testbed_path = "Doczy-Testbed-Official.csv"
results_path = "inv-test-20260205-generic-RESULTS.csv"
print(f"Loading testbed from: {testbed_path}")
testbed = pd.read_csv(testbed_path) # This is the testbed
@@ -37,9 +37,6 @@ metrics_df, comparison_df = testbed_utils.get_one_to_one_analysis(testbed, resul
########################## One-to-N Analysis ##########################
one_to_n_metrics = testbed_utils.get_one_to_n_analysis(testbed, results)
########################## Reimbursement Primary ##########################
reimb_primary_df = testbed_utils.get_reimb_primary(testbed, results)
########################## Order-Independent Provider Analysis ##########################
provider_results, provider_metrics_df = (
testbed_utils.evaluate_provider_fields_separately(testbed, results, verbose=True)
@@ -49,7 +46,7 @@ provider_results, provider_metrics_df = (
print("\nExporting results to Excel...")
# Create Excel writer object - save in testbed folder
output_file = f"data/testbed/{config.TODAY}-testbed-metrics.xlsx"
output_file = f"{config.TODAY}-testbed-metrics.xlsx"
testbed_utils.export_to_excel(
output_file,
testbed,
@@ -57,6 +54,5 @@ testbed_utils.export_to_excel(
metrics_df,
comparison_df,
one_to_n_metrics,
reimb_primary_df,
provider_metrics_df,
)
src/testbed/testbed_utils.py
+64 -243
View File
@@ -461,11 +461,6 @@ def evaluate_provider_fields_separately(testbed_df, results_df, verbose=False):
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():
@@ -690,192 +685,6 @@ def match_rows(fields, labels_df, predictions_df, N):
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.
@@ -884,9 +693,8 @@ def calculate_precision_recall(accuracies):
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.
tuple: (consolidated_df, overall_metrics_df)
- consolidated_df: Dataframe with multi-level headers for precision and recall per file.
- overall_metrics_df: Overall precision, recall, and F1 score for all fields.
"""
precision_dicts, recall_dicts = [], []
@@ -961,7 +769,25 @@ def calculate_precision_recall(accuracies):
precision_df = pd.DataFrame(precision_dicts)
recall_df = pd.DataFrame(recall_dicts)
return precision_df, recall_df, overall_metrics_df
# Create consolidated dataframe with single-level headers
# Get field names (excluding Filename)
fields = [col for col in precision_df.columns if col != "Filename"]
# Create consolidated data with single-level headers
consolidated_data = {"Filename": precision_df["Filename"]}
for field in fields:
# Add precision column with " - Precision" suffix
consolidated_data[f"{field} - Precision"] = precision_df[field]
# Add recall column with " - Recall" suffix
if field in recall_df.columns:
consolidated_data[f"{field} - Recall"] = recall_df[field]
# Create the consolidated dataframe
consolidated_df = pd.DataFrame(consolidated_data)
return consolidated_df, overall_metrics_df
########## TESTBED POSTPROCESSING CODE ##########
@@ -1136,9 +962,28 @@ def get_row_comparison(testbed, results):
row_comparison["testbed"] = testbed.groupby("FILE_NAME").size()
row_comparison["results"] = results.groupby("FILE_NAME").size()
# Add new columns
row_comparison["missing"] = row_comparison["testbed"] - row_comparison["results"]
row_comparison["abs_diff"] = row_comparison["missing"].abs()
row_comparison["pct_diff"] = row_comparison["abs_diff"] / row_comparison["testbed"]
# Add average row at the bottom
avg_row = pd.DataFrame(
{
"testbed": [row_comparison["testbed"].mean()],
"results": [row_comparison["results"].mean()],
"missing": [row_comparison["missing"].mean()],
"abs_diff": [row_comparison["abs_diff"].mean()],
"pct_diff": [row_comparison["pct_diff"].mean()],
},
index=["Average"],
)
row_comparison = pd.concat([row_comparison, avg_row])
print("\n*************** Row Count Comparison ****************")
print(
f"Files with different row counts: {(row_comparison['testbed'] != row_comparison['results']).sum()}"
f"Files with different row counts: {(row_comparison['testbed'] != row_comparison['results']).sum() - 1}" # -1 to exclude Average row
)
return row_comparison
@@ -1340,32 +1185,32 @@ def get_one_to_n_analysis(testbed, results):
.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",
],
# "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"],
# "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():
@@ -1426,10 +1271,9 @@ def get_one_to_n_analysis(testbed, results):
confusion_matrix["Filename"] = file
accuracies.append(confusion_matrix)
precision_df, recall_df, overall = calculate_precision_recall(accuracies)
consolidated_df, overall = calculate_precision_recall(accuracies)
one_to_n_metrics[field_type] = {
"precision": precision_df,
"recall": recall_df,
"consolidated": consolidated_df,
"overall": overall,
}
@@ -1455,19 +1299,6 @@ def get_one_to_n_analysis(testbed, results):
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,
@@ -1475,7 +1306,6 @@ def export_to_excel(
metrics_df,
comparison_df,
one_to_n_metrics,
reimb_primary_df,
provider_metrics_df,
):
with pd.ExcelWriter(output_file, engine="openpyxl") as writer:
@@ -1548,24 +1378,15 @@ def export_to_excel(
)
overall.to_excel(writer, sheet_name=f"{field_type} Overall")
else:
precision_df = metrics["precision"]
recall_df = metrics["recall"]
consolidated_df = metrics["consolidated"]
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
# Write consolidated dataframe with single-level headers to one sheet
consolidated_df.to_excel(
writer, sheet_name=f"{field_type} Metrics", 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)