317 lines
No EOL
10 KiB
Python
317 lines
No EOL
10 KiB
Python
from datetime import date
|
|
from enum import Enum
|
|
from typing import Optional
|
|
from src.models import PHQ, Applicant, Plan, EstimationResponse, EstimationDetails, EstimationResult
|
|
from src.cache.redis_cache import fetch_conditions, fetch_drug, 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): (60, 124),
|
|
(4, 3): (63, 129),
|
|
(4, 4): (65, 135),
|
|
(4, 5): (68, 140),
|
|
(4, 6): (71, 145),
|
|
(4, 7): (73, 151),
|
|
(4, 8): (76, 156),
|
|
(4, 9): (79, 162),
|
|
(4, 10): (81, 167),
|
|
(4, 11): (84, 173),
|
|
(5, 0): (87, 179),
|
|
(5, 1): (90, 185),
|
|
(5, 2): (93, 191),
|
|
(5, 3): (96, 197),
|
|
(5, 4): (99, 204),
|
|
(5, 5): (102, 210),
|
|
(5, 6): (105, 217),
|
|
(5, 7): (109, 223),
|
|
(5, 8): (112, 230),
|
|
(5, 9): (115, 237),
|
|
(5, 10): (118, 244),
|
|
(5, 11): (122, 251),
|
|
(6, 0): (125, 258),
|
|
(6, 1): (129, 265),
|
|
(6, 2): (132, 273),
|
|
(6, 3): (136, 280),
|
|
(6, 4): (140, 288),
|
|
(6, 5): (143, 295),
|
|
(6, 6): (147, 303),
|
|
(6, 7): (151, 311),
|
|
(6, 8): (155, 319),
|
|
(6, 9): (159, 327),
|
|
(6, 10): (163, 335),
|
|
(6, 11): (167, 343),
|
|
(7, 0): (171, 351),
|
|
(7, 1): (175, 360),
|
|
(7, 2): (179, 368),
|
|
(7, 3): (183, 377),
|
|
(7, 4): (187, 386),
|
|
}
|
|
|
|
|
|
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()
|
|
print(f"AI DTQ Response: {response_json}")
|
|
|
|
if "insurable" in response_json:
|
|
dtq = not response_json["message"]["insurable"]
|
|
reason = response_json["message"]["reason"]
|
|
else:
|
|
dtq = False
|
|
reason = ""
|
|
return dtq, reason
|
|
except Exception as e:
|
|
print(f"Error in AI DTQ check: {e}")
|
|
# Fallback to not DTQ on error
|
|
return False, ""
|
|
|
|
def check_dtq(self, phq: PHQ) -> tuple[bool, str]:
|
|
uninsurable_conditions_response = fetch_conditions()
|
|
uninsurable_conditions = [condition["key"] for condition in uninsurable_conditions_response]
|
|
|
|
if phq.pregnancy:
|
|
return True, "Pregnancy"
|
|
|
|
for issue in phq.issues:
|
|
if issue.key == "surgery":
|
|
for detail in issue.details:
|
|
if detail.key == "not_performed":
|
|
return True, "Surgery that was not performed"
|
|
|
|
for condition in phq.conditions:
|
|
if condition.key in uninsurable_conditions:
|
|
return True, f"Condition: {condition.key}"
|
|
|
|
# issues would be checked by AI
|
|
return False, None
|
|
|
|
def calculate_age(self, born):
|
|
today = date.today()
|
|
return today.year - born.year - ((today.month, today.day) < (born.month, born.day))
|
|
|
|
def calculate_rx_spend(self, phq: PHQ, applicant_id: int) -> float:
|
|
rx_spend = 0
|
|
for medication in phq.medications:
|
|
if medication.applicant != applicant_id:
|
|
continue
|
|
try:
|
|
drug_name = medication.name
|
|
drug_url = search_drug(drug_name)
|
|
drug_price = fetch_drug(drug_url)
|
|
rx_spend += drug_price.prices[0].dosages[0].price
|
|
except Exception:
|
|
pass
|
|
return rx_spend
|
|
|
|
def get_tier(self, coverage: int, rx_spend: float) -> Optional[Tier]:
|
|
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_65, Tier.TIER_7
|
|
]
|
|
|
|
rates = {
|
|
1: [122, 127, 133, 139, 145, 151, 158, 164, 172, 182, 196, 209, 222],
|
|
2: [243, 253, 265, 277, 290, 302, 315, 327, 343, 364, 390, 417, 444],
|
|
3: [219, 228, 239, 249, 261, 272, 283, 295, 309, 327, 351, 376, 400],
|
|
4: [365, 379, 397, 415, 435, 452, 472, 491, 514, 545, 585, 626, 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]
|
|
|
|
return None
|
|
|
|
def get_plan_price(self, plan: Plan, tier: Tier, coverage: int):
|
|
plan_data = get_plan_by_id(plan.id)
|
|
prices = plan_data["prices"]
|
|
tier_str = f"Tier {tier.value}"
|
|
for price in prices:
|
|
if price["tier"] == tier_str and price["coverage"] == coverage:
|
|
return price["price"]
|
|
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_dtq, reason = self.check_dtq(phq)
|
|
if not is_dtq:
|
|
is_dtq, reason = await self.check_dtq_ai(phq)
|
|
|
|
max_age = max(self.calculate_age(applicant.dob) for applicant in applicants)
|
|
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_age = self.calculate_age(applicant.dob)
|
|
for base_age, tier in BASE_TIERS.items():
|
|
if applicant_age <= base_age:
|
|
applicant_tier = tier
|
|
break
|
|
|
|
weight_min, weight_max = HEIGHT_WEIGHT_TABLE[(applicant.heightFt, applicant.heightIn)]
|
|
if applicant.weight < weight_min:
|
|
is_dtq = True
|
|
reason = "Declined due to low BMI of one or more applicants"
|
|
elif applicant.weight > weight_max:
|
|
is_dtq = True
|
|
reason = "Declined due to high BMI of one or more applicants"
|
|
|
|
rx_spend_applicant = self.calculate_rx_spend(phq, applicant_id)
|
|
rx_spend += rx_spend_applicant
|
|
|
|
applicant_new_tier = self.get_tier(plans[0].coverage, rx_spend_applicant)
|
|
if applicant_new_tier is None:
|
|
is_dtq = True
|
|
reason = "Declined due to high Rx spend"
|
|
break
|
|
|
|
if applicant_new_tier > applicant_tier:
|
|
applicant_tier = applicant_new_tier
|
|
reason = f"Rx spend increased tier to {applicant_new_tier}."
|
|
|
|
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=reason if reason else f"Tier {applicant_tier} assigned with Rx spend within allowed limits."
|
|
)
|
|
)
|
|
|
|
plan_price = self.get_plan_price(plans[0], base_tier, plans[0].coverage)
|
|
|
|
if is_dtq:
|
|
return EstimationResponse(
|
|
status="rejected",
|
|
details=EstimationDetails(
|
|
dtq=is_dtq,
|
|
reason=reason,
|
|
tier=base_tier.value,
|
|
total_price=plan_price
|
|
),
|
|
results=estimation_results
|
|
)
|
|
|
|
new_tier = self.get_tier(plans[0].coverage, rx_spend)
|
|
|
|
if new_tier is None:
|
|
return EstimationResponse(
|
|
status="rejected",
|
|
details=EstimationDetails(
|
|
dtq=True,
|
|
reason="Declined due to high Rx spend",
|
|
tier=base_tier.value,
|
|
total_price=plan_price
|
|
),
|
|
results=estimation_results
|
|
)
|
|
|
|
if new_tier > base_tier:
|
|
base_tier = new_tier
|
|
|
|
plan_price = self.get_plan_price(plans[0], base_tier, plans[0].coverage)
|
|
|
|
status = "accepted" if base_tier is not None else "rejected"
|
|
|
|
return EstimationResponse(
|
|
status=status,
|
|
details=EstimationDetails(
|
|
dtq=is_dtq,
|
|
reason=reason,
|
|
tier=base_tier.value,
|
|
total_price=plan_price
|
|
),
|
|
results=estimation_results
|
|
)
|
|
|
|
|
|
|
|
|
|
|