shashanks/medical_coding
0
1"""2Randomized Case Generator for the Medical Coding Auditor Environment.3 4Generates procedurally-created task scenarios by composing error templates5from the existing guideline data. Each generated case is deterministic given6the same seed (derived from episode_id), so grading remains reproducible.7 8Supports difficulty targeting: random_easy, random_medium, random_hard, random_expert,9or plain "random" for a random difficulty.10"""11 12from __future__ import annotations13 14import hashlib15import random16from typing import Any, Dict, List, Optional, Tuple17 18# ---------------------------------------------------------------------------19# Patient templates20# ---------------------------------------------------------------------------21 22_FIRST_NAMES_M = ["James", "Robert", "Michael", "David", "William", "Thomas", "Richard", "Charles"]23_FIRST_NAMES_F = ["Mary", "Patricia", "Jennifer", "Linda", "Elizabeth", "Susan", "Karen", "Nancy"]24_LAST_INITIALS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"25_INSURANCES = [26 "Blue Cross PPO", "Medicare Part B", "Aetna HMO",27 "United Healthcare PPO", "Cigna EPO", "Humana Gold Plus",28 "Anthem Blue Shield", "Tricare Standard",29]30 31 32def _make_patient(rng: random.Random, sex: Optional[str] = None) -> Dict[str, Any]:33 """Generate a random patient with demographics."""34 if sex is None:35 sex = rng.choice(["male", "female"])36 age = rng.randint(22, 78)37 first = rng.choice(_FIRST_NAMES_M if sex == "male" else _FIRST_NAMES_F)38 last_init = rng.choice(_LAST_INITIALS)39 return {40 "age": age,41 "sex": sex,42 "mrn": f"MRN-GEN-{rng.randint(1000, 9999)}",43 "insurance": rng.choice(_INSURANCES),44 "_name": f"{first} {last_init}.",45 }46 47 48# ---------------------------------------------------------------------------49# Error templates — each returns (proposed_codes_fragment, expected_error, note_snippet)50# ---------------------------------------------------------------------------51 52def _error_demographic_mismatch(53 rng: random.Random, patient: Dict[str, Any], data: Dict[str, Any],54) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Any], str]:55 """O80 on a male patient."""56 patient["sex"] = "male" # force male for this error57 codes = {58 "O80": {"description": "Encounter for full-term uncomplicated delivery", "code_type": "ICD-10-CM"},59 }60 error = {61 "code": "O80",62 "error_type": "demographic_mismatch",63 "description": "O80 is a maternity code applicable only to female patients. Cannot be assigned to a male patient.",64 "key_terms": ["female", "male", "maternity", "obstetric", "delivery", "sex", "gender", "demographic"],65 }66 title = "Mr." if patient["sex"] == "male" else "Ms."67 note = (68 f"HISTORY OF PRESENT ILLNESS: {title} {patient['_name']} is a "69 f"{patient['age']}-year-old {patient['sex']} presenting for a routine wellness visit. "70 f"No obstetric or gynecological history is documented."71 )72 return codes, error, note73 74 75def _error_excludes1_copd_asthma(76 rng: random.Random, patient: Dict[str, Any], data: Dict[str, Any],77) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Any], str]:78 """J44.1 + J45.20 Excludes1 conflict."""79 codes = {80 "J44.1": {"description": "COPD with acute exacerbation", "code_type": "ICD-10-CM"},81 "J45.20": {"description": "Mild intermittent asthma, uncomplicated", "code_type": "ICD-10-CM"},82 }83 error = {84 "code": "J44.1",85 "error_type": "excludes1_conflict",86 "conflicting_code": "J45.20",87 "description": "J44.1 (COPD) has Excludes1 for J45 (asthma). Mutually exclusive — cannot code together.",88 "key_terms": ["Excludes1", "mutually exclusive", "asthma", "COPD", "J45", "J44", "cannot be coded together"],89 }90 note = (91 f"HISTORY OF PRESENT ILLNESS: Patient has a 15-year history of COPD (GOLD stage II). "92 f"Presenting with acute exacerbation — increased dyspnea and sputum. "93 f"Prior asthma label from primary care was superseded by COPD diagnosis after PFTs "94 f"confirmed irreversible obstruction. Patient does NOT carry a concurrent asthma diagnosis.\n\n"95 f"SPIROMETRY: FEV1 48% predicted, post-bronchodilator improvement <10% (consistent with COPD)."96 )97 return codes, error, note98 99 100def _error_excludes1_diabetes(101 rng: random.Random, patient: Dict[str, Any], data: Dict[str, Any],102) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Any], str]:103 """E10.9 + E11.9 Excludes1 conflict."""104 codes = {105 "E10.9": {"description": "Type 1 diabetes mellitus without complications", "code_type": "ICD-10-CM"},106 "E11.9": {"description": "Type 2 diabetes mellitus without complications", "code_type": "ICD-10-CM"},107 }108 error = {109 "code": "E10.9",110 "error_type": "excludes1_conflict",111 "conflicting_code": "E11.9",112 "description": "E10.9 (Type 1 DM) has Excludes1 for E11 (Type 2 DM). Mutually exclusive.",113 "key_terms": ["Excludes1", "mutually exclusive", "Type 1", "Type 2", "E11", "E10", "diabetes"],114 }115 note = (116 f"ENDOCRINOLOGY ASSESSMENT: Patient has Type 2 diabetes mellitus, managed with metformin. "117 f"C-peptide is normal (2.3 ng/mL), GAD65 autoantibodies are negative, "118 f"ruling out autoimmune Type 1 diabetes. Diagnosis is definitively Type 2 only."119 )120 return codes, error, note121 122 123def _error_ncci_echo(124 rng: random.Random, patient: Dict[str, Any], data: Dict[str, Any],125) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Any], str]:126 """93306 + 93307 or 93306 + 93308 NCCI bundling."""127 limited = rng.choice(["93307", "93308"])128 limited_desc = (129 "Echocardiography, transthoracic, limited study without Doppler"130 if limited == "93307"131 else "Echocardiography, transthoracic, follow-up or limited study"132 )133 codes = {134 "93306": {"description": "Echocardiography, transthoracic, complete with Doppler", "code_type": "CPT"},135 limited: {"description": limited_desc, "code_type": "CPT"},136 }137 error = {138 "code": "93306",139 "error_type": "ncci_edit",140 "conflicting_code": limited,141 "description": f"93306 (complete TTE) and {limited} (limited TTE) are NCCI PTP edit — unbundling violation.",142 "key_terms": ["bundling", "unbundling", "NCCI", "PTP", "complete", "limited", limited, "comprehensive"],143 }144 note = (145 f"SERVICES RENDERED: A complete transthoracic echocardiogram was performed with 2D imaging, "146 f"M-mode, spectral Doppler, and color flow Doppler. LVEF estimated at {rng.randint(40, 60)}%. "147 f"Additionally, a limited echocardiographic study was documented for focused wall motion assessment."148 )149 return codes, error, note150 151 152def _error_ncci_em(153 rng: random.Random, patient: Dict[str, Any], data: Dict[str, Any],154) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Any], str]:155 """99213 + 99214 E/M unbundling."""156 codes = {157 "99213": {"description": "Office visit, low-moderate complexity", "code_type": "CPT"},158 "99214": {"description": "Office visit, moderate-high complexity", "code_type": "CPT"},159 }160 error = {161 "code": "99213",162 "error_type": "ncci_edit",163 "conflicting_code": "99214",164 "description": "99213 and 99214 are different E/M levels for the same service — NCCI edit, only one should be billed.",165 "key_terms": ["bundling", "NCCI", "E/M", "99213", "99214", "same service", "unbundling"],166 }167 note = (168 f"SERVICES RENDERED: An office visit was performed. Medical decision making was "169 f"moderate to high complexity given the patient's multiple chronic conditions."170 )171 return codes, error, note172 173 174def _error_specificity_fracture(175 rng: random.Random, patient: Dict[str, Any], data: Dict[str, Any],176) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Any], str]:177 """S52.501A wrong 7th character (should be D for follow-up)."""178 codes = {179 "S52.501A": {180 "description": "Fracture of lower end of right radius, initial encounter",181 "code_type": "ICD-10-CM",182 },183 }184 error = {185 "code": "S52.501A",186 "error_type": "specificity_error",187 "description": "7th character 'A' (initial) is wrong — this is a follow-up visit. Should be S52.501D (subsequent).",188 "key_terms": ["7th character", "subsequent", "initial", "follow-up", "healing", "S52.501D", "phase of care"],189 }190 weeks = rng.randint(4, 12)191 note = (192 f"HISTORY OF PRESENT ILLNESS: Patient presents for follow-up {weeks} weeks after closed "193 f"fracture of distal right radius. Cast was removed previously. X-ray shows routine healing "194 f"with appropriate callus formation.\n\n"195 f"Note: This is a SUBSEQUENT ENCOUNTER for a healing fracture — active treatment concluded {weeks} weeks ago."196 )197 return codes, error, note198 199 200def _error_untraceable_code(201 rng: random.Random, patient: Dict[str, Any], data: Dict[str, Any],202) -> Tuple[Dict[str, Dict[str, str]], Dict[str, Any], str]:203 """A fabricated ICD-10 code that doesn't exist."""204 fake_codes = ["Z99.999", "E99.99", "J99.999", "M99.999", "R99.999"]205 fake = rng.choice(fake_codes)206 codes = {207 fake: {"description": "Unspecified condition (placeholder)", "code_type": "ICD-10-CM"},208 }209 error = {210 "code": fake,211 "error_type": "untraceable_code",212 "description": f"{fake} does not exist in the official ICD-10-CM tabular list. Untraceable code.",213 "key_terms": ["invalid", "untraceable", "not exist", "not found", "official", "billable"],214 }215 note = "" # untraceable codes don't need clinical note context216 return codes, error, note217 218 219# ---------------------------------------------------------------------------220# Filler codes (valid, no errors — false positive traps)221# ---------------------------------------------------------------------------222 223_FILLER_CODES: List[Dict[str, Any]] = [224 {"code": "Z00.00", "description": "Encounter for general adult medical examination", "code_type": "ICD-10-CM"},225 {"code": "Z87.891", "description": "Personal history of other specified conditions", "code_type": "ICD-10-CM"},226 {"code": "M79.621", "description": "Pain in right upper arm", "code_type": "ICD-10-CM"},227 {"code": "99213", "description": "Office visit, low-moderate complexity", "code_type": "CPT"},228 {"code": "99214", "description": "Office visit, moderate-high complexity", "code_type": "CPT"},229]230 231 232# ---------------------------------------------------------------------------233# Error pools per difficulty234# ---------------------------------------------------------------------------235 236_ERROR_POOL_EASY = [_error_demographic_mismatch]237_ERROR_POOL_MEDIUM = [_error_excludes1_copd_asthma, _error_excludes1_diabetes, _error_ncci_echo, _error_ncci_em]238_ERROR_POOL_HARD = [_error_specificity_fracture, _error_untraceable_code]239 240_DIFFICULTY_CONFIG = {241 "easy": {"error_count": 1, "pools": [_ERROR_POOL_EASY], "filler_count": (1, 2), "max_steps": 10},242 "medium": {"error_count": 1, "pools": [_ERROR_POOL_MEDIUM], "filler_count": (1, 2), "max_steps": 15},243 "hard": {"error_count": 2, "pools": [_ERROR_POOL_MEDIUM, _ERROR_POOL_HARD], "filler_count": (1, 2), "max_steps": 20},244 "expert": {"error_count": 3, "pools": [_ERROR_POOL_EASY, _ERROR_POOL_MEDIUM, _ERROR_POOL_HARD], "filler_count": (1, 3), "max_steps": 25},245}246 247 248# ---------------------------------------------------------------------------249# Clinical note assembly250# ---------------------------------------------------------------------------251 252def _assemble_clinical_note(253 patient: Dict[str, Any],254 error_snippets: List[str],255 rng: random.Random,256) -> str:257 """Compose a synthetic clinical note from patient info and error-specific snippets."""258 title = "Mr." if patient["sex"] == "male" else "Ms."259 age = patient["age"]260 sex = patient["sex"]261 262 parts = [263 f"REASON FOR VISIT: Comprehensive evaluation and management.",264 "",265 ]266 267 # Add error-specific clinical content268 for snippet in error_snippets:269 if snippet:270 parts.append(snippet)271 parts.append("")272 273 # Generic physical exam274 bp_sys = rng.randint(110, 150)275 bp_dia = rng.randint(70, 95)276 hr = rng.randint(62, 98)277 parts.extend([278 f"PHYSICAL EXAM: Vitals: BP {bp_sys}/{bp_dia}, HR {hr}. "279 f"General: {title} {patient['_name']} is a {age}-year-old {sex} "280 f"in no acute distress. Exam otherwise unremarkable for areas not addressed above.",281 "",282 f"ASSESSMENT AND PLAN: See individual assessments above. "283 f"Return for follow-up as indicated.",284 ])285 286 return "\n".join(parts)287 288 289# ---------------------------------------------------------------------------290# Public API291# ---------------------------------------------------------------------------292 293def generate_random_task(294 data: Dict[str, Any],295 seed: str,296 difficulty: Optional[str] = None,297) -> Dict[str, Any]:298 """299 Generate a random task scenario.300 301 Args:302 data: The loaded ground_truth_cases.json data (for guideline reference).303 seed: A string seed for deterministic generation (typically episode_id).304 difficulty: One of 'easy', 'medium', 'hard', 'expert', or None (random).305 306 Returns:307 A task dict in the same format as the fixed tasks in ground_truth_cases.json.308 """309 # Deterministic RNG from seed310 seed_int = int(hashlib.sha256(seed.encode()).hexdigest()[:16], 16)311 rng = random.Random(seed_int)312 313 # Pick difficulty314 if difficulty not in _DIFFICULTY_CONFIG:315 difficulty = rng.choice(list(_DIFFICULTY_CONFIG.keys()))316 config = _DIFFICULTY_CONFIG[difficulty]317 318 # Generate patient319 patient = _make_patient(rng)320 321 # Select error generators — pick one from each pool, up to error_count322 error_generators = []323 pools = list(config["pools"])324 for _ in range(config["error_count"]):325 if not pools:326 break327 pool = pools.pop(0)328 gen = rng.choice(pool)329 error_generators.append(gen)330 331 # Generate errors332 proposed_codes: Dict[str, Dict[str, str]] = {}333 expected_errors: List[Dict[str, Any]] = []334 note_snippets: List[str] = []335 used_codes: set = set()336 337 for gen_fn in error_generators:338 codes_frag, error, snippet = gen_fn(rng, patient, data)339 # Avoid code collisions340 if any(c in used_codes for c in codes_frag):341 continue342 proposed_codes.update(codes_frag)343 used_codes.update(codes_frag.keys())344 expected_errors.append(error)345 note_snippets.append(snippet)346 347 # Add filler codes (false-positive traps)348 filler_min, filler_max = config["filler_count"]349 n_fillers = rng.randint(filler_min, filler_max)350 available_fillers = [f for f in _FILLER_CODES if f["code"] not in used_codes]351 rng.shuffle(available_fillers)352 for filler in available_fillers[:n_fillers]:353 proposed_codes[filler["code"]] = {354 "description": filler["description"],355 "code_type": filler["code_type"],356 }357 358 # Assemble clinical note359 clinical_note = _assemble_clinical_note(patient, note_snippets, rng)360 361 # Build hints362 hints = [363 "This is a procedurally generated case — investigate all proposed codes systematically.",364 f"Expect approximately {len(expected_errors)} error(s) in this case.",365 "Not every code has an error — avoid false positives.",366 ]367 368 # Remove _name from patient before returning (internal only)369 patient_clean = {k: v for k, v in patient.items() if not k.startswith("_")}370 371 return {372 "task_id": f"random_{difficulty}_{seed[:8]}",373 "difficulty": difficulty,374 "description": f"Procedurally generated {difficulty} case with {len(expected_errors)} error(s).",375 "scenario": f"A {patient['age']}-year-old {patient['sex']} patient encounter with {len(proposed_codes)} proposed codes.",376 "patient": patient_clean,377 "clinical_note": clinical_note,378 "proposed_codes": proposed_codes,379 "expected_errors": expected_errors,380 "max_steps": config["max_steps"],381 "hints": hints,382 }383