"""Experience-study functions over the canonical :class:`actuarialpy.Experience`.
These are the analytical consumers of the ecosystem's shared experience
container: each takes an :class:`~actuarialpy.Experience` (which binds the
column roles once) plus the study-specific decisions -- groupings, windows,
bands, claimant keys, expected columns -- and returns a report DataFrame.
Transformations (``complete``, ``adjust``, ``deseasonalize``, ``filter``,
``with_status``) live on the ``Experience`` object itself in ``actuarialpy``;
this module is everything that *reads* one.
"""
from __future__ import annotations
from collections.abc import Iterable
from typing import Any
import pandas as pd
from actuarialpy import Experience, resolve_amount, resolve_date, single_role_or_none
from actuarialpy.columns import validate_columns
from actuarialpy.credibility import credibility_weighted_estimate
from actuarialpy.metrics import per_exposure, safe_divide
from actuarialpy.pooling import pool_losses
from actuarialpy.trend import _comparison_masks
from experiencestudies.banding import summarize_by_band
from experiencestudies.claimants import claim_concentration, summarize_claimants
from experiencestudies.claimants import top_claimants as _top_claimants
from experiencestudies.cohorts import cohort_summary, duration_summary
from experiencestudies.components import component_driver_analysis, summarize_components
from experiencestudies.decomposition import (
decompose_per_exposure_trend,
frequency_severity_summary,
)
from experiencestudies.expected import summarize_actual_vs_expected
from experiencestudies.experience import status_summary, summarize_experience, summarize_views
from experiencestudies.rolling import rolling_summary
__all__ = [
"summary",
"views",
"rolling",
"frequency_severity",
"decompose_trend",
"components",
"component_summary",
"actual_vs_expected",
"claimants",
"top_claimants",
"claimant_concentration",
"cohort",
"duration",
"by_status",
"by_band",
"margin",
"credibility_weighted",
"pool_claimants",
]
def _resolve_count(exp: Experience, count_col: str | None) -> str:
resolved = count_col or single_role_or_none(exp.count)
if resolved is None:
raise ValueError(
"A count column is required. Bind count=... on the Experience or "
"pass count_col=... to this function."
)
validate_columns(exp.data, [resolved])
return resolved
def _resolve_exposure(exp: Experience, exposure_col: str | None) -> str:
if exposure_col is not None:
validate_columns(exp.data, [exposure_col])
return exposure_col
resolved = single_role_or_none(exp.exposure)
if resolved is None:
raise ValueError(
"An exposure column is required for this function. Bind "
"exposure=... on the Experience or pass exposure_col=... here."
)
return resolved
[docs]
def summary(exp: Experience, by: str | list[str] | None = None, **kwargs: Any) -> pd.DataFrame:
"""Summarize experience by optional grouping columns.
``profile`` (e.g. ``"health"``) may be passed to apply profile naming
defaults such as ``mlr`` for the ratio column.
"""
if by is None:
by = kwargs.pop("groupby", None)
return summarize_experience(
exp.data,
groupby=by,
expense_cols=kwargs.pop("expense_cols", kwargs.pop("expense", list(exp.expense) or None)),
revenue_cols=kwargs.pop("revenue_cols", kwargs.pop("revenue", list(exp.revenue) or None)),
exposure_cols=kwargs.pop("exposure_cols", kwargs.pop("exposure", list(exp.exposure) or None)),
**kwargs,
)
[docs]
def views(
exp: Experience, views: dict[str, str | Iterable[str] | None], **kwargs: Any
) -> dict[str, pd.DataFrame]:
"""Create several named grouped experience views."""
return summarize_views(
exp.data,
views=views,
expense_cols=kwargs.pop("expense_cols", kwargs.pop("expense", list(exp.expense) or None)),
revenue_cols=kwargs.pop("revenue_cols", kwargs.pop("revenue", list(exp.revenue) or None)),
exposure_cols=kwargs.pop("exposure_cols", kwargs.pop("exposure", list(exp.exposure) or None)),
**kwargs,
)
[docs]
def rolling(
exp: Experience,
window: int = 12,
*,
groupby: str | list[str] | None = None,
date_col: str | None = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Create a rolling-period experience summary."""
return rolling_summary(
exp.data,
date_col=resolve_date(exp, date_col),
window=window,
groupby=groupby,
expense_cols=kwargs.pop("expense_cols", kwargs.pop("expense", list(exp.expense) or None)),
revenue_cols=kwargs.pop("revenue_cols", kwargs.pop("revenue", list(exp.revenue) or None)),
exposure_cols=kwargs.pop("exposure_cols", kwargs.pop("exposure", list(exp.exposure) or None)),
**kwargs,
)
[docs]
def frequency_severity(
exp: Experience,
*,
count_col: str | None = None,
loss_col: str | None = None,
exposure_col: str | None = None,
groupby: str | list[str] | None = None,
) -> pd.DataFrame:
"""Per-group claim frequency, severity, and per-exposure loss.
Uses the bound ``count``, ``expense`` (as the loss), and ``exposure``
roles. The identity ``loss_per_exposure == frequency * severity`` holds
for every row.
"""
data, resolved_loss = resolve_amount(exp, loss_col)
return frequency_severity_summary(
data,
count_col=_resolve_count(exp, count_col),
loss_col=resolved_loss,
exposure_col=_resolve_exposure(exp, exposure_col),
groupby=groupby,
)
[docs]
def decompose_trend(
exp: Experience,
*,
count_col: str | None = None,
loss_col: str | None = None,
exposure_col: str | None = None,
mix_by: str | Iterable[str] | None = None,
groupby: str | list[str] | None = None,
period_col: str | None = None,
prior_period: Any = None,
current_period: Any = None,
date_col: str | None = None,
prior_start: Any = None,
prior_end: Any = None,
current_start: Any = None,
current_end: Any = None,
prior_filter: Any = None,
current_filter: Any = None,
) -> pd.DataFrame:
"""Decompose the per-exposure loss trend between two periods.
Splits the bound frame into prior and current with the same comparison
modes as :func:`actuarialpy.trend_summary` -- ``period_col`` with
``prior_period`` / ``current_period``, a ``date_col`` with prior/current
ranges (the bound ``date`` is used when no ``date_col`` is passed), or
explicit masks -- then decomposes the change via
:func:`decompose_per_exposure_trend` using the bound ``count``,
``expense`` (as the loss), and ``exposure`` roles. Pass ``mix_by`` to add
the third LMDI mix term; ``groupby`` reports one decomposition per group.
"""
resolved_count = _resolve_count(exp, count_col)
resolved_exposure = _resolve_exposure(exp, exposure_col)
data, resolved_loss = resolve_amount(exp, loss_col)
date_mode = any(
v is not None for v in (date_col, prior_start, prior_end, current_start, current_end)
)
resolved_date = (date_col if date_col is not None else exp.date) if date_mode else None
prior_mask, current_mask, _ = _comparison_masks(
data,
period_col=period_col,
prior_period=prior_period,
current_period=current_period,
date_col=resolved_date,
prior_start=prior_start,
prior_end=prior_end,
current_start=current_start,
current_end=current_end,
prior_filter=prior_filter,
current_filter=current_filter,
)
return decompose_per_exposure_trend(
data.loc[prior_mask],
data.loc[current_mask],
count_col=resolved_count,
loss_col=resolved_loss,
exposure_col=resolved_exposure,
on=groupby,
mix_by=mix_by,
)
[docs]
def components(
exp: Experience,
component_cols: str | list[str],
*,
exposure_col: str | None = None,
groupby: str | list[str] | None = None,
date_col: str | None = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Explain component drivers between two periods."""
# Use the bound date column only for date-range comparisons; with
# period_col it would create two comparison modes and raise.
resolved_date = date_col if date_col is not None else exp.date
if "period_col" in kwargs and date_col is None:
resolved_date = None
return component_driver_analysis(
exp.data,
component_cols=component_cols,
exposure_col=exposure_col or single_role_or_none(exp.exposure),
groupby=groupby,
date_col=resolved_date,
**kwargs,
)
[docs]
def component_summary(
exp: Experience,
component_cols: str | list[str],
*,
groupby: str | list[str] | None = None,
exposure_col: str | None = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Summarize component amounts, per-exposure values, and shares."""
return summarize_components(
exp.data,
groupby=groupby,
component_cols=component_cols,
exposure_col=exposure_col or single_role_or_none(exp.exposure),
**kwargs,
)
[docs]
def actual_vs_expected(
exp: Experience,
expected: str | list[str],
*,
actual: str | list[str] | None = None,
groupby: str | list[str] | None = None,
exposure: str | list[str] | None = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Summarize actual-versus-expected experience.
If ``actual`` is omitted, the bound expense columns are used.
"""
return summarize_actual_vs_expected(
exp.data,
groupby=groupby,
actual_cols=list(exp.expense) if actual is None else actual,
expected_cols=expected,
exposure_cols=list(exp.exposure) if exposure is None else exposure,
**kwargs,
)
[docs]
def claimants(
exp: Experience,
claimant_col: str,
*,
amount_cols: str | list[str] | None = None,
groupby: str | list[str] | None = None,
exposure_col: str | None = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Aggregate the experience to claimant/member/risk level."""
return summarize_claimants(
exp.data,
claimant_col=claimant_col,
amount_cols=list(exp.expense) if amount_cols is None else amount_cols,
groupby=groupby,
exposure_col=exposure_col,
**kwargs,
)
[docs]
def top_claimants(
exp: Experience | pd.DataFrame,
claimant_col: str | None = None,
*,
amount_cols: str | list[str] | None = None,
amount_col: str | None = None,
groupby: str | list[str] | None = None,
n: int = 25,
**kwargs: Any,
) -> pd.DataFrame:
"""Return top claimants by amount.
Accepts an :class:`~actuarialpy.Experience` (bound expense columns are the
default amounts) or a plain DataFrame with explicit ``amount_cols``.
"""
if isinstance(exp, pd.DataFrame):
return _top_claimants(
exp,
claimant_col=claimant_col,
amount_cols=amount_cols,
amount_col=amount_col,
groupby=groupby,
n=n,
**kwargs,
)
return _top_claimants(
exp.data,
claimant_col=claimant_col,
amount_cols=(
list(exp.expense) if amount_cols is None and amount_col is None else amount_cols
),
amount_col=amount_col,
groupby=groupby,
n=n,
**kwargs,
)
[docs]
def claimant_concentration(
exp: Experience,
claimant_col: str,
*,
amount_cols: str | list[str] | None = None,
groupby: str | list[str] | None = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Summarize how concentrated experience is among top claimants."""
claimant_summary = summarize_claimants(
exp.data,
claimant_col=claimant_col,
amount_cols=list(exp.expense) if amount_cols is None else amount_cols,
groupby=groupby,
)
return claim_concentration(claimant_summary, groupby=groupby, **kwargs)
[docs]
def cohort(
exp: Experience,
*,
entity_col: str,
start_date_col: str,
duration_months: int = 12,
groupby: str | list[str] | None = None,
date_col: str | None = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Summarize each entity's first N months or cohort-duration window."""
return cohort_summary(
exp.data,
entity_col=entity_col,
date_col=resolve_date(exp, date_col),
start_date_col=start_date_col,
duration_months=duration_months,
groupby=groupby,
expense_cols=kwargs.pop("expense_cols", kwargs.pop("expense", list(exp.expense) or None)),
revenue_cols=kwargs.pop("revenue_cols", kwargs.pop("revenue", list(exp.revenue) or None)),
exposure_cols=kwargs.pop("exposure_cols", kwargs.pop("exposure", list(exp.exposure) or None)),
**kwargs,
)
[docs]
def duration(
exp: Experience,
*,
entity_col: str,
start_date_col: str,
max_duration_month: int | None = None,
date_col: str | None = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Summarize experience by duration month since entity start."""
return duration_summary(
exp.data,
entity_col=entity_col,
date_col=resolve_date(exp, date_col),
start_date_col=start_date_col,
expense_cols=kwargs.pop("expense_cols", kwargs.pop("expense", list(exp.expense) or None)),
revenue_cols=kwargs.pop("revenue_cols", kwargs.pop("revenue", list(exp.revenue) or None)),
exposure_cols=kwargs.pop("exposure_cols", kwargs.pop("exposure", list(exp.exposure) or None)),
max_duration_month=max_duration_month,
**kwargs,
)
[docs]
def by_status(
exp: Experience, status_col: str, *, entity_col: str | None = None, **kwargs: Any
) -> pd.DataFrame:
"""Summarize experience by a status column (see :meth:`Experience.with_status`)."""
return status_summary(
exp.data,
status_col=status_col,
entity_col=entity_col,
expense_cols=kwargs.pop("expense_cols", kwargs.pop("expense", list(exp.expense) or None)),
revenue_cols=kwargs.pop("revenue_cols", kwargs.pop("revenue", list(exp.revenue) or None)),
exposure_cols=kwargs.pop("exposure_cols", kwargs.pop("exposure", list(exp.exposure) or None)),
**kwargs,
)
[docs]
def by_band(
exp: Experience,
value_col: str,
bands: Any,
*,
labels: Any = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Summarize experience by a size band on ``value_col``."""
return summarize_by_band(
exp.data,
value_col,
bands,
labels=labels,
expense_cols=kwargs.pop("expense_cols", kwargs.pop("expense", list(exp.expense) or None)),
revenue_cols=kwargs.pop("revenue_cols", kwargs.pop("revenue", list(exp.revenue) or None)),
exposure_cols=kwargs.pop("exposure_cols", kwargs.pop("exposure", list(exp.exposure) or None)),
**kwargs,
)
def _experience_first(fn, df_alternative):
import functools
@functools.wraps(fn)
def wrapped(exp, *args, **kwargs):
from actuarialpy import ExperienceSet
if isinstance(exp, ExperienceSet):
exp = exp.tab
if not isinstance(exp, Experience):
raise TypeError(
f"es.{fn.__name__} takes an actuarialpy.Experience as its first "
f"argument (it reads the bound roles); for a plain DataFrame use "
f"{df_alternative}(...) with explicit column names."
)
return fn(exp, *args, **kwargs)
return wrapped
[docs]
def margin(
exp: Experience,
by: str | list[str] | None = None,
*,
margin_col: str = "margin",
ratio_col: str = "margin_ratio",
per_exposure_col: str | None = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Underwriting margin (revenue net of expense) by optional grouping.
Aggregates the bound expense and revenue roles with :func:`summary`, then
adds the margin (``total_revenue - total_expense``), the margin ratio, and
an optional per-exposure margin.
"""
if not exp.revenue:
raise ValueError(
"margin requires a revenue role; bind revenue=... on the Experience"
)
out = summary(exp, by, **kwargs)
out[margin_col] = out["total_revenue"] - out["total_expense"]
out[ratio_col] = safe_divide(out[margin_col], out["total_revenue"])
if per_exposure_col is not None:
exposure = single_role_or_none(exp.exposure)
if exposure is None:
raise ValueError("A single bound exposure is required for per_exposure_col.")
out[per_exposure_col] = per_exposure(out[margin_col], out[exposure])
return out
[docs]
def credibility_weighted(
exp: Experience,
groupby: str | list[str],
*,
z: Any,
metric: str = "loss_ratio",
complement: float | None = None,
out_col: str | None = None,
**kwargs: Any,
) -> pd.DataFrame:
"""Blend each group's ``metric`` with a complement at credibility ``z``.
Computes the grouped summary (:func:`summary`), then blends ``metric``
toward ``complement`` using ``z`` (see
:func:`actuarialpy.credibility_weighted_estimate`). ``z`` may be a scalar
or values aligned to the grouped rows. When ``complement`` is omitted the
book-level value of ``metric`` is used as the complement of credibility.
"""
grouped = summary(exp, groupby, **kwargs)
if metric not in grouped.columns:
raise ValueError(f"metric '{metric}' is not in the summary columns: {list(grouped.columns)}")
if complement is None:
complement = summary(exp, **kwargs)[metric].iloc[0]
name = out_col or f"credibility_weighted_{metric}"
grouped[name] = credibility_weighted_estimate(grouped[metric], complement, z)
return grouped
[docs]
def pool_claimants(
exp: Experience,
claimant_col: str,
pooling_point: float,
*,
amount_cols: str | list[str] | None = None,
groupby: str | list[str] | None = None,
amount_name: str = "total_expense",
**kwargs: Any,
) -> pd.DataFrame:
"""Aggregate to claimant level and split each claimant into pooled/excess.
Summarizes the experience to claimant grain (:func:`claimants`) and caps
each claimant's total at ``pooling_point`` (see
:func:`actuarialpy.pool_losses`), returning pooled and excess columns for
capped experience and the excess hand-off to tail modeling.
"""
claimant_totals = summarize_claimants(
exp.data,
claimant_col=claimant_col,
amount_cols=list(exp.expense) if amount_cols is None else amount_cols,
groupby=groupby,
amount_name=amount_name,
)
return pool_losses(claimant_totals, amount_name, pooling_point, **kwargs)
_DF_ALTERNATIVES = {
"summary": "summarize_experience",
"views": "summarize_views",
"rolling": "rolling_summary",
"margin": "summarize_experience",
"frequency_severity": "frequency_severity_summary",
"decompose_trend": "decompose_per_exposure_trend",
"components": "component_driver_analysis",
"component_summary": "summarize_components",
"actual_vs_expected": "summarize_actual_vs_expected",
"claimants": "summarize_claimants",
"claimant_concentration": "claim_concentration",
"pool_claimants": "actuarialpy.pool_losses",
"cohort": "cohort_summary",
"duration": "duration_summary",
"by_status": "status_summary",
"by_band": "summarize_by_band",
"credibility_weighted": "actuarialpy.credibility_weighted_estimate",
}
for _name, _alt in _DF_ALTERNATIVES.items():
globals()[_name] = _experience_first(globals()[_name], _alt)
del _name, _alt