from datetime import date from enum import Enum import json from typing import Optional, Union from src.cache.drug_cache import fetch_drug_with_dosage from src.models import PHQ, Applicant, Plan, EstimationResponse, EstimationDetails, EstimationResult from src.cache.redis_cache import fetch_conditions, get_plan_by_id, search_drug from src.config import settings import httpx from src.services.session_service import session_service class Tier(Enum): TIER_1 = 1 TIER_15 = 1.5 TIER_2 = 2 TIER_25 = 2.5 TIER_3 = 3 TIER_35 = 3.5 TIER_4 = 4 TIER_45 = 4.5 TIER_5 = 5 TIER_55 = 5.5 TIER_6 = 6 TIER_65 = 6.5 TIER_7 = 7 def __lt__(self, other): if isinstance(other, Tier): return self.value < other.value return NotImplemented def __le__(self, other): if isinstance(other, Tier): return self.value <= other.value return NotImplemented def __gt__(self, other): if isinstance(other, Tier): return self.value > other.value return NotImplemented def __ge__(self, other): if isinstance(other, Tier): return self.value >= other.value return NotImplemented BASE_TIERS = { 30: Tier.TIER_1, 40: Tier.TIER_2, 50: Tier.TIER_3, 60: Tier.TIER_35, 200: Tier.TIER_4, } HEIGHT_WEIGHT_TABLE = { (4, 2): (54, 60, 125, 150), (4, 3): (56, 63, 130, 156), (4, 4): (59, 65, 136, 162), (4, 5): (60, 68, 141, 169), (4, 6): (63, 71, 146, 175), (4, 7): (65, 73, 152, 182), (4, 8): (67, 76, 157, 188), (4, 9): (70, 79, 163, 195), (4, 10): (72, 81, 168, 202), (4, 11): (75, 84, 174, 209), (5, 0): (77, 87, 180, 216), (5, 1): (80, 90, 186, 223), (5, 2): (83, 93, 192, 230), (5, 3): (85, 96, 199, 238), (5, 4): (88, 99, 205, 245), (5, 5): (91, 102, 211, 253), (5, 6): (93, 105, 218, 261), (5, 7): (96, 109, 224, 269), (5, 8): (99, 112, 231, 277), (5, 9): (102, 115, 238, 285), (5, 10): (105, 118, 245, 293), (5, 11): (108, 122, 252, 302), (6, 0): (111, 125, 259, 310), (6, 1): (114, 129, 266, 319), (6, 2): (117, 132, 274, 327), (6, 3): (121, 136, 281, 336), (6, 4): (124, 140, 289, 345), (6, 5): (127, 143, 296, 354), (6, 6): (130, 147, 304, 364), (6, 7): (134, 151, 312, 374), (6, 8): (137, 155, 320, 383), (6, 9): (140, 159, 328, 393), (6, 10): (144, 163, 336, 403), (6, 11): (147, 167, 344, 413), (7, 0): (151, 171, 352, 423), (7, 1): (155, 175, 361, 433), (7, 2): (158, 179, 369, 443), (7, 3): (162, 183, 378, 453), (7, 4): (166, 187, 387, 464), } UNINSURABLE_MEDICATIONS = [ "abacavir", "abarelix", "abciximab", "acamprosate", "adalimumab", "al-721", "albuterol/ipratropium", "alemtuzumab", "alteplase", "altretarnine", "amantadine", "ambrisentan", "amiodarone", "anakinra", "anastrozole", "apomorphine", "aripripazole", "asparaginase", "atazanavir", "auranofin", "aurothiogiucose", "aurothiomalate", "azathioprine", "bog", "becaplermin", "benztropine", "bevacizumab", "bicalutamide", "biperiden hydrochloride", "bleomycin", "bromocriptine", "busulfan", "capecitabine", "carbidopa", "carbidopa/levodopa", "carboplatin", "chlorambucil", "chlorotrianisene", "chlorpromazine", "cilostazol", "cinacalcet", "cisplatin", "cladribine", "clozapine", "cyclophosphamide", "cycloserine", "cyclosporine", "dalteparin", "dantrolene", "darunavir", "delavirdine", "didanosine", "dipyridamole", "dipyridamolekspirin", "disuifram", "donepezil", "doxorubicin", "dronabinol", "efavirenz", "emtricita bine", "emtricitabine", "emtricitabine/tenofovir", "enfuvirtide", "enoxaparin", "entacapone", "entacapone/levodopa/carbidopa", "epoetin alfa", "eptifibatide", "ergoloid mesylates", "estramustine", "etanercept", "ethinyl estradiol", "ethopropazine", "etoposide", "exemestane", "filgrastim", "fecainide", "fluorourcil", "fluphenazine", "flutamide", "fondaparinux", "fosamprenavir", "foscarnet sodium", "fulvestrant", "galantamine", "glatiramer", "gold sodium thiomalate", "goserelin", "haloperidol", "heparin", "hydroxyurea", "imatinib", "indinavir", "infliximab", "insulin > 50 units per day", "interferon", "interferon alfa-2a", "interferon beta 1a", "interferon beta lb", "ipratropiurn", "isoniazid", "lamivudine", "lamivudine/zidovudine/abacavir", "letrozole", "leucovorin", "leuprolide", "levamisole hydrochloride", "levodopa", "lomustine", "lopinavir", "loxapine", "maraviroc", "medroxyprogesterone acetate", "megestrol", "melphalan", "memantine", "methadone", "methotrexate", "mitomycin", "mitoxantrone", "morphine", "mycophenolate", "naltrexone", "natalizumab", "nebulizer device", "nelfinavir", "neostigmine", "nesiritide", "nevirapine", "nilutamide", "nitroglycerine", "nitroglycerine transdermal", "olanzapine", "ondansetron", "oxygen", "paliperidone", "penicillamine", "pergolide mesylate", "perphenazine", "pimozide", "pramipexole", "procainamide", "prochlorperazine", "procyclidine", "pyridostigmine", "quetiapine", "quinidine", "rasagiline", "riluzole", "risperidone", "ritonavir", "rituximab", "rivastig mine", "ropinirole", "rotigotine", "saquinavir", "selegiline", "sotalol", "stavudine", "streptozocin", "tacrine", "tacrolimus", "tamoxifen", "tenofovir", "testolactone", "thioridazine", "thiotepa", "thiothixene", "ticlopidine", "tiotropium", "tipranavir", "tirofiban", "tolcapone", "toremifene", "trastuzumab", "treprosinil", "trifluoperazine", "trihexyphenidyl", "triptorelin", "valganiciclovir", "vincristine", "warfarin", "zalcitabine", "zidovudine", "ziprasidone", "zoledronic acid" ] UNINSURABLE_ISSUES = [ "alcohol", "alzheimer_disease", "chronic_pulmonary_disorder", "copd", "hepatitis", "liver_disorder", "myasthenia_gravis", "organ_transplant", "other_cognitive", "nebulizer", "aids", "als", "amputation", "cancer", "cirrhosis", "emphysema", "multiple_sclerosis", "nervous_disorder", "osteoarthritis", "parkinson_disease", "senile_dementia", "systemic_lupus", ] class EstimationService: def __init__(self): self.base_url = settings.TALESTORM_API_BASE_URL self.api_key = settings.TALESTORM_API_KEY async def get_client(self) -> httpx.AsyncClient: """Get HTTP client for talestorm-ai API""" headers = {} if self.api_key: headers["X-API-Key"] = self.api_key return httpx.AsyncClient( base_url=self.base_url, headers=headers, timeout=httpx.Timeout(60.0, connect=10.0) # 30s total timeout, 10s connect timeout ) async def check_dtq_ai(self, phq: PHQ) -> tuple[bool, str]: try: client = await self.get_client() session_id = await session_service.create_session(agent_id=settings.TALESTORM_DTQ_AGENT_ID) response = await client.post( "/chat/", json={"chat_session_id": session_id, "user_message": f"Check if this applicant eligible for insurance:\n {phq.model_dump_json()}"}, ) response_json = response.json()['message'] print(f"AI DTQ Response: {response_json}") ai_response = json.loads(response_json) dtq = not ai_response["insurable"] reason = "[AI] " + ai_response["reason"] return dtq, reason except Exception as e: print(f"Error in AI DTQ check: {e}") # Fallback to DTQ on error return True, f"Error on AI service: {e}" def check_dtq(self, phq: PHQ) -> tuple[bool, str]: for medication in phq.medications: if medication.name.lower() in UNINSURABLE_MEDICATIONS: return True, f"Uninsurable medication: {medication.name}" uninsurable_conditions_response = fetch_conditions() uninsurable_conditions = [condition["key"] for condition in uninsurable_conditions_response] if phq.pregnancy: return True, "Applicant is pregnant" for issue in phq.issues: if issue.key in UNINSURABLE_ISSUES: return True, f"Uninsurable issue: {issue.key}" if issue.key == "surgery": for detail in issue.details: if detail.key == "not_performed": return True, "Applicant have surgery that was not performed" for condition in phq.conditions: if condition.key in uninsurable_conditions: return True, f"Applicant have uninsurable condition: {condition.key}" # issues would be partially checked by AI return False, "" def calculate_age(self, born): today = date.today() return today.year - born.year - ((today.month, today.day) < (born.month, born.day)) async def calculate_rx_spend(self, phq: PHQ, applicant_id: int) -> tuple[float, list[str]]: rx_spend = 0 review_reasons = [] for medication in phq.medications: if medication.applicant != applicant_id: continue try: drug_name = medication.name.lower() drug_url = search_drug(drug_name) if not drug_url: review_reasons.append(f"Drug not found: {drug_name}") continue drug_dosage = float(medication.dosage) drug_price = await fetch_drug_with_dosage(drug_url, drug_dosage) if not drug_price: review_reasons.append(f"Dosage {drug_dosage} for drug {drug_name} not found") continue if medication.frequency in ["Once daily", "At bedtime"]: month_times = 30 elif medication.frequency == "Twice daily": month_times = 60 elif medication.frequency in ["Three times daily", "After meals", "Before meals"]: month_times = 90 elif medication.frequency == "Four times daily": month_times = 120 elif medication.frequency in ["Weekly", "Once weekly"]: month_times = 4 elif medication.frequency in ["Monthly", "One-time dose"]: month_times = 1 elif medication.frequency == "Every other day": month_times = 15 elif medication.frequency == "Three times per week": month_times = 12 elif medication.frequency == "Every 2 weeks": month_times = 2 elif medication.frequency == "Every X hours": try: freq = int(medication.frequencyDescription) month_times = (24/freq) * 30 except: month_times = 1 review_reasons.append(f"Unclear frequency of drug {medication.name}: Every X hours; X is not defined.") else: review_reasons.append(f"Unclear frequency of drug {medication.name}: {medication.frequency}") month_times = 1 rx_spend += drug_price.unit_price * month_times except Exception as e: print(f"Error calculating rx spend for {medication.name}: {e}") pass return rx_spend, review_reasons def get_tier(self, coverage: int, rx_spend: float) -> tuple[Tier | None, str | None]: tiers = [ Tier.TIER_1, Tier.TIER_15, Tier.TIER_2, Tier.TIER_25, Tier.TIER_3, Tier.TIER_35, Tier.TIER_4, Tier.TIER_45, Tier.TIER_5, Tier.TIER_55, Tier.TIER_6, Tier.TIER_7 ] rates = { 1: [122, 127, 133, 139, 145, 151, 158, 164, 172, 182, 196, 222], 2: [243, 253, 265, 277, 290, 302, 315, 327, 343, 364, 390, 444], 3: [219, 228, 239, 249, 261, 272, 283, 295, 309, 327, 351, 400], 4: [365, 379, 397, 415, 435, 452, 472, 491, 514, 545, 585, 666], } if coverage not in rates: raise ValueError(f"Unsupported coverage type: {coverage}") for i, threshold in enumerate(rates[coverage]): if rx_spend <= threshold: return tiers[i], None return None, f"Rx spend ({rx_spend}) is greater than {rates[coverage][-1]} for coverage {coverage}" def get_plan_coverage(self, plan: Plan): plan_data = get_plan_by_id(plan.id) prices = plan_data["prices"] for price in prices: if price["id"] == plan.priceId: return price["coverage"] return 0 def get_plan_price(self, plan: Plan, tier: Tier, coverage: int): plan_data = get_plan_by_id(plan.id) prices = plan_data["prices"] for price in prices: if price["coverage"] == coverage and price["tier"] == f"Tier {tier.value}": return price["id"] return 0 def calculate_bmi(self, weight, ft, inch): try: inches = ft * 12 + inch return (weight / (inches ** 2)) * 703 except: return None async def estimate_insurance(self, applicants: list[Applicant], phq: PHQ, plan_coverage: int): estimation_results = [] is_review = False review_reasons = [] dtq_reasons = [] accept_reasons = [] is_dtq, reason = self.check_dtq(phq) if not is_dtq: is_dtq, reason = await self.check_dtq_ai(phq) if is_dtq: dtq_reasons.append(reason) max_age = max(self.calculate_age(applicant.dob) for applicant in applicants) base_tier = None for base_age, tier in BASE_TIERS.items(): if max_age <= base_age: base_tier = tier break rx_spend = 0 for applicant_id, applicant in enumerate(applicants): applicant_review_reasons = [] applicant_dtq_reasons = [] applicant_accept_reasons = [] applicant_age = self.calculate_age(applicant.dob) applicant_tier = None for base_age, tier in BASE_TIERS.items(): if applicant_age <= base_age: applicant_tier = tier break if applicant_age >= 65: is_dtq = True reason = f"Applicant age {applicant_age} is over 65" dtq_reasons.append(reason) applicant_dtq_reasons.append(reason) if (applicant.heightFt, applicant.heightIn) in HEIGHT_WEIGHT_TABLE: w1, w2, w3, w4 = HEIGHT_WEIGHT_TABLE[(applicant.heightFt, applicant.heightIn)] if applicant.weight < w1: is_dtq = True reason = f"Applicant weight {applicant.weight}lbs is less than {w1}, at height {applicant.heightFt}' {applicant.heightIn}''" dtq_reasons.append(reason) applicant_dtq_reasons.append(reason) elif w1 <= applicant.weight < w2: is_review = True weight_reason = f"Applicant weight {applicant.weight}lbs is less than {w2}, at height {applicant.heightFt}' {applicant.heightIn}''" review_reasons.append(weight_reason) applicant_review_reasons.append(weight_reason) elif w3 <= applicant.weight < w4: is_review = True weight_reason = f"Applicant weight {applicant.weight}lbs is greater than {w3-1}, at height {applicant.heightFt}' {applicant.heightIn}''" review_reasons.append(weight_reason) applicant_review_reasons.append(weight_reason) elif applicant.weight >= w4: is_dtq = True reason = f"Applicant weight {applicant.weight}lbs is greater than {w4-1}, at height {applicant.heightFt}' {applicant.heightIn}''" dtq_reasons.append(reason) applicant_dtq_reasons.append(reason) rx_spend_applicant, rx_review_reasons = await self.calculate_rx_spend(phq, applicant_id) if rx_review_reasons: is_review = True review_reasons += rx_review_reasons rx_spend += rx_spend_applicant applicant_new_tier, tier_reason = self.get_tier(plan_coverage, rx_spend_applicant) if applicant_new_tier is None: is_dtq = True dtq_reasons.append(tier_reason) applicant_dtq_reasons.append(tier_reason) if applicant_new_tier and applicant_new_tier > applicant_tier: applicant_tier = applicant_new_tier reason = f"Rx spend increased tier to {applicant_new_tier}." accept_reasons.append(reason) applicant_accept_reasons.append(reason) if applicant_dtq_reasons: final_reason = "\n".join(applicant_dtq_reasons) elif applicant_review_reasons: final_reason = "\n".join(applicant_review_reasons) elif applicant_accept_reasons: final_reason = "\n".join(applicant_accept_reasons) else: final_reason = f"Tier {applicant_tier.value} assigned with Rx spend within allowed limits." estimation_results.append( EstimationResult( name=applicant.firstName, applicant_type="", age=applicant_age, bmi=self.calculate_bmi(applicant.weight, applicant.heightFt, applicant.heightIn), tier=applicant_tier.value, rx_spend=rx_spend_applicant, message=final_reason, ) ) # plan_price_id = self.get_plan_price(plans[0], base_tier, plan_coverage) if is_dtq: reason = "\n".join(dtq_reasons) return EstimationResponse( status="rejected", details=EstimationDetails( dtq=is_dtq, reason=reason, # price_id=plan_price_id, tier=f"Tier {base_tier.value}", ), # results=estimation_results ) if is_review: reason = "\n".join(review_reasons) return EstimationResponse( status="human_review", details=EstimationDetails( dtq=is_dtq, reason=reason, # price_id=plan_price_id, tier=f"Tier {base_tier.value}", ), # results=estimation_results ) new_tier, tier_reason = self.get_tier(plan_coverage, rx_spend) if new_tier is None: dtq_reasons.append(tier_reason) reason = "\n".join(dtq_reasons) return EstimationResponse( status="rejected", details=EstimationDetails( dtq=True, reason=reason, # price_id=plan_price_id, tier=f"Tier {base_tier.value}", ), # results=estimation_results ) if new_tier > base_tier: base_tier = new_tier # plan_price_id = self.get_plan_price(plans[0], base_tier, plan_coverage) if base_tier is not None: reason = "\n".join(accept_reasons) return EstimationResponse( status="accepted", details=EstimationDetails( dtq=is_dtq, reason=reason, # price_id=plan_price_id, tier=f"Tier {base_tier.value}", ), # results=estimation_results ) else: reason = "\n".join(dtq_reasons) return EstimationResponse( status="rejected", details=EstimationDetails( dtq=is_dtq, reason=reason, # price_id=plan_price_id, tier=f"Tier {base_tier.value}", ), # results=estimation_results )