AI-Powered Insurance Claims Processing

1from typing import List, Dict
2import cv2
3import pytesseract
4from PIL import Image
5
6class DocumentProcessor:
7    def __init__(self):
8        self.ocr = pytesseract
9        self.vision_model = load_vision_model()
10        
11    async def process_claim_document(
12        self,
13        document: bytes,
14        document_type: str
15    ) -> Dict:
16        """Process claim document and extract structured data"""
17        
18        # Image preprocessing
19        image = self.preprocess_image(document)
20        
21        # Classify document type if unknown
22        if document_type == "unknown":
23            document_type = await self.classify_document(image)
24        
25        # Extract text using OCR
26        text = self.ocr.image_to_string(image)
27        
28        # Extract structured data based on document type
29        if document_type == "police_report":
30            data = self.extract_police_report(text, image)
31        elif document_type == "medical_report":
32            data = self.extract_medical_report(text, image)
33        elif document_type == "damage_photo":
34            data = await self.analyze_damage_photo(image)
35        else:
36            data = self.extract_generic(text)
37        
38        return {
39            "document_type": document_type,
40            "extracted_data": data,
41            "confidence": self.calculate_confidence(data),
42            "requires_review": data.get("confidence", 0) < 0.85
43        }
44    
45    async def analyze_damage_photo(self, image: np.ndarray) -> Dict:
46        """Analyze damage in photos using computer vision"""
47        
48        # Detect vehicle/property
49        detections = self.vision_model.detect(image)
50        
51        # Assess damage severity
52        damage_score = self.assess_damage(detections)
53        
54        # Estimate repair cost
55        estimated_cost = self.estimate_cost(detections, damage_score)
56        
57        return {
58            "damage_type": detections.get("damage_type"),
59            "severity": damage_score,
60            "estimated_cost": estimated_cost,
61            "confidence": detections.get("confidence"),
62            "requires_expert": damage_score > 0.7
63        }
64

1from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
2from sklearn.neural_network import MLPClassifier
3import numpy as np
4
5class FraudDetector:
6    def __init__(self):
7        # Ensemble of models
8        self.models = {
9            'random_forest': RandomForestClassifier(n_estimators=100),
10            'gradient_boost': GradientBoostingClassifier(n_estimators=100),
11            'neural_net': MLPClassifier(hidden_layers=(100, 50))
12        }
13        self.feature_extractor = FeatureExtractor()
14        
15    def extract_features(self, claim: Dict) -> np.ndarray:
16        """Extract features for fraud detection"""
17        
18        features = []
19        
20        # Temporal features
21        features.extend([
22            claim['time_to_report'],  # Time from incident to report
23            claim['day_of_week'],
24            claim['hour_of_day'],
25            claim['days_since_policy_start'],
26        ])
27        
28        # Claim features
29        features.extend([
30            claim['claim_amount'],
31            claim['claim_amount'] / claim['policy_limit'],  # Ratio
32            len(claim['documents']),
33            claim['has_police_report'],
34            claim['has_witnesses'],
35        ])
36        
37        # Historical features
38        history = self.get_claim_history(claim['policyholder_id'])
39        features.extend([
40            len(history),  # Number of previous claims
41            sum(c['amount'] for c in history),  # Total claimed
42            max([c['amount'] for c in history]) if history else 0,
43            self.claim_frequency(history),  # Claims per year
44        ])
45        
46        # Network features
47        network = self.analyze_network(claim)
48        features.extend([
49            network['connected_claimants'],
50            network['shared_addresses'],
51            network['shared_contacts'],
52        ])
53        
54        # Document features
55        features.extend([
56            claim['document_quality_score'],
57            claim['has_metadata_anomalies'],
58            claim['has_image_manipulation'],
59        ])
60        
61        return np.array(features)
62    
63    async def predict_fraud(self, claim: Dict) -> Dict:
64        """Predict fraud probability using ensemble"""
65        
66        features = self.extract_features(claim)
67        
68        # Get predictions from all models
69        predictions = {}
70        for name, model in self.models.items():
71            prob = model.predict_proba(features.reshape(1, -1))[0][1]
72            predictions[name] = prob
73        
74        # Ensemble: weighted average
75        weights = {'random_forest': 0.4, 'gradient_boost': 0.4, 'neural_net': 0.2}
76        fraud_score = sum(predictions[m] * weights[m] for m in predictions)
77        
78        # Explain prediction
79        explanation = self.explain_prediction(features, fraud_score)
80        
81        return {
82            'fraud_score': fraud_score,
83            'is_suspicious': fraud_score > 0.7,
84            'requires_investigation': fraud_score > 0.5,
85            'explanation': explanation,
86            'model_predictions': predictions
87        }
88    
89    def explain_prediction(
90        self,
91        features: np.ndarray,
92        fraud_score: float
93    ) -> List[str]:
94        """Generate human-readable explanation"""
95        
96        explanations = []
97        
98        # Feature importance from random forest
99        importances = self.models['random_forest'].feature_importances_
100        top_features = np.argsort(importances)[-5:]
101        
102        feature_names = [
103            'time_to_report', 'day_of_week', 'claim_amount',
104            'previous_claims', 'network_connections', 'document_quality'
105        ]
106        
107        for idx in top_features:
108            feature_name = feature_names[idx]
109            feature_value = features[idx]
110            
111            if feature_name == 'time_to_report' and feature_value < 1:
112                explanations.append("Claim reported suspiciously quickly")
113            elif feature_name == 'previous_claims' and feature_value > 3:
114                explanations.append(f"High claim frequency: {int(feature_value)} previous claims")
115            elif feature_name == 'network_connections' and feature_value > 2:
116                explanations.append("Connected to other suspicious claimants")
117        
118        return explanations
119

1from typing import Optional
2from decimal import Decimal
3
4class ClaimsDecisionEngine:
5    def __init__(self):
6        self.fraud_detector = FraudDetector()
7        self.policy_validator = PolicyValidator()
8        self.cost_estimator = CostEstimator()
9        
10    async def evaluate_claim(self, claim: Dict) -> Decision:
11        """Evaluate claim and make decision"""
12        
13        # Step 1: Policy validation
14        policy_valid = await self.policy_validator.validate(
15            claim['policy_id'],
16            claim['incident_date']
17        )
18        
19        if not policy_valid.is_valid:
20            return Decision(
21                decision='DENIED',
22                reason=policy_valid.reason,
23                requires_review=False
24            )
25        
26        # Step 2: Fraud check
27        fraud_check = await self.fraud_detector.predict_fraud(claim)
28        
29        if fraud_check['fraud_score'] > 0.7:
30            return Decision(
31                decision='INVESTIGATION',
32                reason='High fraud risk',
33                fraud_score=fraud_check['fraud_score'],
34                requires_review=True
35            )
36        
37        # Step 3: Cost estimation
38        estimated_cost = await self.cost_estimator.estimate(claim)
39        claimed_amount = Decimal(str(claim['amount']))
40        
41        # Check if estimate matches claim
42        difference_ratio = abs(estimated_cost - claimed_amount) / estimated_cost
43        
44        if difference_ratio > 0.3:  # 30% difference
45            return Decision(
46                decision='REVIEW',
47                reason='Claim amount differs significantly from estimate',
48                estimated_cost=estimated_cost,
49                claimed_amount=claimed_amount,
50                requires_review=True
51            )
52        
53        # Step 4: Automated approval criteria
54        auto_approve = (
55            claimed_amount <= Decimal('5000') and
56            fraud_check['fraud_score'] < 0.3 and
57            difference_ratio < 0.1 and
58            claim['has_required_documents']
59        )
60        
61        if auto_approve:
62            return Decision(
63                decision='APPROVED',
64                amount=claimed_amount,
65                requires_review=False,
66                processing_time_seconds=int(time.time() - claim['submitted_at'])
67            )
68        
69        # Step 5: Route to human reviewer
70        return Decision(
71            decision='REVIEW',
72            reason='Requires expert evaluation',
73            estimated_cost=estimated_cost,
74            fraud_score=fraud_check['fraud_score'],
75            requires_review=True,
76            priority=self.calculate_priority(claim, fraud_check)
77        )
78

1from fastapi import FastAPI, UploadFile, File
2from typing import List
3import asyncio
4
5app = FastAPI()
6
7@app.post("/api/v1/claims/submit")
8async def submit_claim(
9    policy_id: str,
10    incident_date: str,
11    claim_amount: float,
12    description: str,
13    documents: List[UploadFile] = File(...)
14) -> Dict:
15    """Submit new insurance claim"""
16    
17    # Create claim record
18    claim = await create_claim({
19        'policy_id': policy_id,
20        'incident_date': incident_date,
21        'amount': claim_amount,
22        'description': description
23    })
24    
25    # Process documents in parallel
26    document_tasks = [
27        process_document(doc, claim['id'])
28        for doc in documents
29    ]
30    processed_docs = await asyncio.gather(*document_tasks)
31    
32    # Evaluate claim
33    decision = await decision_engine.evaluate_claim({
34        **claim,
35        'documents': processed_docs
36    })
37    
38    # Update claim with decision
39    await update_claim(claim['id'], decision)
40    
41    # Notify customer
42    await send_notification(
43        claim['policy_holder_email'],
44        decision
45    )
46    
47    return {
48        'claim_id': claim['id'],
49        'decision': decision['decision'],
50        'estimated_time': decision.get('estimated_resolution_time'),
51        'next_steps': decision.get('next_steps')
52    }
53
54@app.get("/api/v1/claims/{claim_id}/status")
55async def get_claim_status(claim_id: str) -> Dict:
56    """Get current claim status"""
57    
58    claim = await get_claim(claim_id)
59    
60    return {
61        'claim_id': claim_id,
62        'status': claim['status'],
63        'decision': claim.get('decision'),
64        'amount': claim.get('approved_amount'),
65        'timeline': claim['timeline'],
66        'documents_required': claim.get('missing_documents', [])
67    }
68

1┌─────────────────────────────────────┐
2│  Data Ingestion (Apache Airflow)    │
3│  - Claims data                      │
4│  - Historical data                  │
5│  - External data sources            │
6└──────────────┬──────────────────────┘
7               ↓
8┌─────────────────────────────────────┐
9│  Feature Engineering                │
10│  - Temporal features                │
11│  - Network analysis                 │
12│  - Document analysis                │
13└──────────────┬──────────────────────┘
14               ↓
15┌─────────────────────────────────────┐
16│  Model Training (TensorFlow)        │
17│  - Fraud detection                  │
18│  - Cost estimation                  │
19│  - Document classification          │
20└──────────────┬──────────────────────┘
21               ↓
22┌─────────────────────────────────────┐
23│  Model Serving (FastAPI)            │
24│  - REST API                         │
25│  - Real-time predictions            │
26│  - A/B testing                      │
27└─────────────────────────────────────┘
28