virgil-ai/src/services/estimation_service_v2.py
2025-10-01 17:29:49 +03:00

631 lines
No EOL
20 KiB
Python

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, plans: list[Plan]):
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
plan_coverage = self.get_plan_coverage(plans[0])
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,
),
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,
),
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,
),
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,
),
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,
),
results=estimation_results
)