black

calibration_conditions.py

@@ -31,100 +31,158 @@
 import condition_aux
 import isleconfig
 
+
 def condition_stationary_state_cash(logobj):
     """Stationarity test for total cash"""
-    return condition_aux.condition_stationary_state(logobj.history_logs['total_cash'])
-
+    return condition_aux.condition_stationary_state(logobj.history_logs["total_cash"])
+
+
 def condition_stationary_state_excess_capital(logobj):
     """Stationarity test for total excess capital"""
-    return condition_aux.condition_stationary_state(logobj.history_logs['total_excess_capital'])
+    return condition_aux.condition_stationary_state(
+        logobj.history_logs["total_excess_capital"]
+    )
+
 
 def condition_stationary_state_profits_losses(logobj):
     """Stationarity test for total profits and losses"""
-    return condition_aux.condition_stationary_state(logobj.history_logs['total_profitslosses'])
+    return condition_aux.condition_stationary_state(
+        logobj.history_logs["total_profitslosses"]
+    )
+
 
 def condition_stationary_state_contracts(logobj):
     """Stationarity test for total number of contracts"""
-    return condition_aux.condition_stationary_state(logobj.history_logs['total_contracts'])
+    return condition_aux.condition_stationary_state(
+        logobj.history_logs["total_contracts"]
+    )
+
 
 def condition_stationary_state_rein_cash(logobj):
     """Stationarity test for total cash (reinsurers)"""
-    return condition_aux.condition_stationary_state(logobj.history_logs['total_reincash'])
+    return condition_aux.condition_stationary_state(
+        logobj.history_logs["total_reincash"]
+    )
+
 
 def condition_stationary_state_rein_excess_capital(logobj):
     """Stationarity test for total excess capital (reinsurers)"""
-    return condition_aux.condition_stationary_state(logobj.history_logs['total_reinexcess_capital'])
+    return condition_aux.condition_stationary_state(
+        logobj.history_logs["total_reinexcess_capital"]
+    )
+
 
 def condition_stationary_state_rein_profits_losses(logobj):
     """Stationarity test for total profits and losses (reinsurers)"""
-    return condition_aux.condition_stationary_state(logobj.history_logs['total_reinprofitslosses'])
+    return condition_aux.condition_stationary_state(
+        logobj.history_logs["total_reinprofitslosses"]
+    )
+
 
 def condition_stationary_state_rein_contracts(logobj):
     """Stationarity test for total number of reinsured contracts"""
-    return condition_aux.condition_stationary_state(logobj.history_logs['total_reincontracts'])
+    return condition_aux.condition_stationary_state(
+        logobj.history_logs["total_reincontracts"]
+    )
+
 
 def condition_stationary_state_market_premium(logobj):
     """Stationarity test for insurance market premium"""
-    return condition_aux.condition_stationary_state(logobj.history_logs['market_premium'])
+    return condition_aux.condition_stationary_state(
+        logobj.history_logs["market_premium"]
+    )
+
 
 def condition_stationary_state_rein_market_premium(logobj):
     """Stationarity test for reinsurance market premium"""
-    return condition_aux.condition_stationary_state(logobj.history_logs['market_reinpremium'])
+    return condition_aux.condition_stationary_state(
+        logobj.history_logs["market_reinpremium"]
+    )
 
-def condition_defaults_insurance(logobj):          # TODO: develop this into a non-binary measure
+
+def condition_defaults_insurance(
+    logobj
+):  # TODO: develop this into a non-binary measure
     """Test for number of insurance bankruptcies (non zero, not all insurers)"""
-    #series  = logobj.history_logs['total_operational']
-    #if series[-1] != 0 and any(series[i]-series[i-1] < 0 for i in range(1,len(series))):
-    opseries = [logobj.history_logs["insurance_firms_cash"][-1][i][2] for i in \
-                        range(len(logobj.history_logs["insurance_firms_cash"][-1]))]
+    # series  = logobj.history_logs['total_operational']
+    # if series[-1] != 0 and any(series[i]-series[i-1] < 0 for i in range(1,len(series))):
+    opseries = [
+        logobj.history_logs["insurance_firms_cash"][-1][i][2]
+        for i in range(len(logobj.history_logs["insurance_firms_cash"][-1]))
+    ]
     if any(opseries) and not all(opseries):
         return 1
     else:
         return 0
 
-def condition_defaults_reinsurance(logobj):        # TODO: develop this into a non-binary measure
+
+def condition_defaults_reinsurance(
+    logobj
+):  # TODO: develop this into a non-binary measure
     """Test for number of reinsurance bankruptcies (non zero, not all reinsurers)"""
-    #series  = logobj.history_logs['total_reinoperational']
-    #if series[-1] != 0 and any(series[i]-series[i-1] < 0 for i in range(1,len(series))):
-    opseries = [logobj.history_logs["reinsurance_firms_cash"][-1][i][2] for i in \
-                        range(len(logobj.history_logs["reinsurance_firms_cash"][-1]))]
+    # series  = logobj.history_logs['total_reinoperational']
+    # if series[-1] != 0 and any(series[i]-series[i-1] < 0 for i in range(1,len(series))):
+    opseries = [
+        logobj.history_logs["reinsurance_firms_cash"][-1][i][2]
+        for i in range(len(logobj.history_logs["reinsurance_firms_cash"][-1]))
+    ]
     if any(opseries) and not all(opseries):
         return 1
     else:
         return 0
 
+
 def condition_insurance_coverage(logobj):
     """Test for insurance coverage close to 100%"""
-    return logobj.history_logs['total_contracts'][-1] * 1. / isleconfig.simulation_parameters["no_risks"]
+    return (
+        logobj.history_logs["total_contracts"][-1]
+        * 1.0
+        / isleconfig.simulation_parameters["no_risks"]
+    )
+
 
 def condition_reinsurance_coverage(logobj, minimum=0.6):
     """Test for reinsurance coverage close to some minimum that may be less than 100% (default 60%)"""
-    score = logobj.history_logs['total_reincontracts'][-1] * 1. / (minimum * logobj.history_logs['total_contracts'][-1])
-    score = 1 if score>1 else score
+    score = (
+        logobj.history_logs["total_reincontracts"][-1]
+        * 1.0
+        / (minimum * logobj.history_logs["total_contracts"][-1])
+    )
+    score = 1 if score > 1 else score
     return score
 
-def condition_insurance_firm_dist(logobj):                 
+
+def condition_insurance_firm_dist(logobj):
     """Empirical calibration test for insurance firm size (total assets; cash)"""
     """filter operational firms"""
-    #dist = [logobj.history_logs["insurance_firms_cash"][-1][i][0] for i in range(len(logobj.history_logs["insurance_firms_cash"])) if \
+    # dist = [logobj.history_logs["insurance_firms_cash"][-1][i][0] for i in range(len(logobj.history_logs["insurance_firms_cash"])) if \
     #           logobj.history_logs["insurance_firms_cash"][-1][i][0] > isleconfig.simulation_parameters["cash_permanency_limit"]]
-    dist = [logobj.history_logs["insurance_firms_cash"][-1][i][0] for i in \
-                range(len(logobj.history_logs["insurance_firms_cash"][-1])) if \
-                      logobj.history_logs["insurance_firms_cash"][-1][i][2]]
+    dist = [
+        logobj.history_logs["insurance_firms_cash"][-1][i][0]
+        for i in range(len(logobj.history_logs["insurance_firms_cash"][-1]))
+        if logobj.history_logs["insurance_firms_cash"][-1][i][2]
+    ]
     """run two-sided KS test"""
-    KS_statistic, p_value = stats.ks_2samp(condition_aux.scaler(condition_aux.insurance_firm_sizes_empirical_2017), 
-                                           condition_aux.scaler(dist))
+    KS_statistic, p_value = stats.ks_2samp(
+        condition_aux.scaler(condition_aux.insurance_firm_sizes_empirical_2017),
+        condition_aux.scaler(dist),
+    )
     return p_value
 
-def condition_reinsurance_firm_dist(logobj):                 
+
+def condition_reinsurance_firm_dist(logobj):
     """Empirical calibration test for reinsurance firm size (total assets; cash)"""
     """filter operational firms"""
-    #dist = [logobj.history_logs["reinsurance_firms_cash"][-1][i][0] for i in range(len(logobj.history_logs["reinsurance_firms_cash"])) if \
+    # dist = [logobj.history_logs["reinsurance_firms_cash"][-1][i][0] for i in range(len(logobj.history_logs["reinsurance_firms_cash"])) if \
     #           logobj.history_logs["reinsurance_firms_cash"][-1][i][0] > isleconfig.simulation_parameters["cash_permanency_limit"]]
-    dist = [logobj.history_logs["reinsurance_firms_cash"][-1][i][0] for i in \
-                    range(len(logobj.history_logs["reinsurance_firms_cash"][-1])) if 
-                    logobj.history_logs["reinsurance_firms_cash"][-1][i][2]]
+    dist = [
+        logobj.history_logs["reinsurance_firms_cash"][-1][i][0]
+        for i in range(len(logobj.history_logs["reinsurance_firms_cash"][-1]))
+        if logobj.history_logs["reinsurance_firms_cash"][-1][i][2]
+    ]
     """run two-sided KS test"""
-    KS_statistic, p_value = stats.ks_2samp(condition_aux.scaler(condition_aux.reinsurance_firm_sizes_empirical_2017), 
-                                           condition_aux.scaler(dist))
+    KS_statistic, p_value = stats.ks_2samp(
+        condition_aux.scaler(condition_aux.reinsurance_firm_sizes_empirical_2017),
+        condition_aux.scaler(dist),
+    )
     return p_value

calibrationscore.py

@@ -5,9 +5,10 @@
 from inspect import getmembers, isfunction
 import numpy as np
 
-import calibration_conditions   # Test functions
+import calibration_conditions  # Test functions
 
-class CalibrationScore():
+
+class CalibrationScore:
     def __init__(self, L):
         """Constructor method.
             Arguments:
@@ -17,31 +18,41 @@ def __init__(self, L):
         """Assert sanity of log and save log."""
         assert isinstance(L, logger.Logger)
         self.logger = L
-        
+
         """Prepare list of calibration tests from calibration_conditions.py"""
-        self.conditions = [f for f in getmembers(calibration_conditions) if isfunction(f[1])]
-
+        self.conditions = [
+            f for f in getmembers(calibration_conditions) if isfunction(f[1])
+        ]
+
         """Prepare calibration score variable."""
         self.calibration_score = None
-        
+
     def test_all(self):
         """Method to test all calibration tests.
             No arguments.
             Returns combined calibration score as float \in [0,1]."""
-        
+
         """Compute score components"""
-        scores = {condition[0]: condition[1](self.logger) for condition in self.conditions}
+        scores = {
+            condition[0]: condition[1](self.logger) for condition in self.conditions
+        }
         """Print components"""
         print("\n")
         for cond_name, score in scores.items():
             print("{0:47s}: {1:8f}".format(cond_name, score))
         """Compute combined score"""
-        self.calibration_score = self.combine_scores(np.array([*scores.values()], dtype=object))
+        self.calibration_score = self.combine_scores(
+            np.array([*scores.values()], dtype=object)
+        )
         """Print combined score"""
-        print("\n                        Total calibration score: {0:8f}".format(self.calibration_score))
+        print(
+            "\n                        Total calibration score: {0:8f}".format(
+                self.calibration_score
+            )
+        )
         """Return"""
         return self.calibration_score
-    
+
     def combine_scores(self, slist):
         """Method to combine calibration score components. Combination is additive (mean). Change the function
            for other combination methods (multiplicative or minimum).

catbond.py

@@ -1,4 +1,3 @@
-
 import isleconfig
 import numpy as np
 import scipy.stats
@@ -38,23 +37,34 @@ def iterate(self, time):
         self.simulation.bank.award_interest(self, self.cash)
         self.effect_payments(time)
         if isleconfig.verbose:
-            print(time, ":", self.id, len(self.underwritten_contracts), self.cash, self.operational)
-
+            print(
+                time,
+                ":",
+                self.id,
+                len(self.underwritten_contracts),
+                self.cash,
+                self.operational,
+            )
+
             """mature contracts"""
             print("Number of underwritten contracts ", len(self.underwritten_contracts))
-        maturing = [contract for contract in self.underwritten_contracts if contract.expiration <= time]
+        maturing = [
+            contract
+            for contract in self.underwritten_contracts
+            if contract.expiration <= time
+        ]
         for contract in maturing:
             self.underwritten_contracts.remove(contract)
             contract.mature(time)
 
         [contract.check_payment_due(time) for contract in self.underwritten_contracts]
-        
+
         if self.underwritten_contracts == []:
             self.mature_bond()
-        else:   #TODO: dividend should only be payed according to pre-arranged schedule, and only if no risk events have materialized so far
+        else:  # TODO: dividend should only be payed according to pre-arranged schedule, and only if no risk events have materialized so far
             if self.operational:
                 self.pay_dividends(time)
-    
+
     def set_owner(self, owner):
         """Method to set owner of the Cat Bond.
             Accepts:
@@ -63,26 +73,30 @@ def set_owner(self, owner):
         self.owner = owner
         if isleconfig.verbose:
             print("SOLD")
-    
+
     def set_contract(self, contract):
         """Method to record new instances of CatBonds.
             Accepts:
                 owner: Type class
             No return values
         Only one contract is ever added to the list of underwritten contracts as each CatBond is a contract itself."""
         self.underwritten_contracts.append(contract)
-    
+
     def mature_bond(self):
         """Method to mature CatBond.
             No accepted values
             No return values
         When the catbond contract matures this is called which pays the value of the catbond to the simulation, and is
         then deleted from the list of agents."""
         if self.operational:
-            obligation = {"amount": self.cash, "recipient": self.simulation, "due_time": 1, "purpose": 'mature'}
+            obligation = {
+                "amount": self.cash,
+                "recipient": self.simulation,
+                "due_time": 1,
+                "purpose": "mature",
+            }
             self.pay(obligation)
             self.simulation.delete_agents("catbond", [self])
             self.operational = False
-        else: print('CatBond is not operational so cannot mature')
-
-
+        else:
+            print("CatBond is not operational so cannot mature")