"""Insurance pricing module with market cycle support.
This module implements realistic insurance premium calculation based on
frequency and severity distributions, replacing hardcoded premium rates
in simulations. It supports market cycle adjustments and integrates with
existing loss generators and insurance structures.
Example:
Basic usage for pricing an insurance program::
from ergodic_insurance.insurance_pricing import InsurancePricer, MarketCycle
from ergodic_insurance.loss_distributions import ManufacturingLossGenerator
# Initialize loss generator and pricer
loss_gen = ManufacturingLossGenerator()
pricer = InsurancePricer(
loss_generator=loss_gen,
loss_ratio=0.70,
market_cycle=MarketCycle.NORMAL
)
# Price an insurance program
program = InsuranceProgram(layers=[...])
priced_program = pricer.price_insurance_program(
program,
expected_revenue=15_000_000
)
# Get total premium
total_premium = priced_program.calculate_premium()
Attributes:
MarketCycle: Enum representing market conditions (HARD, NORMAL, SOFT)
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
import logging
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
import warnings
import numpy as np
from ._warnings import ErgodicInsuranceDeprecationWarning
from .claim_development import ClaimDevelopment
from .loss_distributions import LossDistribution
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from .exposure_base import ExposureBase
from .insurance import InsuranceLayer, InsurancePolicy
from .insurance_program import EnhancedInsuranceLayer, InsuranceProgram
from .loss_distributions import ManufacturingLossGenerator
[docs]
class MarketCycle(Enum):
"""Market cycle states affecting insurance pricing.
Each state corresponds to a target loss ratio that insurers
use to price coverage. Lower loss ratios (hard markets) result
in higher premiums.
Attributes:
HARD: Seller's market with limited capacity (60% loss ratio)
NORMAL: Balanced market conditions (70% loss ratio)
SOFT: Buyer's market with excess capacity (80% loss ratio)
"""
HARD = 0.60 # 60% loss ratio - higher premiums
NORMAL = 0.70 # 70% loss ratio - standard premiums
SOFT = 0.80 # 80% loss ratio - lower premiums
[docs]
@dataclass
class PricingParameters:
"""Parameters for insurance pricing calculations.
Attributes:
loss_ratio: Target loss ratio for pricing (claims/premium)
expense_ratio: Operating expense ratio excluding LAE (default 0.25).
Covers commissions, overhead, admin, and other non-LAE expenses.
profit_margin: Target profit margin (default 0.05)
risk_loading: Additional loading for uncertainty (default 0.10)
confidence_level: Confidence level for pricing (default 0.95)
simulation_years: Years to simulate for pricing (default 10)
min_premium: Minimum premium floor (default 1000)
max_rate_on_line: Maximum rate on line cap (default 0.50)
alae_ratio: Allocated LAE ratio as fraction of pure premium (default
0.10). Covers claim-specific costs such as legal fees and expert
witnesses. Typical range for commercial lines is 0.08-0.15.
ulae_ratio: Unallocated LAE ratio as fraction of pure premium (default
0.05). Covers general claims department overhead. Typical range
is 0.03-0.08.
development_pattern: Optional claim development pattern for developing
losses to ultimate (default None). When set, simulated losses are
adjusted by age-to-ultimate factors per ASOP 25 so that immature
accident years are brought to their expected ultimate value before
pure premium calculation. Use ``None`` to skip development
(equivalent to assuming all losses are already at ultimate).
"""
loss_ratio: float = 0.70
expense_ratio: float = 0.25
profit_margin: float = 0.05
risk_loading: float = 0.10
confidence_level: float = 0.95
simulation_years: int = 10
min_premium: float = 1000.0
max_rate_on_line: float = 0.50
alae_ratio: float = 0.10
ulae_ratio: float = 0.05
development_pattern: Optional[ClaimDevelopment] = None
def __post_init__(self) -> None:
# Validate that expense ratio + profit margin leave room for losses
if self.expense_ratio + self.profit_margin >= 1.0:
raise ValueError(
f"expense_ratio ({self.expense_ratio:.2f}) + "
f"profit_margin ({self.profit_margin:.2f}) = "
f"{self.expense_ratio + self.profit_margin:.2f} >= 1.0; "
f"no premium can cover losses under these parameters"
)
# Warn if loss_ratio is inconsistent with expense/profit parameters.
# The actuarial identity implies loss_ratio = 1 - V - Q.
indicated_lr = 1 - self.expense_ratio - self.profit_margin
if abs(self.loss_ratio - indicated_lr) > 0.01:
logger.warning(
"loss_ratio (%.2f) is inconsistent with "
"1 - expense_ratio - profit_margin = %.2f. "
"The actuarial pricing identity Premium = Pure_Premium / "
"(1 - V - Q) expects these to match. Adjust loss_ratio or "
"expense_ratio/profit_margin for consistency.",
self.loss_ratio,
indicated_lr,
)
if self.alae_ratio + self.ulae_ratio > self.expense_ratio:
logger.warning(
"LAE ratio (%.2f) exceeds expense ratio (%.2f). This is "
"unusual — verify that expense_ratio accounts for all "
"operating expenses or adjust alae_ratio/ulae_ratio.",
self.alae_ratio + self.ulae_ratio,
self.expense_ratio,
)
@property
def lae_ratio(self) -> float:
"""Combined LAE ratio (ALAE + ULAE) as fraction of pure premium."""
return self.alae_ratio + self.ulae_ratio
[docs]
@dataclass
class LayerPricing:
"""Pricing details for a single insurance layer.
Attributes:
attachment_point: Where coverage starts
limit: Maximum coverage amount
expected_frequency: Expected claims per year hitting this layer
expected_severity: Average severity of claims in this layer
pure_premium: Expected loss cost
technical_premium: Pure premium with expenses and profit
market_premium: Final premium after market adjustments
rate_on_line: Premium as percentage of limit
confidence_interval: (lower, upper) bounds at confidence level
lae_loading: LAE component calculated from dedicated ALAE/ULAE ratios
development_factor: Average age-to-ultimate LDF applied (1.0 = no development)
"""
attachment_point: float
limit: float
expected_frequency: float
expected_severity: float
pure_premium: float
technical_premium: float
market_premium: float
rate_on_line: float
confidence_interval: Tuple[float, float]
lae_loading: float = 0.0
development_factor: float = 1.0
[docs]
class LayerPricer:
"""Analytical layer pricing using limited expected values.
Computes expected losses in an excess layer (attachment, attachment + limit)
directly from a fitted severity distribution, without simulation. This
replaces ad-hoc rate heuristics with actuarially sound formulas based on
the limited expected value (LEV), increased limits factors (ILFs), loss
elimination ratios (LERs), and exposure curves (Lee diagrams).
The fundamental identity is:
E[loss in layer (a, a+l)] = LEV(a + l) - LEV(a)
where LEV(d) = E[min(X, d)] is the limited expected value at *d*.
References:
- Lee (1988) — Loss Distributions (exposure curves)
- Miccolis (1977) — On the Theory of Increased Limits and Excess of Loss Pricing
- Klugman, Panjer, Willmot — *Loss Models*, Chapter 5
Args:
severity_distribution: Fitted severity distribution with a
``limited_expected_value(limit)`` method.
frequency: Expected annual claim frequency for the distribution.
Example:
Analytical pricing of an excess layer::
from ergodic_insurance.loss_distributions import ParetoLoss
from ergodic_insurance.insurance_pricing import LayerPricer
severity = ParetoLoss(alpha=2.5, xm=100_000)
pricer = LayerPricer(severity, frequency=5.0)
# Expected loss in the $1M xs $500K layer
layer_loss = pricer.expected_layer_loss(500_000, 1_000_000)
# Increased Limits Factor at $2M relative to $500K basic limit
ilf = pricer.increased_limits_factor(2_000_000, basic_limit=500_000)
"""
def __init__(
self,
severity_distribution: LossDistribution,
frequency: float = 1.0,
) -> None:
self.severity = severity_distribution
self.frequency = frequency
[docs]
def expected_layer_loss(self, attachment: float, limit: float) -> float:
"""Calculate the expected loss in an excess layer.
E[loss in (a, a+l)] = LEV(a + l) - LEV(a)
This is the pure premium for a single occurrence in the layer,
multiplied by frequency to get the annual expected layer loss.
Args:
attachment: Layer attachment point (deductible).
limit: Layer limit (width of coverage).
Returns:
Annual expected loss cost for the layer.
"""
if limit <= 0:
return 0.0
lev_top = self.severity.limited_expected_value(attachment + limit)
lev_bottom = self.severity.limited_expected_value(attachment)
per_occurrence = lev_top - lev_bottom
return float(self.frequency * max(per_occurrence, 0.0))
[docs]
def increased_limits_factor(self, limit: float, basic_limit: float) -> float:
"""Calculate the Increased Limits Factor (ILF).
ILF(L) = LEV(L) / LEV(B)
where B is the basic limit and L is the desired limit.
Used to price policies at higher limits relative to a base.
Reference: Miccolis (1977); CAS Study Note on ILF ratemaking.
Args:
limit: Target policy limit.
basic_limit: Basic (reference) limit.
Returns:
ILF ratio (>= 1.0 when limit >= basic_limit).
"""
lev_basic = self.severity.limited_expected_value(basic_limit)
if lev_basic <= 0:
return 1.0
lev_limit = self.severity.limited_expected_value(limit)
return float(lev_limit / lev_basic)
[docs]
def loss_elimination_ratio(self, deductible: float) -> float:
"""Calculate the Loss Elimination Ratio (LER).
LER(d) = LEV(d) / E[X]
The proportion of total expected losses eliminated by a deductible *d*.
An LER of 0.40 means the deductible removes 40% of expected losses.
Reference: Klugman, Panjer, Willmot — *Loss Models*, Definition 5.2.
Args:
deductible: Deductible amount.
Returns:
LER in [0, 1]. Returns 0.0 if E[X] is infinite.
"""
ev = self.severity.expected_value()
if not np.isfinite(ev) or ev <= 0:
return 0.0
lev_d = self.severity.limited_expected_value(deductible)
return float(min(lev_d / ev, 1.0))
[docs]
def exposure_curve(self, n_points: int = 100) -> Dict[str, List[float]]:
"""Compute the exposure curve (Lee diagram).
The exposure curve relates the retention as a fraction of the policy
limit to the proportion of expected losses retained. Formally:
G(r) = LEV(r * M) / LEV(M)
where M is the maximum possible loss (or a large practical upper bound)
and r ranges from 0 to 1.
Useful for visualising how much loss is retained vs. ceded at
different attachment points.
Reference: Lee (1988).
Args:
n_points: Number of points on the curve (default 100).
Returns:
Dictionary with 'retention_pct' and 'loss_eliminated_pct' lists,
each of length *n_points* + 1 (including the origin).
"""
ev = self.severity.expected_value()
if not np.isfinite(ev) or ev <= 0:
diagonal = [0.0] + [i / n_points for i in range(1, n_points + 1)]
return {"retention_pct": diagonal, "loss_eliminated_pct": diagonal}
# Use a practical upper bound: 20x expected value
max_loss = ev * 20.0
lev_max = self.severity.limited_expected_value(max_loss)
if lev_max <= 0:
diagonal = [0.0] + [i / n_points for i in range(1, n_points + 1)]
return {"retention_pct": diagonal, "loss_eliminated_pct": diagonal}
retention_pcts: List[float] = []
loss_elim_pcts: List[float] = []
for i in range(n_points + 1):
pct = i / n_points
retention_pcts.append(pct)
d = pct * max_loss
lev_d = self.severity.limited_expected_value(d)
loss_elim_pcts.append(float(lev_d / lev_max))
return {"retention_pct": retention_pcts, "loss_eliminated_pct": loss_elim_pcts}
[docs]
class InsurancePricer:
"""Calculate insurance premiums based on loss distributions and market conditions.
This class provides methods to price individual layers and complete insurance
programs using frequency/severity distributions from loss generators. It supports
market cycle adjustments and maintains backward compatibility with fixed rates.
Args:
loss_generator: Manufacturing loss generator for frequency/severity data
loss_ratio: Target loss ratio for pricing (or use market_cycle)
market_cycle: Market cycle state (overrides loss_ratio if provided)
parameters: Additional pricing parameters
seed: Random seed for reproducible simulations
Example:
Pricing with different market conditions::
# Hard market pricing (higher premiums)
hard_pricer = InsurancePricer(
loss_generator=loss_gen,
market_cycle=MarketCycle.HARD
)
# Soft market pricing (lower premiums)
soft_pricer = InsurancePricer(
loss_generator=loss_gen,
market_cycle=MarketCycle.SOFT
)
"""
def __init__(
self,
loss_generator: Optional["ManufacturingLossGenerator"] = None,
loss_ratio: Optional[float] = None,
market_cycle: Optional[MarketCycle] = None,
parameters: Optional[PricingParameters] = None,
exposure: Optional["ExposureBase"] = None,
seed: Optional[int] = None,
):
"""Initialize the insurance pricer.
Args:
loss_generator: Loss generator for frequency/severity
loss_ratio: Target loss ratio (0-1)
market_cycle: Market cycle state
parameters: Pricing parameters
exposure: Optional exposure object for dynamic revenue tracking
seed: Random seed for simulations
"""
self.loss_generator = loss_generator
self.exposure = exposure
self.parameters = parameters or PricingParameters()
self.rng = np.random.default_rng(seed)
# Set loss ratio based on market cycle or explicit value
if market_cycle is not None:
self.loss_ratio = market_cycle.value
self.market_cycle = market_cycle
elif loss_ratio is not None:
self.loss_ratio = loss_ratio
self.market_cycle = self._infer_market_cycle(loss_ratio)
else:
self.loss_ratio = self.parameters.loss_ratio
self.market_cycle = MarketCycle.NORMAL
def _infer_market_cycle(self, loss_ratio: float) -> MarketCycle:
"""Infer market cycle from loss ratio.
Args:
loss_ratio: Target loss ratio
Returns:
Closest market cycle state
"""
if loss_ratio <= 0.65:
return MarketCycle.HARD
if loss_ratio >= 0.75:
return MarketCycle.SOFT
return MarketCycle.NORMAL
[docs]
def calculate_pure_premium(
self,
attachment_point: float,
limit: float,
expected_revenue: float,
simulation_years: Optional[int] = None,
) -> Tuple[float, Dict[str, Any]]:
"""Calculate pure premium for a layer via mean annual aggregate.
Pure premium is the mean of the simulated annual aggregate losses
in the layer (CAS Exam 5 / Werner & Modlin Chapter 4). When a
``development_pattern`` is configured on the pricing parameters,
each simulation year's aggregate losses are developed to ultimate
using age-to-ultimate factors (ASOP 25 / CAS Ratemaking Chapter 4).
Older simulation years are treated as more mature while the most
recent year is treated as the least mature, mirroring standard
experience-rating practice.
Args:
attachment_point: Where layer coverage starts
limit: Maximum coverage from this layer
expected_revenue: Expected annual revenue for scaling
simulation_years: Years to simulate (default from parameters)
Returns:
Tuple of (pure_premium, statistics_dict) with detailed metrics
Raises:
ValueError: If loss_generator is not configured
"""
if self.loss_generator is None:
raise ValueError("Loss generator required for pure premium calculation")
years = simulation_years or self.parameters.simulation_years
# Run simulations to estimate losses in this layer
layer_losses = []
frequencies = []
severities = []
annual_aggregates = []
for year_idx in range(years):
# Generate annual losses (time enables loss cost trending per ASOP 13)
losses, _stats = self.loss_generator.generate_losses(
duration=1.0,
revenue=expected_revenue,
include_catastrophic=True,
time=float(year_idx),
)
# Calculate losses hitting this layer
annual_layer_losses = []
for loss in losses:
if loss.amount > attachment_point:
layer_loss = min(loss.amount - attachment_point, limit)
annual_layer_losses.append(layer_loss)
# Track annual aggregate (including zero-loss years)
annual_aggregates.append(sum(annual_layer_losses))
# Track statistics
if annual_layer_losses:
layer_losses.extend(annual_layer_losses)
frequencies.append(len(annual_layer_losses))
severities.extend(annual_layer_losses)
else:
frequencies.append(0)
# Calculate expected values
if layer_losses:
expected_frequency = float(np.mean(frequencies))
expected_severity = float(np.mean(severities) if severities else 0)
undeveloped_pure_premium = float(np.mean(annual_aggregates))
# ---------------------------------------------------------
# Develop losses to ultimate (Issue #714, ASOP 25)
# ---------------------------------------------------------
# Each simulation year is treated as an accident year at a
# specific maturity. Year 0 has had ``years`` years of
# development; the most recent year (``years - 1``) has had
# only 1 year. Dividing each year's observed aggregate by
# the cumulative percent developed produces the estimated
# ultimate aggregate (standard assumed-pattern chain-ladder).
pattern = self.parameters.development_pattern
if pattern is not None:
developed_aggregates: List[float] = []
for year_idx, agg in enumerate(annual_aggregates):
dev_age = years - year_idx # maturity in years
pct_developed = pattern.get_cumulative_paid(dev_age)
if pct_developed > 0:
developed_aggregates.append(agg / pct_developed)
else:
# No development information — use raw aggregate
developed_aggregates.append(agg)
pure_premium = float(np.mean(developed_aggregates))
development_factor = (
pure_premium / undeveloped_pure_premium if undeveloped_pure_premium > 0 else 1.0
)
# Update annual aggregates for CI calculation
annual_aggregates = developed_aggregates
else:
pure_premium = undeveloped_pure_premium
development_factor = 1.0
# Bootstrap confidence interval on mean annual aggregate (Issue #614)
# CI reflects aggregate pricing uncertainty, not individual loss variability
aggregates_arr = np.array(annual_aggregates)
n_bootstrap = 10_000
indices = self.rng.integers(0, years, size=(n_bootstrap, years))
boot_means = aggregates_arr[indices].mean(axis=1)
alpha = 1 - self.parameters.confidence_level
lower = float(np.percentile(boot_means, 100 * alpha / 2))
upper = float(np.percentile(boot_means, 100 * (1 - alpha / 2)))
confidence_interval = (lower, upper)
else:
expected_frequency = 0.0
expected_severity = 0.0
pure_premium = 0.0
undeveloped_pure_premium = 0.0
development_factor = 1.0
confidence_interval = (0.0, 0.0)
statistics = {
"expected_frequency": expected_frequency,
"expected_severity": expected_severity,
"pure_premium": pure_premium,
"undeveloped_pure_premium": undeveloped_pure_premium,
"development_factor": development_factor,
"confidence_interval": confidence_interval,
"years_simulated": years,
"total_losses_in_layer": len(layer_losses),
"max_loss_in_layer": max(layer_losses) if layer_losses else 0,
"attachment_point": attachment_point,
"limit": limit,
"annual_aggregates": annual_aggregates,
}
return pure_premium, statistics
[docs]
def calculate_technical_premium(
self,
pure_premium: float,
limit: float,
) -> float:
"""Convert pure premium to technical premium with risk and LAE loading.
Technical premium adds a risk loading for parameter uncertainty
to the pure premium, plus LAE (loss adjustment expense) as a known
cost component per ASOP 29. Expense and profit margins are applied
separately via the actuarial pricing identity in
calculate_market_premium() to avoid double-counting.
Formula:
technical_premium = pure_premium * (1 + risk_loading)
+ pure_premium * lae_ratio
Args:
pure_premium: Expected loss cost
limit: Layer limit for rate capping
Returns:
Technical premium amount
"""
# Add risk loading for uncertainty
risk_loading = 1 + self.parameters.risk_loading
# LAE is a known cost component added on top of risk-loaded premium
lae_loading = pure_premium * self.parameters.lae_ratio
# Calculate technical premium (expense/profit applied via loss ratio)
technical_premium = (pure_premium * risk_loading) + lae_loading
# Apply minimum premium
technical_premium = max(technical_premium, self.parameters.min_premium)
# Cap at maximum rate on line
max_premium = limit * self.parameters.max_rate_on_line
technical_premium = min(technical_premium, max_premium)
return technical_premium
[docs]
def calculate_market_premium(
self,
technical_premium: float,
market_cycle: Optional[MarketCycle] = None,
) -> float:
"""Apply expense, profit, and market cycle loadings to technical premium.
Uses the standard actuarial pricing identity:
Premium = Pure_Premium / (1 - V - Q)
where *V* is the expense ratio and *Q* is the profit margin
(Werner & Modlin, *Basic Ratemaking*, Ch. 7). The market cycle
then scales this indicated premium to reflect competitive pressure.
With default parameters (V=0.25, Q=0.05, loss_ratio=0.70) the
formula reduces to ``technical_premium / cycle_loss_ratio``,
preserving backward compatibility.
Args:
technical_premium: Premium with risk and LAE loadings
market_cycle: Optional market cycle override
Returns:
Market-adjusted premium incorporating expenses, profit,
and competitive cycle effects
"""
import warnings
cycle = market_cycle or self.market_cycle
cycle_lr = cycle.value
V = self.parameters.expense_ratio
Q = self.parameters.profit_margin
indicated_lr = 1 - V - Q
# Actuarial base premium: covers losses, expenses, and profit
base_premium = technical_premium / indicated_lr
# Market cycle adjustment relative to the target loss ratio.
# Hard market (cycle_lr < target): premiums rise
# Soft market (cycle_lr > target): premiums fall
target_lr = self.parameters.loss_ratio
cycle_factor = target_lr / cycle_lr
market_premium = base_premium * cycle_factor
# Warn when the market cycle implies a combined ratio > 100%,
# meaning the insurer would operate at an underwriting loss.
combined_ratio = cycle_lr + V + Q
if combined_ratio > 1.0 + 1e-9:
logger.warning(
"Combined ratio (%.1f%%) exceeds 100%%: "
"market cycle loss ratio (%.0f%%) + "
"expense ratio (%.0f%%) + profit margin (%.0f%%). "
"The insurer would operate at an underwriting loss "
"under current market conditions.",
combined_ratio * 100,
cycle_lr * 100,
V * 100,
Q * 100,
)
return market_premium
[docs]
def price_layer(
self,
attachment_point: float,
limit: float,
expected_revenue: float,
market_cycle: Optional[MarketCycle] = None,
) -> LayerPricing:
"""Price a single insurance layer.
Complete pricing process from pure premium through market adjustment.
Args:
attachment_point: Where coverage starts
limit: Maximum coverage amount
expected_revenue: Expected annual revenue
market_cycle: Optional market cycle override
Returns:
LayerPricing object with all pricing details
"""
# Calculate pure premium
pure_premium, stats = self.calculate_pure_premium(attachment_point, limit, expected_revenue)
# Calculate technical premium with loadings
technical_premium = self.calculate_technical_premium(pure_premium, limit)
# Apply market cycle adjustment
market_premium = self.calculate_market_premium(technical_premium, market_cycle=market_cycle)
# Calculate rate on line
rate_on_line = market_premium / limit if limit > 0 else 0.0
# LAE loading from dedicated ALAE/ULAE ratios (Issue #616)
lae_loading = pure_premium * self.parameters.lae_ratio
return LayerPricing(
attachment_point=attachment_point,
limit=limit,
expected_frequency=stats["expected_frequency"],
expected_severity=stats["expected_severity"],
pure_premium=pure_premium,
technical_premium=technical_premium,
market_premium=market_premium,
rate_on_line=rate_on_line,
confidence_interval=stats["confidence_interval"],
lae_loading=lae_loading,
development_factor=stats["development_factor"],
)
[docs]
def price_insurance_program(
self,
program: "InsuranceProgram",
expected_revenue: Optional[float] = None,
time: float = 0.0,
market_cycle: Optional[MarketCycle] = None,
update_program: bool = True,
) -> "InsuranceProgram":
"""Price a complete insurance program.
Prices all layers in the program and optionally updates their rates.
Args:
program: Insurance program to price
expected_revenue: Expected annual revenue (optional if using exposure)
time: Time for exposure calculation (default 0.0)
market_cycle: Optional market cycle override
update_program: Whether to update program layer rates
Returns:
Program with updated pricing (original or copy based on update_program)
"""
from .insurance_program import InsuranceProgram
# Create a copy if not updating in place
if not update_program:
import copy
program = copy.deepcopy(program)
# Get actual revenue from exposure if available, otherwise use expected_revenue
if self.exposure is not None:
actual_revenue = self.exposure.get_exposure(time)
elif expected_revenue is not None:
actual_revenue = expected_revenue
else:
raise ValueError("Either expected_revenue or exposure must be provided")
# Price each layer
pricing_results = []
for layer in program.layers:
layer_pricing = self.price_layer(
attachment_point=layer.attachment_point,
limit=layer.limit,
expected_revenue=actual_revenue,
market_cycle=market_cycle,
)
pricing_results.append(layer_pricing)
# Update layer premium rate if requested
if update_program:
layer.base_premium_rate = layer_pricing.rate_on_line
# Store pricing results in program for reference
if not hasattr(program, "pricing_results"):
program.pricing_results = []
program.pricing_results = pricing_results
return program
[docs]
def price_insurance_policy(
self,
policy: "InsurancePolicy",
expected_revenue: float,
market_cycle: Optional[MarketCycle] = None,
update_policy: bool = True,
) -> "InsurancePolicy":
"""Price a basic insurance policy.
.. deprecated::
Use :meth:`price_insurance_program` instead.
Prices all layers in the policy and optionally updates their rates.
Args:
policy: Insurance policy to price
expected_revenue: Expected annual revenue
market_cycle: Optional market cycle override
update_policy: Whether to update policy layer rates
Returns:
Policy with updated pricing (original or copy based on update_policy)
"""
warnings.warn(
"price_insurance_policy() is deprecated. "
"Use price_insurance_program() with an InsuranceProgram instead.",
ErgodicInsuranceDeprecationWarning,
stacklevel=2,
)
from .insurance import InsurancePolicy
# Create a copy if not updating in place
if not update_policy:
import copy
policy = copy.deepcopy(policy)
# Price each layer
pricing_results = []
for layer in policy.layers:
layer_pricing = self.price_layer(
attachment_point=layer.attachment_point,
limit=layer.limit,
expected_revenue=expected_revenue,
market_cycle=market_cycle,
)
pricing_results.append(layer_pricing)
# Update layer rate if requested
if update_policy:
layer.rate = layer_pricing.rate_on_line
# Store pricing results for reference
if not hasattr(policy, "pricing_results"):
policy.pricing_results = []
policy.pricing_results = pricing_results
return policy
[docs]
def compare_market_cycles(
self,
attachment_point: float,
limit: float,
expected_revenue: float,
) -> Dict[str, LayerPricing]:
"""Compare pricing across different market cycles.
Useful for understanding market impact on premiums.
Args:
attachment_point: Where coverage starts
limit: Maximum coverage amount
expected_revenue: Expected annual revenue
Returns:
Dictionary mapping market cycle names to pricing results
"""
# Calculate pure premium ONCE (it doesn't change by market cycle)
pure_premium, stats = self.calculate_pure_premium(attachment_point, limit, expected_revenue)
# Calculate technical premium ONCE (also doesn't change by market cycle)
technical_premium = self.calculate_technical_premium(pure_premium, limit)
results = {}
# LAE loading from dedicated ALAE/ULAE ratios (Issue #616)
lae_loading = pure_premium * self.parameters.lae_ratio
# Apply different market cycle adjustments to the same technical premium
for cycle in MarketCycle:
# Only the market adjustment changes
market_premium = self.calculate_market_premium(technical_premium, market_cycle=cycle)
rate_on_line = market_premium / limit if limit > 0 else 0.0
results[cycle.name] = LayerPricing(
attachment_point=attachment_point,
limit=limit,
expected_frequency=stats["expected_frequency"],
expected_severity=stats["expected_severity"],
pure_premium=pure_premium,
technical_premium=technical_premium,
market_premium=market_premium,
rate_on_line=rate_on_line,
confidence_interval=stats["confidence_interval"],
lae_loading=lae_loading,
development_factor=stats["development_factor"],
)
return results
[docs]
def simulate_cycle_transition(
self,
program: "InsuranceProgram",
expected_revenue: float,
years: int = 10,
transition_probs: Optional[Dict[str, float]] = None,
) -> List[Dict[str, Any]]:
"""Simulate insurance pricing over market cycle transitions.
Models how premiums change as markets transition between states.
Args:
program: Insurance program to simulate
expected_revenue: Expected annual revenue
years: Number of years to simulate
transition_probs: Market transition probabilities
Returns:
List of annual results with cycle states and premiums
"""
if transition_probs is None:
# Default transition probabilities
transition_probs = {
"hard_to_normal": 0.4,
"hard_to_soft": 0.1,
"normal_to_hard": 0.2,
"normal_to_soft": 0.2,
"soft_to_normal": 0.3,
"soft_to_hard": 0.1,
}
results = []
current_cycle = self.market_cycle
for year in range(years):
# Price program for current cycle
priced_program = self.price_insurance_program(
program=program,
expected_revenue=expected_revenue,
market_cycle=current_cycle,
update_program=False,
)
# Calculate total premium from actual pricing results
if hasattr(priced_program, "pricing_results") and priced_program.pricing_results:
total_premium = sum(pr.market_premium for pr in priced_program.pricing_results)
else:
# Fallback to calculating from layers
total_premium = priced_program.calculate_premium()
# Store results
results.append(
{
"year": year,
"market_cycle": current_cycle.name,
"loss_ratio": current_cycle.value,
"total_premium": total_premium,
"layer_premiums": [pr.market_premium for pr in priced_program.pricing_results],
}
)
# Transition to next cycle
current_cycle = self._transition_cycle(current_cycle, transition_probs)
return results
def _transition_cycle(
self,
current: MarketCycle,
probs: Dict[str, float],
) -> MarketCycle:
"""Simulate market cycle transition.
Args:
current: Current market cycle
probs: Transition probabilities
Returns:
Next market cycle state
"""
rand = self.rng.random()
if current == MarketCycle.HARD:
if rand < probs.get("hard_to_normal", 0.4):
return MarketCycle.NORMAL
if rand < probs.get("hard_to_normal", 0.4) + probs.get("hard_to_soft", 0.1):
return MarketCycle.SOFT
return MarketCycle.HARD
if current == MarketCycle.NORMAL:
if rand < probs.get("normal_to_hard", 0.2):
return MarketCycle.HARD
if rand < probs.get("normal_to_hard", 0.2) + probs.get("normal_to_soft", 0.2):
return MarketCycle.SOFT
return MarketCycle.NORMAL
# SOFT
if rand < probs.get("soft_to_normal", 0.3):
return MarketCycle.NORMAL
if rand < probs.get("soft_to_normal", 0.3) + probs.get("soft_to_hard", 0.1):
return MarketCycle.HARD
return MarketCycle.SOFT
[docs]
@staticmethod
def create_from_config(
config: Dict[str, Any],
loss_generator: Optional["ManufacturingLossGenerator"] = None,
) -> "InsurancePricer":
"""Create pricer from configuration dictionary.
Args:
config: Configuration dictionary. Supports an optional
``development_pattern`` key whose value is the name of a
standard ``DevelopmentPatternType`` (e.g. ``"long_tail_10yr"``)
or a dict with ``factors`` and optional ``tail_factor``.
loss_generator: Optional loss generator
Returns:
Configured InsurancePricer instance
"""
# Resolve development pattern
dev_cfg = config.get("development_pattern")
dev_pattern: Optional[ClaimDevelopment] = None
if dev_cfg is not None:
if isinstance(dev_cfg, ClaimDevelopment):
dev_pattern = dev_cfg
elif isinstance(dev_cfg, str):
_PATTERN_FACTORIES = {
"immediate": ClaimDevelopment.create_immediate,
"medium_tail_5yr": ClaimDevelopment.create_medium_tail_5yr,
"long_tail_10yr": ClaimDevelopment.create_long_tail_10yr,
"very_long_tail_15yr": ClaimDevelopment.create_very_long_tail_15yr,
}
factory = _PATTERN_FACTORIES.get(dev_cfg.lower())
if factory is None:
raise ValueError(
f"Unknown development pattern '{dev_cfg}'. "
f"Valid names: {list(_PATTERN_FACTORIES.keys())}"
)
dev_pattern = factory()
elif isinstance(dev_cfg, dict):
dev_pattern = ClaimDevelopment(
pattern_name=dev_cfg.get("name", "custom"),
development_factors=dev_cfg["factors"],
tail_factor=dev_cfg.get("tail_factor", 0.0),
)
# Extract pricing parameters
params = PricingParameters(
loss_ratio=config.get("loss_ratio", 0.70),
expense_ratio=config.get("expense_ratio", 0.25),
profit_margin=config.get("profit_margin", 0.05),
risk_loading=config.get("risk_loading", 0.10),
confidence_level=config.get("confidence_level", 0.95),
simulation_years=config.get("simulation_years", 10),
min_premium=config.get("min_premium", 1000.0),
max_rate_on_line=config.get("max_rate_on_line", 0.50),
alae_ratio=config.get("alae_ratio", 0.10),
ulae_ratio=config.get("ulae_ratio", 0.05),
development_pattern=dev_pattern,
)
# Get market cycle
cycle_str = config.get("market_cycle", "NORMAL")
try:
market_cycle = MarketCycle[cycle_str.upper()]
except KeyError:
market_cycle = MarketCycle.NORMAL
return InsurancePricer(
loss_generator=loss_generator,
market_cycle=market_cycle,
parameters=params,
seed=config.get("seed"),
)