"""Rolling-period actuarial summaries.
A ``window``-month rolling summary is *calendar*-aware by default: the window
spans ``window`` contiguous calendar periods (at ``freq``), so a gap in the data
is a genuine gap. Under the earlier row-count behaviour a 12-row window that
straddled a missing month was silently labelled a complete 12-month period and
summed rows from further back in time. Pass ``window_basis="observations"`` to
recover the row-count behaviour explicitly.
"""
from __future__ import annotations
from collections.abc import Iterable
import pandas as pd
from actuarialpy.columns import as_list, per_exposure_name, validate_columns
from actuarialpy.metrics import loss_ratio, per_exposure
from experiencestudies.experience import summarize_experience
[docs]
def rolling_summary(
df: pd.DataFrame,
*,
date_col: str,
window: int = 12,
groupby: str | Iterable[str] | None = None,
expense_cols: str | Iterable[str],
revenue_cols: str | Iterable[str],
exposure_cols: str | Iterable[str] | None = None,
min_periods: int | None = None,
drop_incomplete: bool = True,
ratio_col: str = "loss_ratio",
freq: str = "M",
window_basis: str = "calendar",
) -> pd.DataFrame:
"""Rolling sums and ratios by period and optional grouping.
``window_basis`` controls what the window counts:
- ``"calendar"`` (default): ``window`` contiguous calendar periods at
``freq``. Each group is aggregated to ``freq`` and reindexed onto its
complete period range, so a missing period is a real gap -- it is summed
over the periods actually present and, unless the whole ``window`` is
present, the row is treated as incomplete. This is the correct basis for a
summary described in months.
- ``"observations"``: ``window`` consecutive *rows*, regardless of the
calendar spacing between them (the historical behaviour).
``period_start`` and ``period_end`` bound each window. ``min_periods`` is the
minimum number of populated periods required to emit a window's sum (default:
``window``); ``drop_incomplete`` (default ``True``) keeps only windows with a
full ``window`` periods populated.
"""
if window <= 0:
raise ValueError("window must be positive")
if window_basis not in ("calendar", "observations"):
raise ValueError("window_basis must be 'calendar' or 'observations'")
groups = as_list(groupby)
expenses = as_list(expense_cols)
revenues = as_list(revenue_cols)
exposures = as_list(exposure_cols)
validate_columns(df, groups + [date_col] + expenses + revenues + exposures)
min_periods = window if min_periods is None else min_periods
base = summarize_experience(
df,
groupby=groups + [date_col],
expense_cols=expenses,
revenue_cols=revenues,
exposure_cols=exposures,
ratio_col="period_ratio",
)
# Convert dates BEFORE sorting: sorting raw non-ISO date strings orders them
# lexicographically, which is not chronological.
base[date_col] = pd.to_datetime(base[date_col])
base = base.sort_values(groups + [date_col] if groups else [date_col])
amount_cols = ["total_expense", "total_revenue"] + exposures
iterator = base.groupby(groups, dropna=False, sort=False) if groups else [(None, base)]
pieces = []
for key, part in iterator:
if window_basis == "calendar":
out = _calendar_window(part, date_col, amount_cols, window, min_periods, freq)
else:
out = _observation_window(part, date_col, amount_cols, window, min_periods)
if groups:
key_tuple = key if isinstance(key, tuple) else (key,)
for col, val in zip(groups, key_tuple, strict=True):
out.insert(0, col, val)
out[ratio_col] = loss_ratio(out["total_expense"], out["total_revenue"])
for exposure in exposures:
out[per_exposure_name("total_expense", exposure)] = per_exposure(out["total_expense"], out[exposure])
out[per_exposure_name("total_revenue", exposure)] = per_exposure(out["total_revenue"], out[exposure])
if drop_incomplete:
out = out[out["months_available"] >= window].copy()
pieces.append(out)
if not pieces:
return pd.DataFrame()
result = pd.concat(pieces, ignore_index=True)
if drop_incomplete:
result = result.drop(columns=["months_available"])
return result
def _observation_window(part, date_col, amount_cols, window, min_periods):
"""Historical row-count window: ``window`` consecutive rows."""
part = part.sort_values(date_col).reset_index(drop=True)
rolled = part[amount_cols].rolling(window=window, min_periods=min_periods).sum()
months_available = part["total_expense"].rolling(window=window, min_periods=1).count().astype(int)
dates = pd.to_datetime(part[date_col]).reset_index(drop=True)
starts = [dates.iloc[max(0, i - window + 1)] for i in range(len(part))]
out = pd.DataFrame(index=part.index)
out["period_start"] = starts
out["period_end"] = dates.to_numpy()
out["months_available"] = months_available.to_numpy()
for col in amount_cols:
out[col] = rolled[col].to_numpy()
return out.reset_index(drop=True)
def _calendar_window(part, date_col, amount_cols, window, min_periods, freq):
"""Calendar window: reindex to the complete period grid, then roll over it."""
periods = pd.PeriodIndex(pd.to_datetime(part[date_col]), freq=freq)
# collapse any sub-period duplicates, then fill the complete calendar range
grid = part[amount_cols].groupby(periods, sort=True).sum()
full = pd.period_range(grid.index.min(), grid.index.max(), freq=freq)
grid = grid.reindex(full)
rolled = grid.rolling(window=window, min_periods=min_periods).sum()
# count populated calendar periods in each window (missing periods are NaN rows)
months_available = grid[amount_cols[0]].rolling(window=window, min_periods=1).count().astype(int)
ends = full.to_timestamp()
starts = [full[max(0, i - window + 1)].to_timestamp() for i in range(len(full))]
out = pd.DataFrame(index=range(len(full)))
out["period_start"] = starts
out["period_end"] = ends
out["months_available"] = months_available.to_numpy()
for col in amount_cols:
out[col] = rolled[col].to_numpy()
return out