diff --git a/chainladder/core/pandas.py b/chainladder/core/pandas.py index 94df66c3..ee352b6b 100644 --- a/chainladder/core/pandas.py +++ b/chainladder/core/pandas.py @@ -1,6 +1,7 @@ """ Mirror pandas API onto the Triangle class. """ + # This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at https://mozilla.org/MPL/2.0/. @@ -13,15 +14,9 @@ __dt64_dtype__, _warn_dask_parallel_deprecated, ) -from chainladder.utils.utility_functions import ( - concat, - num_to_nan -) +from chainladder.utils.utility_functions import concat, num_to_nan -from typing import ( - cast, - TYPE_CHECKING -) +from typing import cast, TYPE_CHECKING try: @@ -35,23 +30,16 @@ from chainladder.core.typing import BackendArray, TriangleProtocol from collections.abc import Callable from numpy import ndarray - from pandas import ( - DataFrame, - Series - ) + from pandas import DataFrame, Series from types import ModuleType from pandas._typing import IndexLabel - from typing import ( - Any, - Literal, - Type - ) + from typing import Any, Literal, Type + _TrianglePandasBase = TriangleProtocol else: _TrianglePandasBase = object - class TriangleGroupBy: def __init__(self, obj: Triangle, by, axis=0, **kwargs): self.obj = obj.copy() @@ -76,23 +64,22 @@ def __getitem__(self, key): class TrianglePandas(_TrianglePandasBase): - def to_frame( - self, - origin_as_datetime: bool = True, - keepdims: bool = False, - implicit_axis: bool = False, + self, + origin_as_datetime: bool = True, + keepdims: bool = False, + implicit_axis: bool = False, ) -> DataFrame | Series: - """ Converts a triangle to a pandas.DataFrame. + """Converts a triangle to a pandas.DataFrame. Parameters ---------- origin_as_datetime : bool (default = True) - When all dimensions are returned, whether the origin vector - should be converted from PeriodIndex into a datetime dtype. + When all dimensions are returned, whether the origin vector + should be converted from PeriodIndex into a datetime dtype. keepdims : bool (default = False) Converted DataFrame will keep all dimensions intact and maintain a consistent - format regardless of whether any dimensions are of length 1. + format regardless of whether any dimensions are of length 1. Ignored when 3 or more dimensions (index, column, origin, and development) have lengths greater than 1 @@ -115,7 +102,7 @@ def to_frame( values: COO = cast("COO", obj.values) out: DataFrame = pd.DataFrame(obj.index.iloc[values.coords[0]]) out["columns"] = obj.columns[values.coords[1]] - missing_cols: list = list(set(self.columns) - set(out['columns'])) + missing_cols: list = list(set(self.columns) - set(out["columns"])) if origin_as_datetime: out["origin"] = obj.odims[values.coords[2]] else: @@ -131,36 +118,40 @@ def to_frame( ) valuation_series = pd.DataFrame( - obj.valuation.values.reshape(obj.shape[-2:], order='F'), - index=obj.odims if origin_as_datetime else obj.origin, - columns=obj.ddims + obj.valuation.values.reshape(obj.shape[-2:], order="F"), + index=obj.odims if origin_as_datetime else obj.origin, + columns=obj.ddims, ).unstack() - valuation_series.name = 'valuation' + valuation_series.name = "valuation" valuation: DataFrame = valuation_series.reset_index().rename( - columns={ - 'level_0': 'development', - 'level_1': 'origin'} + columns={"level_0": "development", "level_1": "origin"} + ) + val_dict: dict = dict( + zip( + list(zip(valuation["origin"], valuation["development"])), + valuation["valuation"], + ) ) - val_dict: dict = dict(zip(list(zip( - valuation['origin'], valuation['development'])), - valuation['valuation'])) if len(out) > 0: - out['valuation'] = out.apply( - lambda x: val_dict[(x['origin'], x['development'])], axis=1) + out["valuation"] = out.apply( + lambda x: val_dict[(x["origin"], x["development"])], axis=1 + ) else: - out['valuation'] = self.valuation_date + out["valuation"] = self.valuation_date col_order: list = list(self.columns) if implicit_axis: - col_order: list = ['origin', 'development', 'valuation'] + col_order + col_order: list = ["origin", "development", "valuation"] + col_order else: if is_val_tri: - col_order: list = ['origin', 'valuation'] + col_order + col_order: list = ["origin", "valuation"] + col_order else: - col_order: list = ['origin', 'development'] + col_order + col_order: list = ["origin", "development"] + col_order for col in set(missing_cols) - self.virtual_columns.columns.keys(): out[col] = np.nan # Create physical columns out of virtual ones. - for col in set(missing_cols).intersection(self.virtual_columns.columns.keys()): + for col in set(missing_cols).intersection( + self.virtual_columns.columns.keys() + ): # Fill na to enable floating-point computation. out[col] = out.fillna(0).apply(self.virtual_columns.columns[col], 1) # Coerce 0 to np.nan. @@ -202,7 +193,7 @@ def to_frame( return self.to_frame( origin_as_datetime=origin_as_datetime, keepdims=True, - implicit_axis=implicit_axis + implicit_axis=implicit_axis, ) def plot(self, *args: Any, **kwargs: Any) -> None: @@ -246,7 +237,9 @@ def hvplot(self, *args: Any, **kwargs: Any) -> Any: return df.hvplot(*args, **kwargs) @staticmethod - def _get_axis(axis: Literal['index', 'columns', 'origin', 'development'] | int | None) -> int: + def _get_axis( + axis: Literal["index", "columns", "origin", "development"] | int | None, + ) -> int: """ Returns the integer representation of the requested axis. @@ -524,13 +517,11 @@ def dropna(self) -> Triangle: # Case when triangle has multiple development periods, e.g., not latest diagonal or ultimate. if obj.shape[-1] != 1: # Flag the development periods that have data. - ddim = list( - (xp.nansum(obj.values[0, 0, :], -2) != 0).astype("int")) + ddim = list((xp.nansum(obj.values[0, 0, :], -2) != 0).astype("int")) ddim = obj.development[pd.Series(ddim).astype(bool)] # Slice the Triangle by the development periods that have data. obj = self[ - (self.development >= ddim.min()) & ( - self.development <= ddim.max()) + (self.development >= ddim.min()) & (self.development <= ddim.max()) ] obj = cast("TriangleProtocol", cast(object, obj)) # Slice the triangle by the origin periods that have data. @@ -566,7 +557,9 @@ def fillna(self, value: int | float | ndarray, inplace: bool = False) -> Triangl return cast("Triangle", cast(object, self)) else: new_obj = self.copy() - cast("TriangleProtocol", cast(object, new_obj)).fillna(value=value, inplace=True) + cast("TriangleProtocol", cast(object, new_obj)).fillna( + value=value, inplace=True + ) return new_obj def fillzero(self, inplace: bool = False) -> Triangle: @@ -587,7 +580,8 @@ def fillzero(self, inplace: bool = False) -> Triangle: # Fill the NaNs by locating their positions within the triangle. self.values = np.where( (xp.nan_to_num(self.values) == 0) * (self.nan_triangle == 1), - self.nan_triangle * 0, self.values + self.nan_triangle * 0, + self.values, ) return cast("Triangle", cast(object, self)) else: @@ -595,6 +589,48 @@ def fillzero(self, inplace: bool = False) -> Triangle: cast("TriangleProtocol", cast(object, new_obj)).fillzero(inplace=True) return new_obj + @staticmethod + def _validate_contiguous_drop( + axis_series: pd.Series, + drop_labels: list[Any], + axis_name: str, + errors: str, + ) -> np.ndarray: + """Validate and return boolean keep mask for dropping contiguous edge periods. + + Parameters + ---------- + axis_series : pd.Series + The existing labels along the axis. + drop_labels : list + Labels requested to be dropped. + axis_name : str + Name of the axis ('origin' or 'development') for error messages. + errors : {'raise', 'ignore'} + Whether to raise or ignore missing labels. + + Returns + ------- + np.ndarray + Boolean mask of labels to keep. + """ + axis_labels = np.array(axis_series.astype(str)) + str_drop_labels = [str(label) for label in drop_labels] + missing = [label for label in str_drop_labels if label not in axis_labels] + if missing and errors == "raise": + raise KeyError(f"{missing} not found in the {axis_name} axis.") + keep = ~np.isin(axis_labels, str_drop_labels) + kept_positions = np.flatnonzero(keep) + if len(kept_positions) and not np.array_equal( + kept_positions, + np.arange(kept_positions[0], kept_positions[-1] + 1), + ): + raise ValueError( + f"Only the first or last {axis_name} periods may be dropped; " + f"dropping an interior {axis_name} period would leave a gap." + ) + return keep + def drop( self, labels: str | int | list | None = None, @@ -774,23 +810,9 @@ def drop( [item for item in result.columns if item not in ax_labels] ] elif ax == 2: - origin_labels = np.array(result.origin.astype(str)) - drop_labels = [str(label) for label in ax_labels] - missing = [ - label for label in drop_labels if label not in origin_labels - ] - if missing and errors == "raise": - raise KeyError(f"{missing} not found in the origin axis.") - keep = ~np.isin(origin_labels, drop_labels) - kept_positions = np.flatnonzero(keep) - if len(kept_positions) and not np.array_equal( - kept_positions, - np.arange(kept_positions[0], kept_positions[-1] + 1), - ): - raise ValueError( - "Only the first or last origin periods may be dropped; " - "dropping an interior origin period would leave a gap." - ) + keep = self._validate_contiguous_drop( + result.origin, ax_labels, "origin", errors + ) result = result[keep] # Trim any development periods that were left entirely NaN by # the origin drop, so dropping origins trims the triangle @@ -800,34 +822,20 @@ def drop( if result.shape[-1] > 1: agg = result.sum(axis=0).sum(axis=1) vals = agg.values[0, 0, :] - vals_np = vals.todense() if hasattr(vals, "todense") else np.asarray(vals) + vals_np = ( + vals.todense() if hasattr(vals, "todense") else np.asarray(vals) + ) arr = np.nan_to_num(vals_np) dev_has_data = list((arr.sum(axis=-2) != 0).astype(int)) - dev_labels = agg.development[ - pd.Series(dev_has_data).astype(bool) - ] + dev_labels = agg.development[pd.Series(dev_has_data).astype(bool)] result = result[ (result.development >= dev_labels.min()) & (result.development <= dev_labels.max()) ] elif ax == 3: - dev_labels = np.array(result.development.astype(str)) - drop_labels = [str(label) for label in ax_labels] - missing = [ - label for label in drop_labels if label not in dev_labels - ] - if missing and errors == "raise": - raise KeyError(f"{missing} not found in the development axis.") - keep = ~np.isin(dev_labels, drop_labels) - kept_positions = np.flatnonzero(keep) - if len(kept_positions) and not np.array_equal( - kept_positions, - np.arange(kept_positions[0], kept_positions[-1] + 1), - ): - raise ValueError( - "Only the first or last development periods may be dropped; " - "dropping an interior development period would leave a gap." - ) + keep = self._validate_contiguous_drop( + result.development, ax_labels, "development", errors + ) result = result._slice(keep, "ddims") if result.is_val_tri: result.valuation_date = min( @@ -908,7 +916,7 @@ def rename( Triangle Triangle with relabeled axis. """ - + if isinstance(value, dict): if axis == "columns" or axis == 1: full_dict = dict(zip(self.columns.values, self.columns.values)) @@ -916,7 +924,7 @@ def rename( self.columns = self.columns.map(full_dict) else: raise ValueError( - "Invalid value provided to the 'value' parameter. Accepted values for index, origin, and development axes are a str or a list" + "Invalid value provided to the 'value' parameter. Accepted values for index, origin, and development axes are a str or a list" ) else: value = [value] if type(value) is str else value @@ -990,8 +998,7 @@ def sort_index(self, *args, **kwargs) -> Triangle: Triangle """ sorted_index: DataFrame = cast( - "DataFrame", - self.index.sort_values(self.key_labels, *args, **kwargs) + "DataFrame", self.index.sort_values(self.key_labels, *args, **kwargs) ) return self.iloc[sorted_index.index] @@ -1059,12 +1066,13 @@ def xs( self, index_key: IndexLabel, level: IndexLabel | None = None, - drop_level: bool = True) -> Triangle: + drop_level: bool = True, + ) -> Triangle: """ - Mimics xs from pandas. key difference is that this function only slices + Mimics xs from pandas. key difference is that this function only slices the index, therefore axis is always 0 and not an argument in the function - - Main use case for this function is when slicing beyond the first field in + + Main use case for this function is when slicing beyond the first field in the index (such as LOB in the clrd dataset) Parameters @@ -1095,23 +1103,17 @@ def xs( new_ax_df = new_ax.to_frame(index=None)[new_ax.names] result.index = new_ax_df else: - result.index = pd.DataFrame(data=['Total'], columns=pd.Index(['Total'])) + result.index = pd.DataFrame(data=["Total"], columns=pd.Index(["Total"])) return result -def add_triangle_agg_func( - cls: Type[TrianglePandas], - k: str, - v: str -): + +def add_triangle_agg_func(cls: Type[TrianglePandas], k: str, v: str): """ Aggregate Overrides in Triangle """ def agg_func( - self: Triangle, - axis: str | int | None = None, - *args, - **kwargs + self: Triangle, axis: str | int | None = None, *args, **kwargs ) -> Triangle | ndarray: """ Applies the aggregation function specified by k from the outer function. @@ -1176,6 +1178,7 @@ def add_groupby_agg_func(cls, k: str, v: str): def agg_func(self, *args, **kwargs): from chainladder.utils import concat + obj = self.obj.copy() auto_sparse = kwargs.pop("auto_sparse", True) if db and obj.array_backend == "sparse": @@ -1192,8 +1195,7 @@ def aggregate(i, obj, axis, v): else: values = [ getattr( - obj.iloc.__getitem__( - tuple([slice(None)] * self.axis + [i])), v + obj.iloc.__getitem__(tuple([slice(None)] * self.axis + [i])), v )(self.axis, auto_sparse=False, keepdims=True) for i in self.groups.indices.values() ] @@ -1202,7 +1204,8 @@ def aggregate(i, obj, axis, v): if self.axis == 0: if isinstance(group_index, pd.MultiIndex): index = ( - pd.DataFrame( + pd + .DataFrame( np.zeros(len(group_index)), index=group_index, columns=["_"], @@ -1230,11 +1233,7 @@ def aggregate(i, obj, axis, v): obj = obj._auto_sparse() return obj - set_method( - cls=cls, - func=agg_func, - k=k - ) + set_method(cls=cls, func=agg_func, k=k) def add_df_passthru(cls, k): @@ -1247,9 +1246,7 @@ def df_passthru(self, *args, **kwargs): def set_method( - cls: Type[TrianglePandas | TriangleGroupBy], - func: Callable, - k: str + cls: Type[TrianglePandas | TriangleGroupBy], func: Callable, k: str ) -> None: """ Assigns methods to a class. diff --git a/chainladder/core/tests/test_triangle.py b/chainladder/core/tests/test_triangle.py index 8faf004e..087a7c09 100644 --- a/chainladder/core/tests/test_triangle.py +++ b/chainladder/core/tests/test_triangle.py @@ -835,6 +835,43 @@ def test_drop_index_axis_not_implemented_raises(clrd): clrd.drop(index="Agway Ins Co") +def test_validate_contiguous_drop_helper(raa): + """Direct test for TrianglePandas._validate_contiguous_drop.""" + from chainladder.core.pandas import TrianglePandas + + # Valid drop of last development period + keep = TrianglePandas._validate_contiguous_drop( + raa.development, [120], "development", errors="raise" + ) + assert np.array_equal(keep, np.array([True] * 9 + [False])) + + # Valid drop of first origin period + keep_orig = TrianglePandas._validate_contiguous_drop( + raa.origin, [raa.origin[0]], "origin", errors="raise" + ) + assert np.array_equal(keep_orig, np.array([False] + [True] * 9)) + + # Missing label with errors="raise" + with pytest.raises(KeyError, match=r"\['999'\] not found in the development axis"): + TrianglePandas._validate_contiguous_drop( + raa.development, [999], "development", errors="raise" + ) + + # Missing label with errors="ignore" + keep_ignore = TrianglePandas._validate_contiguous_drop( + raa.development, [999], "development", errors="ignore" + ) + assert np.all(keep_ignore) + + # Interior label drop raises ValueError + with pytest.raises( + ValueError, match="Only the first or last development periods may be dropped" + ): + TrianglePandas._validate_contiguous_drop( + raa.development, [36], "development", errors="raise" + ) + + def test_hvplot_passthrough(genins, monkeypatch): """TrianglePandas.hvplot() passthrough test for patch coverage.""" monkeypatch.setattr( @@ -2696,19 +2733,17 @@ def test_set_development_no_development_column() -> None: def test_set_development_age_in_months() -> None: """Development given as an age in months (not a date) resolves to the valuation date that many months after the origin's period start.""" - df = pd.DataFrame( - { - 'origin': [1995, 1996], - 'development': [12, 24], - 'reported': [1.0, 2.0] - } - ) + df = pd.DataFrame({ + "origin": [1995, 1996], + "development": [12, 24], + "reported": [1.0, 2.0], + }) tri = cl.Triangle( data=df, - origin='origin', - development='development', - columns='reported', - cumulative=True + origin="origin", + development="development", + columns="reported", + cumulative=True, ) assert list(tri.development) == [12, 24, 36] frame = tri.to_frame(origin_as_datetime=False) @@ -2719,19 +2754,17 @@ def test_set_development_age_in_months() -> None: def test_set_development_age_respects_mid_period_origin() -> None: """Age is relative to the start of the origin's own period, not the literal recorded origin date.""" - df = pd.DataFrame( - { - 'origin': ['2018-06-15', '2018-06-15'], - 'development': [12, 24], - 'reported': [100.0, 150.0] - } - ) + df = pd.DataFrame({ + "origin": ["2018-06-15", "2018-06-15"], + "development": [12, 24], + "reported": [100.0, 150.0], + }) tri = cl.Triangle( data=df, - origin='origin', - development='development', - columns='reported', - cumulative=True + origin="origin", + development="development", + columns="reported", + cumulative=True, ) assert list(tri.development) == [12, 24] @@ -2739,44 +2772,40 @@ def test_set_development_age_respects_mid_period_origin() -> None: def test_set_development_age_semiannual_origin() -> None: """Age works when the origin grain is semiannual, using the calendar (Jan/Jul) anchor to place the valuation date.""" - df = pd.DataFrame( - { - 'origin': ['2017-01-01', '2017-01-01', '2017-07-01', '2018-01-01'], - 'development': [6, 12, 6, 6], - 'reported': [1.0, 2.0, 3.0, 5.0] - } - ) + df = pd.DataFrame({ + "origin": ["2017-01-01", "2017-01-01", "2017-07-01", "2018-01-01"], + "development": [6, 12, 6, 6], + "reported": [1.0, 2.0, 3.0, 5.0], + }) tri = cl.Triangle( data=df, - origin='origin', - development='development', - columns='reported', - cumulative=True + origin="origin", + development="development", + columns="reported", + cumulative=True, ) - assert tri.origin_grain == 'S' + assert tri.origin_grain == "S" assert list(tri.development) == [6, 12, 18] frame = tri.to_frame(origin_as_datetime=False) - assert frame.loc['2017H1', 6] == 1.0 - assert frame.loc['2017H2', 6] == 3.0 + assert frame.loc["2017H1", 6] == 1.0 + assert frame.loc["2017H2", 6] == 3.0 def test_set_development_age_non_calendar_semiannual_raises() -> None: """A semiannual origin grain that isn't calendar-anchored (Jan/Jul) has no native pandas period, so an age can't be placed - raise clearly.""" - df = pd.DataFrame( - { - 'origin': ['2017-02-01', '2017-02-01', '2017-08-01'], - 'development': [6, 12, 6], - 'reported': [1.0, 2.0, 3.0] - } - ) - with pytest.raises(ValueError, match='non-calendar semiannual'): + df = pd.DataFrame({ + "origin": ["2017-02-01", "2017-02-01", "2017-08-01"], + "development": [6, 12, 6], + "reported": [1.0, 2.0, 3.0], + }) + with pytest.raises(ValueError, match="non-calendar semiannual"): cl.Triangle( data=df, - origin='origin', - development='development', - columns='reported', - cumulative=True + origin="origin", + development="development", + columns="reported", + cumulative=True, ) @@ -2784,19 +2813,17 @@ def test_set_development_bare_years_unaffected_by_age_support() -> None: """A development column that is genuinely a bare calendar year (e.g. the literal year 1970) must still parse as a date, not get reinterpreted as an age.""" - df = pd.DataFrame( - { - 'origin': [1969, 1970], - 'development': [1970, 1970], - 'reported': [1.0, 2.0] - } - ) + df = pd.DataFrame({ + "origin": [1969, 1970], + "development": [1970, 1970], + "reported": [1.0, 2.0], + }) tri = cl.Triangle( data=df, - origin='origin', - development='development', - columns='reported', - cumulative=True + origin="origin", + development="development", + columns="reported", + cumulative=True, ) assert list(tri.development) == ["1970"]