Merged in feature/standardized-services (pull request #958)

Feature/standardized services

* service term standardization

* prompt update

* prompt updates for standardization

* only service standardization

* new file

* add supporting files and test scripts for standardization work

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* remove old files

* Merge remote-tracking branch 'origin/dev' into feature/standardized-services

* final fixes

* Merged dev into feature/standardized-services

* additional features

* removed  unwanted files

* remove unwanted files

* Merge branch 'dev' into feature/standardized-services

* Merge remote-tracking branch 'origin/dev' into feature/standardized-services

* reversed  vendor changes

* Merged dev into feature/standardized-services

* addressed PR comments

* Merge branch 'dev' into feature/standardized-services

* Merged dev into feature/standardized-services

* addressed 3 remaining comments inPR

* black formatting

* incorporated feedback from AI code reviewer

* reused existing  functionality


Approved-by: Katon Minhas

documentation/SERVICE_TERM_STANDARDIZATION.md

@@ -0,0 +1,132 @@
+# Service Term Standardization
+
+## Overview
+
+The service term standardization feature ensures that raw, inconsistently-named `SERVICE_TERM` values ingested from payer PC pipeline outputs are normalized into a canonical form stored in a new column: `AARETE_DERIVED_SERVICE_TERM`. This improves downstream reporting accuracy, cross-payer comparability, and analytics consistency.
+
+---
+
+## Problem Statement
+
+Payer PC output files contain a `SERVICE_TERM` column populated by source systems with no enforced vocabulary. The same concept (e.g., "Medical/Surgical") may appear as dozens of variants: mixed case, abbreviations, typos, or legacy codes. Without standardization, grouping and aggregating by service term produces fragmented, unreliable results.
+
+---
+
+## Solution
+
+A post-processing step (`standardize_service_terms`) is applied after the PC pipeline runs. It:
+
+1. Reads the raw `SERVICE_TERM` column from the combined pipeline output.
+2. Optionally reads a `PC_Details` sheet from a PC Excel file to enable **per-group** mapping (different payers or file groups may use different term vocabularies).
+3. Maps each raw term to a canonical `AARETE_DERIVED_SERVICE_TERM` value.
+4. Appends the new column to the DataFrame without modifying the original `SERVICE_TERM`.
+
+The logic lives in:
+
+```
+src/pipelines/shared/postprocessing/service_term_standardization.py
+```
+
+---
+
+## Standalone Script
+
+For manual inspection and retroactive application, a standalone script is provided:
+
+```
+src/scripts/retroactively_standardize_service_term.py
+```
+
+### Usage
+
+```bash
+python src/scripts/retroactively_standardize_service_term.py <input_file> [pc_excel_path]
+```
+
+| Argument | Required | Description |
+|---|---|---|
+| `input_file` | Yes | CSV or Excel file with a `SERVICE_TERM` column |
+| `pc_excel_path` | No | Separate PC Excel file containing a `PC_Details` sheet for per-group mode |
+
+**Auto-detection:** If `input_file` is an Excel file that already contains a `PC_Details` sheet, per-group mode is enabled automatically — no second argument needed.
+
+### Output
+
+Both output files are written to the **same folder as the input file**:
+
+| File | Description |
+|---|---|
+| `<input>_standardized_services.csv/.xlsx` | Full dataset with the new `AARETE_DERIVED_SERVICE_TERM` column appended |
+| `<input>_standardized_services_mapping.csv` | Unique `SERVICE_TERM → AARETE_DERIVED_SERVICE_TERM` mapping table for audit |
+
+The output format (CSV vs. Excel) matches the input file format.
+
+### Console Summary
+
+The script prints a summary to stdout:
+
+```
+======================================================================
+STANDARDIZATION SUMMARY
+======================================================================
+  Total unique mappings : 142
+  Changed               : 89
+  Unchanged (pass-thru) : 53
+
+CHANGED MAPPINGS (first 50):
+  RAW SERVICE_TERM                                         →  AARETE_DERIVED_SERVICE_TERM
+  ...
+
+UNCHANGED TERMS (first 20):
+  ...
+```
+
+---
+
+## Modes of Operation
+
+| Mode | When it applies | Behavior |
+|---|---|---|
+| **Global** | No `PC_Details` sheet found | Single mapping applied to all rows |
+| **Per-group** | `PC_Details` sheet present (auto-detected or explicitly provided) | Mapping scoped per file group / payer using `FILE_NAME` and `grouping_key` from `PC_Details` |
+
+---
+
+## Column Reference
+
+| Column | Source | Description |
+|---|---|---|
+| `SERVICE_TERM` | Raw payer data | Original term as received from the payer |
+| `AARETE_DERIVED_SERVICE_TERM` | Post-processing | Canonical standardized term added by this feature |
+
+---
+
+## Running Standalone (Quickstart)
+
+The script can be invoked from **any working directory** — it automatically adds the repo root to `sys.path`:
+
+```bash
+# From repo root
+python src/scripts/retroactively_standardize_service_term.py \
+    /path/to/payer_output.xlsx
+
+# With a separate PC Excel for grouping
+python src/scripts/retroactively_standardize_service_term.py \
+    /path/to/combined_output.csv \
+    /path/to/pc_output.xlsx
+```
+
+Outputs will appear alongside the input file:
+```
+/path/to/payer_output_standardized_services.xlsx
+/path/to/payer_output_standardized_services_mapping.csv
+```
+
+---
+
+## Related Files
+
+| Path | Purpose |
+|---|---|
+| `src/pipelines/shared/postprocessing/service_term_standardization.py` | Core standardization logic |
+| `src/scripts/retroactively_standardize_service_term.py` | Standalone inspection / retroactive application script |

src/constants/investment_columns.py

@@ -105,6 +105,7 @@ FIELD_FORMAT_MAPPING = {
     "FEE_SCHEDULE_VERSION": "str",
     "AARETE_DERIVED_FEE_SCHEDULE_VERSION": "str",
     "SERVICE_TERM": "str",
+    "AARETE_DERIVED_SERVICE_TERM": "str",
     "CPT4_PROC_CD": "list[str]",
     "CPT4_PROC_CD_DESC": "list[str]",
     "CPT4_PROC_MOD": "list[str]",

src/parent_child/__main__.py

@@ -83,7 +83,7 @@ def main(
     doczy_output_for_pc: str,
     client: Optional[str] = None,
     output_dir: Optional[str] = None,
-) -> tuple[int, str]:
+) -> tuple[int, str, pd.DataFrame]:
     """
     Main entry point for parent-child mapping.
 
@@ -97,7 +97,7 @@ def main(
                 When None (standalone run), uses outputs/parent_child/{run_timestamp}.
 
     Returns:
-        Tuple of (row_count, local_path) where local_path is the written Excel file.
+        Tuple of (row_count, local_path, pc_df) where local_path is the written Excel file.
     """
     # Extract client name from input file for output naming
     client_name = extract_client_name(doczy_output_for_pc)
@@ -154,13 +154,13 @@ def main(
             original_df_with_pc, pc_df = parent_child_mapping(cleaned_df, original_df)
 
         # Save detailed results
-        pc_df1 = qc_main(pc_df, client)
+        pc_df_qc = qc_main(pc_df, client)
 
         output_filename = f"{client_name}_generic_hierarchy_lineage_output.xlsx"
-        local_path = write_pc_excel(df_read, pc_df1, output_dir, client_name)
+        local_path = write_pc_excel(df_read, pc_df_qc, output_dir, client_name)
 
         row_count = pc_df.shape[0] if pc_df is not None else mapped_df.shape[0]
-        return row_count, local_path
+        return row_count, local_path, pc_df_qc
     except Exception as e:
         logging.error(f"Parent child mapping failed due to {e}", exc_info=True)
         raise
@@ -174,7 +174,7 @@ if __name__ == "__main__":
     # Generate timestamp for this run
     run_timestamp = f"run_{datetime.now().strftime('%Y%m%d_%H-%M')}_{config.BATCH_ID}"
     if config.DOCZY_OUTPUT_FOR_PC:
-        row_count, _ = main(config.DOCZY_OUTPUT_FOR_PC)
+        row_count, _, _pc_df = main(config.DOCZY_OUTPUT_FOR_PC)
         logging.info(f"PC Mapping complete. Processed {row_count} files.")
     else:
         logging.warning("No input file specified in config.DOCZY_OUTPUT_FOR_PC")

src/parent_child/column_mapper.py

@@ -236,6 +236,16 @@ class ColumnMapper:
             DataFrame with standardized column names
         """
         reverse_mapping = {v: k for k, v in self.column_mapping.items()}
+        # If renaming column A → B but B already exists as a different column,
+        # drop the pre-existing B first. The mapper has already declared A as the
+        # canonical source for standard name B, so the stale B column is superseded.
+        cols_to_drop = [
+            dst
+            for src, dst in reverse_mapping.items()
+            if src != dst and dst in df.columns
+        ]
+        if cols_to_drop:
+            df = df.drop(columns=cols_to_drop)
         return df.rename(columns=reverse_mapping)
 
     def has_column(self, standard_name: str) -> bool:

src/pipelines/runner.py

@@ -42,6 +42,9 @@ from src.constants.constants import Constants
 from src import config
 from src.core.registry import get_registry
 from src.parent_child import main as parent_child_main
+from src.pipelines.shared.postprocessing.service_term_standardization import (
+    standardize_service_terms,
+)
 from src.document_classification import main as dtc_main
 from src.qc_qa.pipeline import (
     generate_statistics,
@@ -468,7 +471,7 @@ def main(client: str = "saas", testing=False, test_params={}):
                         run_timestamp,
                         "parent-child",
                     )
-                    _, pc_local_path = parent_child_main.main(
+                    _, pc_local_path, pc_details_df = parent_child_main.main(
                         doczy_output_for_pc,
                         client=client,
                         output_dir=pc_output_dir,
@@ -478,6 +481,29 @@ def main(client: str = "saas", testing=False, test_params={}):
                             pc_local_path, run_timestamp, "parent-child"
                         )
 
+                # Standardize SERVICE_TERM per provider group (SaaS/client pipelines only).
+                if not registry.is_vendor(client):
+                    logging.info("=" * 80)
+                    logging.info("Standardizing Service Terms per provider group...")
+                    logging.info("=" * 80)
+                    FINAL_RESULT_DF_CC = standardize_service_terms(
+                        FINAL_RESULT_DF_CC, pc_local_path
+                    )
+                    if "AARETE_DERIVED_SERVICE_TERM" not in FINAL_RESULT_DF_CC.columns:
+                        logging.warning(
+                            "Service term standardization failed; AARETE_DERIVED_SERVICE_TERM column not present"
+                        )
+                    else:
+                        logging.info("Service Term standardization complete.")
+                if config.WRITE_TO_S3:
+                    io_utils.write_s3(
+                        FINAL_RESULT_DF_CC, "", run_timestamp, "cc_results_full"
+                    )
+                else:
+                    io_utils.write_local(
+                        FINAL_RESULT_DF_CC, "", run_timestamp, "cc_results_full"
+                    )
+
         if not per_file_usage_df.empty and not batch_summary_df.empty:
             logging.info("Exporting usage and cost tracking data locally...")
             io_utils.write_local(per_file_usage_df, "", run_timestamp, "usage")

src/pipelines/saas/main.py

@@ -17,6 +17,79 @@ def _resolve_client() -> str:
     return "saas"
 
 
+from src.parent_child import main as parent_child_main
+from src.document_classification import main as dtc_main
+from src.pipelines.shared.postprocessing.service_term_standardization import (
+    standardize_service_terms,
+)
+from src.qc_qa.pipeline import (
+    generate_statistics,
+    generate_tin_contract_summary,
+    generate_tin_statistics,
+    run_qc_qa_pipeline,
+    save_qc_qa_outputs,
+)
+
+
+def safe_process_file(item, constants, run_timestamp):
+    """Process a single file with comprehensive error handling.
+
+    This function wraps the main file processing logic to ensure that:
+    - Individual file failures don't crash the entire batch
+    - Detailed error information is captured for debugging
+
+    Args:
+        item: Tuple of (filename, contract_text) representing the file to process
+        constants: Constants object containing configuration and processing rules
+        run_timestamp: Timestamp string for output file naming and tracking
+
+    Returns:
+        tuple: Either:
+            - (cc_df, dashboard_df) tuple with extracted field results (success case)
+            - (error_df, error_df) tuple with error details (failure case)
+
+    Note:
+        Returns both CC and dashboard versions from file processing.
+        With FAISS migration, no special resource cleanup is needed.
+    """
+    file_id = (
+        item[0] if isinstance(item, (list, tuple)) and len(item) > 0 else item
+    )  # unpack file_id from item (passed in as a tuple below)
+    try:
+        cc_df, dashboard_df = file_processing.process_file(
+            item, constants, run_timestamp
+        )
+        return cc_df, dashboard_df
+    except (
+        Exception
+    ) as e:  # When there's an issue with the processing inside the future
+        error_type = type(e).__name__
+        error_message = str(e)
+        full_traceback = traceback.format_exc()
+
+        logging.error(f"Error processing file {file_id}")
+        logging.error(f"Error type: {error_type}")
+        logging.error(f"Error Message: {error_message}")
+
+        logging.error(f"Full traceback:\n{full_traceback}")
+
+        error_df = pd.DataFrame(
+            [
+                {
+                    "FILE_NAME": file_id,
+                    "error": error_message.replace(",", ";").replace(
+                        "\n", " "
+                    ),  # Replace problematic characters for CSV compatibility
+                    "error_type": error_type,
+                    "traceback": full_traceback.replace(",", ";").replace(
+                        "\n", " | "
+                    ),  # keep line breaks as separators
+                }
+            ]
+        )  # Return a single-row dataframe so we can still concat it later
+        return error_df, error_df  # Return tuple for consistency
+
+
 def main(testing=False, test_params={}):
     client = _resolve_client()
     return runner.main(client=client, testing=testing, test_params=test_params)

src/pipelines/shared/postprocessing/service_term_standardization.py

@@ -0,0 +1,616 @@
+"""
+SERVICE_TERM Standardization
+
+After all files have been processed and combined, this module standardizes the
+SERVICE_TERM field by mapping each unique value to a canonical service name via LLM.
+The result is stored in a new AARETE_DERIVED_SERVICE_TERM column inserted immediately
+after SERVICE_TERM.
+
+Two-pass design:
+  Pass 1 — Taxonomy derivation
+    All unique terms are sent to the LLM (in chunks when needed) with the sole
+    purpose of producing a stable canonical name list.  No mapping happens here.
+    If chunks are used, each chunk contributes candidate names which are then
+    consolidated into one final taxonomy via a short deduplication call.
+
+  Pass 2 — Mapping
+    Every unique term is mapped to exactly one canonical name from the taxonomy
+    derived in Pass 1.  Mapping is batched so very large vocabularies stay within
+    token limits.
+
+This guarantees the canonical name vocabulary is determined by the full dataset —
+not just the first N terms — before any row is mapped.
+
+Grouping mode (optional):
+  When a pc_excel_path is provided, the PC_Details sheet is read to obtain a
+  FILE_NAME → grouping_key mapping.  The two-pass standardization then runs
+  independently for each grouping_key so that canonical names are derived from and
+  tuned to the specific service mix of each provider group.  Rows with no matching
+  grouping_key are standardized globally as a fallback.
+
+On any LLM failure the column falls back to the original SERVICE_TERM value so
+downstream processing is never blocked.
+"""
+
+import json
+import logging
+
+import pandas as pd
+
+import src.utils.json_utils as json_utils
+import src.utils.llm_utils as llm_utils
+import src.utils.usage_tracking as usage_tracking
+
+# ── Tuneable constants ────────────────────────────────────────────────────────
+
+_MODEL_ID = "haiku_latest"
+_MAX_TOKENS = 8192
+
+# How many terms to send per chunk in Pass 1 (taxonomy derivation)
+_TAXONOMY_CHUNK_SIZE = 400
+
+# How many candidate names to consolidate per LLM call (output token guard)
+_CONSOLIDATION_CHUNK_SIZE = 250
+
+# How many terms to send per batch in Pass 2 (mapping)
+_MAPPING_BATCH_SIZE = 100
+
+# PC_Details constants (mirrors latest_contract_for_service.py)
+_PC_SHEET = "PC_Details"
+_PC_JOIN_KEY = "FILE_NAME"
+_PC_GROUP_COL = "grouping_key"
+
+
+# ── Prompt functions (cacheable INSTRUCTION/PROMPT pattern) ───────────────────
+
+
+def SERVICE_TERM_STANDARDIZATION_INSTRUCTION() -> str:
+    return (
+        "You are a healthcare contract data analyst specializing in normalizing medical "
+        "reimbursement terminology. Your task is to standardize raw SERVICE_TERM strings "
+        "extracted from payer-provider contracts into clean, consistent canonical names "
+        "so that downstream analysis can reliably group and compare services across contracts."
+    )
+
+
+# Pass 1a — derive candidate canonical names from a chunk of terms
+def SERVICE_TERM_TAXONOMY_CHUNK_PROMPT(terms_json: str) -> str:
+    return f"""\
+Below is a sample of unique SERVICE_TERM values from a healthcare payer-provider contract group.
+
+Produce a concise list of CANONICAL standardized service names that cover every service in
+this list.
+
+==============================================================================
+RULE 1 — MEDICAL BILLING CODES (HIGHEST PRIORITY — APPLY BEFORE ALL OTHER RULES)
+==============================================================================
+If a term contains ANY medical billing code — CPT (5-digit number), HCPCS (letter followed
+by 4 digits), NDC, or equivalent — the canonical name MUST contain ONLY the code(s).
+Strip EVERYTHING else: descriptions, drug names, gene names, test names, category labels,
+dashes, prefixes, and suffixes.
+
+Output format — preserve the prefix used in the original term:
+  - If the original term contains the word "CPT", prefix the code(s) with "CPT".
+  - Otherwise (HCPCS codes, bare numeric codes, or category-label phrases), prefix with "HCPCS"
+    for letter-prefixed codes (e.g. G0431, H0001, S3854) and "CPT" for pure 5-digit numbers.
+  - Single code → "CPT XXXXX" or "HCPCS X0000"
+  - Multiple codes in one term → each code gets its own prefix, separated by semicolons.
+
+Recognize codes even when buried in descriptive phrases or prefixed with labels.
+
+  "CPT 81250 - G6PC GENE"                              → "CPT 81250"
+  "CPT 81220 - CFTR GENE COM VARIANTS"                 → "CPT 81220"
+  "CPT S3854 - PROSIGNA BREAST CANCER ASSAY"           → "CPT S3854"
+  "Drug screening confirmation codes 80150, 80152"     → "CPT 80150; CPT 80152"
+  "Drug screening confirmation codes 80150, 80152, 80154, 80182"
+                                                       → "CPT 80150; CPT 80152; CPT 80154; CPT 80182"
+  "CPT Code 99213"                                     → "CPT 99213"
+  "HCPCS Code H0001"                                   → "HCPCS H0001"
+  "Drug screening codes G0431"                         → "HCPCS G0431"
+  "Lab panel code 80053"                               → "CPT 80053"
+  "Genetic test CPT 81401"                             → "CPT 81401"
+
+Do NOT include any description, name, label, or dash after the code number.
+A canonical name that looks like "CPT 81220 - Anything" is WRONG.
+A canonical name that looks like "CPT 81220" is CORRECT.
+
+==============================================================================
+RULES 2–6 apply ONLY to terms that contain NO billing codes
+==============================================================================
+
+2. STRIP VERBOSE DESCRIPTORS — remove generic filler words that add no meaning, such as
+   "Covered", "Eligible", "Allowable", "Services", "Benefit", "Care", "Treatment" when they
+   surround a core service name.  Keep the core name.
+   Example: "Covered Inpatient Hospital Services" → "Inpatient Hospital"
+   Example: "Eligible Physical Therapy Services" → "Physical Therapy"
+
+3. PRESERVE PHASE / RATE SUFFIXES — if a term includes a numbered or lettered phase, tier,
+   or rate qualifier, retain it with a dash separator.  Treat each variant as a distinct name.
+   Example: "Physical Therapy Phase 1" → "Physical Therapy - Phase 1"
+   Example: "Chemotherapy Phase 2" → "Chemotherapy - Phase 2"
+
+4. APPEND PRODUCT LINE & PROVIDER TYPE KEYWORDS AS SUFFIX — detect plan or provider keywords
+   such as HMO, PPO, EPO, POS, Medicare, Medicaid, Behavioral Health, Mental Health,
+   Specialist, PCP, and append them in parentheses after the core name.
+   Example: "Inpatient Hospital Medicare" → "Inpatient Hospital (Medicare)"
+   Example: "PCP Office Visit" → "Office Visit (PCP)"
+
+5. TITLE CASE — all output labels must be in Title Case.  Fix ALL-CAPS raw terms.
+   Example: "INPATIENT HOSPITAL SERVICES" → "Inpatient Hospital"
+
+6. PASS-THROUGH — if a raw term is already clean and standardized (correct casing, no filler
+   words, no code format issues), output it with only minimal adjustments (Title Case, strip
+   trailing "Services" if redundant).
+
+Return ONLY a valid JSON array of canonical service name strings — no explanation, no markdown.
+
+SERVICE_TERMS:
+{terms_json}"""
+
+
+# Pass 1b — consolidate candidate canonical names from multiple chunks into one final list
+def SERVICE_TERM_CONSOLIDATION_PROMPT(categories_json: str) -> str:
+    return f"""\
+The following candidate canonical service names were proposed from different subsets of the
+same contract group.  Many are duplicates or near-duplicates.
+
+Merge and deduplicate into one final, minimal set applying these rules:
+- MEDICAL BILLING CODES TAKE PRIORITY — any candidate that is already in code-only form
+  ("CPT XXXXX", "HCPCS X0000", or a semicolon-separated list of such codes) must be kept
+  exactly as-is.  Never re-attach descriptions, drug names, gene names, or test labels to
+  a code-only entry.  Discard any near-duplicate that contains the same code(s) plus extra text.
+  Example: keep "CPT 81250", discard "CPT 81250 - G6PC Gene".
+  Example: keep "CPT S3854", discard "HCPCS S3854" or "CPT S3854 - Prosigna Assay".
+- Keep the most descriptive but concise form for non-code entries (prefer "Inpatient Hospital"
+  over "Inpatient" or "Inpatient Hospital Services").
+- Keep Phase / Rate variants separate ("Physical Therapy - Phase 1" is distinct from
+  "Physical Therapy - Phase 2").
+- Keep product line / provider type parenthetical suffixes when present and meaningful.
+- All labels must be in Title Case.
+- Drop near-duplicates for non-code entries (e.g., "Laboratory" supersedes "Lab",
+  "Lab Services", "Laboratory Services").
+
+Return ONLY a valid JSON array of the final canonical name strings — no explanation,
+no markdown.
+
+CANDIDATE NAMES:
+{categories_json}"""
+
+
+# Pass 2 — map terms to the established canonical taxonomy
+def SERVICE_TERM_MAPPING_BATCH_PROMPT(categories_json: str, terms_json: str) -> str:
+    return f"""\
+Map each raw SERVICE_TERM below to exactly one canonical name from the established list.
+Do NOT create new canonical names.
+
+Mapping rules:
+1. STRIP FILLER — ignore generic filler words ("Covered", "Eligible", "Allowable",
+   "Services", "Benefit") when matching a raw term to a canonical name.
+2. TITLE CASE — normalize ALL-CAPS raw terms before matching.
+3. PRESERVE PHASE / RATE SUFFIXES — "Phase 1" raw terms must map to the "- Phase 1"
+   canonical variant; "Phase 2" must map to "- Phase 2".  Never collapse phase variants.
+4. PRESERVE PRODUCT / PROVIDER DISTINCTIONS — if both "Inpatient Hospital (HMO)" and
+   "Inpatient Hospital (PPO)" exist as canonical names, route accordingly.
+5. MEDICAL BILLING CODES — when a raw term contains one or more medical billing codes
+   (CPT 5-digit numeric, HCPCS alphanumeric starting with a letter, NDC, etc.), map it
+   to the canonical entry that contains ONLY those code(s) — strip ALL surrounding text
+   (descriptions, drug names, gene names, test labels, category phrases).
+   - Preserve the prefix from the original term: if the original says "CPT", use "CPT";
+     otherwise use "HCPCS" for letter-prefixed codes, "CPT" for 5-digit numeric codes.
+   - If multiple codes appear in one raw term, the canonical name will be a
+     semicolon-separated list.
+   - Recognize codes even when buried in descriptive phrases.
+   Example: "CPT 81250 - G6PC GENE" → "CPT 81250"
+   Example: "CPT S3854 - PROSIGNA BREAST CANCER ASSAY" → "CPT S3854"
+   Example: "Drug screening confirmation codes 80150, 80152" → "CPT 80150; CPT 80152"
+   Example: "Drug screening codes G0431" → "HCPCS G0431"
+6. PASS-THROUGH — if a raw term is already in its canonical form (or very close), map it
+   to the matching canonical name directly.
+7. CONSISTENCY — within this batch, if two raw terms represent the same service, map them
+   both to the same canonical name.
+
+ESTABLISHED CANONICAL NAMES:
+{categories_json}
+
+SERVICE_TERMS:
+{terms_json}
+
+Return ONLY a valid JSON object where each key is the original SERVICE_TERM string
+(exactly as given) and each value is its canonical name — no explanation, no markdown."""
+
+
+# ── Internal helpers ──────────────────────────────────────────────────────────
+
+
+def _call_llm(prompt: str) -> str:
+    """Invoke the LLM; returns empty string on any error."""
+    try:
+        return llm_utils.invoke_claude(
+            prompt=prompt,
+            model_id=_MODEL_ID,
+            filename="service_term_standardization",
+            max_tokens=_MAX_TOKENS,
+            instruction=SERVICE_TERM_STANDARDIZATION_INSTRUCTION(),
+            usage_label="service_term_standardization",
+        )
+    except Exception as exc:
+        logging.error(
+            f"SERVICE_TERM standardization: LLM call failed — {exc}", exc_info=True
+        )
+        return ""
+
+
+def _read_grouping_map(pc_excel_path: str) -> dict[str, str]:
+    """Read FILE_NAME → grouping_key from the PC_Details sheet.
+
+    Mirrors the read logic in add_latest_contract_for_service so that the same
+    grouping_key values are used in both modules.
+
+    Returns:
+        Dict mapping FILE_NAME to grouping_key.  Returns {} on any failure so
+        callers can gracefully fall back to global standardization.
+    """
+    try:
+        pc_df = pd.read_excel(pc_excel_path, sheet_name=_PC_SHEET)
+    except Exception as exc:
+        logging.warning(
+            f"SERVICE_TERM standardization: could not read '{_PC_SHEET}' from "
+            f"{pc_excel_path} — {exc}; falling back to global standardization."
+        )
+        return {}
+
+    missing = {_PC_JOIN_KEY, _PC_GROUP_COL} - set(pc_df.columns)
+    if missing:
+        logging.warning(
+            f"SERVICE_TERM standardization: PC_Details missing column(s) {sorted(missing)}; "
+            "falling back to global standardization."
+        )
+        return {}
+
+    slim = pc_df[[_PC_JOIN_KEY, _PC_GROUP_COL]].drop_duplicates(
+        subset=[_PC_JOIN_KEY], keep="first"
+    )
+    return dict(zip(slim[_PC_JOIN_KEY], slim[_PC_GROUP_COL]))
+
+
+# ── Pass 1 — Taxonomy derivation ──────────────────────────────────────────────
+
+
+def _derive_taxonomy(unique_terms: list[str]) -> list[str]:
+    """Pass 1: derive a stable canonical name list from ALL unique terms.
+
+    When terms fit in a single chunk, one LLM call returns the canonical name list
+    directly.  When multiple chunks are needed, each chunk proposes candidates; a
+    final consolidation call merges them into a single deduplicated taxonomy.
+
+    Args:
+        unique_terms: All distinct non-empty SERVICE_TERM values.
+
+    Returns:
+        Sorted list of canonical name strings, or [] on complete failure.
+    """
+    chunks = [
+        unique_terms[i : i + _TAXONOMY_CHUNK_SIZE]
+        for i in range(0, len(unique_terms), _TAXONOMY_CHUNK_SIZE)
+    ]
+
+    all_candidate_names: list[str] = []
+
+    for idx, chunk in enumerate(chunks):
+        prompt = SERVICE_TERM_TAXONOMY_CHUNK_PROMPT(
+            json.dumps(chunk, ensure_ascii=False)
+        )
+        response = _call_llm(prompt)
+        if not response:
+            logging.warning(
+                f"SERVICE_TERM standardization Pass 1: empty response for chunk {idx + 1}; skipping"
+            )
+            continue
+        try:
+            candidates = [
+                str(v).strip()
+                for v in json_utils.parse_json_list(response)
+                if str(v).strip()
+            ]
+        except (ValueError, TypeError):
+            logging.warning(
+                f"SERVICE_TERM standardization Pass 1: failed to parse list response "
+                f"(chunk {idx + 1}) — raw (first 500 chars): {response[:500]!r}"
+            )
+            candidates = []
+        all_candidate_names.extend(candidates)
+
+    if not all_candidate_names:
+        logging.warning(
+            "SERVICE_TERM standardization Pass 1: no canonical names derived; "
+            "will fall back to original terms"
+        )
+        return []
+
+    # Single chunk — no consolidation needed
+    if len(chunks) == 1:
+        return sorted(set(all_candidate_names))
+
+    # Multiple chunks — consolidate to remove duplicates / near-duplicates.
+    # Process in sub-chunks so each LLM call stays within the output token limit.
+    deduped = sorted(set(all_candidate_names))
+
+    consolidated: list[str] = []
+    con_chunks = [
+        deduped[i : i + _CONSOLIDATION_CHUNK_SIZE]
+        for i in range(0, len(deduped), _CONSOLIDATION_CHUNK_SIZE)
+    ]
+    for idx, con_chunk in enumerate(con_chunks):
+        con_prompt = SERVICE_TERM_CONSOLIDATION_PROMPT(
+            json.dumps(con_chunk, ensure_ascii=False)
+        )
+        con_response = _call_llm(con_prompt)
+        if con_response:
+            try:
+                consolidated.extend(
+                    str(v).strip()
+                    for v in json_utils.parse_json_list(con_response)
+                    if str(v).strip()
+                )
+            except (ValueError, TypeError):
+                logging.warning(
+                    f"SERVICE_TERM standardization Pass 1: failed to parse consolidation response "
+                    f"(con_chunk {idx + 1}) — raw (first 500 chars): {con_response[:500]!r}"
+                )
+                consolidated.extend(con_chunk)
+        else:
+            # Keep originals for this chunk on failure
+            consolidated.extend(con_chunk)
+
+    # If multiple sub-chunks were used, do a final merge pass on the consolidated output
+    if len(con_chunks) > 1 and consolidated:
+        merge_prompt = SERVICE_TERM_CONSOLIDATION_PROMPT(
+            json.dumps(sorted(set(consolidated)), ensure_ascii=False)
+        )
+        merge_response = _call_llm(merge_prompt)
+        if merge_response:
+            try:
+                merged = [
+                    str(v).strip()
+                    for v in json_utils.parse_json_list(merge_response)
+                    if str(v).strip()
+                ]
+                if merged:
+                    consolidated = merged
+            except (ValueError, TypeError):
+                logging.warning(
+                    f"SERVICE_TERM standardization Pass 1: failed to parse final merge response "
+                    f"— raw (first 500 chars): {merge_response[:500]!r}"
+                )
+
+    if consolidated:
+        return sorted(set(consolidated))
+
+    # Consolidation failed — fall back to raw deduplication
+    taxonomy = sorted(set(all_candidate_names))
+    logging.warning(
+        f"SERVICE_TERM standardization Pass 1: consolidation failed; "
+        f"using raw deduplicated list ({len(taxonomy)} names)"
+    )
+    return taxonomy
+
+
+# ── Pass 2 — Mapping ──────────────────────────────────────────────────────────
+
+
+def _map_terms_to_taxonomy(
+    unique_terms: list[str], taxonomy: list[str]
+) -> dict[str, str]:
+    """Pass 2: map every unique term to a canonical name from the established taxonomy.
+
+    Args:
+        unique_terms: All distinct non-empty SERVICE_TERM values.
+        taxonomy: Canonical name list produced by Pass 1.
+
+    Returns:
+        Dict mapping each term to its canonical name.  Terms the LLM omits fall
+        back to the original value.
+    """
+    mapping: dict[str, str] = {}
+    categories_json = json.dumps(taxonomy, ensure_ascii=False)
+
+    batches = [
+        unique_terms[i : i + _MAPPING_BATCH_SIZE]
+        for i in range(0, len(unique_terms), _MAPPING_BATCH_SIZE)
+    ]
+
+    for idx, batch in enumerate(batches):
+        prompt = SERVICE_TERM_MAPPING_BATCH_PROMPT(
+            categories_json,
+            json.dumps(batch, ensure_ascii=False),
+        )
+        response = _call_llm(prompt)
+        if response:
+            try:
+                result = json_utils.parse_json_dict(response)
+                batch_mapping = {str(k): str(v) for k, v in result.items()}
+            except (ValueError, TypeError):
+                logging.warning(
+                    f"SERVICE_TERM standardization Pass 2: failed to parse mapping response "
+                    f"(batch {idx + 1}) — raw (first 500 chars): {response[:500]!r}"
+                )
+                batch_mapping = {}
+        else:
+            batch_mapping = {}
+
+        for term in batch:
+            mapping[term] = batch_mapping.get(term, term)
+
+        unmapped = [t for t in batch if t not in batch_mapping]
+        if unmapped:
+            logging.warning(
+                f"SERVICE_TERM standardization Pass 2: batch {idx + 1} — "
+                f"{len(unmapped)} term(s) not mapped by LLM, falling back to original"
+            )
+
+    return mapping
+
+
+# ── Public API ────────────────────────────────────────────────────────────────
+
+
+def build_standardization_mapping(unique_terms: list[str]) -> dict[str, str]:
+    """Two-pass LLM standardization: derive taxonomy then map all terms.
+
+    Args:
+        unique_terms: Deduplicated list of SERVICE_TERM strings.
+
+    Returns:
+        Dict mapping each original term to its canonical name.
+    """
+    if not unique_terms:
+        return {}
+
+    taxonomy = _derive_taxonomy(unique_terms)
+    if not taxonomy:
+        # Full fallback — return identity mapping
+        return {t: t for t in unique_terms}
+
+    return _map_terms_to_taxonomy(unique_terms, taxonomy)
+
+
+def standardize_service_terms(
+    df: pd.DataFrame,
+    pc_excel_path: str | None = None,
+) -> pd.DataFrame:
+    """Add AARETE_DERIVED_SERVICE_TERM column immediately after SERVICE_TERM.
+
+    When pc_excel_path is provided, the PC_Details sheet is read to obtain a
+    FILE_NAME → grouping_key mapping.  The two-pass standardization then runs
+    independently per grouping_key so that canonical names are derived from the
+    specific service mix of each provider group.  Rows whose FILE_NAME has no
+    matching grouping_key are standardized globally as a fallback.
+
+    When pc_excel_path is absent (or the file cannot be read), all unique terms
+    are standardized together in a single global pass.
+
+    Args:
+        df: Combined output DataFrame containing a SERVICE_TERM column.
+        pc_excel_path: Optional path to the parent-child Excel output file whose
+            PC_Details sheet contains FILE_NAME and grouping_key columns.
+
+    Returns:
+        DataFrame with AARETE_DERIVED_SERVICE_TERM inserted right after SERVICE_TERM.
+        Returns df unchanged if SERVICE_TERM column is absent.
+    """
+    if "SERVICE_TERM" not in df.columns:
+        logging.warning(
+            "SERVICE_TERM standardization: SERVICE_TERM column not found; skipping."
+        )
+        return df
+
+    if "AARETE_DERIVED_SERVICE_TERM" in df.columns:
+        df = df.drop(columns=["AARETE_DERIVED_SERVICE_TERM"])
+
+    # ── Snapshot usage before LLM calls (for cost delta at the end) ───────────
+    _usage_before_snapshot = (
+        usage_tracking.get_usage_data()
+        .get("per_file_per_model", {})
+        .get("service_term_standardization", {})
+    )
+    _cost_before = sum(v["total_cost"] for v in _usage_before_snapshot.values())
+    _tokens_before = sum(v["total_tokens"] for v in _usage_before_snapshot.values())
+    _requests_before = sum(v["request_count"] for v in _usage_before_snapshot.values())
+
+    # ── Resolve grouping map ──────────────────────────────────────────────────
+    grouping_map: dict[str, str] = {}
+    if pc_excel_path:
+        grouping_map = _read_grouping_map(pc_excel_path)
+
+    # ── Build term → canonical name mapping ───────────────────────────────────
+    if grouping_map:
+        # Assign grouping_key per row via FILE_NAME (NaN when no match)
+        has_join_key = _PC_JOIN_KEY in df.columns
+        row_groups = (
+            df[_PC_JOIN_KEY].map(grouping_map)
+            if has_join_key
+            else pd.Series(index=df.index, dtype=object)
+        )
+
+        ungrouped_count = row_groups.isna().sum()
+        if ungrouped_count > 0:
+            logging.info(
+                f"SERVICE_TERM standardization: {ungrouped_count} rows have no matching "
+                f"grouping_key; using global fallback"
+            )
+
+        combined_mapping: dict[str, str] = {}
+
+        # Per-group standardization
+        unique_groups = row_groups.dropna().unique().tolist()
+        for group_key in unique_groups:
+            group_terms = [
+                t
+                for t in df.loc[row_groups == group_key, "SERVICE_TERM"]
+                .dropna()
+                .unique()
+                .tolist()
+                if isinstance(t, str) and t.strip()
+            ]
+            if not group_terms:
+                continue
+            group_mapping = build_standardization_mapping(group_terms)
+            combined_mapping.update(group_mapping)
+
+        # Global fallback for rows with no matching grouping_key
+        ungrouped_terms = [
+            t
+            for t in df.loc[row_groups.isna(), "SERVICE_TERM"]
+            .dropna()
+            .unique()
+            .tolist()
+            if isinstance(t, str) and t.strip() and t not in combined_mapping
+        ]
+        if ungrouped_terms:
+            combined_mapping.update(build_standardization_mapping(ungrouped_terms))
+
+        mapping = combined_mapping
+
+    else:
+        # Global mode — original behaviour
+        unique_terms = [
+            t
+            for t in df["SERVICE_TERM"].dropna().unique().tolist()
+            if isinstance(t, str) and t.strip()
+        ]
+        mapping = build_standardization_mapping(unique_terms)
+
+    # ── Apply mapping ─────────────────────────────────────────────────────────
+    df["AARETE_DERIVED_SERVICE_TERM"] = df["SERVICE_TERM"].apply(
+        lambda v: (
+            mapping.get(str(v), str(v))
+            if pd.notna(v) and str(v).strip()
+            else (v if pd.isna(v) else "")
+        )
+    )
+
+    # Insert right after SERVICE_TERM
+    cols = df.columns.tolist()
+    st_idx = cols.index("SERVICE_TERM")
+    cols.remove("AARETE_DERIVED_SERVICE_TERM")
+    cols.insert(st_idx + 1, "AARETE_DERIVED_SERVICE_TERM")
+    df = df[cols]
+
+    # ── Log cost incurred by this run ─────────────────────────────────────────
+    _usage_after = (
+        usage_tracking.get_usage_data()
+        .get("per_file_per_model", {})
+        .get("service_term_standardization", {})
+    )
+    _cost_after = sum(v["total_cost"] for v in _usage_after.values())
+    _tokens_after = sum(v["total_tokens"] for v in _usage_after.values())
+    _requests_after = sum(v["request_count"] for v in _usage_after.values())
+    _cost_delta = _cost_after - _cost_before
+    _tokens_delta = _tokens_after - _tokens_before
+    _requests_delta = _requests_after - _requests_before
+    logging.debug(
+        f"SERVICE_TERM standardization cost: ${_cost_delta:.6f} "
+        f"({_tokens_delta} tokens, {_requests_delta} LLM requests)"
+    )
+
+    return df

src/scripts/retroactively_standardize_service_term.py

@@ -0,0 +1,184 @@
+"""
+Service Term Standardization — manual test script
+
+Loads a PC pipeline output file, runs standardize_service_terms, and writes
+results so you can visually inspect SERVICE_TERM → AARETE_DERIVED_SERVICE_TERM mappings.
+
+Usage:
+    python service_standardization_test.py <pc_output_path> [pc_excel_path]
+
+    pc_output_path  — CSV or Excel file that is the combined pipeline output
+                      (must contain a SERVICE_TERM column).
+                      If this is an Excel file that also contains a PC_Details
+                      sheet, per-group mode is enabled automatically — no second
+                      argument needed.
+    pc_excel_path   — optional: path to a separate PC Excel file whose PC_Details
+                      sheet contains FILE_NAME and grouping_key.  Only needed when
+                      the grouping info lives in a different file from the output.
+
+Examples:
+    python service_standardization_test.py output/combined.csv
+    python service_standardization_test.py output/combined.xlsx
+    python service_standardization_test.py output/combined.csv path/to/pc_output.xlsx
+"""
+
+import difflib
+import logging
+import sys
+from pathlib import Path
+
+import pandas as pd
+
+# ── Standalone path fix ───────────────────────────────────────────────────────
+# Ensure the repo root (two levels up from src/scripts/) is on sys.path so the
+# script can be invoked directly from any working directory.
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+if str(_REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(_REPO_ROOT))
+# ─────────────────────────────────────────────────────────────────────────────
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s  %(levelname)-8s  %(message)s",
+    datefmt="%H:%M:%S",
+)
+
+
+def has_pc_details_sheet(path: str) -> bool:
+    """Return True if path is an Excel file that contains a PC_Details sheet."""
+    p = Path(path)
+    if p.suffix.lower() not in {".xlsx", ".xls"}:
+        return False
+    try:
+        xl = pd.ExcelFile(path)
+        return "PC_Details" in xl.sheet_names
+    except Exception:
+        return False
+
+
+def main():
+    if len(sys.argv) < 2:
+        logging.error(__doc__)
+        sys.exit(1)
+
+    pc_output_path = sys.argv[1]
+    pc_excel_path = sys.argv[2] if len(sys.argv) > 2 else None
+
+    # ── Load PC output ────────────────────────────────────────────────────────
+    from src.utils import io_utils
+
+    _raw = io_utils.read_local(pc_output_path)
+    if _raw is None:
+        logging.error(f"Could not load file — {pc_output_path}")
+        sys.exit(1)
+    if not isinstance(_raw, pd.DataFrame):
+        logging.error(
+            f"Expected a DataFrame from {pc_output_path}, got {type(_raw).__name__}"
+        )
+        sys.exit(1)
+    df = _raw
+    logging.info(f"Loaded {len(df):,} rows from {pc_output_path}")
+
+    if "SERVICE_TERM" not in df.columns:
+        logging.error("SERVICE_TERM column not found in the provided file.")
+        cols = df.columns.tolist()
+        close = difflib.get_close_matches("SERVICE_TERM", cols, n=3, cutoff=0.5)
+        if close:
+            logging.error(f"Did you mean one of: {close}")
+        else:
+            logging.error(f"Columns present: {cols}")
+        sys.exit(1)
+
+    # Auto-detect: if the output file itself has a PC_Details sheet, use it for
+    # grouping without requiring a second argument.
+    if pc_excel_path is None and has_pc_details_sheet(pc_output_path):
+        pc_excel_path = pc_output_path
+        logging.info(
+            f"Auto-detected PC_Details sheet in {pc_output_path} — enabling per-group mode"
+        )
+
+    unique_before = df["SERVICE_TERM"].dropna().nunique()
+    logging.info(
+        f"Unique SERVICE_TERM values before standardization: {unique_before:,}"
+    )
+    logging.debug(
+        "Sample raw terms: %s", df["SERVICE_TERM"].dropna().unique()[:20].tolist()
+    )
+
+    if pc_excel_path:
+        logging.info(f"Grouping mode: using PC Excel at {pc_excel_path}")
+    else:
+        logging.info("Global mode: no PC Excel provided")
+
+    # ── Run standardization ───────────────────────────────────────────────────
+    logging.info("Running standardize_service_terms...")
+    from src.pipelines.shared.postprocessing.service_term_standardization import (
+        standardize_service_terms,
+    )
+
+    df_out = standardize_service_terms(df, pc_excel_path=pc_excel_path)
+
+    # ── Log mapping summary ───────────────────────────────────────────────────
+    if "AARETE_DERIVED_SERVICE_TERM" not in df_out.columns:
+        logging.error(
+            "AARETE_DERIVED_SERVICE_TERM column was not added — check logs above."
+        )
+        sys.exit(1)
+
+    mapping_df = (
+        df_out[["SERVICE_TERM", "AARETE_DERIVED_SERVICE_TERM"]]
+        .drop_duplicates()
+        .sort_values("SERVICE_TERM")
+        .reset_index(drop=True)
+    )
+
+    changed = mapping_df[
+        mapping_df["SERVICE_TERM"].astype(str)
+        != mapping_df["AARETE_DERIVED_SERVICE_TERM"].astype(str)
+    ]
+    unchanged = mapping_df[
+        mapping_df["SERVICE_TERM"].astype(str)
+        == mapping_df["AARETE_DERIVED_SERVICE_TERM"].astype(str)
+    ]
+
+    logging.info("=" * 70)
+    logging.info("STANDARDIZATION SUMMARY")
+    logging.info("=" * 70)
+    logging.info(f"  Total unique mappings : {len(mapping_df):,}")
+    logging.info(f"  Changed               : {len(changed):,}")
+    logging.info(f"  Unchanged (pass-thru) : {len(unchanged):,}")
+
+    if not changed.empty:
+        logging.info("CHANGED MAPPINGS (first 50):")
+        for _, row in changed.head(50).iterrows():
+            raw = str(row["SERVICE_TERM"])[:53]
+            canonical = str(row["AARETE_DERIVED_SERVICE_TERM"])
+            logging.info(f"  {raw:<55}  →  {canonical}")
+
+    if not unchanged.empty:
+        logging.info("UNCHANGED TERMS (first 20):")
+        for _, row in unchanged.head(20).iterrows():
+            logging.info(f"  {row['SERVICE_TERM']!r}")
+
+    # ── Write outputs ─────────────────────────────────────────────────────────
+    input_path = Path(pc_output_path)
+    out_dir = input_path.parent
+    stem = input_path.stem
+    suffix = input_path.suffix.lower()
+
+    out_ext = suffix if suffix in {".xlsx", ".xls"} else ".csv"
+    output_file = out_dir / f"{stem}_standardized_services{out_ext}"
+    mapping_file = out_dir / f"{stem}_standardized_services_mapping.csv"
+
+    if out_ext in {".xlsx", ".xls"}:
+        df_out.to_excel(output_file, index=False)
+    else:
+        df_out.to_csv(output_file, index=False)
+    logging.info(f"Full output written to   : {output_file}")
+
+    mapping_df.to_csv(mapping_file, index=False)
+    logging.info(f"Mapping table written to : {mapping_file}")
+
+
+if __name__ == "__main__":
+    main()