risk_disc -> risk_disc_rates, fixes return period mechanism

Author: spjuhel

climada/trajectories/interpolated_trajectory.py

@@ -40,7 +40,7 @@
     ImpactComputationStrategy,
 )
 from climada.trajectories.snapshot import Snapshot
-from climada.trajectories.trajectory import RiskTrajectory
+from climada.trajectories.trajectory import DEFAULT_RP, RiskTrajectory
 from climada.util import log_level
 
 LOGGER = logging.getLogger(__name__)
@@ -69,16 +69,18 @@ def __init__(
         self,
         snapshots_list: list[Snapshot],
         *,
+        return_periods: list[int] = DEFAULT_RP,
         time_resolution: str = "Y",
         all_groups_name: str = "All",
-        risk_disc: DiscRates | None = None,
+        risk_disc_rates: DiscRates | None = None,
         interpolation_strategy: InterpolationStrategyBase | None = None,
         impact_computation_strategy: ImpactComputationStrategy | None = None,
     ):
         super().__init__(
             snapshots_list,
+            return_periods=return_periods,
             all_groups_name=all_groups_name,
-            risk_disc=risk_disc,
+            risk_disc_rates=risk_disc_rates,
         )
         self._risk_metrics_up_to_date: bool = False
         self.start_date = min([snapshot.date for snapshot in snapshots_list])
@@ -247,8 +249,10 @@ def _generic_metrics(
             ]
             setattr(self, attr_name, tmp)
 
-            if self._risk_disc:
-                return self.npv_transform(getattr(self, attr_name), self._risk_disc)
+            if self._risk_disc_rates:
+                return self.npv_transform(
+                    getattr(self, attr_name), self._risk_disc_rates
+                )
 
             return getattr(self, attr_name)
 

climada/trajectories/snapshot.py

@@ -49,7 +49,6 @@ class Snapshot:
         a datetime object or a string representation of a datetime object
         with format YYYY-MM-DD.
 
-
     Attributes
     ----------
     date : datetime

climada/trajectories/static_trajectory.py

@@ -32,7 +32,7 @@
     ImpactComputationStrategy,
 )
 from climada.trajectories.snapshot import Snapshot
-from climada.trajectories.trajectory import RiskTrajectory
+from climada.trajectories.trajectory import DEFAULT_RP, RiskTrajectory
 from climada.util import log_level
 
 LOGGER = logging.getLogger(__name__)
@@ -57,14 +57,16 @@ def __init__(
         self,
         snapshots_list: list[Snapshot],
         *,
+        return_periods: list[int] = DEFAULT_RP,
         all_groups_name: str = "All",
-        risk_disc: DiscRates | None = None,
+        risk_disc_rates: DiscRates | None = None,
         impact_computation_strategy: ImpactComputationStrategy | None = None,
     ):
         super().__init__(
             snapshots_list,
+            return_periods=return_periods,
             all_groups_name=all_groups_name,
-            risk_disc=risk_disc,
+            risk_disc_rates=risk_disc_rates,
         )
         self._risk_metrics_calculators = CalcRiskMetricsPoints(
             self._snapshots,
@@ -140,8 +142,10 @@ def _generic_metrics(
             ]
             setattr(self, attr_name, tmp)
 
-            if self._risk_disc:
-                return self.npv_transform(getattr(self, attr_name), self._risk_disc)
+            if self._risk_disc_rates:
+                return self.npv_transform(
+                    getattr(self, attr_name), self._risk_disc_rates
+                )
 
             return getattr(self, attr_name)
 

climada/trajectories/test/test_risk_trajectory.py

@@ -88,7 +88,7 @@ def test_init_basic(self):
         )
         self.assertEqual(rt.start_date, self.mock_snapshot1.date)
         self.assertEqual(rt.end_date, self.mock_snapshot3.date)
-        self.assertIsNone(rt._risk_disc)
+        self.assertIsNone(rt._risk_disc_rates)
         self.assertEqual(rt._interpolation_strategy, self.mock_interpolation_strategy)
         self.assertEqual(
             rt._impact_computation_strategy, self.mock_impact_computation_strategy
@@ -106,13 +106,13 @@ def test_init_with_custom_params(self):
             self.snapshots_list,
             time_resolution="MS",
             all_groups_name="CustomAll",
-            risk_disc=mock_disc,
+            risk_disc_rates=mock_disc,
             interpolation_strategy=Mock(),
             impact_computation_strategy=Mock(),
         )
         self.assertEqual(rt._time_resolution, "MS")
         self.assertEqual(rt._all_groups_name, "CustomAll")
-        self.assertEqual(rt._risk_disc, mock_disc)
+        self.assertEqual(rt._risk_disc_rates, mock_disc)
 
     ## Test Properties (`@property` and `@setter`)
     def test_default_rp_getter_setter(self):
@@ -212,7 +212,7 @@ def test_calc_risk_periods_sorting(self, MockCalcRiskPeriod):
     def test_generic_metrics_basic_flow(self, mock_npv_transform, mock_risk_periods):
         rt = InterpolatedRiskTrajectory(self.snapshots_list)
         rt._all_groups_name = "All"  # Ensure default
-        rt._risk_disc = self.mock_disc_rates  # For NPV transform check
+        rt._risk_disc_rates = self.mock_disc_rates  # For NPV transform check
 
         # Mock CalcRiskPeriod instances returned by risk_periods property
         mock_calc_period1 = Mock()
@@ -267,7 +267,7 @@ def test_generic_metrics_basic_flow(self, mock_npv_transform, mock_risk_periods)
             ["group", "date", "measure", "metric", "risk"]
         ]
         # npv_transform should be called with the correctly formatted (concatenated and ordered) DataFrame
-        # and the risk_disc attribute
+        # and the risk_disc_rates attribute
         mock_npv_transform.assert_called_once()
         pd.testing.assert_frame_equal(
             mock_npv_transform.call_args[0][0].reset_index(drop=True),

climada/trajectories/trajectory.py

@@ -28,36 +28,34 @@
 import pandas as pd
 
 from climada.entity.disc_rates.base import DiscRates
-from climada.trajectories.riskperiod import (
-    ImpactCalcComputation,
-    ImpactComputationStrategy,
-)
 from climada.trajectories.snapshot import Snapshot
 
 LOGGER = logging.getLogger(__name__)
 
+DEFAULT_RP = [20, 50, 100]
+
 
 class RiskTrajectory(ABC):
     _grouper = ["measure", "metric"]
     """Results dataframe grouper"""
 
     POSSIBLE_METRICS = []
-    DEFAULT_RP = [20, 50, 100]
 
     def __init__(
         self,
         snapshots_list: list[Snapshot],
         *,
+        return_periods: list[int],
         all_groups_name: str = "All",
-        risk_disc: DiscRates | None = None,
+        risk_disc_rates: DiscRates | None = None,
     ):
         self._reset_metrics()
         self._snapshots = snapshots_list
         self._all_groups_name = all_groups_name
-        self._return_periods = self.DEFAULT_RP
+        self._return_periods = return_periods
         self.start_date = min([snapshot.date for snapshot in snapshots_list])
         self.end_date = max([snapshot.date for snapshot in snapshots_list])
-        self._risk_disc = risk_disc
+        self._risk_disc_rates = risk_disc_rates
 
     def _reset_metrics(self):
         for metric in self.POSSIBLE_METRICS:
@@ -103,7 +101,7 @@ def return_periods(self, value, /):
         self._return_periods = value
 
     @property
-    def risk_disc(self) -> DiscRates | None:
+    def risk_disc_rates(self) -> DiscRates | None:
         """The discount rate applied to compute net present values.
         None means no discount rate.
 
@@ -112,25 +110,27 @@ def risk_disc(self) -> DiscRates | None:
 
         Changing its value resets the metrics.
         """
-        return self._risk_disc
+        return self._risk_disc_rates
 
-    @risk_disc.setter
-    def risk_disc(self, value, /):
+    @risk_disc_rates.setter
+    def risk_disc_rates(self, value, /):
         if not isinstance(value, DiscRates):
             raise ValueError("Risk discount needs to be a `DiscRates` object.")
 
         self._reset_metrics()
-        self._risk_disc = value
+        self._risk_disc_rates = value
 
     @classmethod
-    def npv_transform(cls, df: pd.DataFrame, risk_disc: DiscRates) -> pd.DataFrame:
+    def npv_transform(
+        cls, df: pd.DataFrame, risk_disc_rates: DiscRates
+    ) -> pd.DataFrame:
         """Apply discount rate to a metric `DataFrame`.
 
         Parameters
         ----------
         df : pd.DataFrame
             The `DataFrame` of the metric to discount.
-        risk_disc : DiscRate
+        risk_disc_rates : DiscRate
             The discount rate to apply.
 
         Returns
@@ -156,15 +156,15 @@ def _npv_group(group, disc):
             as_index=False,
             group_keys=False,
             observed=True,
-        )["risk"].transform(_npv_group, risk_disc)
+        )["risk"].transform(_npv_group, risk_disc_rates)
         df = df.reset_index()
         return df
 
     @staticmethod
     def _calc_npv_cash_flows(
         cash_flows: pd.DataFrame,
         start_date: datetime.date,
-        disc: DiscRates | None = None,
+        disc_rates: DiscRates | None = None,
     ):
         """Apply discount rate to cash flows.
 
@@ -187,7 +187,7 @@ def _calc_npv_cash_flows(
         A dataframe (copy) of `cash_flows` where values are discounted according to `disc`
         """
 
-        if not disc:
+        if not disc_rates:
             return cash_flows
 
         if not isinstance(cash_flows.index, pd.DatetimeIndex):
@@ -198,7 +198,7 @@ def _calc_npv_cash_flows(
 
         # Merge with the discount rates based on the year
         tmp = df.merge(
-            pd.DataFrame({"year": disc.years, "rate": disc.rates}),
+            pd.DataFrame({"year": disc_rates.years, "rate": disc_rates.rates}),
             on="year",
             how="left",
         )