Chapter 9: InsurTech Solutions

# Example: AI-Powered Auto Insurance Underwriting
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier, GradientBoostingRegressor
from sklearn.preprocessing import StandardScaler
import joblib

class AutoInsuranceUnderwriter:
    def __init__(self):
        self.risk_classifier = None
        self.premium_regressor = None
        self.scaler = StandardScaler()
        self.feature_weights = {
            'driver_age': 0.15,
            'driving_history': 0.25,
            'vehicle_value': 0.20,
            'location_risk': 0.15,
            'credit_score': 0.10,
            'telematics_score': 0.15
        }

    def load_models(self):
        """Load pre-trained models"""
        self.risk_classifier = joblib.load('models/risk_classifier.pkl')
        self.premium_regressor = joblib.load('models/premium_regressor.pkl')
        self.scaler = joblib.load('models/feature_scaler.pkl')

    def extract_features(self, application_data):
        """Extract and engineer features from application"""
        features = {}

        # Demographic features
        features['driver_age'] = application_data['age']
        features['gender_risk'] = self._calculate_gender_risk(
            application_data['gender'],
            application_data['age']
        )

        # Driving history features
        features['accidents_3_years'] = len(application_data.get('accidents', []))
        features['violations_3_years'] = len(application_data.get('violations', []))
        features['years_licensed'] = application_data['years_licensed']

        # Vehicle features
        features['vehicle_age'] = 2024 - application_data['vehicle_year']
        features['vehicle_value'] = application_data['vehicle_value']
        features['safety_rating'] = application_data.get('safety_rating', 3)

        # Location-based risk
        features['zip_accident_rate'] = self._get_zip_accident_rate(
            application_data['zip_code']
        )
        features['weather_risk'] = self._calculate_weather_risk(
            application_data['zip_code']
        )

        # Credit-based features (where legally allowed)
        if application_data.get('credit_authorized'):
            features['credit_score'] = application_data.get('credit_score', 650)

        # Telematics data (if available)
        if application_data.get('telematics_data'):
            features['telematics_score'] = self._calculate_telematics_score(
                application_data['telematics_data']
            )

        return features

    def underwrite_application(self, application_data):
        """
        Complete underwriting process for auto insurance application
        """
        try:
            # Extract features
            features = self.extract_features(application_data)
            feature_vector = self._prepare_feature_vector(features)

            # Risk classification
            risk_probability = self.risk_classifier.predict_proba(
                feature_vector.reshape(1, -1)
            )[0]

            # Premium calculation
            base_premium = self.premium_regressor.predict(
                feature_vector.reshape(1, -1)
            )[0]

            # Apply business rules
            decision = self._apply_business_rules(
                features, risk_probability, base_premium
            )

            return {
                'decision': decision['status'],
                'premium': decision.get('premium'),
                'risk_score': risk_probability[1],  # High risk probability
                'confidence': max(risk_probability),
                'factors': self._explain_decision(features, risk_probability),
                'referral_reasons': decision.get('referral_reasons', [])
            }

        except Exception as e:
            return {
                'decision': 'REFER',
                'reason': f'Processing error: {str(e)}',
                'referral_reasons': ['Technical processing error']
            }

    def _calculate_telematics_score(self, telematics_data):
        """Calculate risk score from telematics data"""
        if not telematics_data:
            return 0.5  # Neutral score

        # Extract key metrics
        avg_speed = np.mean(telematics_data.get('speeds', []))
        hard_braking = telematics_data.get('hard_braking_events', 0)
        rapid_acceleration = telematics_data.get('acceleration_events', 0)
        night_driving = telematics_data.get('night_driving_pct', 0)
        miles_driven = telematics_data.get('total_miles', 0)

        # Calculate composite score (0-1, lower is better)
        speed_score = min(avg_speed / 80, 1.0)  # Normalized to speed limit
        safety_score = (hard_braking + rapid_acceleration) / max(miles_driven / 1000, 1)
        time_score = night_driving / 100

        composite_score = (speed_score * 0.4 + safety_score * 0.4 + time_score * 0.2)
        return min(max(composite_score, 0), 1)

    def _apply_business_rules(self, features, risk_probability, base_premium):
        """Apply business rules for final decision"""
        high_risk_threshold = 0.7
        low_risk_threshold = 0.3

        referral_reasons = []

        # High-risk automatic decline
        if risk_probability[1] > high_risk_threshold:
            if features.get('accidents_3_years', 0) > 2:
                return {'status': 'DECLINE', 'reason': 'Excessive accident history'}

        # Referral conditions
        if features.get('driver_age', 25) < 21:
            referral_reasons.append('Young driver')

        if features.get('credit_score', 700) < 550:
            referral_reasons.append('Low credit score')

        if base_premium > 5000:
            referral_reasons.append('High premium amount')

        if referral_reasons:
            return {
                'status': 'REFER',
                'referral_reasons': referral_reasons,
                'premium': base_premium
            }

        # Auto-approval for low risk
        if risk_probability[1] < low_risk_threshold:
            return {
                'status': 'APPROVE',
                'premium': base_premium
            }

        # Standard approval
        return {
            'status': 'APPROVE',
            'premium': base_premium * (1 + risk_probability[1] * 0.5)  # Risk adjustment
        }

# Example: Weather-Based Parametric Insurance
import requests
import json
from datetime import datetime, timedelta
import asyncio

class ParametricInsuranceProcessor:
    def __init__(self):
        self.weather_api_key = "your_weather_api_key"
        self.satellite_api_key = "your_satellite_api_key"
        self.blockchain_endpoint = "your_blockchain_endpoint"

    async def process_weather_claims(self, policy_data):
        """
        Process parametric weather insurance claims
        """
        active_policies = self._get_active_policies(policy_data['region'])

        for policy in active_policies:
            try:
                # Get weather data for policy location
                weather_data = await self._get_weather_data(
                    policy['coordinates'],
                    policy['coverage_period']
                )

                # Check trigger conditions
                trigger_result = self._evaluate_triggers(
                    policy['triggers'],
                    weather_data
                )

                if trigger_result['triggered']:
                    # Calculate payout
                    payout_amount = self._calculate_payout(
                        policy,
                        trigger_result
                    )

                    # Initiate automatic payout
                    await self._process_payout(
                        policy['policy_id'],
                        payout_amount,
                        trigger_result['evidence']
                    )

                    # Log to blockchain for transparency
                    await self._log_to_blockchain(
                        policy['policy_id'],
                        trigger_result,
                        payout_amount
                    )

            except Exception as e:
                await self._handle_processing_error(policy['policy_id'], str(e))

    async def _get_weather_data(self, coordinates, period):
        """Fetch weather data from multiple sources"""
        weather_apis = [
            f"https://api.openweathermap.org/data/2.5/weather?lat={coordinates['lat']}&lon={coordinates['lon']}&appid={self.weather_api_key}",
            f"https://api.weather.gov/points/{coordinates['lat']},{coordinates['lon']}"
        ]

        weather_data = {}
        for api_url in weather_apis:
            try:
                response = requests.get(api_url, timeout=10)
                if response.status_code == 200:
                    weather_data.update(response.json())
            except requests.RequestException:
                continue

        return weather_data

    def _evaluate_triggers(self, triggers, weather_data):
        """Evaluate if policy triggers are met"""
        triggered_conditions = []

        for trigger in triggers:
            if trigger['type'] == 'rainfall':
                actual_rainfall = weather_data.get('rainfall_mm', 0)
                if trigger['condition'] == 'less_than':
                    if actual_rainfall < trigger['threshold']:
                        triggered_conditions.append({
                            'type': 'rainfall',
                            'threshold': trigger['threshold'],
                            'actual': actual_rainfall,
                            'severity': (trigger['threshold'] - actual_rainfall) / trigger['threshold']
                        })

            elif trigger['type'] == 'temperature':
                actual_temp = weather_data.get('temperature_c', 20)
                if trigger['condition'] == 'exceeds':
                    if actual_temp > trigger['threshold']:
                        triggered_conditions.append({
                            'type': 'temperature',
                            'threshold': trigger['threshold'],
                            'actual': actual_temp,
                            'severity': (actual_temp - trigger['threshold']) / trigger['threshold']
                        })

        return {
            'triggered': len(triggered_conditions) > 0,
            'conditions': triggered_conditions,
            'evidence': weather_data
        }

    def _calculate_payout(self, policy, trigger_result):
        """Calculate payout based on trigger severity"""
        base_payout = policy['coverage_amount']
        total_severity = sum(cond['severity'] for cond in trigger_result['conditions'])

        # Apply severity multiplier (capped at 100% payout)
        payout_percentage = min(total_severity, 1.0)
        return base_payout * payout_percentage

    async def _process_payout(self, policy_id, amount, evidence):
        """Process automatic payout"""
        payout_data = {
            'policy_id': policy_id,
            'amount': amount,
            'evidence': evidence,
            'timestamp': datetime.now().isoformat(),
            'processed_by': 'automated_system'
        }

        # Send to payment processing system
        payment_response = await self._initiate_payment(payout_data)

        # Update policy status
        await self._update_policy_status(policy_id, 'claim_paid', amount)

        return payment_response