refactor(risktraj): lays better foundations for MeasureAppraiser

Author: spjuhel

climada/trajectories/risk_trajectory.py

@@ -105,7 +105,6 @@ def _reset_metrics(self):
         self._risk_components_metrics = None
         self._aai_per_group_metrics = None
         self._all_risk_metrics = None
-        self._metrics_up_to_date = False
 
     @property
     def default_rp(self):
@@ -249,48 +248,67 @@ def _generic_metrics(
 
         return getattr(self, attr_name)
 
-    def aai_metrics(self, npv=True):
-        return self._generic_metrics(
-            npv=npv, metric_name="aai", metric_meth="calc_aai_metric"
+    def _compute_metrics(
+        self, metric_name, metric_meth, total=False, npv=True, *args, **kwargs
+    ):
+        """Helper method to compute metrics and optionally return total risk."""
+        df = self._generic_metrics(
+            npv=npv, metric_name=metric_name, metric_meth=metric_meth, *args, **kwargs
         )
+        if total:
+            return self._per_period_risk(df)
+        return df
 
-    def return_periods_metrics(self, return_periods=None, npv=True):
+    def aai_metrics(self, total=False, npv=True, *args, **kwargs):
+        return self._compute_metrics(
+            total=total,
+            npv=npv,
+            metric_name="aai",
+            metric_meth="calc_aai_metric",
+            *args,
+            **kwargs,
+        )
+
+    def return_periods_metrics(
+        self, total=False, return_periods=None, npv=True, *args, **kwargs
+    ):
         return_periods = return_periods if return_periods else self.default_rp
-        return self._generic_metrics(
+        return self._compute_metrics(
             npv=npv,
             metric_name="return_periods",
             metric_meth="calc_return_periods_metric",
             return_periods=return_periods,
+            *args,
+            **kwargs,
         )
 
-    def aai_per_group_metrics(self, npv=True):
-        return self._generic_metrics(
+    def aai_per_group_metrics(self, npv=True, *args, **kwargs):
+        return self._compute_metrics(
             npv=npv,
             metric_name="aai_per_group",
             metric_meth="calc_aai_per_group_metric",
+            *args,
+            **kwargs,
         )
 
-    def risk_components_metrics(self, npv=True):
-        return self._generic_metrics(
+    def risk_components_metrics(self, npv=True, *args, **kwargs):
+        return self._compute_metrics(
             npv=npv,
             metric_name="risk_components",
             metric_meth="calc_risk_components_metric",
+            *args,
+            **kwargs,
         )
 
     def all_risk_metrics(
-        self, return_periods=[50, 100, 500], npv=True
+        self, return_periods=[50, 100, 500], npv=True, *args, **kwargs
     ) -> pd.DataFrame | pd.Series:
-        if not self._metrics_up_to_date or self._all_risk_metrics is None:
-            aai = self.aai_metrics(npv)
-            rp = self.return_periods_metrics(return_periods, npv)
-            aai_per_group = self.aai_per_group_metrics(npv)
-            risk_components = self.risk_components_metrics(npv)
-            self._all_risk_metrics = pd.concat(
-                [aai, rp, aai_per_group, risk_components]
-            )
-            self._metrics_up_to_date = True
 
-        return self._all_risk_metrics
+        aai = self.aai_metrics(npv, *args, **kwargs)
+        rp = self.return_periods_metrics(return_periods, npv, *args, **kwargs)
+        aai_per_group = self.aai_per_group_metrics(npv, *args, **kwargs)
+        risk_components = self.risk_components_metrics(npv, *args, **kwargs)
+        return pd.concat([aai, rp, aai_per_group, risk_components])
 
     @staticmethod
     def _get_risk_periods(
@@ -321,7 +339,7 @@ def identify_continuous_periods(group, time_unit):
             return group
 
         grouper = cls._grouper
-        if "group" in df.columns:
+        if "group" in df.columns and "group" not in grouper:
             grouper = ["group"] + grouper
 
         df_sorted = df.sort_values(by=cls._grouper + ["date"])
@@ -330,14 +348,20 @@ def identify_continuous_periods(group, time_unit):
             identify_continuous_periods, time_unit
         )
 
+        if isinstance(colname, str):
+            colname = [colname]
+
+        agg_dict = {
+            "start_date": pd.NamedAgg(column="date", aggfunc="min"),
+            "end_date": pd.NamedAgg(column="date", aggfunc="max"),
+        }
+        for col in colname:
+            agg_dict[col] = pd.NamedAgg(column=col, aggfunc="sum")
         # Group by the identified periods and calculate start and end dates
+        print(df_periods)
         df_periods = (
             df_periods.groupby(grouper + ["period_id"], dropna=False)
-            .agg(
-                start_date=pd.NamedAgg(column="date", aggfunc="min"),
-                end_date=pd.NamedAgg(column="date", aggfunc="max"),
-                total=pd.NamedAgg(column=colname, aggfunc="sum"),
-            )
+            .agg(**agg_dict)
             .reset_index()
         )
 
@@ -346,28 +370,22 @@ def identify_continuous_periods(group, time_unit):
             + " to "
             + df_periods["end_date"].astype(str)
         )
-        df_periods = df_periods.rename(columns={"total": f"{colname}"})
+        # df_periods = df_periods.rename(columns={"total": f"{colname}"})
         df_periods = df_periods.drop(["period_id", "start_date", "end_date"], axis=1)
         return df_periods[
             ["period"] + [col for col in df_periods.columns if col != "period"]
         ]
 
     @property
-    def per_date_risk_metrics(self) -> pd.DataFrame | pd.Series:
+    def per_date_risk_metrics(self, *args, **kwargs) -> pd.DataFrame | pd.Series:
         """Returns a tidy dataframe of the risk metrics for all dates."""
-        return self._prepare_risk_metrics(total=False, npv=True)
+        return self.all_risk_metrics(*args, **kwargs)
 
     @property
-    def total_risk_metrics(self) -> pd.DataFrame | pd.Series:
+    def total_risk_metrics(self, *args, **kwargs) -> pd.DataFrame | pd.Series:
         """Returns a tidy dataframe of the risk metrics with the total for each different period."""
-        return self._prepare_risk_metrics(total=True, npv=True)
-
-    def _prepare_risk_metrics(self, total=False, npv=True) -> pd.DataFrame | pd.Series:
-        df = self.all_risk_metrics(npv=npv)
-        if total:
-            return self._per_period_risk(df)
-
-        return df
+        df = self.all_risk_metrics(*args, **kwargs)
+        return self._per_period_risk(df)
 
     def _calc_waterfall_plot_data(self, start_date=None, end_date=None, npv=True):
         start_date = self.start_date if start_date is None else start_date

climada/trajectories/riskperiod.py

@@ -46,6 +46,10 @@ def lazy_property(method):
     @property
     def _lazy(self):
         if getattr(self, attr_name) is None:
+            meas_n = self.measure.name if self.measure else "no_measure"
+            LOGGER.debug(
+                f"Computing {method.__name__} for {self._snapshot0.date}-{self._snapshot1.date} with {meas_n}."
+            )
             setattr(self, attr_name, method(self))
         return getattr(self, attr_name)
 
@@ -67,13 +71,14 @@ def __init__(
         risk_transf_cover: float | None = None,
         calc_residual: bool = False,
     ):
+        LOGGER.info("Instantiating new CalcRiskPeriod.")
         self._snapshot0 = snapshot0
         self._snapshot1 = snapshot1
-        self.date_idx = pd.date_range(
-            snapshot0.date,
-            snapshot1.date,
+        self.date_idx = CalcRiskPeriod._set_date_idx(
+            date1=snapshot0.date,
+            date2=snapshot1.date,
             periods=time_points,
-            freq=interval_freq,  # type: ignore
+            freq=interval_freq,
             name="date",
         )
         self.interpolation_strategy = interpolation_strategy or LinearInterpolation()
@@ -89,13 +94,66 @@ def __init__(
         self._group_id_E1 = self.snapshot1.exposure.gdf["group_id"].values
 
     def _reset_impact_data(self):
-        self._impacts_arrays = None, None, None, None
+        self._impacts_arrays = None
         self._imp_mats_H0, self._imp_mats_H1 = None, None
         self._imp_mats_E0, self._imp_mats_E1 = None, None
         self._per_date_eai_H0, self._per_date_eai_H1 = None, None
         self._per_date_aai_H0, self._per_date_aai_H1 = None, None
         self._per_date_return_periods_H0, self._per_date_return_periods_H1 = None, None
 
+    @staticmethod
+    def _set_date_idx(
+        date1: str | pd.Timestamp,
+        date2: str | pd.Timestamp,
+        periods: int | None = None,
+        freq: str | None = None,
+        name: str | None = None,
+    ) -> pd.DatetimeIndex:
+        """
+        Generate a date range index based on the provided parameters.
+
+        Parameters
+        ----------
+        date1 : str or pd.Timestamp
+            The start date of the date range.
+        date2 : str or pd.Timestamp
+            The end date of the date range.
+        periods : int, optional
+            Number of date points to generate. If None, `freq` must be provided.
+        freq : str, optional
+            Frequency string for the date range. If None, `periods` must be provided.
+        name : str, optional
+            Name of the resulting date range index.
+
+        Returns
+        -------
+        pd.DatetimeIndex
+            A DatetimeIndex representing the date range.
+
+        Raises
+        ------
+        ValueError
+            If the number of periods and frequency given to date_range are inconsistent.
+        """
+        if periods is not None and freq is not None:
+            points = None
+        else:
+            points = periods
+
+        ret = pd.date_range(
+            date1,
+            date2,
+            periods=points,
+            freq=freq,  # type: ignore
+            name=name,
+        )
+        if periods is not None and len(ret) != periods:
+            raise ValueError(
+                "Number of periods and frequency given to date_range are inconsistant"
+            )
+
+        return ret
+
     @property
     def snapshot0(self):
         return self._snapshot0