"""Expense projection workflow."""
from __future__ import annotations
from collections.abc import Iterable
from dataclasses import dataclass
import numpy as np
import pandas as pd
from .actuarialpy_adapter import actuarialpy_function
from .adjustments import Scenario
from .assumptions import AssumptionSet, TrendAssumption
from .calculations import Calculation, CashFlow
from .data import ProjectionData, ProjectionDataset, ProjectionDates
from .exceptions import ValidationError
from .horizon import ProjectionHorizon
from .model import ProjectionModel
from .results import ProjectionResults
def _as_tuple(value: str | Iterable[str]) -> tuple[str, ...]:
if isinstance(value, str):
return (value,)
return tuple(value)
[docs]
@dataclass
class ExpenseProjection:
"""Project expenses on per-exposure, fixed-monthly, percent-of-premium,
and percent-of-claims bases, one table for all four.
``trend`` may be keyed by expense type, so "contractually flat" is a
per-type choice: a zero-trend type projects at its base value while the
types around it trend (see ``examples/expenses.py``). The ``claims`` and
``premium`` inputs are keyed value streams joined on the projection keys
and period — nothing requires the claims stream to contain claims, so a
percentage of any per-period quantity (a claims subset such as hospital
claims only, or another expense run's output) is the same mechanism fed
a different table. Percentage bases are not multiplied by
``active_fraction``: proration is already embedded in the stream they
reference, and prorating again would double count.
"""
expenses: pd.DataFrame
projection_keys: tuple[str, ...] | list[str]
expense_type_col: str
base_value_col: str
basis_col: str
base_date_col: str
horizon: ProjectionHorizon
trend: TrendAssumption
exposure: pd.DataFrame | None = None
premium: pd.DataFrame | None = None
claims: pd.DataFrame | None = None
exposure_col: str = "exposure"
premium_col: str = "premium"
claims_col: str = "projected_claims"
dates: ProjectionDates | None = None
def __post_init__(self) -> None:
self.projection_keys = _as_tuple(self.projection_keys)
required = [
*self.projection_keys,
self.expense_type_col,
self.base_value_col,
self.basis_col,
self.base_date_col,
]
missing = [column for column in required if column not in self.expenses.columns]
if missing:
raise ValidationError(f"expenses is missing columns: {missing}")
allowed = {"per_exposure", "fixed_monthly", "percent_premium", "percent_claims"}
unknown = sorted(set(self.expenses[self.basis_col].dropna()) - allowed)
if unknown:
raise ValidationError(f"unknown expense bases: {unknown}")
def _model(self) -> ProjectionModel:
record_grain = self.projection_keys + (self.expense_type_col,)
def trend_months(context):
base = pd.to_datetime(context[self.base_date_col])
target = pd.to_datetime(context["period_midpoint"])
return (
(target.dt.year - base.dt.year) * 12
+ (target.dt.month - base.dt.month)
+ (target.dt.day - base.dt.day) / 30.4375
)
def rate(context):
factor = actuarialpy_function("trend_factor")(
context[self.trend.name], trend_months(context)
)
return context[self.base_value_col] * factor
def expense(context):
basis = context[self.basis_col]
projected_rate = context["projected_expense_rate"]
result = pd.Series(np.nan, index=context.frame.index, dtype=float)
mask = basis.eq("per_exposure")
if mask.any():
result.loc[mask] = (
projected_rate.loc[mask]
* context[self.exposure_col].loc[mask]
* context["active_fraction"].loc[mask]
)
mask = basis.eq("fixed_monthly")
if mask.any():
result.loc[mask] = (
projected_rate.loc[mask]
* context["active_fraction"].loc[mask]
)
mask = basis.eq("percent_premium")
if mask.any():
result.loc[mask] = (
projected_rate.loc[mask] * context[self.premium_col].loc[mask]
)
mask = basis.eq("percent_claims")
if mask.any():
result.loc[mask] = (
projected_rate.loc[mask] * context[self.claims_col].loc[mask]
)
return result
return ProjectionModel(
assumptions=AssumptionSet(self.trend),
calculations=[
Calculation(
"projected_expense_rate",
formula=rate,
aggregation="mean",
grain=record_grain,
),
CashFlow(
"projected_expense",
formula=expense,
aggregation="sum",
grain=record_grain,
reporting_role="expense",
depends_on=("projected_expense_rate",),
),
],
)
def project(
self,
*,
scenarios: Scenario | Iterable[Scenario] | None = None,
) -> ProjectionResults:
records = ProjectionData(
self.expenses.copy(),
projection_keys=self.projection_keys,
component_keys=[self.expense_type_col],
dates=self.dates,
)
dataset = ProjectionDataset(records)
for name, table, value_col in (
("exposure", self.exposure, self.exposure_col),
("premium", self.premium, self.premium_col),
("claims", self.claims, self.claims_col),
):
if table is not None:
required = [*self.projection_keys, "projection_period", value_col]
missing = [column for column in required if column not in table.columns]
if missing:
raise ValidationError(f"{name} table is missing columns: {missing}")
dataset.add_table(
name,
table,
keys=[*self.projection_keys, "projection_period"],
)
bases = set(self.expenses[self.basis_col])
if "per_exposure" in bases and self.exposure is None:
raise ValidationError("Per-exposure expenses require an exposure table")
if "percent_premium" in bases and self.premium is None:
raise ValidationError("percent_premium expenses require a premium table")
if "percent_claims" in bases and self.claims is None:
raise ValidationError("percent_claims expenses require a claims table")
return self._model().project(dataset, self.horizon, scenarios=scenarios)