From 44cecb5d00bc2200230399acb786f4b4e37414c3 Mon Sep 17 00:00:00 2001 From: Karan Desai Date: Mon, 20 Apr 2026 21:26:47 +0000 Subject: [PATCH] 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 * 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 | 132 ++++ src/constants/investment_columns.py | 1 + src/parent_child/__main__.py | 12 +- src/parent_child/column_mapper.py | 10 + src/pipelines/runner.py | 28 +- src/pipelines/saas/main.py | 73 +++ .../service_term_standardization.py | 616 ++++++++++++++++++ .../retroactively_standardize_service_term.py | 184 ++++++ 8 files changed, 1049 insertions(+), 7 deletions(-) create mode 100644 documentation/SERVICE_TERM_STANDARDIZATION.md create mode 100644 src/pipelines/shared/postprocessing/service_term_standardization.py create mode 100644 src/scripts/retroactively_standardize_service_term.py diff --git a/documentation/SERVICE_TERM_STANDARDIZATION.md b/documentation/SERVICE_TERM_STANDARDIZATION.md new file mode 100644 index 0000000..2a5e23d --- /dev/null +++ b/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 [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 | +|---|---| +| `_standardized_services.csv/.xlsx` | Full dataset with the new `AARETE_DERIVED_SERVICE_TERM` column appended | +| `_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 | diff --git a/src/constants/investment_columns.py b/src/constants/investment_columns.py index 4c5f573..cdecffb 100644 --- a/src/constants/investment_columns.py +++ b/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]", diff --git a/src/parent_child/__main__.py b/src/parent_child/__main__.py index a69e449..53472b6 100644 --- a/src/parent_child/__main__.py +++ b/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") diff --git a/src/parent_child/column_mapper.py b/src/parent_child/column_mapper.py index 3d6d4c5..90d4c25 100644 --- a/src/parent_child/column_mapper.py +++ b/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: diff --git a/src/pipelines/runner.py b/src/pipelines/runner.py index 689d03b..38184c4 100644 --- a/src/pipelines/runner.py +++ b/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") diff --git a/src/pipelines/saas/main.py b/src/pipelines/saas/main.py index 9d79c3c..c9cdc18 100644 --- a/src/pipelines/saas/main.py +++ b/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) diff --git a/src/pipelines/shared/postprocessing/service_term_standardization.py b/src/pipelines/shared/postprocessing/service_term_standardization.py new file mode 100644 index 0000000..1fe4ce3 --- /dev/null +++ b/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 diff --git a/src/scripts/retroactively_standardize_service_term.py b/src/scripts/retroactively_standardize_service_term.py new file mode 100644 index 0000000..ad66b47 --- /dev/null +++ b/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_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()