Removes global risk transfer for now

Author: spjuhel

climada/trajectories/impact_calc_strat.py

@@ -21,12 +21,7 @@
 
 """
 
-import copy
 from abc import ABC, abstractmethod
-from typing import Any, Optional, Union
-
-import numpy as np
-from scipy import sparse
 
 from climada.engine.impact import Impact
 from climada.engine.impact_calc import ImpactCalc
@@ -56,9 +51,6 @@ def compute_impacts(
         exp: Exposures,
         haz: Hazard,
         vul: ImpactFuncSet,
-        risk_transf_attach: Optional[float] = None,
-        risk_transf_cover: Optional[float] = None,
-        calc_residual: bool = True,
     ) -> Impact:
         """
         Calculates the total impact, including optional risk transfer application.
@@ -71,16 +63,6 @@ def compute_impacts(
             The hazard data (e.g., event intensity).
         vul : ImpactFuncSet
             The set of vulnerability functions.
-        risk_transf_attach : float, optional
-            The attachment point (deductible) for the global risk transfer mechanism.
-            If None, default to 0.
-        risk_transf_cover : float, optional
-            The cover limit for the risk transfer mechanism. If None, the cover
-            is assumed to be infinite (only the attachment applies).
-        calc_residual : bool, default=True
-            If True, the function returns the residual impact (after risk transfer).
-            If False, it returns the transferred impact (the part covered by the
-            risk transfer).
 
         Returns
         -------
@@ -99,28 +81,7 @@ class ImpactCalcComputation(ImpactComputationStrategy):
     Default impact computation strategy using the core engine of climada.
 
     This strategy first calculates the raw impact using the standard
-    :class:`ImpactCalc` logic and then applies a global risk transfer mechanism.
-
-    This risk transfer is distinct and non-exclusive with the risk transfer
-    defined through the Exposures object. While in the Exposures case, the transfer
-    is defined on a per coordinate basis, this one is applied total impact on the whole region
-    considered, and proportionally rescales the impacts per coordinate accordingly.
-
-    Notes
-    -----
-    The calculation is performed event-wise:
-
-    1. **Total Impact**: Calculate the total impact for each event
-       (sum of impacts across all exposure points).
-    2. **Transferred Risk per Event**: Defined as:
-       $$\min(\max(0, \text{Total Impact} - \text{attachement}), \text{cover})$$
-    3. **Residual Risk per Event**:
-       $$\text{Total Impact} - \text{Transferred Risk per Event}$$
-    4. **Adjustment**: The original impact per exposure point is scaled
-       by the ratio of (Residual Risk / Total Impact) or
-       (Transferred Risk / Total Impact) for that event.
-       This ensures the risk transfer is shared proportionally among all
-       impacted exposure points.
+    :class:`ImpactCalc` logic.
 
     """
 
@@ -129,9 +90,6 @@ def compute_impacts(
         exp: Exposures,
         haz: Hazard,
         vul: ImpactFuncSet,
-        risk_transf_attach: Optional[float] = None,
-        risk_transf_cover: Optional[float] = None,
-        calc_residual: bool = False,
     ) -> Impact:
         """
         Calculates the impact and applies the "global" risk transfer mechanism.
@@ -144,22 +102,13 @@ def compute_impacts(
             The hazard data.
         vul : ImpactFuncSet
             The set of vulnerability functions.
-        risk_transf_attach : float, optional
-            The attachment point (deductible) for the risk transfer mechanism.
-        risk_transf_cover : float, optional
-            The cover limit for the risk transfer mechanism.
-        calc_residual : bool, default=False
-            If True, returns the residual impact. If False, returns the transferred impact.
 
         Returns
         -------
         Impact
-            The final impact object (either residual or transferred).
+            The final impact object.
         """
         impact = self.compute_impacts_pre_transfer(exp, haz, vul)
-        self._apply_risk_transfer(
-            impact, risk_transf_attach, risk_transf_cover, calc_residual
-        )
         return impact
 
     def compute_impacts_pre_transfer(
@@ -186,106 +135,3 @@ def compute_impacts_pre_transfer(
             An Impact object containing the raw, pre-transfer impact matrix.
         """
         return ImpactCalc(exposures=exp, impfset=vul, hazard=haz).impact()
-
-    def _apply_risk_transfer(
-        self,
-        impact: Impact,
-        risk_transf_attach: Optional[float],
-        risk_transf_cover: Optional[float],
-        calc_residual: bool,
-    ) -> None:
-        """
-        Applies risk transfer logic and modifies the Impact object in-place.
-
-        Parameters
-        ----------
-        impact : Impact
-            The Impact object whose impact matrix will be modified.
-        risk_transf_attach : float, optional
-            The attachment point.
-        risk_transf_cover : float, optional
-            The cover limit.
-        calc_residual : bool
-            Determines whether to set the matrix to the residual or transferred impact.
-        """
-        if risk_transf_attach is not None or risk_transf_cover is not None:
-            impact.imp_mat = self.calculate_residual_or_risk_transfer_impact_matrix(
-                impact.imp_mat, risk_transf_attach, risk_transf_cover, calc_residual
-            )
-
-    def calculate_residual_or_risk_transfer_impact_matrix(
-        self,
-        imp_mat: sparse.csr_matrix,
-        attachement: Optional[float],
-        cover: Optional[float],
-        calc_residual: bool,
-    ) -> sparse.csr_matrix:
-        r"""
-        Calculates either the residual or the risk transfer impact matrix
-        based on a global risk transfer mechanism.
-
-        This function modifies the original impact matrix values proportionally
-        based on the total event impact relative to the attachment and cover.
-
-        Parameters
-        ----------
-        imp_mat : scipy.sparse.csr_matrix or object with .data
-            The original impact matrix (events x exposure points).
-        attachement : float, optional
-            The attachment point (deductible).
-        cover : float, optional
-            The cover limit.
-        calc_residual : bool
-            If True, the function returns the residual impact matrix.
-            If False, it returns the transferred risk impact matrix.
-
-        Returns
-        -------
-        scipy.sparse.csr_matrix or object with .data
-            The adjusted impact matrix (residual or transferred).
-
-        Notes
-        -----
-        The calculation is performed event-wise:
-
-        1. **Total Impact**: Calculate the total impact for each event
-           (sum of impacts across all exposure points).
-        2. **Transferred Risk per Event**: Defined as:
-           $$\min(\max(0, \text{Total Impact} - \text{attachement}), \text{cover})$$
-        3. **Residual Risk per Event**:
-           $$\text{Total Impact} - \text{Transferred Risk per Event}$$
-        4. **Adjustment**: The original impact per exposure point is scaled
-           by the ratio of (Residual Risk / Total Impact) or
-           (Transferred Risk / Total Impact) for that event.
-           This ensures the risk transfer is shared proportionally among all
-           impacted exposure points.
-        """
-        imp_mat = copy.deepcopy(imp_mat)
-        # Calculate the total impact per event
-        total_at_event = imp_mat.sum(axis=1).A1  # type: ignore
-        # Risk layer at event
-        attachement = 0 if attachement is None else attachement
-        cover = np.inf if cover is None else cover
-        transfer_at_event = np.minimum(
-            np.maximum(total_at_event - attachement, 0), cover
-        )
-        residual_at_event = np.maximum(total_at_event - transfer_at_event, 0)
-
-        # Calculate either the residual or transfer impact matrix
-        # Choose the denominator to rescale the impact values
-        if calc_residual:
-            numerator = residual_at_event
-        else:
-            numerator = transfer_at_event
-
-        rescale_impact_values = np.divide(
-            numerator,
-            total_at_event,
-            out=np.zeros_like(numerator, dtype=float),
-            where=total_at_event != 0,
-        )
-
-        # The multiplication is broadcasted across the columns for each row
-        result_matrix = imp_mat.multiply(rescale_impact_values[:, np.newaxis])
-
-        return result_matrix

climada/trajectories/test/test_impact_calc_strat.py

@@ -48,20 +48,14 @@ def setUp(self):
         self.impact_calc_computation = ImpactCalcComputation()
 
     @patch.object(ImpactCalcComputation, "compute_impacts_pre_transfer")
-    @patch.object(ImpactCalcComputation, "_apply_risk_transfer")
-    def test_compute_impacts(
-        self, mock_apply_risk_transfer, mock_calculate_impacts_for_snapshots
-    ):
+    def test_compute_impacts(self, mock_calculate_impacts_for_snapshots):
         mock_impacts = MagicMock(spec=Impact)
         mock_calculate_impacts_for_snapshots.return_value = mock_impacts
 
         result = self.impact_calc_computation.compute_impacts(
             exp=self.mock_snapshot0.exposure,
             haz=self.mock_snapshot0.hazard,
             vul=self.mock_snapshot0.impfset,
-            risk_transf_attach=0.1,
-            risk_transf_cover=0.9,
-            calc_residual=False,
         )
 
         self.assertEqual(result, mock_impacts)
@@ -70,7 +64,6 @@ def test_compute_impacts(
             self.mock_snapshot0.hazard,
             self.mock_snapshot0.impfset,
         )
-        mock_apply_risk_transfer.assert_called_once_with(mock_impacts, 0.1, 0.9, False)
 
     def test_calculate_impacts_for_snapshots(self):
         mock_imp_E0H0 = MagicMock(spec=Impact)
@@ -88,37 +81,6 @@ def test_calculate_impacts_for_snapshots(self):
 
             self.assertEqual(result, mock_imp_E0H0)
 
-    def test_apply_risk_transfer(self):
-        mock_imp_E0H0 = MagicMock(spec=Impact)
-        mock_imp_E0H0.imp_mat = MagicMock(spec=csr_matrix)
-        mock_imp_resi = MagicMock(spec=csr_matrix)
-
-        with patch.object(
-            self.impact_calc_computation,
-            "calculate_residual_or_risk_transfer_impact_matrix",
-        ) as mock_calc_risk_transfer:
-            mock_calc_risk_transfer.return_value = mock_imp_resi
-            self.impact_calc_computation._apply_risk_transfer(
-                mock_imp_E0H0, 0.1, 0.9, False
-            )
-
-            self.assertIs(mock_imp_E0H0.imp_mat, mock_imp_resi)
-
-    def test_calculate_residual_or_risk_transfer_impact_matrix(self):
-        imp_mat = MagicMock()
-        imp_mat.sum.return_value.A1 = np.array([100, 200, 300])
-        imp_mat.multiply.return_value = "rescaled_matrix"
-
-        result = self.impact_calc_computation.calculate_residual_or_risk_transfer_impact_matrix(
-            imp_mat, 0.1, 0.9, True
-        )
-        self.assertEqual(result, "rescaled_matrix")
-
-        result = self.impact_calc_computation.calculate_residual_or_risk_transfer_impact_matrix(
-            imp_mat, 0.1, 0.9, False
-        )
-        self.assertEqual(result, "rescaled_matrix")
-
 
 if __name__ == "__main__":
     TESTS = unittest.TestLoader().loadTestsFromTestCase(TestImpactCalcComputation)