import pandas as pd
import numpy as np
from .utils_cat_bonds import multi_level_es
from climada_petals.util.config import LOGGER
[docs]
class SingleCountryBondSimulation:
"""
Single-country catastrophe bond simulation utilities.
This class simulates losses and returns for a single country's bond
based on per-event payouts and damages.
"""
[docs]
def __init__(self, subarea_calc, term: int, start_year: int, number_of_terms: int):
"""
Initialize with subarea data and simulation horizon.
Parameters
----------
subarea_calc : object
Subarea calculation instance.
term : int
Bond term in years for each simulated period.
start_year : int
The starting year of the simulation.
number_of_terms : int
Number of consecutive terms to simulate.
"""
self.term = term
self.start_year = start_year
self.number_of_terms = number_of_terms
self.subarea_calc = subarea_calc
'''Simulate one term of bond to derive losses'''
[docs]
def init_bond_loss(self, events_per_year: list[pd.DataFrame]):
"""
Calculates the expected losses for a catastrophe bond over its term.
This function simulates the bond's loss experience given a sequence of event data per year and month,
tracking payouts, damages, remaining princpal value, and the timing of losses. It returns the relative
losses, the total payouts and damages per term, and a DataFrame detailing losses and their corresponding months.
Parameters
----------
events_per_year : list[pandas.DataFrame]
A list where each element is a DataFrame representing events in a year. Each DataFrame must
contain at least 'month' and 'pay' columns, where 'pay' is the payout for each event.
Returns
-------
rel_annual_losses : numpy.ndarray
Array of relative payouts/losses per year (losses divided by principal).
rel_monthly_loss : pandas.DataFrame
DataFrame with columns 'losses' and 'months', detailing the losses and their corresponding
months for each year.
summed_payouts : float
The total summed payouts over the bond's term.
summed_damages : float
The total summed damages over the bond's term.
"""
principal0 = self.subarea_calc.principal
principal = principal0
# Use Python lists only for month-level output (tiny)
df_monthly = pd.DataFrame(columns=[
"losses", "months"], dtype=object
)
annual_losses = pd.Series(0.0, index=range(self.term))
summed_damages = 0.0
for year, ev in enumerate(events_per_year):
# Extract arrays
months = ev["month"].to_numpy()
pays = ev["pay"].to_numpy()
damages = ev["damage"].to_numpy()
summed_damages += damages.sum()
# Running cumulative payout to detect exhaustion
cum = np.cumsum(pays)
# Identify first index where principal is exceeded
exhaust_idx = np.searchsorted(cum, principal, side="right")
if exhaust_idx == len(pays):
# principal never exhausted → no capping needed
payouts = pays.copy()
principal -= payouts.sum()
else:
# principal exhausted at this index
payouts = np.zeros_like(pays, dtype=float)
# All payouts before exhaustion are exact
if exhaust_idx > 0:
payouts[:exhaust_idx] = pays[:exhaust_idx]
# Payout at exhaustion month: whatever principal remains
prev_cum = cum[exhaust_idx-1] if exhaust_idx > 0 else 0
payouts[exhaust_idx] = principal - prev_cum
# After that → principal is 0, so payouts remain 0
principal = 0.0
# Store relative losses and months as arrays for consistent indexing
df_monthly.loc[year, "losses"] = list(payouts / principal0)
df_monthly.loc[year, "months"] = list(months)
# Sum for annual loss
annual_losses[year] = payouts.sum()
rel_annual_losses = annual_losses / principal0
summed_payouts = annual_losses.sum()
return rel_annual_losses, df_monthly, summed_payouts, summed_damages
[docs]
def init_loss_simulation(self, confidence_levels: list[float] = [0.95, 0.99]):
"""
Simulate losses, payouts, damages, and risk metrics for a catastrophe bond.
Parameters
----------
confidence_levels : list[float], optional
Confidence levels for VaR and ES calculation.
Sets
-------
df_loss_month : pd.DataFrame
Monthly loss data for all simulations.
loss_metrics : dict
Expected loss, attachment probability, total payouts/damages,
VaR and ES metrics for given confidence levels.
"""
pay_vs_dam = self.subarea_calc.pay_vs_dam
annual_losses = []
list_loss_month = []
total_payouts = 0
total_damages = 0
# Iterate over non-overlapping term blocks
for bond_date in range(self.start_year, self.start_year + self.number_of_terms * self.term, self.term):
# Collect events for the full term (vectorized selection)
events_per_year = [
pay_vs_dam[pay_vs_dam['year'] == (bond_date + offset)].groupby(['month', 'year']).sum().reset_index().sort_values(by=['year','month'])
for offset in range(self.term)
]
ann_losses_term, monthly_losses, summed_payouts, summed_damages = (
self.init_bond_loss(events_per_year)
)
annual_losses.extend(ann_losses_term)
list_loss_month.append(monthly_losses)
total_payouts += summed_payouts
total_damages += summed_damages
# Combine monthly losses
self.df_loss_month = pd.concat(list_loss_month, ignore_index=True)
annual_losses = pd.Series(annual_losses)
exp_loss_ann = annual_losses.mean()
att_prob = (annual_losses > 0).mean()
# Save metrics
self.loss_metrics = {
'Expected_annual_loss': exp_loss_ann,
'Annual_attachment_probability': att_prob,
'Total_payout': total_payouts,
'Total_damages': total_damages,
}
var_list, es_list = multi_level_es(annual_losses, confidence_levels)
for cl, var, es in zip(confidence_levels, var_list, es_list):
self.loss_metrics[f'Annual_Value_at_risk_{int(cl*100)}'] = var
self.loss_metrics[f'Annual_Expected_shortfall_{int(cl*100)}'] = es
LOGGER.info(f'Expected Loss = {exp_loss_ann}')
LOGGER.info(f'Attachment Probability = {att_prob}')
'''Simulate over all terms of bond to derive returns'''
[docs]
def init_return_simulation(self, premium: float):
"""
Simulates the performance of a catastrophe bond over the simulation period, premiums and returns.
This function models the bond's payouts, premiums, and returns over a series of simulated years.
It aggregates annual and total returns and computes Sharpe ratios.
Parameters
----------
premium : float
Annual premium rate for the bond.
Returns
-------
return_metrics : dict
Dictionary with annual premiums, annual returns, total returns,
total premiums, and Sharpe ratio.
"""
premiums_tot = []
ncf_tot = []
cur_nominal = 1
for i in range(len(self.df_loss_month)):
losses = self.df_loss_month['losses'].iloc[i]
months = self.df_loss_month['months'].iloc[i]
if np.sum(losses) == 0:
prem_tmp = cur_nominal * premium
premiums_tot.append(prem_tmp)
ncf_tot.append(prem_tmp)
else:
ncf_tot_tmp = []
premiums_tot_tmp = []
prem_tmp = cur_nominal * premium / 12 * months[0]
premiums_tot_tmp.append(prem_tmp)
ncf_tot_tmp.append(prem_tmp)
for j in range(len(losses)):
loss = losses[j]
month = months[j]
cur_nominal -= loss
if cur_nominal < 0:
loss += cur_nominal
cur_nominal = 0
else:
pass
if j + 1 < len(losses):
next_month = months[j+1]
prem_tmp = ((cur_nominal * premium) / 12 * (next_month - month))
premiums_tot_tmp.append(prem_tmp)
ncf_tot_tmp.append(prem_tmp - loss)
else:
prem_tmp = ((cur_nominal * premium) / 12 * (12- month))
premiums_tot_tmp.append(prem_tmp)
ncf_tot_tmp.append(prem_tmp - loss)
ncf_tot.append(np.sum(ncf_tot_tmp))
premiums_tot.append(np.sum(premiums_tot_tmp))
if (i + 1) % self.term == 0:
cur_nominal = 1
sharpe_ratio = (np.mean(ncf_tot) / np.std(ncf_tot)) if np.std(ncf_tot) != 0 else np.nan
self.return_metrics = {'annual_premiums': np.array(premiums_tot), 'annual_returns': np.array(ncf_tot),
'total_returns': np.sum(np.array(ncf_tot)) * self.subarea_calc.principal , 'total_premiums': np.sum(np.array(premiums_tot)) * self.subarea_calc.principal,
'sharpe_ratio': sharpe_ratio}
