"""CMS Pricing Provider for Medicare fee schedules and cost transparency.""" import logging from typing import List, Dict, Any, Callable import httpx from core.base_provider import BaseProvider from core.decorators import safe_json_return from providers import register_provider logger = logging.getLogger(__name__) # Sample Medicare fee schedule data MEDICARE_FEE_SCHEDULE = { "99213": { "description": "Office visit, established patient, level 3", "medicare_payment": 93.47, "facility_rate": 93.47, "non_facility_rate": 131.20, "rvu": 1.92 }, "99214": { "description": "Office visit, established patient, level 4", "medicare_payment": 131.20, "facility_rate": 131.20, "non_facility_rate": 183.19, "rvu": 2.80 }, "43239": { "description": "Upper GI endoscopy with biopsy", "medicare_payment": 248.54, "facility_rate": 248.54, "non_facility_rate": 450.00, "rvu": 5.28 }, "45378": { "description": "Colonoscopy, diagnostic", "medicare_payment": 365.00, "facility_rate": 365.00, "non_facility_rate": 650.00, "rvu": 7.75 }, "70450": { "description": "CT scan, head without contrast", "medicare_payment": 178.00, "facility_rate": 178.00, "non_facility_rate": 320.00, "rvu": 3.78 }, "70553": { "description": "MRI brain with contrast", "medicare_payment": 485.00, "facility_rate": 485.00, "non_facility_rate": 875.00, "rvu": 10.30 }, "80053": { "description": "Comprehensive metabolic panel", "medicare_payment": 14.00, "facility_rate": 14.00, "non_facility_rate": 18.00, "rvu": 0.30 }, "85025": { "description": "Complete blood count with differential", "medicare_payment": 11.00, "facility_rate": 11.00, "non_facility_rate": 14.00, "rvu": 0.23 }, "71046": { "description": "Chest X-ray, 2 views", "medicare_payment": 38.00, "facility_rate": 38.00, "non_facility_rate": 62.00, "rvu": 0.81 } } # Sample procedure cost estimates PROCEDURE_COST_ESTIMATES = { "colonoscopy": { "procedure": "Diagnostic Colonoscopy", "hcpcs_code": "45378", "average_cost": 2750, "low_estimate": 1500, "high_estimate": 4500, "medicare_rate": 365, "factors": ["Facility type", "Anesthesia", "Biopsy performed", "Location"] }, "mri": { "procedure": "MRI Scan", "hcpcs_code": "70553", "average_cost": 1420, "low_estimate": 400, "high_estimate": 3500, "medicare_rate": 485, "factors": ["Body part", "With/without contrast", "Facility type"] }, "ct scan": { "procedure": "CT Scan", "hcpcs_code": "70450", "average_cost": 820, "low_estimate": 300, "high_estimate": 1800, "medicare_rate": 178, "factors": ["Body part", "With/without contrast", "Emergency vs scheduled"] }, "blood test": { "procedure": "Comprehensive Metabolic Panel", "hcpcs_code": "80053", "average_cost": 45, "low_estimate": 15, "high_estimate": 120, "medicare_rate": 14, "factors": ["Lab facility", "Stat vs routine", "Insurance negotiated rate"] }, "x-ray": { "procedure": "Chest X-Ray", "hcpcs_code": "71046", "average_cost": 125, "low_estimate": 50, "high_estimate": 300, "medicare_rate": 38, "factors": ["Number of views", "Facility type", "Interpretation included"] }, "office visit": { "procedure": "Office Visit - Established Patient", "hcpcs_code": "99213", "average_cost": 150, "low_estimate": 100, "high_estimate": 300, "medicare_rate": 93, "factors": ["Complexity level", "Time spent", "Geographic location"] } } # Sample facility comparison data FACILITY_COSTS = { "45378": [ # Colonoscopy {"facility": "University Hospital", "gross_charge": 4500, "negotiated_rate": 2800, "cash_price": 2500}, {"facility": "Community Medical Center", "gross_charge": 3200, "negotiated_rate": 2100, "cash_price": 1800}, {"facility": "Ambulatory Surgery Center", "gross_charge": 2800, "negotiated_rate": 1600, "cash_price": 1500}, ], "70450": [ # CT Scan {"facility": "University Hospital", "gross_charge": 1800, "negotiated_rate": 820, "cash_price": 600}, {"facility": "Community Medical Center", "gross_charge": 1200, "negotiated_rate": 650, "cash_price": 500}, {"facility": "Imaging Center", "gross_charge": 900, "negotiated_rate": 450, "cash_price": 400}, ], "70553": [ # MRI {"facility": "University Hospital", "gross_charge": 3500, "negotiated_rate": 1800, "cash_price": 1500}, {"facility": "Community Medical Center", "gross_charge": 2800, "negotiated_rate": 1400, "cash_price": 1200}, {"facility": "Imaging Center", "gross_charge": 2200, "negotiated_rate": 1000, "cash_price": 900}, ] } # Procedure code database PROCEDURE_CODES = [ {"code": "99213", "description": "Office visit, established patient, level 3", "category": "Evaluation & Management"}, {"code": "99214", "description": "Office visit, established patient, level 4", "category": "Evaluation & Management"}, {"code": "99215", "description": "Office visit, established patient, level 5", "category": "Evaluation & Management"}, {"code": "99203", "description": "Office visit, new patient, level 3", "category": "Evaluation & Management"}, {"code": "45378", "description": "Colonoscopy, diagnostic", "category": "Gastroenterology"}, {"code": "43239", "description": "Upper GI endoscopy with biopsy", "category": "Gastroenterology"}, {"code": "70450", "description": "CT scan, head without contrast", "category": "Radiology"}, {"code": "70553", "description": "MRI brain with contrast", "category": "Radiology"}, {"code": "71046", "description": "Chest X-ray, 2 views", "category": "Radiology"}, {"code": "80053", "description": "Comprehensive metabolic panel", "category": "Laboratory"}, {"code": "85025", "description": "Complete blood count with differential", "category": "Laboratory"}, {"code": "82947", "description": "Glucose blood test", "category": "Laboratory"}, {"code": "83036", "description": "Hemoglobin A1C", "category": "Laboratory"}, ] @register_provider("cms_pricing") class CMSPricingProvider(BaseProvider): """Provider for Medicare pricing and cost transparency data.""" def __init__(self, client: httpx.AsyncClient): super().__init__("cms_pricing", client) async def initialize(self) -> None: """Initialize CMS pricing provider.""" logger.info("CMS Pricing provider initialized") def get_tools(self) -> List[Callable]: """Return all CMS pricing tools.""" return [ self.cms_get_medicare_fee, self.cms_estimate_procedure_cost, self.cms_search_procedure_codes, self.cms_compare_facility_costs, self.cms_estimate_out_of_pocket, ] @safe_json_return async def cms_get_medicare_fee( self, hcpcs_code: str, locality: str = "National" ) -> Dict[str, Any]: """ Get Medicare fee schedule pricing for a procedure by HCPCS/CPT code. Args: hcpcs_code: HCPCS procedure code (e.g., "99213", "43239") locality: Geographic locality (default: "National") Returns: Medicare reimbursement rates with facility vs non-facility pricing """ result = MEDICARE_FEE_SCHEDULE.get(hcpcs_code, { "description": f"Procedure {hcpcs_code}", "medicare_payment": 150.00, "facility_rate": 150.00, "non_facility_rate": 200.00, "rvu": 3.0, "note": "Sample data - actual rates vary by locality and year" }) result["hcpcs_code"] = hcpcs_code result["locality"] = locality result["year"] = 2024 return result @safe_json_return async def cms_estimate_procedure_cost( self, procedure_name: str, region: str = "National", insurance_type: str = "Medicare" ) -> Dict[str, Any]: """ Estimate typical cost range for a procedure in a geographic region. Args: procedure_name: Common procedure name (e.g., "colonoscopy", "MRI", "blood test") region: Geographic region (default: "National") insurance_type: Type of insurance (Medicare, Commercial, Cash) Returns: Estimated cost range with average, low, and high estimates """ procedure_key = procedure_name.lower() result = PROCEDURE_COST_ESTIMATES.get(procedure_key, { "procedure": procedure_name.title(), "hcpcs_code": "Unknown", "average_cost": 500, "low_estimate": 200, "high_estimate": 1000, "medicare_rate": 150, "factors": ["Procedure complexity", "Facility type", "Geographic location"] }) result["region"] = region result["insurance_type"] = insurance_type result["note"] = "Estimates based on national averages. Actual costs vary significantly by provider and location." # Adjust for insurance type if insurance_type.lower() == "commercial": result["estimated_patient_cost"] = result["average_cost"] elif insurance_type.lower() == "medicare": result["estimated_patient_cost"] = result["medicare_rate"] * 1.2 # 20% coinsurance elif insurance_type.lower() == "cash": result["estimated_patient_cost"] = result["average_cost"] * 0.7 # Cash discount return result @safe_json_return async def cms_search_procedure_codes( self, query: str, limit: int = 10 ) -> Dict[str, Any]: """ Search for HCPCS/CPT codes by procedure description or category. Args: query: Search term (e.g., "office visit", "colonoscopy") limit: Maximum number of results (default: 10) Returns: List of matching procedures with codes, descriptions, and categories """ query_lower = query.lower() matches = [ p for p in PROCEDURE_CODES if query_lower in p["description"].lower() or query_lower in p["category"].lower() ] return { "query": query, "results": matches[:limit], "total": len(matches) } @safe_json_return async def cms_compare_facility_costs( self, procedure_code: str, facilities: List[str] = None ) -> Dict[str, Any]: """ Compare procedure costs across different healthcare facilities. Args: procedure_code: HCPCS/CPT code facilities: List of facility names (uses sample data if not provided) Returns: Cost comparison showing gross charges, negotiated rates, and cash prices """ facilities_data = FACILITY_COSTS.get(procedure_code, [ {"facility": "Sample Facility A", "gross_charge": 2000, "negotiated_rate": 1200, "cash_price": 1000}, {"facility": "Sample Facility B", "gross_charge": 1500, "negotiated_rate": 900, "cash_price": 800}, {"facility": "Sample Facility C", "gross_charge": 1800, "negotiated_rate": 1000, "cash_price": 900}, ]) # Calculate savings cash_prices = [f["cash_price"] for f in facilities_data] lowest_cash = min(cash_prices) highest_cash = max(cash_prices) potential_savings = highest_cash - lowest_cash return { "procedure_code": procedure_code, "facilities": facilities_data, "comparison": { "lowest_cash_price": lowest_cash, "highest_cash_price": highest_cash, "potential_savings": potential_savings, "average_cash_price": sum(cash_prices) / len(cash_prices) }, "note": "Sample data for demonstration. Actual facility prices vary. Patients should request price estimates directly." } @safe_json_return async def cms_estimate_out_of_pocket( self, procedure_code: str, insurance_scenario: str = "Medicare", deductible_met: bool = False ) -> Dict[str, Any]: """ Estimate patient out-of-pocket costs for common insurance scenarios. Args: procedure_code: HCPCS/CPT code insurance_scenario: Insurance type (Medicare, Commercial High Deductible, Commercial PPO, Cash) deductible_met: Whether annual deductible has been met Returns: Estimated patient financial responsibility """ # Get base Medicare cost medicare_data = await self.cms_get_medicare_fee(procedure_code) base_cost = medicare_data.get("medicare_payment", 200) # Define insurance scenarios scenarios = { "Medicare": { "coinsurance": 0.20, "copay": 0, "description": "Medicare Part B (20% coinsurance after deductible)", "deductible": 240 # 2024 Part B deductible }, "Commercial PPO": { "coinsurance": 0.20, "copay": 0, "description": "Commercial PPO (20% coinsurance after deductible)", "deductible": 2000, "multiplier": 2.5 # Commercial pays ~2.5x Medicare }, "Commercial High Deductible": { "coinsurance": 0.30, "copay": 0, "description": "High Deductible Health Plan (30% coinsurance)", "deductible": 5000, "multiplier": 2.5 }, "Cash": { "coinsurance": 0, "copay": 0, "description": "Cash/Self-pay (negotiated rate)", "deductible": 0, "multiplier": 1.8 # Cash discount from commercial } } scenario = scenarios.get(insurance_scenario, scenarios["Medicare"]) # Calculate costs allowed_amount = base_cost * scenario.get("multiplier", 1.0) if deductible_met: patient_responsibility = allowed_amount * scenario["coinsurance"] + scenario["copay"] else: # Patient pays deductible first, then coinsurance on remainder if allowed_amount <= scenario["deductible"]: patient_responsibility = allowed_amount else: remaining = allowed_amount - scenario["deductible"] patient_responsibility = scenario["deductible"] + (remaining * scenario["coinsurance"]) return { "procedure_code": procedure_code, "procedure_description": medicare_data.get("description", "Unknown procedure"), "insurance_scenario": insurance_scenario, "deductible_met": deductible_met, "cost_breakdown": { "allowed_amount": round(allowed_amount, 2), "deductible": scenario["deductible"], "coinsurance_rate": f"{int(scenario['coinsurance'] * 100)}%", "estimated_patient_responsibility": round(patient_responsibility, 2), "insurance_pays": round(allowed_amount - patient_responsibility, 2) }, "scenario_details": scenario["description"], "note": "Estimates only. Actual costs depend on specific plan benefits and negotiated rates." }