Merged in bugfix/generic_dynamic_jan26 (pull request #859)

Bugfix/generic dynamic jan26 * Refactor: Extract dynamic primary metrics logic to separate module - Created src/testbed/dynamic_primary_metrics.py with all dynamic primary field analysis logic - Moved tally-based analysis functions to new module for better modularity - Updated testbed_utils.py to import and use functions from dynamic_primary_metrics - Simplified testbed_metrics_dynamic_only.py to use new module - Removed unused analyze_dynamic_primary_fields function (400+ lines) - Added field filtering in match_rows to handle missing columns gracefully - Minimal changes to testbed_utils.py (~100 lines vs 845 before) * feat: Add debug print blocks and prompt changes for dynamic primary fields in 1:1 processing This commit adds prompt changes for when we pass dynamic primary to 1:1 and comprehensive debug logging for dynamic primary fields when they are escalated to 1:1 processing via Hybrid Smart Chunking (HSC) or Full Context methods. Changes: - Added debug print blocks in hybrid_smart_chunking_funcs.py to log: * Retrieval question, context chunks, and final prompt for HSC processing * Raw LLM output (reasoning + answer) for dynamic primary fields - Added debug print blocks in prompt_calls.py to log: * Contract context, final prompt for Full Context processing * Raw LLM output (reasoning + answer) for dynamic primary fields - Added FULL_CONTEXT_DYNAMIC_PRIMARY_INSTRUCTION() to prompt_templates.py: * Provides specific guidance for dynamic primary fields in full context * Emphasizes extracting values only if they refer to entire contract * Includes pipe-delimited formatting instructions - Updated dynamic_fun… * Ran black, for formatting * Refine dynamic primary field prompts and add Duals auto-detection - Enhanced DYNAMIC_ASSIGNMENT_INSTRUCTION with structural boundary rules - Added exhibit header and subsection context hierarchy for LOB assignment - Implemented Duals auto-detection in dynamic_primary discovery (Medicare+Medicaid -> Duals) - Added Duals value formatting instruction to prevent oversimplification - Commented out DUAL_LOB_CHECK for performance testing - Removed verbose debug print blocks from HSC and Full Context processing - Added focused debug print for dynamic assignment raw LLM output * Remove debug print blocks from pipeline files - Removed all debug print blocks from file_processing.py (8 blocks) - Removed all debug print blocks from dynamic_funcs.py (8 blocks) - Removed debug print block from prompt_calls.py (dynamic assignment) - DUAL_LOB_CHECK remains commented as requested - Total: 188 lines of debug code removed * Merge remote-tracking branch 'origin/main' into bugfix/generic_dynamic_jan26 * Ran black for CI * Remove update_lob_for_duals * Merge branch 'DEV' into bugfix/generic_dynamic_jan26 Approved-by: Katon Minhas
2026-01-30 23:43:48 +00:00
parent 89515ca5cc
commit 1bc9e6aaf3
11 changed files with 723 additions and 127 deletions
@@ -376,6 +376,7 @@ def prompt_full_context(
        ],
        questions=prompt_questions,
    )
+
    try:
        claude_answer_raw = llm_utils.invoke_claude(
            full_context_prompt,
@@ -385,6 +386,7 @@ def prompt_full_context(
            cache=True,
            instruction=prompt_templates.ONE_TO_ONE_MULTI_FIELD_INSTRUCTION(),
        )
+
        full_context_answers_dict = string_utils.universal_json_load(claude_answer_raw)
    except Exception as e:
        logging.error(f"Error processing high accuracy fields: {str(e)}")
@@ -52,6 +52,32 @@ def dynamic_primary(
        )
        # If there is at least ONE Non-N/A answers in the Exhibit
        if not string_utils.is_empty(exhibit_text_answer):
+            # Special handling for LOB: If both Medicare and Medicaid are detected, add Duals
+            if field.field_name == "LOB":
+                # Parse the discovered values (pipe-delimited or comma-separated)
+                values_list = [
+                    val.strip()
+                    for val in exhibit_text_answer.replace(",", "|").split("|")
+                    if val.strip()
+                ]
+                values_lower = [val.lower() for val in values_list]
+
+                # Check if both Medicare and Medicaid are present (case-insensitive)
+                has_medicare = any("medicare" in val for val in values_lower)
+                has_medicaid = any("medicaid" in val for val in values_lower)
+
+                # If both are present and Duals is not already in the list, add it
+                if has_medicare and has_medicaid:
+                    if "duals" not in values_lower:
+                        exhibit_text_answer = (
+                            exhibit_text_answer + " | Duals"
+                            if exhibit_text_answer
+                            else "Duals"
+                        )
+                        logging.debug(
+                            f"Added 'Duals' to LOB valid values because both Medicare and Medicaid were detected"
+                        )
+
            field.update_valid_values(exhibit_text_answer + " | N/A")
            dynamic_reimbursement_fields.add_field(field)
            dynamic_primary_fields.remove_field(field.field_name)
@@ -213,8 +239,7 @@ def add_one_to_one_field(
    # Special Handling for Program, Product, and Network:
    # If any LOB value has been found, skip PROGRAM, PRODUCT, and NETWORK
    # Only search for these fields when NO LOB has been found
-    # NOTE: Check raw LOB field, not AARETE_DERIVED_LOB, because AARETE_DERIVED_LOB
-    # is populated later via crosswalk and may be derived from PROGRAM/PRODUCT
+
    if field_to_add.field_name in ["PROGRAM", "PRODUCT", "NETWORK"]:
        lob_values = [answer_dict.get("LOB", "") for answer_dict in answer_dicts]
        unique_lobs = set([lob for lob in lob_values if not string_utils.is_empty(lob)])
@@ -227,6 +252,15 @@ def add_one_to_one_field(
            + prompt_templates.FULL_CONTEXT_CLAIM_TYPES_ADDITIONAL_INSTRUCTION()
        )

+    # Special instructions for dynamic primary fields when passed to 1:1
+    # Add specific template-language guidance, then append general full context instructions
+    if field_to_add.field_name in ["LOB", "PRODUCT", "PROGRAM", "NETWORK"]:
+        field_to_add.prompt = (
+            field_to_add.prompt
+            + prompt_templates.FULL_CONTEXT_DYNAMIC_PRIMARY_INSTRUCTION()
+        )
+
+    # Apply general full context instructions to all fields
    field_to_add.prompt = (
        field_to_add.prompt
        + prompt_templates.FULL_CONTEXT_ADDITIONAL_INSTRUCTIONS(field_to_add.allow_na)
@@ -286,5 +320,8 @@ def get_dynamic_one_to_one_fields(
            one_to_one_fields = add_one_to_one_field(
                one_to_one_fields, field_to_add, answer_dicts, constants
            )
+        else:
+            # Field found in some rows - keep in 1:N
+            pass

    return one_to_one_fields
@@ -789,9 +789,6 @@ def one_to_n_cleaning(
    ################################ Fill NA Mapping ################################
    all_exhibit_rows = aarete_derived.fill_na_mapping(all_exhibit_rows)

-    ################################ Update LOB for Duals ################################
-    all_exhibit_rows = postprocessing_funcs.update_lob_for_duals(all_exhibit_rows)
-
    ################################ Split REIMB_DATES ################################
    all_exhibit_rows = split_reimb_dates(all_exhibit_rows, filename)

@@ -526,45 +526,6 @@ def add_aarete_derived_product(df):
    return df


-def update_lob_for_duals(answer_dicts):
-    final_answer_dicts = []
-    # Iterate through each row in the dataframe
-    for answer_dict in answer_dicts:
-        # Get SERVICE_TERM value
-        service_term = answer_dict.get("SERVICE_TERM", "").lower()
-        # Check if the SERVICE_TERM contains "Medicare" and "Medicaid" - case insensitive
-        # AND check that there's no explicit PROGRAM or PRODUCT that would override this logic
-        if (
-            "medicare" in service_term
-            and "medicaid" in service_term
-            and string_utils.is_empty(answer_dict.get("AARETE_DERIVED_PROGRAM"))
-            and string_utils.is_empty(answer_dict.get("AARETE_DERIVED_PRODUCT"))
-        ):
-            prompt = prompt_templates.DUAL_LOB_CHECK(service_term)
-            logging.debug(f"Prompt for dual LOB check: {prompt}")
-            claude_answer_raw = llm_utils.invoke_claude(
-                prompt,
-                model_id="legacy_sonnet",
-                filename="",
-                max_tokens=200,
-                cache=True,
-                instruction=prompt_templates.DUAL_LOB_CHECK_INSTRUCTION(),
-            )
-            logging.debug(f"Claude response for dual LOB check: {claude_answer_raw}")
-            claude_answer_extracted = string_utils.extract_text_from_delimiters(
-                claude_answer_raw, Delimiter.PIPE
-            )
-            if (
-                claude_answer_extracted.lower() == "neither"
-                or claude_answer_extracted.lower() == "n/a"
-            ):  # If neither, take original AARETE_DERIVED_LOB
-                pass
-            else:
-                answer_dict["AARETE_DERIVED_LOB"] = claude_answer_extracted
-        final_answer_dicts.append(answer_dict)
-    return final_answer_dicts
-
-
 def fill_empty_reimb_pct_rate(df):
    """
    Fill 'REIMB_PCT_RATE' based on reimbursement method:
@@ -1,3 +1,5 @@
+from typing import Optional
+
 import src.config as config
 from src.prompts.fieldset import FieldSet
 from src.constants.constants import Constants
@@ -359,15 +361,44 @@ You will be shown a Reimbursement Term, and the section of a contract from which
 Closely examine the contract text, find the Reimbursement Term, and decide which field value or values apply to the Reimbursement Term.

 CRITICAL CONSIDERATIONS:
-1. Look for what field the Service actually *applies to*. This is not the same as the Schedule used as a basis for calculation. For example, many LOBs use Medicaid as a basis for their calculation. The presence of 'Medicaid' as a rate does not on it's own guarantee that the Reimbursement Term is 'Medicaid'.
+1. EXHIBIT HEADER CONTEXT - Pay close attention to exhibit headers (e.g., "MEDICARE PRODUCT ATTACHMENT") that appear at the beginning of the exhibit section. These headers indicate the scope and context for ALL reimbursement terms within that exhibit. If an exhibit header clearly specifies a line of business (e.g., "MEDICARE PRODUCT ATTACHMENT"), assign values consistent with that header (e.g., "Medicare" for LOB) for all terms in that exhibit. The exhibit header value should be ADDITIVE - it should be included as one of multiple values when other sources (service term, reimbursement term, or subsection context) also provide relevant values. For example, if the exhibit header indicates "Medicare" and the service term or subsection context indicates "Duals", the final assignment should be "Medicare|Duals" (combining both values). The exhibit header value is always included and does not override other valid values.

-2. USE THE SERVICE_TERM AS PRIMARY CONTEXT: The Service Term shows which programs are mentioned together. If the Service Term explicitly lists specific programs (e.g., "STAR, CHIP HMO, CHIP PERINATE, and STAR+PLUS"), those are the programs that apply to reimbursement terms in that section. Do NOT assign programs that are mentioned in completely separate sections of the contract.
+2. SUBSECTION CONTEXT - Examine the exhibit text for the subsection header, introductory paragraph, or section introduction that is CLOSEST to the reimbursement term.

-3. LOOK FOR PROXIMITY: Programs must be explicitly mentioned in the SAME sentence or paragraph as the Reimbursement Term. If a reimbursement term appears in a section that lists specific programs (e.g., "STAR, CHIP HMO, CHIP PERINATE, and STAR+PLUS: Covered Services..."), only assign those programs, NOT programs mentioned in separate sections (e.g., "Medicare Advantage (Molina Medicare Options) and MA-SNP...").
+   RELATIONSHIP HIERARCHY:
+   - The exhibit header (instruction #1) is ALWAYS additive - it is included in the final assignment.
+   - The subsection context is ADDITIVE with respect to the exhibit header (combines with it).
+   - The subsection context OVERRIDES conflicting information in the reimbursement term or service term.
+   - The reimbursement term (instruction #3) takes LEAST precedence when exhibit header and subsection context provide clear answers.

-4. AVOID CROSS-CONTAMINATION: If the Service Term mentions Medicaid programs (STAR, CHIP, CHIPP, STAR+PLUS, etc.), do NOT assign Medicare programs (MA, MASNP) unless they are explicitly mentioned together in the same context. Similarly, if the Service Term mentions Medicare programs, do NOT assign Medicaid programs unless explicitly mentioned together.
+   STRUCTURAL BOUNDARY RULE - FOLLOW THESE STEPS:
+   1. Identify the structural element containing the reimbursement term (e.g., "Table 1", "Table 2").
+   2. Trace backwards from that element to find the MOST RECENT section marker (e.g., "4.", "3.", "Section A", numbered items like "1.", "2.", "3.").
+   3. If you find a section marker, that creates a HARD BOUNDARY - subsection context from BEFORE that marker is COMPLETELY INVALIDATED and must be ignored, even if it mentioned the table.
+   4. Only consider subsection context that appears AFTER the most recent section marker and BEFORE the structural element.
+   5. If no section marker is found, then use the closest subsection context before the structural element.
+   
+    Section markers (numbered items like "1.", "2.", "3.", "4.") create hard boundaries. Example: If "4. Allowable Charges...for any Medicare inpatient admission" appears before "Table 2", you MUST use ONLY context from section 4. Completely ignore subsection context from earlier sections (like "For Covered Services rendered to Covered Person who are eligible for Medicare and enrolled in a Medicare Plan that may include coverage for both Medicare and Medicaid") even if that earlier text mentioned "Table 2" - the section 4 marker invalidates all earlier context.

-5. "ALL PROGRAMS" AMBIGUITY - CRITICAL: If you see phrases like "for all Health Plan products & programs" in a table or section, this phrase means "all programs in THIS SECTION", NOT "all programs in the entire document".
+   When a relevant subsection context is found WITHIN THE SAME structural boundary with extremely clear wording about eligibility or program coverage, it should:
+   - ADD to the exhibit header value (e.g., exhibit header "Medicare" + subsection "both Medicare and Medicaid" = "Medicare|Duals")
+   - OVERRIDE conflicting information in the service term or reimbursement term (e.g., if subsection says "Duals" but service term says "Medicaid only", assign "Duals" based on subsection)
+
+3. REIMBURSEMENT TERM CONTEXT - Examine the Reimbursement Term itself for context about what value should be assigned to the field. This is a broad instruction to consider the reimbursement term as a primary source of information. However, if the exhibit header (instruction #1) and nearest subsection context (instruction #2) provide clear answers, the reimbursement term context takes LEAST precedence. The reimbursement term context should be considered in addition to the context provided by the exhibit header and subsection context, but should not override them when they are clear and explicit.
+
+4. Look for what field the Service actually *applies to*. This is not the same as the Schedule used as a basis for calculation. For example, many LOBs use Medicaid or Medicare as a basis for their calculation. The presence of 'Medicaid' or 'Medicare' as a rate does not on it's own guarantee that the Reimbursement Term is 'Medicaid' or 'Medicare'.
+
+5. DUALS VALUE FORMATTING: If the answer appears to be "Medicare|Duals" or "Medicaid|Duals", do NOT simplify to just "Duals" - the pipe-delimited format adds important context. Only simplify to "Duals" alone when it is obvious that Duals alone is sufficient and no additional context is needed.
+
+[PROGRAM FIELD SPECIFIC INSTRUCTIONS - Instructions 6-9 below are PRIMARILY for PROGRAM field assignment, though they may loosely apply to other fields. For LOB field assignment, prioritize instructions 1-5 above these instructions.]
+
+6. USE THE SERVICE_TERM AS CONTEXT: The Service Term shows which programs are mentioned together. If the Service Term explicitly lists specific programs (e.g., "STAR, CHIP HMO, CHIP PERINATE, and STAR+PLUS"), those are the programs that apply to reimbursement terms in that section. Do NOT assign programs that are mentioned in completely separate sections of the contract.
+
+7. LOOK FOR PROXIMITY: Programs must be explicitly mentioned in the SAME sentence or paragraph as the Reimbursement Term. If a reimbursement term appears in a section that lists specific programs (e.g., "STAR, CHIP HMO, CHIP PERINATE, and STAR+PLUS: Covered Services..."), only assign those programs, NOT programs mentioned in separate sections (e.g., "Medicare Advantage (Molina Medicare Options) and MA-SNP...").
+
+8. AVOID CROSS-CONTAMINATION: If the Service Term mentions Medicaid programs (STAR, CHIP, CHIPP, STAR+PLUS, etc.), do NOT assign Medicare programs (MA, MASNP) unless they are explicitly mentioned together in the same context. Similarly, if the Service Term mentions Medicare programs, do NOT assign Medicaid programs unless explicitly mentioned together.
+
+9. "ALL PROGRAMS" AMBIGUITY - CRITICAL: If you see phrases like "for all Health Plan products & programs" in a table or section, this phrase means "all programs in THIS SECTION", NOT "all programs in the entire document".
   - If the table appears AFTER a section that says "STAR, CHIP HMO, CHIP PERINATE, and STAR+PLUS" and BEFORE a section that says "Medicare Advantage", the table applies ONLY to the Medicaid programs (STAR, CHIP, CHIPP, STAR+PLUS) mentioned in that section.
   - If the table appears AFTER a section that says "Medicare Advantage", the table applies ONLY to the Medicare programs mentioned in that section.
   - NEVER assign programs from separate sections just because you see "all programs" - always determine which section the reimbursement term belongs to based on its position in the document."""
@@ -1253,6 +1284,25 @@ def FULL_CONTEXT_PAYER_NAME_ADDITIONAL_INSTRUCTION():
    return " Additionally check the contract text for the payer name or payer organization name or payer entity or name of organization issuing the health plan/policy and Choose the most appropriate answer based on the context."


+def FULL_CONTEXT_DYNAMIC_PRIMARY_INSTRUCTION():
+    """Additional instructions for dynamic primary fields (LOB, PRODUCT, PROGRAM, NETWORK) when passed to 1:1 processing.
+
+    These instructions emphasize that values should only be extracted if they explicitly
+    refer to the entire contract and all terms, and to ignore template language or general
+    definitions that don't specifically apply to this contract.
+    """
+    return (
+        " CRITICAL: Only extract values if they explicitly refer to the ENTIRE contract "
+        "and apply to ALL terms and services covered by this specific contract. "
+        "IGNORE any mentions that are part of template language, general definitions, "
+        "or boilerplate text that describes general capabilities but does not specifically "
+        "apply to this contract. If the value is only mentioned in definitions, examples, "
+        "or general descriptions without explicit application to this contract's terms, "
+        "return 'N/A'. Only extract values that are clearly stated as applying to this "
+        "specific contract and all its terms."
+    )
+
+
 def CHECK_PROVIDER_NAME_MATCH_INSTRUCTION() -> str:
    """Static instruction for CHECK_PROVIDER_NAME_MATCH prompt caching.
    Contains all matching rules and critical exclusions.
@@ -1627,22 +1677,6 @@ Service: {service_term}
 Reimbursement Term: {reimb_term.replace('"', "'")}"""


-def DUAL_LOB_CHECK(service_term):
-    return f"""Examine the given Service and determine if the Service applies to Medicare, Medicaid, or Both.
-
-If the Service is only for Medicaid, return 'Medicaid'
-If the Service is only for Medicare, return 'Medicare'
-If the Service is for both Medicare and Medicaid, return 'Duals'
-If the service is for neither Medicare nor Medicaid, return 'neither'
-If it is unclear, write 'N/A'.
-
-Here is the Service:
-{service_term}
-
-Explain your answer in one or two sentences, but put your final answer in |pipes|.
-"""
-
-
 def EFFECTIVE_DATE_FIX_PROMPT() -> str:
    return """Extract the effective date of the contract or contract effective date or Effective Date of Agreement or Effective Date of Amendment or the amendment effective date from the given image data. ALSO LOOK FOR dates in phrases that semantically indicate an effective date (may or may not be explictly labelled as effective date), even if they do not exactly match the phrases above.\nExamples include but ARE NOT LIMITED TO:\nPhrases indicating when an contract or amendment begins or takes effect, Phrases specifying the start date of the contract or amendment, Phrases defining when the contract or amendment becomes operational, or Any phrase indicating contract or amendment creation or execution date.\nIt is possible that the contract will specify that the effective date is derived from elsewhere in the contract. Give the date in a JSON FORMAT with AARETE_DERIVED_EFFECTIVE_DATE as the key and effective date value in YYYY/MM/DD format as the value. 
    Rules:
@@ -0,0 +1,412 @@
+"""Dynamic Primary Fields Metrics Analysis.
+
+This module provides tally-based, order-independent analysis for dynamic primary fields:
+- AARETE_DERIVED_LOB
+- AARETE_DERIVED_PROGRAM
+- AARETE_DERIVED_NETWORK
+- AARETE_DERIVED_PRODUCT
+"""
+
+from collections import Counter
+
+import pandas as pd
+import src.config as config
+import src.utils.string_utils as string_utils
+
+
+def analyze_dynamic_primary_fields_tally_based(
+    testbed_file: pd.DataFrame,
+    results_file: pd.DataFrame,
+    fields: list[str],
+):
+    """Analyze dynamic primary fields using a tally/count-based approach.
+
+    This approach is order-independent and focuses on value counts rather than
+    specific row matching. It provides insights into:
+    1. Correct values: Counts match between testbed and results
+    2. Missing values: Testbed has more occurrences of a value
+    3. Extra values: Results have more occurrences of a value
+
+    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
+
+    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]]
+
+        # 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)
+
+        # Count occurrences
+        testbed_counts = Counter(testbed_expanded)
+        results_counts = Counter(results_expanded)
+
+        # Calculate metrics
+        all_values = set(testbed_counts.keys()) | set(results_counts.keys())
+
+        tp = 0  # Values with correct counts
+        missing_details = (
+            []
+        )  # Values in testbed but not in results (or fewer in results)
+        extra_details = []  # Values in results but not in testbed (or more in results)
+
+        for value in all_values:
+            testbed_count = testbed_counts.get(value, 0)
+            results_count = results_counts.get(value, 0)
+
+            # TP is the minimum count (the matching portion)
+            matching_count = min(testbed_count, results_count)
+            if matching_count > 0:
+                tp += matching_count
+
+            # Missing: testbed has more than results
+            if testbed_count > results_count:
+                missing_count = testbed_count - results_count
+                missing_details.append(
+                    (value, testbed_count, results_count, missing_count)
+                )
+
+            # Extra: results have more than testbed
+            if results_count > testbed_count:
+                extra_count = results_count - testbed_count
+                extra_details.append((value, testbed_count, results_count, extra_count))
+
+        # Calculate totals
+        missing_count = sum(diff for _, _, _, diff in missing_details)
+        extra_count = sum(diff for _, _, _, diff in extra_details)
+
+        analysis_results[field] = {
+            "tp": tp,
+            "missing_values": missing_details,
+            "extra_values": extra_details,
+            "missing_count": missing_count,
+            "extra_count": extra_count,
+            "testbed_counts": dict(testbed_counts),
+            "results_counts": dict(results_counts),
+        }
+
+    return analysis_results
+
+
+def get_dynamic_primary_analysis(testbed, results, fields):
+    """Analyze dynamic primary fields across all files using tally-based approach.
+
+    Args:
+        testbed: DataFrame containing ground truth
+        results: DataFrame containing predictions
+        fields: List of dynamic primary field names to analyze
+
+    Returns:
+        dict: Dictionary with "summary", "details", and "overall" DataFrames
+    """
+    print("\n*************** Dynamic Primary Fields Analysis ****************")
+
+    dynamic_primary_results = []
+    dynamic_primary_summary = []
+
+    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
+
+        # Use tally-based analysis for dynamic primary (order-independent)
+        file_analysis = analyze_dynamic_primary_fields_tally_based(
+            testbed_file, results_file, fields
+        )
+
+        # Create detailed results for this file
+        file_result = {
+            "Filename": file,
+            "testbed_row_count": len(testbed_file),
+            "results_row_count": len(results_file),
+        }
+        file_summary = {
+            "Filename": file,
+            "testbed_row_count": len(testbed_file),
+            "results_row_count": len(results_file),
+        }
+
+        for field in fields:
+            field_data = file_analysis[field]
+
+            tp = field_data["tp"]
+            missing = field_data["missing_count"]
+            extra = field_data["extra_count"]
+            mismatch = missing + extra
+            total = tp + missing + extra
+
+            # Calculate error score: (Missing × 3 + Extra × 1) / (TP + Missing + Extra)
+            # Missing is weighted 3× more than Extra since missing values are more problematic
+            if total > 0:
+                weighted_errors = missing * 3 + extra * 1
+                error_score = min(weighted_errors / total, 1.0)
+            else:
+                error_score = 0.0
+
+            # Store counts
+            file_result[f"{field}_tp"] = tp
+            file_result[f"{field}_missing_count"] = missing
+            file_result[f"{field}_extra_count"] = extra
+            file_result[f"{field}_mismatch_count"] = mismatch
+            file_result[f"{field}_error_score"] = error_score
+
+            # Store testbed and results counts in dictionary format
+            testbed_counts_str = str(field_data["testbed_counts"])
+            results_counts_str = str(field_data["results_counts"])
+            file_result[f"{field}_testbed_counts"] = testbed_counts_str
+            file_result[f"{field}_results_counts"] = results_counts_str
+
+            # Store detailed lists (format for readability in Excel)
+            missing_str = "\n".join(
+                [
+                    f"{val}: Testbed={t_count}, Results={r_count}, Missing={diff}"
+                    for val, t_count, r_count, diff in field_data["missing_values"]
+                ]
+            )
+            file_result[f"{field}_missing_values"] = missing_str if missing_str else ""
+
+            extra_str = "\n".join(
+                [
+                    f"{val}: Testbed={t_count}, Results={r_count}, Extra={diff}"
+                    for val, t_count, r_count, diff in field_data["extra_values"]
+                ]
+            )
+            file_result[f"{field}_extra_values"] = extra_str if extra_str else ""
+
+            # Summary counts for aggregation
+            file_summary[f"{field}_tp"] = tp
+            file_summary[f"{field}_missing"] = missing
+            file_summary[f"{field}_extra"] = extra
+            file_summary[f"{field}_mismatch"] = mismatch
+            file_summary[f"{field}_error_score"] = error_score
+
+        dynamic_primary_results.append(file_result)
+        dynamic_primary_summary.append(file_summary)
+
+    # Create overall summary across all files
+    overall_summary = {}
+    for field in fields:
+        total_tp = sum(r[f"{field}_tp"] for r in dynamic_primary_summary)
+        total_missing = sum(
+            r.get(f"{field}_missing", 0) for r in dynamic_primary_summary
+        )
+        total_extra = sum(r.get(f"{field}_extra", 0) for r in dynamic_primary_summary)
+        total_mismatch = total_missing + total_extra
+
+        total_errors = total_mismatch
+        total_cases = total_tp + total_errors
+
+        # Precision: Of all value occurrences, how many had correct counts?
+        precision = (
+            total_tp / (total_tp + total_missing + total_extra)
+            if (total_tp + total_missing + total_extra) > 0
+            else 0
+        )
+
+        # Recall: Of all testbed value occurrences, how many were correctly extracted?
+        recall = (
+            total_tp / (total_tp + total_missing)
+            if (total_tp + total_missing) > 0
+            else 0
+        )
+        f1 = (
+            (2 * precision * recall) / (precision + recall)
+            if (precision + recall) > 0
+            else 0
+        )
+
+        # Overall error score (weighted average across files)
+        total_weighted_errors = total_missing * 3 + total_extra * 1
+        overall_error_score = (
+            total_weighted_errors / total_cases if total_cases > 0 else 0
+        )
+
+        overall_summary[field] = {
+            "tp": total_tp,
+            "missing": total_missing,
+            "extra": total_extra,
+            "mismatch": total_mismatch,
+            "error_score": overall_error_score,
+            "prec": precision,
+            "rec": recall,
+            "f1": f1,
+        }
+
+    overall_df = pd.DataFrame(overall_summary).T
+    overall_df.index.name = "Field"
+
+    summary_df = pd.DataFrame(dynamic_primary_summary)
+    details_df = pd.DataFrame(dynamic_primary_results)
+
+    return {
+        "summary": summary_df,
+        "details": details_df,
+        "overall": overall_df,
+    }
+
+
+def export_dynamic_primary_to_excel(output_file, dynamic_primary_metrics):
+    """Export dynamic primary fields analysis to Excel with separate sheets per field.
+
+    Args:
+        output_file: Path to output Excel file
+        dynamic_primary_metrics: Dictionary with "summary", "details", and "overall" DataFrames
+    """
+    summary_df = dynamic_primary_metrics["summary"]
+    details_df = dynamic_primary_metrics["details"]
+    overall_df = dynamic_primary_metrics["overall"]
+
+    dynamic_fields = [
+        "AARETE_DERIVED_LOB",
+        "AARETE_DERIVED_PROGRAM",
+        "AARETE_DERIVED_NETWORK",
+        "AARETE_DERIVED_PRODUCT",
+    ]
+
+    with pd.ExcelWriter(output_file, engine="openpyxl") as writer:
+        # Create separate sheets for each dynamic primary field
+        for field in dynamic_fields:
+            if f"{field}_tp" not in summary_df.columns:
+                continue
+
+            # Create field-specific sheet
+            field_df = pd.DataFrame(
+                {
+                    "Filename": summary_df["Filename"],
+                    "TP": summary_df[f"{field}_tp"],
+                    "Missing": summary_df[f"{field}_missing"],
+                    "Extra": summary_df[f"{field}_extra"],
+                    "Mismatch": summary_df[f"{field}_mismatch"],
+                    "Error_Score": summary_df[f"{field}_error_score"],
+                    "Testbed_Counts": details_df[f"{field}_testbed_counts"],
+                    "Results_Counts": details_df[f"{field}_results_counts"],
+                }
+            )
+
+            # Sort by Error_Score descending (most problematic first)
+            field_df = field_df.sort_values("Error_Score", ascending=False)
+
+            # Use a clean sheet name
+            sheet_name = field.replace("AARETE_DERIVED_", "").replace("_", " ").title()
+            field_df.to_excel(writer, sheet_name=sheet_name, index=False)
+
+        # Overall summary sheet with aggregated statistics across all files
+        overall_df.to_excel(writer, sheet_name="Overall Summary", index=True)
+
+        # Detailed summary sheet (all files, all fields)
+        summary_df.to_excel(writer, sheet_name="Summary", index=False)
+
+        # Details sheet (all files, all fields with counts)
+        details_df.to_excel(writer, sheet_name="Details", 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 = min(max_length + 2, 50)
+                worksheet.column_dimensions[column[0].column_letter].width = (
+                    adjusted_width
+                )
+
+
+def run_dynamic_primary_analysis_only(testbed_path, results_path, output_path=None):
+    """Run dynamic primary fields analysis only (standalone function).
+
+    Args:
+        testbed_path: Path to testbed CSV file
+        results_path: Path to results CSV file
+        output_path: Optional path for output Excel file. If None, uses default naming.
+
+    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}")
+    testbed = pd.read_csv(testbed_path)
+    print(f"Testbed loaded: {testbed.shape[0]} rows, {testbed.shape[1]} columns")
+
+    print(f"Loading results from: {results_path}")
+    results = pd.read_csv(results_path)
+    print(f"Results loaded: {results.shape[0]} rows, {results.shape[1]} columns")
+
+    # Preprocess
+    testbed = testbed_utils.testbed_preprocess(testbed)
+    results = testbed_utils.testbed_preprocess(results)
+    testbed, results = testbed_utils.filter_file_names(testbed, results)
+
+    print(f"\nAfter filtering: {len(testbed['FILE_NAME'].unique())} files in common")
+
+    # Define fields to analyze
+    dynamic_primary_fields = [
+        "AARETE_DERIVED_LOB",
+        "AARETE_DERIVED_PROGRAM",
+        "AARETE_DERIVED_NETWORK",
+        "AARETE_DERIVED_PRODUCT",
+    ]
+
+    # Filter to only fields that exist in both dataframes
+    fields_to_analyze = [
+        f
+        for f in dynamic_primary_fields
+        if f in testbed.columns and f in results.columns
+    ]
+
+    print(f"Analyzing {len(fields_to_analyze)} fields: {fields_to_analyze}")
+
+    # Run the analysis
+    metrics = get_dynamic_primary_analysis(testbed, results, fields_to_analyze)
+
+    # Export to Excel if output path provided
+    if output_path:
+        print("\nExporting results to Excel...")
+        export_dynamic_primary_to_excel(output_path, metrics)
+        print(f"Results exported to: {output_path}")
+    elif output_path is None:
+        # Use default naming
+        output_path = f"data/testbed/{config.TODAY}-testbed-metrics-dynamic-only.xlsx"
+        print("\nExporting results to Excel...")
+        export_dynamic_primary_to_excel(output_path, metrics)
+        print(f"Results exported to: {output_path}")
+
+    return metrics
@@ -9,10 +9,16 @@ warnings.filterwarnings("ignore")

 ########################## READ AND PREPROCESS ##########################
 # Read files
-testbed = pd.read_csv("Doczy-Testbed-Official.csv")  # This is the testbed
-results = pd.read_csv(
-    "inv-test-20260122-generic-RESULTS.csv"
-)  # This is the doczy output
+testbed_path = "data/testbed/Doczy-Testbed-Official.csv"
+results_path = "data/testbed/inv-test-20260122-generic-RESULTS-CLEAN.csv"
+
+print(f"Loading testbed from: {testbed_path}")
+testbed = pd.read_csv(testbed_path)  # This is the testbed
+print(f"Testbed loaded: {testbed.shape[0]} rows, {testbed.shape[1]} columns")
+
+print(f"Loading results from: {results_path}")
+results = pd.read_csv(results_path)  # This is the doczy output
+print(f"Results loaded: {results.shape[0]} rows, {results.shape[1]} columns")

 ########################## Preprocess ##########################
 testbed = testbed_utils.testbed_preprocess(testbed)
@@ -42,8 +48,8 @@ provider_results, provider_metrics_df = (
 ########################## Export to Excel ##########################
 print("\nExporting results to Excel...")

-# Create Excel writer object
-output_file = f"{config.TODAY}-testbed-metrics.xlsx"
+# Create Excel writer object - save in testbed folder
+output_file = f"data/testbed/{config.TODAY}-testbed-metrics.xlsx"
 testbed_utils.export_to_excel(
    output_file,
    testbed,
@@ -0,0 +1,30 @@
+"""Run dynamic primary fields analysis only.
+
+This script analyzes only the 4 dynamic primary fields:
+- AARETE_DERIVED_LOB
+- AARETE_DERIVED_PROGRAM
+- AARETE_DERIVED_NETWORK
+- AARETE_DERIVED_PRODUCT
+"""
+
+import warnings
+
+import src.testbed.dynamic_primary_metrics as dynamic_primary_metrics
+
+warnings.filterwarnings("ignore")
+
+if __name__ == "__main__":
+    # File paths
+    testbed_path = "data/testbed/Doczy-Testbed-Official.csv"
+    results_path = "data/testbed/inv-test-20260122-generic-RESULTS-CLEAN.csv"
+
+    print("\n" + "=" * 80)
+    print("ANALYZING DYNAMIC PRIMARY FIELDS ONLY")
+    print("=" * 80)
+
+    # Run the analysis (includes preprocessing, analysis, and export)
+    metrics = dynamic_primary_metrics.run_dynamic_primary_analysis_only(
+        testbed_path, results_path
+    )
+
+    print("\nAnalysis complete!")
@@ -2,8 +2,8 @@
 # function `testbed_postprocess` from the `testbed_utils` module (similar to
 # postprocess.postprocess() but skipping some steps)

-# Run this from the fieldExtraction directory:
-# poetry run python -m src.testbed.testbed_postprocess
+# Run this from the project root:
+# uv run python -m src.testbed.testbed_postprocess

 #### CONFIG ####
 testbed_filename = "src/testbed/v6_DoczyAI_all_test_bed_LIVE_5.6.25.xlsx"
@@ -12,6 +12,13 @@ from rapidfuzz import fuzz
 from src.pipelines.shared.postprocessing import postprocessing_funcs
 from src.core.fieldset import FieldSet

+# Import dynamic primary metrics functions
+from src.testbed.dynamic_primary_metrics import (
+    analyze_dynamic_primary_fields_tally_based,
+    export_dynamic_primary_to_excel,
+    get_dynamic_primary_analysis,
+)
+

 def filter_file_names(testbed, results):
    # Ensure FILE_NAME values match between testbed and results
@@ -579,6 +586,19 @@ def match_rows(fields, labels_df, predictions_df, N):

        return False

+    # Filter fields to only include those that exist in both dataframes
+    original_fields = fields
+    fields = [
+        f for f in fields if f in labels_df.columns and f in predictions_df.columns
+    ]
+
+    if not fields:
+        # Return empty confusion matrix for all original fields
+        return {
+            field_name: {"tp": 0, "fp": 0, "fn": 0, "unmatched_labels": []}
+            for field_name in original_fields
+        }
+
    confusion_matrix = {
        field_name: {"tp": 0, "fp": 0, "unmatched_labels": []} for field_name in fields
    }
@@ -1294,6 +1314,22 @@ def get_one_to_one_analysis(testbed, results):
    return metrics_df, comparison_df


+def get_dynamic_primary_analysis_only(testbed, results, fields):
+    """Analyze only dynamic primary fields using tally-based approach.
+
+    This is a wrapper that calls the function from dynamic_primary_metrics module.
+
+    Args:
+        testbed: DataFrame containing ground truth
+        results: DataFrame containing predictions
+        fields: List of dynamic primary field names to analyze
+
+    Returns:
+        dict: Dictionary with "summary", "details", and "overall" DataFrames
+    """
+    return get_dynamic_primary_analysis(testbed, results, fields)
+
+
 def get_one_to_n_analysis(testbed, results):
    print("\n*************** One-to-N Stats ****************")

@@ -1333,48 +1369,88 @@ def get_one_to_n_analysis(testbed, results):
    }
    one_to_n_metrics = {}
    for field_type, fields in one_to_n_fields.items():
-        N = len(fields)
-        accuracies = []
+        # Filter fields to only include those that exist in both dataframes
+        fields = [f for f in fields if f in testbed.columns and f in results.columns]

-        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,
-                },
+        # Skip this field type if no fields are available
+        if not fields:
+            print(
+                f"\nSkipping {field_type}: No matching fields found in both testbed and results"
+            )
+            continue
+
+        # Special handling for dynamic_primary fields - use tally-based analysis
+        if field_type == "dynamic_primary":
+            # Use the imported function from dynamic_primary_metrics module
+            dynamic_primary_metrics_result = get_dynamic_primary_analysis(
+                testbed, results, fields
+            )
+            one_to_n_metrics[field_type] = dynamic_primary_metrics_result
+
+            print(f"\nOverall Metrics: {field_type}")
+            print(
+                dynamic_primary_metrics_result["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,
+                        "missing": 15,
+                        "extra": 15,
+                        "mismatch": 15,
+                        "error_score": 15,
+                        "prec": 15,
+                        "rec": 15,
+                        "f1": 15,
+                    },
+                )
+            )
+        else:
+            # Standard handling for other field types
+            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

@@ -1428,16 +1504,62 @@ def export_to_excel(

        # 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"]
+            # Special handling for dynamic_primary - create separate sheets per field
+            if field_type == "dynamic_primary":
+                summary_df = metrics["summary"]
+                details_df = metrics["details"]
+                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")
+                # Create separate sheets for each dynamic primary field
+                dynamic_fields = [
+                    "AARETE_DERIVED_LOB",
+                    "AARETE_DERIVED_PROGRAM",
+                    "AARETE_DERIVED_NETWORK",
+                    "AARETE_DERIVED_PRODUCT",
+                ]
+
+                for field in dynamic_fields:
+                    if f"{field}_tp" not in summary_df.columns:
+                        continue
+
+                    field_df = pd.DataFrame(
+                        {
+                            "Filename": summary_df["Filename"],
+                            "TP": summary_df[f"{field}_tp"],
+                            "Missing": summary_df[f"{field}_missing"],
+                            "Extra": summary_df[f"{field}_extra"],
+                            "Mismatch": summary_df[f"{field}_mismatch"],
+                            "Error_Score": summary_df[f"{field}_error_score"],
+                            "Testbed_Counts": details_df[f"{field}_testbed_counts"],
+                            "Results_Counts": details_df[f"{field}_results_counts"],
+                        }
+                    )
+                    field_df = field_df.sort_values("Error_Score", ascending=False)
+                    sheet_name = (
+                        field.replace("AARETE_DERIVED_", "").replace("_", " ").title()
+                    )
+                    field_df.to_excel(writer, sheet_name=sheet_name, index=False)
+
+                summary_df.to_excel(
+                    writer, sheet_name=f"{field_type} Summary", index=False
+                )
+                details_df.to_excel(
+                    writer, sheet_name=f"{field_type} Details", index=False
+                )
+                overall.to_excel(writer, sheet_name=f"{field_type} Overall")
+            else:
+                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(
@@ -659,9 +659,6 @@ class TestOneToNFuncs(unittest.TestCase):
    @patch(
        "src.pipelines.shared.extraction.one_to_n_funcs.aarete_derived.fill_na_mapping"
    )
-    @patch(
-        "src.pipelines.shared.extraction.one_to_n_funcs.postprocessing_funcs.update_lob_for_duals"
-    )
    @patch("src.pipelines.shared.extraction.one_to_n_funcs.split_reimb_dates")
    def test_one_to_n_cleaning_full_pipeline(
        self,
@@ -677,7 +674,6 @@ class TestOneToNFuncs(unittest.TestCase):
        mock_crosswalk.return_value = input_data
        mock_get_lob.return_value = input_data
        mock_fill_na.return_value = input_data
-        mock_update_duals.return_value = input_data
        mock_split_dates.return_value = input_data

        result = one_to_n_funcs.one_to_n_cleaning(
@@ -688,7 +684,6 @@ class TestOneToNFuncs(unittest.TestCase):
        mock_crosswalk.assert_called_once()
        mock_get_lob.assert_called_once()
        mock_fill_na.assert_called_once()
-        mock_update_duals.assert_called_once()
        mock_split_dates.assert_called_once()

        self.assertEqual(result, input_data)