Merged in feature/DAIP2-1562-add-aarete_derived_payer_name (pull request #867)

Feature/DAIP2-1562 add aarete derived payer name

* black format

* Merged DEV into feature/DAIP2-1562-add-aarete_derived_payer_name

* llm_choose_derived_payer_name function added

* threshold updated

* Merge branch 'DEV' into feature/DAIP2-1562-add-aarete_derived_payer_name

* aarete_derived_payer_name column added

* prompt structure updated

* state_flag added

* pipeline error fixed

* pipeline error fixed

* Merged DEV into feature/DAIP2-1562-add-aarete_derived_payer_name

* remove debug print statement for similarity matrix in clustering function

* Merged DEV into feature/DAIP2-1562-add-aarete_derived_payer_name

* state logic added in clustering

* removed print statements

* pull request updates

* black format fix

* Merged DEV into feature/DAIP2-1562-add-aarete_derived_payer_name

* added scalability feature and optimization

* black format

* Pull Request Changes

* updated config parameters

* updated main

* no payer name column changes

* Business team feedback: always run derived payer name with state_flag=True

- Add empty DataFrame guard and proper docstring
- Clean up not needed flags


Approved-by: Siddhant Medar

src/constants/investment_columns.py

@@ -31,6 +31,7 @@ FIELD_FORMAT_MAPPING = {
     "AARETE_DERIVED_AMENDMENT_NUM": "str",
     "CLIENT_NAME": "str",
     "PAYER_NAME": "str",
+    "AARETE_DERIVED_PAYER_NAME": "str",
     "PAYER_STATE": "list[str]",
     "PROVIDER_STATE": "list[str]",
     "FILENAME_TIN": "list[str]",

src/constants/regex_patterns.py

@@ -73,3 +73,6 @@ OCR_SUBSTITUTIONS = {
     "B": "8",  # Letter B to eight
     "b": "8",
 }
+
+# dba patterns
+DBA_PATTERNS = [r"\bD/B/A\b", r"\bDBA\b", r"\bDOING BUSINESS AS\b", r"\bD B A\b"]

src/pipelines/saas/main.py

@@ -11,6 +11,9 @@ import pandas as pd
 random.seed(42)
 
 import src.pipelines.saas.file_processing as file_processing
+from src.pipelines.shared.extraction import one_to_one_funcs
+from src.pipelines.shared.postprocessing import postprocessing_funcs
+from src.constants.investment_columns import FIELD_FORMAT_MAPPING
 import src.prompts.prompt_templates as prompt_templates
 import src.utils.io_utils as io_utils
 import src.utils.llm_utils as llm_utils
@@ -221,6 +224,20 @@ def main(testing=False, test_params={}):
             FINAL_RESULT_DF_CC = pd.DataFrame()
             FINAL_RESULT_DF_DASHBOARD = pd.DataFrame()
 
+        # Add derived payer name and reorder columns for both CC and dashboard results
+        FINAL_RESULT_DF_CC = one_to_one_funcs.add_aarete_derived_payer_name(
+            FINAL_RESULT_DF_CC, state_flag=True
+        )
+        FINAL_RESULT_DF_CC = postprocessing_funcs.reorder_columns(
+            FINAL_RESULT_DF_CC, FIELD_FORMAT_MAPPING
+        )
+        FINAL_RESULT_DF_DASHBOARD = one_to_one_funcs.add_aarete_derived_payer_name(
+            FINAL_RESULT_DF_DASHBOARD, state_flag=True
+        )
+        FINAL_RESULT_DF_DASHBOARD = postprocessing_funcs.reorder_columns(
+            FINAL_RESULT_DF_DASHBOARD, FIELD_FORMAT_MAPPING
+        )
+
         if len(error_results) > 0:
             ERROR_RESULT_DF = pd.concat(error_results, ignore_index=True)
         else:

src/pipelines/saas/prompts/prompt_calls.py

@@ -1080,3 +1080,56 @@ def prompt_provider_info(
         logging.error(f"Raw response: {llm_answer_raw}")
         # Return a default value
         return [{"TIN": "N/A", "NPI": "N/A", "NAME": "N/A"}]
+
+
+def prompt_aarete_derived_payer_name(cluster_rows: list[dict], state_flag: bool) -> str:
+    """
+    Select the canonical payer name from a cluster of similar names (e.g., merging 'AvMed, Inc' and 'AvMed Health Plans').
+    Expects a list of standardized payer name dictionaries.
+
+    Args:
+        cluster_rows (list[dict]): A list of dictionaries, each containing standardized payer name information.
+
+    Returns:
+        str: The canonical payer name selected by the LLM, or "N/A" if selection fails.
+    """
+
+    # If only one record in cluster, avoid LLM call
+    if len(cluster_rows) == 1:
+        base_name = cluster_rows[0].get("base_payer_name", "UNKNOWN")
+        return str(base_name).upper()
+
+    prompt, _parser = prompt_templates.AARETE_DERIVED_PAYER_NAME(
+        cluster_rows, state_flag
+    )
+    logging.debug(f"Aarete Derived Payer Name Prompt: {prompt}")
+    llm_answer_raw = llm_utils.invoke_claude(
+        prompt,
+        "sonnet_latest",
+        "all_files",
+        cache=True,
+        instruction=prompt_templates.AARETE_DERIVED_PAYER_NAME_INSTRUCTION(),
+        usage_label="AARETE_DERIVED_PAYER_NAME",
+    )
+    logging.debug(f"LLM Response for Aarete Derived Payer Name: {llm_answer_raw}")
+    try:
+        llm_answer_final = _parser(llm_answer_raw)
+        if llm_answer_final and not string_utils.is_empty(llm_answer_final):
+            derived_name = (
+                llm_answer_final["AARETE_DERIVED_PAYER_NAME"]
+                .replace("```", "")
+                .replace("**", "")
+                .strip()
+                .upper()
+            )
+            return derived_name
+        else:
+            # If LLM returns N/A or empty, fallback to the programmatic base_payer_name
+            fallback = cluster_rows[0].get("base_payer_name", "UNKNOWN")
+            return str(fallback).upper()
+
+    except Exception as e:
+        logging.error(
+            f"Error parsing LLM response for Aarete Derived Payer Name: {str(e)}"
+        )
+        return str(cluster_rows[0].get("base_payer_name", "UNKNOWN")).upper()

src/pipelines/shared/extraction/one_to_one_funcs.py

@@ -2,6 +2,10 @@ import logging
 import os
 import re
 import json
+import numpy as np
+from typing import Any
+from rapidfuzz import fuzz
+from collections import defaultdict
 
 import pandas as pd
 import src.pipelines.shared.postprocessing.postprocessing_funcs as postprocessing_funcs
@@ -395,3 +399,306 @@ def add_signature_count(one_to_one_results: dict, contract_txt: str) -> dict:
 
     one_to_one_results["NUM_SIGNED_SIGNATORY_LINE_COUNT"] = total_signatures
     return one_to_one_results
+
+
+def add_aarete_derived_payer_name(df: pd.DataFrame, state_flag: bool) -> pd.DataFrame:
+    """
+    Derive canonical payer names from PAYER_NAME column.
+
+    This function clusters similar payer names, standardizes them, and selects
+    the most appropriate canonical name for each cluster.
+
+    Args:
+        df: Input DataFrame containing a PAYER_NAME column.
+        state_flag: Controls whether payer names are treated as state-specific.
+            Always True (confirmed with business team).
+            True  → Preserve state distinctions.
+                    Example: "Molina Healthcare of Texas" and
+                            "Molina Healthcare of Utah" remain separate.
+            False → Collapse regional state suffixes into national brand.
+                    Example: "Molina Healthcare of Texas" and
+                            "Molina Healthcare of Utah" both become "Molina Healthcare".
+                    Leading state-based brand names (e.g., "Oklahoma Complete Health")
+                    are preserved.
+
+    Returns:
+        DataFrame with added AARETE_DERIVED_PAYER_NAME column.
+    """
+
+    if df is None or df.empty:
+        logging.warning("Input DataFrame is empty. Returning empty DataFrame.")
+        return df
+
+    df_payer = df.copy()
+    # Check if PAYER_NAME column exists in the DataFrame, if not add an empty AARETE_DERIVED_PAYER_NAME column and return the DataFrame
+    if "PAYER_NAME" not in df_payer.columns:
+        logging.warning(
+            "PAYER_NAME column not found in DataFrame. skipping AARETE_DERIVED_PAYER_NAME column."
+        )
+        return df_payer
+    # 1. Extract unique payer names
+    payer_names = df_payer["PAYER_NAME"].dropna().unique().tolist()
+    unique_payer_names = [
+        payer for payer in payer_names if not string_utils.is_empty(payer)
+    ]
+    if not unique_payer_names:
+        df_payer["AARETE_DERIVED_PAYER_NAME"] = ""
+        return df_payer
+
+    # 2. Standardize OR normalizing unique payer names (list of dicts)
+    standardized_unique_payer_name_rows = standardize_unique_names(
+        unique_payer_names, state_flag
+    )
+
+    standardized_unique_payer_name_rows = propagate_dba_to_legal(
+        standardized_unique_payer_name_rows
+    )
+
+    # Combining Step`s 3-4: Build similarity matrix and cluster with blocking to handle large volumes efficiently
+    clusters = cluster_with_blocking(
+        standardized_unique_payer_name_rows,
+        state_flag=state_flag,
+        threshold=90,
+    )
+
+    # 5. Build mapping using LLM placeholder
+    mapping = {}
+    for cluster in clusters:
+        derived_payer_name = prompt_calls.prompt_aarete_derived_payer_name(
+            cluster, state_flag
+        )
+
+        for record in cluster:
+            mapping[record["original_payer_name"]] = derived_payer_name
+
+    df_payer["AARETE_DERIVED_PAYER_NAME"] = df_payer["PAYER_NAME"].map(mapping)
+
+    return df_payer
+
+
+def standardize_unique_names(
+    unique_names: list[str], state_flag: bool
+) -> list[dict[str, Any]]:
+    """
+    Normalize payer names and enrich them with DBA and token signatures.
+    Runs only on unique payer names for efficiency.
+
+    Block key is derived from the FIRST token of the base payer name
+    (not from sorted token signature) to preserve brand identity.
+    """
+    rows = []
+
+    for name in unique_names:
+        clean = string_utils.normalize_text(name)
+
+        state = string_utils.detect_state(clean)
+
+        legal, dba = string_utils.extract_dba_payer_name(clean)
+        base = dba if dba else clean
+        legal = legal if legal else clean
+
+        base = string_utils.remove_legal_suffixes(base)
+        legal = string_utils.remove_legal_suffixes(legal)
+
+        if not state_flag:
+            base = string_utils.remove_states(base)
+            legal = string_utils.remove_states(legal)
+
+        signature = string_utils.token_signature(base)
+        tokens = base.split()
+        block_key = tokens[0] if tokens else ""
+
+        rows.append(
+            {
+                "original_payer_name": name,
+                "base_payer_name": base,
+                "signature": signature if signature else (),
+                "has_dba_payer_name": dba is not None,
+                "dba_payer_name": dba,
+                "legal_payer_name": legal,
+                "state": state,
+                "block_key": block_key,
+            }
+        )
+
+    return rows
+
+
+def build_similarity_matrix(rows: list[dict]) -> np.ndarray:
+    """
+    Compute a pairwise similarity matrix for base payer names.
+
+    This function calculates fuzzy string similarity scores between
+    all pairs of 'base_payer_name' values in the input rows using
+    multiple fuzzy matching strategies. The highest score is used
+    as the final similarity value.
+
+    Args:
+        rows (list[dict]): A list of dictionaries, each containing
+            a 'base_payer_name' key.
+
+    Returns:
+        np.ndarray: A square (n x n) matrix of similarity scores
+        ranging from 0 to 100.
+    """
+    # Number of payer name records
+    n = len(rows)
+
+    # Initialize an empty similarity matrix
+    sim_matrix = np.zeros((n, n))
+
+    # Compute pairwise similarities
+    for i in range(n):
+        for j in range(i, n):
+            if i == j:
+                # A string is always 100% similar to itself
+                sim_matrix[i, j] = 100
+            else:
+                # Compute similarity using multiple fuzzy matching methods and different combinations
+                score = max(
+                    fuzz.ratio(rows[i]["base_payer_name"], rows[j]["base_payer_name"]),
+                    fuzz.token_set_ratio(
+                        rows[i]["base_payer_name"], rows[j]["base_payer_name"]
+                    ),
+                    fuzz.ratio(rows[i]["legal_payer_name"], rows[j]["base_payer_name"]),
+                    fuzz.token_set_ratio(
+                        rows[i]["legal_payer_name"], rows[j]["base_payer_name"]
+                    ),
+                    fuzz.ratio(rows[i]["base_payer_name"], rows[j]["legal_payer_name"]),
+                    fuzz.token_set_ratio(
+                        rows[i]["base_payer_name"], rows[j]["legal_payer_name"]
+                    ),
+                )
+
+                # Fill both symmetric positions in the matrix
+                sim_matrix[i, j] = score
+                sim_matrix[j, i] = score
+
+    return sim_matrix
+
+
+def cluster_from_similarity(
+    rows: list[dict[str, Any]],
+    sim_matrix: np.ndarray,
+    state_flag: bool,
+    threshold: int = 90,
+) -> list[list[dict[str, Any]]]:
+    """
+    Cluster payer records using a similarity matrix and threshold.
+
+    Each row is grouped with other rows whose similarity score
+    meets or exceeds the given threshold. Once a row is assigned
+    to a cluster, it is not reconsidered.
+
+    Args:
+        rows (list[dict]): List of row dictionaries containing payer data.
+        sim_matrix (np.ndarray): Pairwise similarity matrix (n x n).
+        threshold (int): Minimum similarity score required to cluster.
+
+    Returns:
+        list[list[dict]]: A list of clusters, where each cluster
+        is a list of row dictionaries.
+    """
+    n = len(rows)
+    visited: set[int] = set()
+    clusters: list[list[dict[str, Any]]] = []
+
+    # Iterate through each record
+    for i in range(n):
+        # Skip rows that have already been clustered
+        if i in visited:
+            continue
+
+        # Start a new cluster with the current index
+        cluster_indices = {i}
+
+        # Add all rows similar enough to the current one
+        for j in range(n):
+            if sim_matrix[i, j] >= threshold:
+
+                # --- STATE LOGIC ---
+                if state_flag:
+                    state_i = rows[i]["state"]
+                    state_j = rows[j]["state"]
+
+                    # If one has state and the other doesn't → do NOT cluster
+                    if state_i != state_j:
+                        continue
+                cluster_indices.add(j)
+
+        # Mark clustered indices as visited
+        visited.update(cluster_indices)
+
+        # Build the cluster from the original rows
+        clusters.append([rows[idx] for idx in cluster_indices])
+
+    return clusters
+
+
+def cluster_with_blocking(
+    rows: list[dict[str, Any]],
+    state_flag: bool,
+    threshold: int = 90,
+) -> list[list[dict[str, Any]]]:
+    """
+    Cluster payer names using first-token blocking to scale to large volumes.
+    """
+
+    # 1. Group rows by block key
+    blocks = defaultdict(list)
+    for row in rows:
+        blocks[row["block_key"]].append(row)
+
+    all_clusters = []
+
+    # 2. Process each block independently
+    for block_rows in blocks.values():
+
+        # Small blocks don't need matrix
+        if len(block_rows) == 1:
+            all_clusters.append(block_rows)
+            continue
+
+        sim_matrix = build_similarity_matrix(block_rows)
+
+        block_clusters = cluster_from_similarity(
+            block_rows,
+            sim_matrix,
+            state_flag=state_flag,
+            threshold=threshold,
+        )
+
+        all_clusters.extend(block_clusters)
+
+    return all_clusters
+
+
+def propagate_dba_to_legal(rows: list[dict]) -> list[dict]:
+    """
+    If a legal entity has an associated DBA in any row,
+    propagate that DBA to all rows sharing the same legal entity.
+    """
+
+    legal_to_dba = {}
+
+    # First pass: build mapping
+    for row in rows:
+        legal = row.get("legal_payer_name")
+        dba = row.get("dba_payer_name")
+
+        if legal and dba:
+            legal_to_dba[legal] = dba
+
+    # Second pass: propagate
+    for row in rows:
+        legal = row.get("legal_payer_name")
+
+        if legal in legal_to_dba:
+            new_base = legal_to_dba[legal]
+            row["base_payer_name"] = new_base
+
+            # Update block key
+            tokens = new_base.split()
+            row["block_key"] = tokens[0] if tokens else ""
+
+    return rows

src/prompts/investment_prompts.json

@@ -240,7 +240,7 @@
         "field_name": "PAYER_NAME",
         "relationship": "one_to_one",
         "field_type": "smart_chunked",
-        "prompt": "What is the name of the payer that is a party to the contract as stated in the Preamble?",
+        "prompt": "What is the name of the payer that is a party to the contract as stated in the Preamble? Extract the name exactly as it appears in the contract.",
         "retrieval_question": "What is the name of the payer, health plan, insurance company, or payor that is a party to this contract? Payer name in the preamble, parties section, or beginning of agreement."
     },
     {

src/prompts/prompt_templates.py

@@ -2378,3 +2378,61 @@ Reimbursement: {reimbursement_dict["REIMB_TERM"]}
 Return the integer of the {special_case_term} that is most associated with the REIMBURSEMENT TERM."""
 
     return (prompt, _json_list_parser)
+
+
+def AARETE_DERIVED_PAYER_NAME_INSTRUCTION():
+    return f"""You are an expert in healthcare entity data normalization.
+    Your goal is to take a list of similar payer names and determine the single,
+    most accurate canonical legal or brand name.
+
+    Given variants string contains variations of a single Payer (insurance company) name
+    extracted from different contracts. They all refer to the same entity.
+
+    TASK:
+    1. Select or derive the most professional canonical name.
+    2. If given STATE_FLAG is FALSE:
+        - Collapse regional entities into the national brand (e.g., 'Molina Healthcare').
+        - DO NOT remove the word 'State' if it is part of a proper noun (e.g., 'New York State Catholic Health Plan').
+    3. If given STATE_FLAG is TRUE:
+        - Maintain state distinctions (e.g., 'Molina Healthcare of Texas').
+    4. Prefer the 'DBA' (Doing Business As) name if it is present.
+    5. Cleaned name should be preferred over original name, but use your judgment to determine if the cleaned name is actually more accurate or if it has removed important context.
+    6. Remove excessive noise like "a corporation of...", "an Illinois company", or "d/b/a" prefixes.
+    7. Maintain essential legal identifiers only if they are part of the standard brand identity.
+    8. Return only the cleaned canonical name without any explanation.
+
+    Example Variants Input:
+    VARIANTS:
+    - AvMed Health Plan, Inc
+    - AvMED, INC., d/b/a AvMED HEALTH PLANS
+    Example Output:
+    {{"AARETE_DERIVED_PAYER_NAME":"AvMed Health Plans"}}
+
+    [OUTPUT FORMAT]
+    Return ONLY the JSON dictionary with no explanatory text. Use "AARETE_DERIVED_PAYER_NAME" as the key. 
+    {JSON_DICT_FORMAT_INSTRUCTIONS}
+    """
+
+
+def AARETE_DERIVED_PAYER_NAME(
+    cluster_rows: list[dict], state_flag: bool
+) -> Tuple[str, Callable[[str], dict]]:
+    """
+    Constructs the prompt using both original and cleaned names for maximum context.
+    """
+    variant_list = []
+    for r in cluster_rows:
+        # Using your preferred logic to show the LLM both states of the data
+        variant = (
+            f"- Original: {r['original_payer_name']} (Cleaned: {r['base_payer_name']})"
+        )
+        variant_list.append(variant)
+
+    variants_string = "\n".join(variant_list)
+    prompt = f"""
+    VARIANTS:
+    {variants_string}
+    STATE_FLAG:
+    {state_flag}
+    """
+    return (prompt, _json_dict_parser)

src/tests/test_derived_payer_name.py

@@ -0,0 +1,206 @@
+import unittest
+import pandas as pd
+import numpy as np
+from unittest.mock import patch
+
+from src.utils.string_utils import (
+    normalize_text,
+    extract_dba_payer_name,
+    remove_legal_suffixes,
+    remove_states,
+    token_signature,
+)
+from src.pipelines.shared.extraction.one_to_one_funcs import (
+    standardize_unique_names,
+    add_aarete_derived_payer_name,
+    cluster_from_similarity,
+    build_similarity_matrix,
+    cluster_with_blocking,
+    propagate_dba_to_legal,
+)
+
+
+class TestPayerNameTextUtilities(unittest.TestCase):
+
+    def test_normalize_payer_name_text(self):
+        text = "Molina Healthcare, Inc."
+        result = normalize_text(text)
+        self.assertEqual(result, "molina healthcare inc")
+
+    def test_extract_dba_payer_name_from_text(self):
+        text = "ABC Corp d/b/a XYZ Health"
+        normalized = normalize_text(text)
+        result = extract_dba_payer_name(normalized)
+        self.assertEqual(result, ("abc corp", "xyz health"))
+
+    def test_remove_legal_suffixes_from_payer_name(self):
+        text = "molina healthcare inc"
+        result = remove_legal_suffixes(text)
+        self.assertEqual(result, "molina healthcare")
+
+    def test_generate_token_signature_for_payer_name(self):
+        text = "molina healthcare of utah"
+        result = token_signature(text)
+        self.assertEqual(result, ("healthcare", "molina", "utah"))
+
+
+class TestPayerNameStandardization(unittest.TestCase):
+
+    def test_standardize_unique_payer_names(self):
+        names = ["Molina Healthcare of Utah, Inc."]
+        rows = standardize_unique_names(names, state_flag=True)
+
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0]["original_payer_name"], names[0])
+        self.assertIn("base_payer_name", rows[0])
+        self.assertIn("signature", rows[0])
+        self.assertIn("block_key", rows[0])
+        self.assertFalse(rows[0]["has_dba_payer_name"])
+
+
+class TestRemoveStatesFunction(unittest.TestCase):
+
+    def test_remove_trailing_state_suffix(self):
+        self.assertEqual(
+            remove_states("community health choice texas"), "community health choice"
+        )
+
+        self.assertEqual(remove_states("fidelis care new york"), "fidelis care")
+
+    def test_preserve_leading_state_brand(self):
+        self.assertEqual(
+            remove_states("oklahoma complete health"), "oklahoma complete health"
+        )
+
+        self.assertEqual(
+            remove_states("texas childrens health plan"), "texas childrens health plan"
+        )
+
+    def test_remove_of_state_pattern(self):
+        self.assertEqual(
+            remove_states("molina healthcare of utah"), "molina healthcare of"
+        )
+
+        self.assertEqual(
+            remove_states("molina healthcare of texas"), "molina healthcare of"
+        )
+
+    def test_no_state_present(self):
+        self.assertEqual(remove_states("centene corporation"), "centene corporation")
+
+
+class TestPropagateDbaToLegal(unittest.TestCase):
+
+    def test_propagate_dba_to_matching_legal_entity(self):
+        rows = [
+            {
+                "original_payer_name": "NY State Catholic Health Plan dba Fidelis Care New York",
+                "base_payer_name": "fidelis care new york",
+                "legal_payer_name": "new york state catholic health plan",
+                "dba_payer_name": "fidelis care new york",
+                "block_key": "fidelis",
+            },
+            {
+                "original_payer_name": "New York State Catholic Health Plan",
+                "base_payer_name": "new york state catholic health plan",
+                "legal_payer_name": "new york state catholic health plan",
+                "dba_payer_name": None,
+                "block_key": "new",
+            },
+        ]
+
+        updated = propagate_dba_to_legal(rows)
+
+        # Both should now share same base_payer_name
+        self.assertEqual(updated[0]["base_payer_name"], "fidelis care new york")
+        self.assertEqual(updated[1]["base_payer_name"], "fidelis care new york")
+
+        # Block key should also update
+        self.assertEqual(updated[1]["block_key"], "fidelis")
+
+    def test_no_propagation_when_no_dba_present(self):
+        rows = [
+            {
+                "original_payer_name": "ODS Health Plan",
+                "base_payer_name": "ods health plan",
+                "legal_payer_name": "ods health plan",
+                "dba_payer_name": None,
+                "block_key": "ods",
+            },
+            {
+                "original_payer_name": "Moda Health Plan",
+                "base_payer_name": "moda health plan",
+                "legal_payer_name": "moda health plan",
+                "dba_payer_name": None,
+                "block_key": "moda",
+            },
+        ]
+
+        updated = propagate_dba_to_legal(rows)
+
+        # Should remain unchanged
+        self.assertEqual(updated[0]["base_payer_name"], "ods health plan")
+        self.assertEqual(updated[1]["base_payer_name"], "moda health plan")
+
+
+class TestPayerNameClustering(unittest.TestCase):
+
+    def test_cluster_payer_names_with_state_flag_true(self):
+        names = ["Molina Healthcare of Utah", "Molina Healthcare"]
+
+        rows = standardize_unique_names(names, state_flag=True)
+
+        clusters = cluster_with_blocking(rows, state_flag=True, threshold=85)
+
+        # Should remain separate clusters
+        self.assertEqual(len(clusters), 2)
+
+    def test_cluster_payer_names_with_state_flag_false(self):
+        names = ["Molina Healthcare of Utah", "Molina Healthcare"]
+
+        rows = standardize_unique_names(names, state_flag=False)
+
+        clusters = cluster_with_blocking(rows, state_flag=False, threshold=85)
+
+        # Should cluster together
+        self.assertEqual(len(clusters), 1)
+
+
+class TestPayerNameEndToEndPipeline(unittest.TestCase):
+
+    @patch("src.pipelines.saas.prompts.prompt_calls.prompt_aarete_derived_payer_name")
+    def test_derive_payer_names_end_to_end(self, mock_llm):
+
+        data = {
+            "PAYER_NAME": [
+                "NEW YORK STATE CATHOLIC HEALTH PLAN, INC. dba FIDELIS CARE NEW YORK",
+                "NEW YORK STATE CATHOLIC HEALTH PLAN, INC.",
+                "NEW YORK STATE CATHOLIC HEALTH PLAN, INC. d/b/a FIDELIS CARE NEW YORK",
+            ]
+        }
+
+        # Mock LLM to return canonical name
+        mock_llm.return_value = "FIDELIS CARE NEW YORK"
+
+        df = pd.DataFrame(data)
+
+        result = add_aarete_derived_payer_name(df, state_flag=True)
+
+        self.assertIn("AARETE_DERIVED_PAYER_NAME", result.columns)
+
+        # All rows should map to same canonical value
+        self.assertEqual(
+            result.loc[0, "AARETE_DERIVED_PAYER_NAME"], "FIDELIS CARE NEW YORK"
+        )
+
+        self.assertEqual(
+            result.loc[1, "AARETE_DERIVED_PAYER_NAME"], "FIDELIS CARE NEW YORK"
+        )
+
+        self.assertEqual(
+            result.loc[2, "AARETE_DERIVED_PAYER_NAME"], "FIDELIS CARE NEW YORK"
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

src/utils/string_utils.py

@@ -14,8 +14,79 @@ from src.constants.regex_patterns import (
     BACKTICK_PATTERN,
     PIPE_PATTERN,
     TRIPLE_BACKTICK_PATTERN,
+    DBA_PATTERNS,
 )
 
+LEGAL_SUFFIXES = {
+    "inc",
+    "inc.",
+    "incorporated",
+    "corp",
+    "corporation",
+    "llc",
+    "l.l.c",
+    "ltd",
+    "limited",
+    "co",
+    "company",
+    "affiliates",
+}
+
+STOPWORDS = {"of", "the", "and", "its"}
+
+US_STATES = [
+    "alabama",
+    "alaska",
+    "arizona",
+    "arkansas",
+    "california",
+    "colorado",
+    "connecticut",
+    "delaware",
+    "florida",
+    "georgia",
+    "hawaii",
+    "idaho",
+    "illinois",
+    "indiana",
+    "iowa",
+    "kansas",
+    "kentucky",
+    "louisiana",
+    "maine",
+    "maryland",
+    "massachusetts",
+    "michigan",
+    "minnesota",
+    "mississippi",
+    "missouri",
+    "montana",
+    "nebraska",
+    "nevada",
+    "new hampshire",
+    "new jersey",
+    "new mexico",
+    "new york",
+    "north carolina",
+    "north dakota",
+    "ohio",
+    "oklahoma",
+    "oregon",
+    "pennsylvania",
+    "rhode island",
+    "south carolina",
+    "south dakota",
+    "tennessee",
+    "texas",
+    "utah",
+    "vermont",
+    "virginia",
+    "washington",
+    "west virginia",
+    "wisconsin",
+    "wyoming",
+]
+
 
 def extract_text_from_delimiters(
     raw_output: str, delimiter: Delimiter, match_index: int = -1
@@ -806,3 +877,167 @@ def normalize_to_json_list(val):
         result = json.dumps(expanded)
         return result
     return val
+
+
+def normalize_text(text: str) -> str:
+    """
+    Normalize a text string by applying common preprocessing steps.
+
+    This function:
+    - Converts all characters to lowercase
+    - Removes punctuation and special characters
+    - Replaces multiple whitespace characters with a single space
+    - Strips leading and trailing whitespace
+
+    Args:
+        text (str): The input text to normalize.
+
+    Returns:
+        str: The normalized text.
+    """
+    # Convert the text to lowercase for case-insensitive processing
+    text = text.lower()
+
+    # Replace all characters that are not letters, numbers, or whitespace with a space
+    text = re.sub(r"[^\w\s]", " ", text)
+
+    # Replace multiple whitespace characters with a single space
+    # and remove leading/trailing spaces
+    text = re.sub(r"\s+", " ", text).strip()
+
+    return text
+
+
+def remove_legal_suffixes(text: str) -> str:
+    """
+    Remove legal entity suffixes from a text string.
+
+    This function splits the input text into tokens (words) and
+    removes any tokens that match known legal suffixes
+    (e.g., "llc", "inc", "corp").
+
+    Args:
+        text (str): The input text containing a business name.
+
+    Returns:
+        str: The text with legal suffixes removed.
+    """
+    # Split the text into individual words
+    tokens = text.split()
+
+    # Filter out any tokens that are legal suffixes
+    tokens = [t for t in tokens if t not in LEGAL_SUFFIXES]
+
+    # Rejoin the remaining tokens into a single string
+    return " ".join(tokens)
+
+
+def extract_dba_payer_name(text: str) -> tuple[str | None, str | None]:
+    """
+    Extract a DBA (Doing Business As) name from a text string.
+
+    The function checks the input text against a list of predefined
+    DBA patterns (e.g., "dba", "doing business as"). If a pattern is found,
+    it returns the text that appears after the pattern.
+
+    Ex:
+    Input: NEW YORK STATE CATHOLIC HEALTH PLAN, INC. d/b/a FIDELIS CARE NEW YORK
+
+    Output: FIDELIS CARE NEW YORK
+
+    Args:
+        text (str): The input text that may contain a DBA reference.
+
+    Returns:
+        tuple[str | None, str | None]: A tuple containing:
+            - The text before the DBA pattern (potentially the legal name)
+            - The text after the DBA pattern (the extracted DBA name)
+            If no DBA pattern is found, both elements of the tuple will be None.
+    """
+    # Iterate through all known DBA patterns
+    for pattern in DBA_PATTERNS:
+        # Split the text using the current pattern (case-insensitive)
+        match = re.split(pattern, text, flags=re.I)
+
+        # If the pattern was found, the split will produce more than one part
+        if len(match) > 1:
+            # Return the text after the matched pattern, trimmed of whitespace
+            return match[0].strip(), match[-1].strip()
+
+    # Return None if no DBA pattern is found
+    return None, None
+
+
+def remove_states(text: str) -> str:
+    """
+    Remove U.S. state names or abbreviations from a text string.
+
+    The function iterates through a list of U.S. states and removes
+    any exact (word-boundary matched) occurrences from the input text.
+    Extra whitespace created by removals is cleaned up before returning.
+
+    Remove state names when they appear as a regional suffix
+    (i.e., at the end of the string), but preserve them when
+    they are leading brand identifiers (e.g., 'Oklahoma Complete Health').
+
+    Args:
+        text (str): The input text that may contain U.S. state names.
+
+    Returns:
+        str: The text with U.S. state references removed.
+    """
+
+    tokens = text.split()
+    if not tokens:
+        return text
+
+    for state in US_STATES:
+        state_tokens = state.split()
+
+        # Check if state appears at END of string
+        if tokens[-len(state_tokens) :] == state_tokens:
+
+            # Do NOT remove if state is the first token
+            if tokens[0] == state_tokens[0]:
+                return text
+
+            # Remove trailing state tokens
+            new_tokens = tokens[: -len(state_tokens)]
+            return " ".join(new_tokens).strip()
+
+    # Also handle "of <state>" pattern anywhere in the string, as it often indicates a regional suffixe.g., "Health Plan of New York" should become "Health Plan"
+    # Remove each state using a case-insensitive regex match. It looks for the word "of" followed by the state name.
+    for state in US_STATES:
+        text = re.sub(rf"\bof\s+{state}\b", "", text, flags=re.I)
+
+    # Normalize whitespace after removals
+    return re.sub(r"\s+", " ", text).strip()
+
+
+def detect_state(text: str) -> str | None:
+    """Return state name if found in text."""
+    for state in US_STATES:
+        if re.search(rf"\b{state}\b", text, re.I):
+            return state
+    return None
+
+
+def token_signature(text: str) -> tuple:
+    """
+    Generate a normalized token signature from a text string.
+
+    The function removes stopwords, sorts the remaining tokens,
+    and returns them as an immutable tuple. This signature can be
+    used for fast comparison or deduplication of similar texts.
+
+    Args:
+        text (str): The input text to tokenize.
+
+    Returns:
+        tuple: A sorted tuple of tokens excluding stopwords.
+    """
+    # Split the text into tokens and remove stopwords
+    tokens = [t for t in text.split() if t not in STOPWORDS]
+
+    # Sort tokens to ensure order-independent comparison
+    return tuple(sorted(tokens))