File: /home/clawbot-insurance/Skills/claims/reserve-calculator/reserve_calculator.py
#!/usr/bin/env python3"""Reserve Calculator SkillCalculates claim reserves based on analysis"""
import jsonfrom datetime import datetimefrom pathlib import Pathfrom typing import Dict
class ReserveCalculator: def __init__(self, insurer_id: str): self.insurer_id = insurer_id self.memory_dir = Path(f"/home/clawbot-insurance/Memory") def calculate_reserve(self, claim_id: str) -> Dict: """Calculate recommended reserve for claim""" # Load claim data claim_file = self.memory_dir / f"Claims/{claim_id}/metadata.json" with open(claim_file) as f: claim = json.load(f) # Load policy coverage policy = self._lookup_policy(claim["policy_number"]) # Load document analysis analysis = self._load_document_analysis(claim_id) # Base reserve calculation by claim type claim_type = claim.get("claim_type", "auto_collision") if claim_type == "auto_collision": reserve = self._calculate_auto_reserve(claim, policy, analysis) elif claim_type == "property_damage": reserve = self._calculate_property_reserve(claim, policy, analysis) elif claim_type == "liability": reserve = self._calculate_liability_reserve(claim, policy, analysis) elif claim_type == "workers_comp": reserve = self._calculate_wc_reserve(claim, policy, analysis) else: reserve = self._calculate_generic_reserve(claim, policy) # Apply coverage limits reserve = self._apply_coverage_limits(reserve, policy) # Save reserve reserve_record = { "claim_id": claim_id, "calculated_at": datetime.now().isoformat(), "reserve": reserve, "methodology": reserve.get("methodology", "standard"), "confidence": reserve.get("confidence", "medium") } reserve_dir = self.memory_dir / f"Claims/{claim_id}/reserve" reserve_dir.mkdir(parents=True, exist_ok=True) with open(reserve_dir / "calculation.json", 'w') as f: json.dump(reserve_record, f, indent=2) return reserve_record def _calculate_auto_reserve(self, claim: Dict, policy: Dict, analysis: Dict) -> Dict: """Calculate auto claim reserve""" reserve = { "total": 0, "breakdown": {}, "methodology": "auto_collision_standard" } # Property damage if claim.get("intake_data", {}).get("vehicle_damage"): vehicle_damage = self._parse_amount(claim["intake_data"]["vehicle_damage"]) reserve["breakdown"]["property_damage"] = min(vehicle_damage, 25000) reserve["total"] += reserve["breakdown"]["property_damage"] # Injury reserve if claim.get("intake_data", {}).get("injuries_reported") == "yes": # Base injury reserve injury_reserve = 15000 # Adjust based on medical treatment mentioned treatment = claim.get("intake_data", {}).get("medical_treatment", "") if "hospital" in treatment.lower(): injury_reserve = 50000 elif "emergency room" in treatment.lower(): injury_reserve = 25000 reserve["breakdown"]["bodily_injury"] = injury_reserve reserve["total"] += injury_reserve # Rental car reserve["breakdown"]["rental_reimbursement"] = 500 reserve["total"] += 500 # Add 10% for unknowns reserve["breakdown"]["unknown"] = int(reserve["total"] * 0.10) reserve["total"] += reserve["breakdown"]["unknown"] reserve["confidence"] = "medium" return reserve def _calculate_property_reserve(self, claim: Dict, policy: Dict, analysis: Dict) -> Dict: """Calculate property damage reserve""" reserve = { "total": 0, "breakdown": {}, "methodology": "property_damage_standard" } # Base on reported damage if claim.get("intake_data", {}).get("damage_estimate"): estimate = self._parse_amount(claim["intake_data"]["damage_estimate"]) reserve["breakdown"]["dwelling_damage"] = estimate reserve["total"] += estimate # Contents if claim.get("intake_data", {}).get("contents_damaged") == "yes": reserve["breakdown"]["contents"] = 10000 reserve["total"] += 10000 # Additional living expenses reserve["breakdown"]["ale"] = 5000 reserve["total"] += 5000 reserve["confidence"] = "medium" return reserve def _calculate_liability_reserve(self, claim: Dict, policy: Dict, analysis: Dict) -> Dict: """Calculate liability claim reserve""" reserve = { "total": 0, "breakdown": {}, "methodology": "liability_standard" } # Base liability reserve base_reserve = 25000 # Adjust based on injuries claimed injuries = claim.get("intake_data", {}).get("injuries_claimed", "") if "serious" in injuries.lower() or "permanent" in injuries.lower(): base_reserve = 100000 reserve["breakdown"]["indemnity"] = base_reserve reserve["breakdown"]["defense"] = int(base_reserve * 0.20) # 20% for defense reserve["total"] = reserve["breakdown"]["indemnity"] + reserve["breakdown"]["defense"] reserve["confidence"] = "low" # Liability is uncertain return reserve def _calculate_wc_reserve(self, claim: Dict, policy: Dict, analysis: Dict) -> Dict: """Calculate workers comp reserve""" reserve = { "total": 0, "breakdown": {}, "methodology": "workers_comp_standard" } # Medical reserve["breakdown"]["medical"] = 15000 # Indemnity (lost wages) reserve["breakdown"]["indemnity"] = 10000 # Expense reserve["breakdown"]["expense"] = 2000 reserve["total"] = sum(reserve["breakdown"].values()) reserve["confidence"] = "medium" return reserve def _calculate_generic_reserve(self, claim: Dict, policy: Dict) -> Dict: """Generic reserve calculation""" return { "total": 10000, "breakdown": {"general": 10000}, "methodology": "generic", "confidence": "low" } def _apply_coverage_limits(self, reserve: Dict, policy: Dict) -> Dict: """Apply policy coverage limits to reserve""" # Get coverage limit limit = policy.get("coverage_limit", 100000) if reserve["total"] > limit: reserve["total"] = limit reserve["capped_at_limit"] = True return reserve def _load_document_analysis(self, claim_id: str) -> Dict: """Load document analysis if available""" analysis_file = self.memory_dir / f"Claims/{claim_id}/analysis/document-analysis.json" if analysis_file.exists(): with open(analysis_file) as f: return json.load(f) return {} def _lookup_policy(self, policy_number: str) -> Dict: """Look up policy details""" policies_dir = self.memory_dir / "Policies" for policy_dir in policies_dir.iterdir(): if policy_dir.is_dir(): policy_file = policy_dir / "policy-data.json" if policy_file.exists(): with open(policy_file) as f: policy = json.load(f) if policy.get("policy_number") == policy_number: return policy return {"coverage_limit": 100000} def _parse_amount(self, amount_str: str) -> int: """Parse amount string to integer""" # Remove non-numeric characters except decimal cleaned = ''.join(c for c in amount_str if c.isdigit() or c == '.') try: return int(float(cleaned)) except: return 0
# CLI interfaceif __name__ == "__main__": import sys insurer_id = sys.argv[1] claim_id = sys.argv[2] calculator = ReserveCalculator(insurer_id) result = calculator.calculate_reserve(claim_id) print(json.dumps(result, indent=2))