Automating Repetitive Calculations for Actuaries #

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import logging
from typing import Dict, List, Optional, Tuple

class PolicyRenewalAnalyzer:
    """
    Comprehensive automation solution for policy renewal analysis.
    
    This class demonstrates a complete approach to automating a common
    actuarial calculation that involves multiple steps and produces
    insights across various business dimensions.
    """
    
    def __init__(self, policy_data: pd.DataFrame, config: Dict):
        """
        Initialize the renewal analyzer with policy data and configuration.
        
        Parameters:
        policy_data (pd.DataFrame): Policy-level data with renewal information
        config (Dict): Configuration parameters for the analysis
        """
        self.policy_data = policy_data
        self.config = config
        self.results = {}
        self.logger = logging.getLogger(__name__)
        
        # Validate input data upon initialization
        self._validate_input_data()
        
    def _validate_input_data(self) -> None:
        """Validates the structure and content of input policy data."""
        required_columns = [
            'policy_id', 'issue_date', 'expiration_date', 
            'renewal_date', 'product_type', 'region', 
            'premium', 'agent_id', 'customer_id'
        ]
        
        missing_columns = set(required_columns) - set(self.policy_data.columns)
        if missing_columns:
            raise ValueError(f"Missing required columns: {missing_columns}")
            
        # Validate date columns
        date_columns = ['issue_date', 'expiration_date', 'renewal_date']
        for col in date_columns:
            if not pd.api.types.is_datetime64_any_dtype(self.policy_data[col]):
                try:
                    self.policy_data[col] = pd.to_datetime(self.policy_data[col])
                except:
                    raise ValueError(f"Cannot convert {col} to datetime")
    
    def calculate_renewal_metrics(self, analysis_date: datetime) -> Dict:
        """
        Calculates comprehensive renewal metrics as of the analysis date.
        
        This method demonstrates how complex analytical logic can be
        systematized and made repeatable through automation.
        
        Parameters:
        analysis_date (datetime): Date for which to perform the analysis
        
        Returns:
        Dict: Comprehensive renewal statistics and metrics
        """
        self.logger.info(f"Starting renewal analysis for {analysis_date}")
        
        # Step 1: Prepare data for analysis
        analysis_data = self._prepare_analysis_data(analysis_date)
        
        # Step 2: Calculate basic renewal statistics
        basic_stats = self._calculate_basic_renewal_stats(analysis_data)
        
        # Step 3: Perform dimensional analysis
        dimensional_analysis = self._perform_dimensional_analysis(analysis_data)
        
        # Step 4: Calculate trend metrics
        trend_metrics = self._calculate_trend_metrics(analysis_data, analysis_date)
        
        # Step 5: Generate risk indicators
        risk_indicators = self._generate_risk_indicators(analysis_data)
        
        # Compile comprehensive results
        comprehensive_results = {
            'analysis_date': analysis_date,
            'basic_statistics': basic_stats,
            'dimensional_analysis': dimensional_analysis,
            'trend_metrics': trend_metrics,
            'risk_indicators': risk_indicators,
            'data_quality_summary': self._summarize_data_quality(analysis_data)
        }
        
        self.results[analysis_date.strftime('%Y-%m-%d')] = comprehensive_results
        self.logger.info("Renewal analysis completed successfully")
        
        return comprehensive_results
    
    def _prepare_analysis_data(self, analysis_date: datetime) -> pd.DataFrame:
        """Prepares and enriches data for renewal analysis."""
        # Create a copy to avoid modifying original data
        data = self.policy_data.copy()
        
        # Calculate key metrics for each policy
        data['days_since_issue'] = (analysis_date - data['issue_date']).dt.days
        data['days_to_expiration'] = (data['expiration_date'] - analysis_date).dt.days
        data['policy_age_years'] = data['days_since_issue'] / 365.25
        
        # Determine renewal status
        data['renewed'] = data['renewal_date'].notna()
        data['renewal_within_window'] = (
            (data['renewal_date'] >= data['expiration_date'] - timedelta(days=30)) &
            (data['renewal_date'] <= data['expiration_date'] + timedelta(days=30))
        )
        
        # Create analytical segments
        data['premium_segment'] = pd.qcut(
            data['premium'], 
            q=4, 
            labels=['Low', 'Medium', 'High', 'Premium']
        )
        
        data['age_segment'] = pd.cut(
            data['policy_age_years'],
            bins=[0, 1, 3, 5, 10, float('inf')],
            labels=['New', 'Young', 'Mature', 'Established', 'Veteran']
        )
        
        return data
    
    def _calculate_basic_renewal_stats(self, data: pd.DataFrame) -> Dict:
        """Calculates fundamental renewal statistics."""
        total_policies = len(data)
        renewed_policies = data['renewed'].sum()
        renewal_rate = renewed_policies / total_policies if total_policies > 0 else 0
        
        # Calculate weighted metrics
        total_premium = data['premium'].sum()
        renewed_premium = data[data['renewed']]['premium'].sum()
        premium_weighted_renewal_rate = renewed_premium / total_premium if total_premium > 0 else 0
        
        return {
            'total_policies_analyzed': total_policies,
            'renewed_policies': int(renewed_policies),
            'lapsed_policies': total_policies - int(renewed_policies),
            'overall_renewal_rate': round(renewal_rate, 4),
            'premium_weighted_renewal_rate': round(premium_weighted_renewal_rate, 4),
            'average_premium_renewed': data[data['renewed']]['premium'].mean(),
            'average_premium_lapsed': data[~data['renewed']]['premium'].mean()
        }
    
    def _perform_dimensional_analysis(self, data: pd.DataFrame) -> Dict:
        """Performs renewal analysis across multiple business dimensions."""
        dimensions = ['product_type', 'region', 'premium_segment', 'age_segment']
        dimensional_results = {}
        
        for dimension in dimensions:
            dimension_stats = (
                data.groupby(dimension)
                .agg({
                    'policy_id': 'count',
                    'renewed': ['sum', 'mean'],
                    'premium': ['sum', 'mean']
                })
                .round(4)
            )
            
            # Flatten column names
            dimension_stats.columns = [
                'policy_count', 'renewed_count', 'renewal_rate', 
                'total_premium', 'average_premium'
            ]
            
            dimensional_results[dimension] = dimension_stats.to_dict('index')
        
        return dimensional_results
    
    def _calculate_trend_metrics(self, data: pd.DataFrame, analysis_date: datetime) -> Dict:
        """Calculates trend metrics to identify patterns over time."""
        # This would typically involve historical data comparison
        # For this example, we'll create cohort-based trends
        
        cohort_analysis = (
            data.groupby('age_segment')
            .agg({
                'renewed': 'mean',
                'premium': 'mean',
                'policy_id': 'count'
            })
            .round(4)
        )
        
        return {
            'cohort_renewal_rates': cohort_analysis.to_dict('index'),
            'trend_direction': 'stable',  # Would be calculated from historical data
            'seasonal_adjustment': 1.0   # Would be derived from seasonal patterns
        }
    
    def _generate_risk_indicators(self, data: pd.DataFrame) -> Dict:
        """Generates risk indicators for renewal management."""
        # Calculate concentration risks
        product_concentration = (
            data.groupby('product_type')['premium'].sum() / 
            data['premium'].sum()
        ).max()
        
        regional_concentration = (
            data.groupby('region')['premium'].sum() / 
            data['premium'].sum()
        ).max()
        
        return {
            'product_concentration_risk': round(product_concentration, 4),
            'regional_concentration_risk': round(regional_concentration, 4),
            'low_renewal_segments': self._identify_low_renewal_segments(data),
            'at_risk_premium': self._calculate_at_risk_premium(data)
        }
    
    def _identify_low_renewal_segments(self, data: pd.DataFrame) -> List[str]:
        """Identifies segments with renewal rates below threshold."""
        threshold = self.config.get('low_renewal_threshold', 0.7)
        
        segments = []
        for dimension in ['product_type', 'region']:
            segment_rates = data.groupby(dimension)['renewed'].mean()
            low_segments = segment_rates[segment_rates < threshold].index.tolist()
            segments.extend([f"{dimension}: {seg}" for seg in low_segments])
        
        return segments
    
    def _calculate_at_risk_premium(self, data: pd.DataFrame) -> float:
        """Calculates premium amount at risk based on renewal patterns."""
        # Simplified risk calculation based on historical patterns
        at_risk_data = data[
            (data['days_to_expiration'] <= 60) & 
            (data['days_to_expiration'] > 0) &
            (~data['renewed'])
        ]
        
        return float(at_risk_data['premium'].sum())
    
    def _summarize_data_quality(self, data: pd.DataFrame) -> Dict:
        """Summarizes data quality metrics for the analysis."""
        return {
            'total_records': len(data),
            'missing_renewal_dates': data['renewal_date'].isna().sum(),
            'invalid_date_sequences': (
                data['expiration_date'] < data['issue_date']
            ).sum(),
            'data_completeness_score': (
                1 - data.isna().sum().sum() / (len(data) * len(data.columns))
            )
        }

# Usage example and configuration
def run_renewal_analysis_example():
    """Demonstrates how to use the PolicyRenewalAnalyzer."""
    
    # Configuration for the analysis
    config = {
        'low_renewal_threshold': 0.7,
        'renewal_window_days': 30,
        'reporting_segments': ['product_type', 'region', 'premium_segment']
    }
    
    # In practice, this would load from your data source
    # sample_data = load_policy_data_from_database()
    
    # Initialize the analyzer
    # analyzer = PolicyRenewalAnalyzer(sample_data, config)
    
    # Run the analysis
    # results = analyzer.calculate_renewal_metrics(datetime.now())
    
    # Generate reports or take action based on results
    # generate_renewal_report(results)
    
    pass

from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Tuple
import pandas as pd
import numpy as np
from dataclasses import dataclass
import logging

@dataclass
class ReserveCalculationConfig:
    """Configuration class for reserve calculations."""
    data_source: str
    calculation_date: str
    methods: List[str]
    confidence_levels: List[float]
    development_periods: int
    tail_factor_method: str
    
class ReserveMethodBase(ABC):
    """
    Abstract base class for reserve calculation methods.
    
    This demonstrates how to create a modular framework that can
    accommodate different calculation methodologies while maintaining
    consistency in interfaces and data handling.
    """
    
    def __init__(self, config: ReserveCalculationConfig):
        self.config = config
        self.logger = logging.getLogger(self.__class__.__name__)
        
    @abstractmethod
    def calculate_reserves(self, triangle_data: pd.DataFrame) -> Dict:
        """Calculate reserves using the specific method."""
        pass
    
    @abstractmethod
    def get_method_diagnostics(self) -> Dict:
        """Return method-specific diagnostic information."""
        pass
    
    def _validate_triangle_data(self, data: pd.DataFrame) -> bool:
        """Common validation logic for triangle data."""
        required_columns = ['accident_year', 'development_period', 'cumulative_paid']
        
        if not all(col in data.columns for col in required_columns):
            raise ValueError(f"Missing required columns: {required_columns}")
            
        if data.isna().any().any():
            self.logger.warning("Triangle contains missing values")
            
        return True

class ChainLadderMethod(ReserveMethodBase):
    """
    Implementation of the Chain Ladder reserve calculation method.
    
    This class demonstrates how specific calculation methods can inherit
    from a common base while implementing their unique logic.
    """
    
    def __init__(self, config: ReserveCalculationConfig):
        super().__init__(config)
        self.development_factors = None
        self.tail_factor = None
        
    def calculate_reserves(self, triangle_data: pd.DataFrame) -> Dict:
        """
        Calculates reserves using the chain ladder method.
        
        Returns:
        Dict: Comprehensive results including reserves, factors, and diagnostics
        """
        self._validate_triangle_data(triangle_data)
        self.logger.info("Starting Chain Ladder calculation")
        
        # Step 1: Create development triangle
        triangle = self._create_development_triangle(triangle_data)
        
        # Step 2: Calculate development factors
        self.development_factors = self._calculate_development_factors(triangle)
        
        # Step 3: Calculate tail factor
        self.tail_factor = self._calculate_tail_factor()
        
        # Step 4: Project ultimate values
        ultimate_values = self._project_ultimate_values(triangle)
        
        # Step 5: Calculate reserves
        reserves = self._calculate_point_estimates(triangle, ultimate_values)
        
        # Step 6: Calculate confidence intervals if requested
        confidence_intervals = None
        if self.config.confidence_levels:
            confidence_intervals = self._calculate_confidence_intervals(triangle)
        
        return {
            'method_name': 'Chain Ladder',
            'reserves_by_year': reserves,
            'total_reserve': reserves.sum(),
            'development_factors': self.development_factors,
            'tail_factor': self.tail_factor,
            'ultimate_values': ultimate_values,
            'confidence_intervals': confidence_intervals,
            'diagnostics': self.get_method_diagnostics()
        }
    
    def _create_development_triangle(self, data: pd.DataFrame) -> pd.DataFrame:
        """Creates a properly formatted development triangle."""
        triangle = data.pivot_table(
            index='accident_year',
            columns='development_period',
            values='cumulative_paid',
            aggfunc='sum'
        )
        
        # Ensure proper ordering and completeness
        max_dev_period = triangle.columns.max()
        expected_periods = list(range(1, max_dev_period + 1))
        triangle = triangle.reindex(columns=expected_periods)
        
        return triangle
    
    def _calculate_development_factors(self, triangle: pd.DataFrame) -> pd.Series:
        """Calculates age-to-age development factors."""
        factors = pd.Series(index=triangle.columns[:-1], dtype=float)
        
        for i in range(len(triangle.columns) - 1):
            current_period = triangle.columns[i]
            next_period = triangle.columns[i + 1]
            
            # Calculate volume-weighted average factor
            current_values = triangle[current_period].dropna()
            next_values = triangle[next_period].reindex(current_values.index).dropna()
            
            if len(current_values) > 0 and len(next_values) > 0:
                factor = next_values.sum() / current_values.sum()
                factors[current_period] = factor
            else:
                factors[current_period] = 1.0
                self.logger.warning(f"Insufficient data for period {current_period}")
        
        return factors
    
    def _calculate_tail_factor(self) -> float:
        """Calculates tail factor using configured method."""
        if self.config.tail_factor_method == 'curve_fit':
            return self._curve_fit_tail_factor()
        elif self.config.tail_factor_method == 'fixed':
            return 1.05  # Industry standard default
        else:
            return 1.0
    
    def _curve_fit_tail_factor(self) -> float:
        """Calculates tail factor using curve fitting approach."""
        # Simplified implementation - in practice would use more sophisticated methods
        last_few_factors = self.development_factors.iloc[-3:].values
        if len(last_few_factors) >= 2:
            decay_rate = np.mean(np.diff(last_few_factors))
            tail_contribution = max(0.05, abs(decay_rate * 2))
            return 1.0 + tail_contribution
        return 1.05
    
    def _project_ultimate_values(self, triangle: pd.DataFrame) -> pd.Series:
        """Projects ultimate values for each accident year."""
        ultimate_values = pd.Series(index=triangle.index, dtype=float)
        
        for accident_year in triangle.index:
            # Find the last non-null value
            year_data = triangle.loc[accident_year].dropna()
            if len(year_data) > 0:
                last_period = year_data.index[-1]
                last_value = year_data.iloc[-1]
                
                # Apply remaining development factors
                remaining_factors = self.development_factors.loc[last_period:].prod()
                ultimate_value = last_value * remaining_factors * self.tail_factor
                ultimate_values[accident_year] = ultimate_value
            else:
                ultimate_values[accident_year] = 0.0
                
        return ultimate_values
    
    def _calculate_point_estimates(self, triangle: pd.DataFrame, ultimate_values: pd.Series) -> pd.Series:
        """Calculates point estimate reserves."""
        reserves = pd.Series(index=triangle.index, dtype=float)
        
        for accident_year in triangle.index:
            latest_paid = triangle.loc[accident_year].dropna().iloc[-1] if not triangle.loc[accident_year].dropna().empty else 0
            ultimate = ultimate_values[accident_year]
            reserves[accident_year] = max(0, ultimate - latest_paid)
            
        return reserves
    
    def _calculate_confidence_intervals(self, triangle: pd.DataFrame) -> Dict:
        """Calculates confidence intervals using bootstrap or other methods."""
        # Simplified implementation - would use more sophisticated statistical methods
        confidence_intervals = {}
        
        for confidence_level in self.config.confidence_levels:
            # This would typically involve bootstrap resampling or parametric methods
            intervals = {}  # Would contain actual interval calculations
            confidence_intervals[confidence_level] = intervals
            
        return confidence_intervals
    
    def get_method_diagnostics(self) -> Dict:
        """Returns diagnostic information for the Chain Ladder method."""
        diagnostics = {
            'factor_stability': self._assess_factor_stability(),
            'data_completeness': self._assess_data_completeness(),
            'outlier_analysis': self._detect_outliers(),
            'goodness_of_fit': self._calculate_goodness_of_fit()
        }
        
        return diagnostics
    
    def _assess_factor_stability(self) -> Dict:
        """Assesses the stability of development factors."""
        if self.development_factors is None:
            return {'error': 'Development factors not calculated'}
            
        # Calculate coefficient of variation for factors
        cv = self.development_factors.std() / self.development_factors.mean()
        
        return {
            'coefficient_of_variation': cv,
            'stability_rating': 'High' if cv < 0.1 else 'Medium' if cv < 0.2 else 'Low'
        }
    
    def _assess_data_completeness(self) -> Dict:
        """Assesses completeness of the data triangle."""
        # Implementation would analyze missing values and data patterns
        return {'completeness_score': 0.95}  # Placeholder
    
    def _detect_outliers(self) -> List:
        """Detects outlier values in the development pattern."""
        # Implementation would identify statistical outliers
        return []  # Placeholder
    
    def _calculate_goodness_of_fit(self) -> float:
        """Calculates goodness of fit measure for the model."""
        # Implementation would calculate R-squared or similar measure
        return 0.85  # Placeholder

class ReserveCalculationEngine:
    """
    Main engine for orchestrating reserve calculations across multiple methods.
    
    This class demonstrates how modular components can be orchestrated
    to provide comprehensive analytical capabilities.
    """
    
    def __init__(self, config: ReserveCalculationConfig):
        self.config = config
        self.methods = {}
        self.results = {}
        self.logger = logging.getLogger(__name__)
        
        # Initialize requested methods
        self._initialize_methods()
    
    def _initialize_methods(self):
        """Initialize the requested calculation methods."""
        method_map = {
            'chain_ladder': ChainLadderMethod,
            # Additional methods would be added here:
            # 'bornhuetter_ferguson': BornhuetterFergusonMethod,
            # 'expected_loss_ratio': ExpectedLossRatioMethod,
        }
        
        for method_name in self.config.methods:
            if method_name in method_map:
                self.methods[method_name] = method_map[method_name](self.config)
                self.logger.info(f"Initialized method: {method_name}")
            else:
                self.logger.warning(f"Unknown method requested: {method_name}")
    
    def run_calculations(self, triangle_data: pd.DataFrame) -> Dict:
        """
        Runs all configured calculation methods and compiles results.
        
        Parameters:
        triangle_data (pd.DataFrame): Loss development triangle data
        
        Returns:
        Dict: Comprehensive results from all methods
        """
        self.logger.info("Starting reserve calculation process")
        
        for method_name, method_instance in self.methods.items():
            try:
                self.logger.info(f"Running {method_name} calculation")
                method_results = method_instance.calculate_reserves(triangle_data)
                self.results[method_name] = method_results
                
            except Exception as e:
                self.logger.error(f"Error in {method_name}: {str(e)}")
                self.results[method_name] = {'error': str(e)}
        
        # Generate comparative analysis
        comparison = self._generate_method_comparison()
        
        # Compile final results
        final_results = {
            'calculation_date': self.config.calculation_date,
            'method_results': self.results,
            'method_comparison': comparison,
            'recommended_estimate': self._select_recommended_estimate(),
            'summary_statistics': self._generate_summary_statistics()
        }
        
        self.logger.info("Reserve calculation process completed")
        return final_results
    
    def _generate_method_comparison(self) -> Dict:
        """Generates comparison statistics across methods."""
        comparison = {