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_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]: 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" # TODO: Add other DTQ conditions 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 )