add function to derive optimal fixed pools

Author: KaiOBerg

climada_petals/engine/cat_bonds/pooling_functions.py

@@ -3,6 +3,85 @@
 from math import comb
 from pymoo.core.problem import ElementwiseProblem
 from pymoo.core.variable import Integer
+import pandas as pd
+from pymoo.operators.sampling.rnd import IntegerRandomSampling
+from pymoo.operators.mutation.pm import PolynomialMutation
+from pymoo.operators.crossover.hux import HalfUniformCrossover
+from pymoo.algorithms.soo.nonconvex.ga import GA
+from pymoo.optimize import minimize
+from pymoo.operators.repair.rounding import RoundingRepair
+
+def process_n(n, cntry_names, sng_ann_losses, principal_sng_dic, n_opt_rep=100):
+    """
+    Runs risk concentration minimization for a given number of pools using a genetic algorithm,
+    processes the optimization results, and generates convergence plots.
+    Parameters
+    ----------
+    n : int
+        Number of pools to optimize.
+    cntry_names : list of str
+        List of country names corresponding to the columns in the loss data.
+    sng_ann_losses : pandas.DataFrame
+        Dataframe of annual loss data for each country.
+    principal_sng_dic : dict
+        Principal values for each country.
+    n_opt_rep : int, optional
+        Number of optimization repetitions for seed analysis (default is 100).
+    Returns
+    -------
+    country_allocation : pandas.DataFrame
+        DataFrame containing a optimal country allocation for the minimum concentration solution.
+    algorithm_result: pymoo.core.result.Result
+        Result object from the optimization containing details of the optimization process.
+    """
+    opt_rep = range(0,n_opt_rep,1)
+    df_losses = pd.DataFrame(sng_ann_losses)
+
+    ### TRANSFROM PRINCIPAL VALUES TO LIST ###
+    principal_sng = principal_sng_dic.set_index('Key').loc[cntry_names, 'Value'].tolist()
+
+    ### CALCULATE ALPHA FOR RISK CONCENTRATION OPTIMIZATION ###
+    RT = len(df_losses[cntry_names[0]])
+    alpha = 1-1/RT 
+    
+    bools = df_losses >= np.quantile(df_losses, alpha, axis=0)
+
+    risk_concentration = 1.0
+    # Loop through repetitions for seed analysis
+    for index in opt_rep:
+        # Define Problem and Algorithm (same as inside the loop)
+        problem = PoolOptimizationFixedNumber(principal_sng, df_losses, bools, alpha, n, calc_pool_conc)
+        algorithm = GA(
+            pop_size=2000,
+            sampling=IntegerRandomSampling(),
+            crossover=HalfUniformCrossover(),
+            mutation=PolynomialMutation(repair=RoundingRepair()),
+            eliminate_duplicates=True,
+        )
+
+        # Solve the problem
+        res_reg = minimize(problem, algorithm, verbose=False, save_history=True)
+
+        # Process results (same code as inside the loop)
+        x = res_reg.X
+        risk_concentration_new = res_reg.F
+        if risk_concentration_new is not None and risk_concentration is not None and risk_concentration_new < risk_concentration:
+            algorithm_result = res_reg
+            risk_concentration = risk_concentration_new
+            sorted_unique = sorted(set(x))
+            rank_dict = {value: rank + 1 for rank, value in enumerate(sorted_unique)}
+            x = [rank_dict[value] for value in x]
+
+            # Add to dump dataframe
+            country_allocation = pd.DataFrame(columns=[cntry_names, 'min_conc'])
+            country_allocation = pd.DataFrame([x], columns=cntry_names)
+            country_allocation['min_conc'] = pd.DataFrame([res_reg.F], columns=['min_conc'])
+
+        return country_allocation, algorithm_result
+
+
+
+
 
 def calc_pool_conc(x, data_arr, bools, alpha):
     """Calculate diversification of a given pool. Used to