class CLVStats:
"""Prepares the per-customer BTYD summary for pymc-marketing CLV models.
Aggregates transaction data into the ``frequency`` / ``recency`` / ``T`` /
``monetary_value`` summary that ``ParetoNBDModel`` and ``GammaGammaModel`` consume.
Results are accessible via the ``table`` attribute (ibis Table) or the ``df`` property
(materialized pandas DataFrame). See the module docstring for column definitions.
Constructing ``CLVStats`` runs one aggregate query against the backend to resolve and
validate the observation window (plus, when ``customer_attributes`` is given, one aggregate to
check its customer_id is unique and a single distinct-value query enumerating all ``one_hot_col``
categories); the full per-customer summary stays lazy and is materialized only on first
``df`` access.
Args:
data (pd.DataFrame | ibis.Table): Transaction data with the ``customer_id``,
``transaction_date`` (a date/datetime type), and ``unit_spend`` columns. Undated rows and
returns-only days (net spend <= 0) are excluded; a customer left with none drops out.
period (str, optional): The time unit for ``recency`` and ``T``. One of ``"day"``,
``"week"``, or ``"month"`` (case-insensitive; short forms like ``"d"``/``"m"`` accepted).
A month is a fixed 365.25/12-day unit. Defaults to ``"week"``.
observation_period_end (str | datetime.date | None, optional): The end of the
observation window, from which each customer's age ``T`` is measured. An
ISO-8601 string or a ``datetime.date``. Defaults to the latest transaction date
in ``data``.
customer_attributes (pd.DataFrame | ibis.Table | None, optional): A per-customer table (one
row per ``customer_id``) of extra columns to attach to the summary via a left join —
covariates such as signup channel, region, or a pre-computed ``stores_shopped`` count.
Build it however you like (e.g. ``SegTransactionStats`` grouped on ``customer_id``). Must
cover every customer with a non-NULL value for each (non-one-hot) covariate; a missing
customer or NULL value is rejected on :attr:`df` access. Defaults to ``None``.
one_hot_col (str | list[str] | None, optional): Column(s) of ``customer_attributes`` to
one-hot encode into 0/1 dummy columns suitable for ``ParetoNBDModel`` covariates. For
each column one level is dropped as the reference level (dummies would otherwise be
collinear with the model intercept) — NULL when the column contains NULLs, otherwise the
first value in sorted order — and the original column is removed. Warns (``UserWarning``)
if a column yields more than 32 dummies (a likely high-cardinality mistake) or zero dummies
(a single-category column that contributes no covariate). Defaults to ``None``.
Raises:
TypeError: If ``data`` or ``customer_attributes`` is not a pandas DataFrame or an Ibis Table,
if ``transaction_date`` is not a date/datetime type, or if ``observation_period_end`` is
not a valid date-like value.
ValueError: If required columns are missing, if ``period`` is not ``"day"``, ``"week"``, or
``"month"``, if ``observation_period_end`` is before the latest transaction, if
``customer_attributes`` lacks the ``customer_id`` column or has a duplicate ``customer_id``,
if ``one_hot_col`` is given without ``customer_attributes`` or names a column absent from
it, if an attached column name collides with a reserved summary column, or if ``data``
contains no transactions.
"""
#: A "month" is a fixed 365.25/12-day unit (the average Gregorian month), not a calendar month, so
#: fractional age stays well defined and 12 months is exactly one year.
_DAYS_PER_MONTH: ClassVar[float] = 365.25 / 12
VALID_PERIODS: ClassVar[tuple[str, ...]] = ("day", "week", "month")
_DAYS_PER_PERIOD: ClassVar[dict[str, float]] = {"day": 1, "week": 7, "month": _DAYS_PER_MONTH}
#: Maps each period to pymc-marketing's ``time_unit`` code; exposed via ``pymc_time_unit``.
_PYMC_TIME_UNIT: ClassVar[dict[str, str]] = {"day": "D", "week": "W", "month": "M"}
#: |Pearson r| between frequency and monetary_value above which ``repeat_buyers`` warns: GammaGamma
#: assumes the two are independent, and a stronger correlation biases its spend estimates (0.10-0.15
#: is the practical "weak enough" band in BTYD guidance).
_MONETARY_FREQUENCY_CORR_WARN: ClassVar[float] = 0.15
#: The base BTYD output columns (literal pymc-marketing names); anything else on the summary is a
#: covariate attached via customer_attributes / one_hot_col. See ``covariate_cols``.
_BASE_SUMMARY_COLUMNS: ClassVar[tuple[str, ...]] = ("customer_id", "frequency", "recency", "T", "monetary_value")
def __init__(
self,
data: pd.DataFrame | ibis.Table,
*,
period: str = "week",
observation_period_end: str | datetime.date | None = None,
customer_attributes: pd.DataFrame | ibis.Table | None = None,
one_hot_col: str | list[str] | None = None,
) -> None:
"""Initializes and computes the BTYD summary."""
cols = ColumnHelper()
data = ensure_ibis_table(data, "data")
ensure_data_has_columns(data, [cols.customer_id, cols.transaction_date, cols.unit_spend])
ensure_tznaive_datetime(data, cols.transaction_date)
period = ensure_period(period, self.VALID_PERIODS, "period")
attributes = self._prepare_customer_attributes(customer_attributes)
one_hot_cols = self._resolve_one_hot_cols(attributes, one_hot_col)
# Drop undated rows: a NULL date forms its own occasion and skews recency/T.
data = data.filter(data[cols.transaction_date].notnull()) # noqa: PD004 (ibis API, not pandas)
# Cast the single max scalar to a date (not every row) — same idiom as segmentation/rfm.py.
latest_raw = data[cols.transaction_date].max().cast("date").execute()
if pd.isna(latest_raw): # None or NaT: the table has no transactions to summarize.
msg = "data contains no transactions; cannot build a CLV summary."
raise ValueError(msg)
latest_transaction = _coerce_to_date(latest_raw)
if observation_period_end is None:
observation_period_end = latest_transaction
else:
observation_period_end = _coerce_to_date(observation_period_end)
if observation_period_end < latest_transaction:
msg = (
f"observation_period_end ({observation_period_end}) must be on or after the latest "
f"transaction date ({latest_transaction}); an earlier end would produce a negative age."
)
raise ValueError(msg)
summary = self._compute_summary(data, period, observation_period_end)
table = self._attach_customer_attributes(summary, attributes, one_hot_cols)
# pymc-marketing requires the id column named literally "customer_id" (no remap arg), like the
# other four. Rename so an overridden customer_id option still yields a model-ready summary.
if cols.customer_id != "customer_id":
table = table.rename({"customer_id": cols.customer_id})
self.period = period
self.table = table
@staticmethod
def _prepare_customer_attributes(customer_attributes: pd.DataFrame | ibis.Table | None) -> ibis.Table | None:
"""Validate and normalize the customer_attributes table to an Ibis Table.
Args:
customer_attributes (pd.DataFrame | ibis.Table | None): The caller-supplied per-customer
attribute table (one row per ``customer_id``), or ``None``.
Returns:
ibis.Table | None: The attributes as an Ibis Table, or ``None`` if none were supplied.
Raises:
TypeError: If ``customer_attributes`` is neither a pandas DataFrame nor an Ibis Table.
ValueError: If it lacks the ``customer_id`` column, or has a duplicate ``customer_id``
(a left join on a non-unique key would fan out, i.e. duplicate, the summary rows).
"""
if customer_attributes is None:
return None
cols = ColumnHelper()
attributes = ensure_ibis_table(customer_attributes, "customer_attributes")
if cols.customer_id not in attributes.columns:
msg = f"customer_attributes must contain the customer_id column '{cols.customer_id}'."
raise ValueError(msg)
counts = attributes.aggregate(
rows=attributes.count(),
customers=attributes[cols.customer_id].nunique(),
).execute()
if counts["rows"].iloc[0] != counts["customers"].iloc[0]:
msg = "customer_attributes must have one row per customer_id; found duplicate customer ids."
raise ValueError(msg)
return attributes
@staticmethod
def _resolve_one_hot_cols(
attributes: ibis.Table | None,
one_hot_col: str | list[str] | None,
) -> list[str]:
"""Normalize ``one_hot_col`` to a de-duplicated list of columns present in ``attributes``.
Args:
attributes (ibis.Table | None): The prepared customer_attributes table, or ``None``.
one_hot_col (str | list[str] | None): The user-supplied one-hot column(s).
Returns:
list[str]: The normalized, de-duplicated one-hot column names (empty if ``one_hot_col`` is None).
Raises:
ValueError: If ``one_hot_col`` is given without ``customer_attributes``, or names a column
absent from ``customer_attributes``.
"""
if one_hot_col is None:
return []
if attributes is None:
msg = "one_hot_col requires customer_attributes; there are no columns to encode."
raise ValueError(msg)
one_hot_cols = ensure_columns(attributes, one_hot_col, "one_hot_col")
# De-duplicate while preserving order: a repeated column would be encoded twice, and the first
# pass drops it, so the second would fail to find it.
return list(dict.fromkeys(one_hot_cols))
@staticmethod
def _attach_customer_attributes(
summary: ibis.Table,
attributes: ibis.Table | None,
one_hot_cols: list[str],
) -> ibis.Table:
"""Left-join the per-customer attributes onto the summary, one-hot encoding the requested columns.
Args:
summary (ibis.Table): The base BTYD summary (one row per customer).
attributes (ibis.Table | None): The prepared customer_attributes table, or ``None``.
one_hot_cols (list[str]): Normalized one-hot column names (a subset of the attribute columns).
Returns:
ibis.Table: The summary with the attribute and one-hot columns attached.
Raises:
ValueError: If an attached column name collides with a reserved BTYD summary column
(``customer_id``, ``frequency``, ``recency``, ``T``, ``monetary_value``).
"""
if attributes is None:
return summary
cols = ColumnHelper()
if len(one_hot_cols) > 0:
attributes = _one_hot_encode(attributes, one_hot_cols)
attached = [col for col in attributes.columns if col != cols.customer_id]
# Reserve the literal "customer_id" too (the output id name __init__ renames to); else under an
# overridden customer_id option an attribute named "customer_id" silently collides.
collisions = sorted(set(attached) & (set(summary.columns) | {"customer_id"}))
if len(collisions) > 0:
msg = f"customer_attributes / one_hot_col columns collide with reserved BTYD summary columns: {collisions}"
raise ValueError(msg)
# Reselect the base columns plus the attached ones, dropping the duplicated join key ibis
# appends (e.g. customer_id_right); selecting by name avoids hardcoding the join suffix.
return summary.left_join(attributes, cols.customer_id)[[*summary.columns, *attached]]
@classmethod
def _compute_summary(
cls,
data: ibis.Table,
period: str,
observation_period_end: datetime.date,
) -> ibis.Table:
"""Computes the BTYD summary table.
Args:
data (ibis.Table): The validated transaction data.
period (str): The validated period unit (``"day"``, ``"week"``, or ``"month"``).
observation_period_end (datetime.date): The resolved observation window end.
Returns:
ibis.Table: One row per customer with the BTYD summary columns.
"""
cols = ColumnHelper()
# Collapse to one purchase occasion per customer per calendar day, summing spend
# within the day (same-day baskets are a single occasion under the BTYD convention).
day = data[cols.transaction_date].cast("date").name("_day")
daily = data.group_by([cols.customer_id, day]).aggregate(_day_spend=data[cols.unit_spend].sum())
# A day is an occasion only if its net spend is positive; returns-only days are not purchases.
daily = daily.filter(daily._day_spend > 0)
# A repeat occasion is any purchase day after the customer's first purchase day. The flag is
# an int rather than a bool because SQL Server has no boolean type to project into a SELECT.
first_day = daily["_day"].min().over(ibis.window(group_by=daily[cols.customer_id]))
daily = daily.mutate(_is_repeat=ibis.ifelse(daily["_day"] > first_day, 1, 0).cast("int8"))
summary = daily.group_by(cols.customer_id).aggregate(
_first_day=daily["_day"].min(),
_last_day=daily["_day"].max(),
_occasions=daily.count(),
_repeat_spend=daily._day_spend.sum(where=daily._is_repeat == 1),
)
days_per_period = cls._DAYS_PER_PERIOD[period]
observation_end = ibis.literal(observation_period_end)
frequency = summary._occasions - 1
# frequency / recency / T / monetary_value are pymc-marketing's required literal names
# (deliberately not options.py names). The id keeps its configured name here for the join
# keys; __init__ renames the final column to the literal "customer_id" the models also need.
return summary.mutate(
frequency=frequency.cast("int64"),
# Day-granularity delta (Oracle-safe) divided by days-per-period for a fractional age.
recency=summary._last_day.delta(summary._first_day, unit="day") / days_per_period,
T=observation_end.delta(summary._first_day, unit="day") / days_per_period,
monetary_value=summary._repeat_spend / frequency.nullif(0),
).select(
cols.customer_id,
"frequency",
"recency",
"T",
"monetary_value",
)
@functools.cached_property
def df(self) -> pd.DataFrame:
"""Returns the materialized BTYD summary as a pandas DataFrame.
The ``customer_id`` is a column (not the index) so the frame can be passed straight
to ``ParetoNBDModel`` / ``GammaGammaModel``, which expect it as a column.
Returns:
pd.DataFrame: One row per customer with columns ``customer_id``, ``frequency``,
``recency``, ``T``, and ``monetary_value``, followed by any ``customer_attributes``
and one-hot columns requested at construction.
Raises:
ValueError: If a covariate column is NULL for any customer (a customer missing from
``customer_attributes``, or a NULL attribute value). NULL covariates silently break
``ParetoNBDModel``'s fit, which does not validate for them.
"""
result = self.table.execute()
# Checking one-hot dummies too is safe: a NULL *source* value became the reference level (0), so a
# dummy reads NULL only for a customer missing from the join -- exactly the case worth rejecting.
null_covariates = [col for col in self.covariate_cols if result[col].isna().any()]
if len(null_covariates) > 0:
msg = (
f"customer_attributes leaves NULL in covariate column(s) {null_covariates}: a customer is "
"missing from customer_attributes, or has a NULL attribute value. NULL covariates make "
"ParetoNBDModel's fit fail. Provide non-NULL covariates for every customer, or drop those "
"customers first."
)
raise ValueError(msg)
return result
@property
def repeat_buyers(self) -> pd.DataFrame:
"""The GammaGamma-ready subset of :attr:`df` — the repeat buyers (``frequency > 0``, index reset).
One-time buyers (``NaN`` monetary_value) are excluded; ``GammaGammaModel`` cannot fit them. No
spend filter is needed because ``monetary_value`` is always positive here. Warns (``UserWarning``)
if ``|Pearson r|`` between ``frequency`` and ``monetary_value`` exceeds
``_MONETARY_FREQUENCY_CORR_WARN``, breaking GammaGamma's independence assumption.
Returns:
pd.DataFrame: The ``frequency > 0`` rows of :attr:`df`, index reset.
"""
summary = self.df
gg_ready = summary[summary["frequency"] > 0].reset_index(drop=True)
if len(gg_ready) == 0: # no repeat buyers: nothing to correlate
return gg_ready
# Correlation is only defined when both columns vary; a constant column would divide by a zero
# standard deviation (NaN, plus a numpy warning), so skip the check in that degenerate case.
frequency, monetary_value = gg_ready["frequency"], gg_ready["monetary_value"]
both_vary = frequency.min() != frequency.max() and monetary_value.min() != monetary_value.max()
if both_vary:
corr = frequency.corr(monetary_value)
if abs(corr) > self._MONETARY_FREQUENCY_CORR_WARN:
warnings.warn(
f"frequency and monetary_value are correlated (Pearson r={corr:.2f}); GammaGamma assumes "
"they are independent, so its spend estimates may be biased.",
stacklevel=2,
)
return gg_ready
@property
def covariate_cols(self) -> list[str]:
"""The covariate columns attached to the summary (customer_attributes and one-hot dummies).
Every column beyond the base BTYD summary, ready to pass as ``purchase_covariate_cols`` /
``dropout_covariate_cols`` to ``ParetoNBDModel``.
Returns:
list[str]: The covariate column names in summary order (empty if none were requested).
"""
return [col for col in self.table.columns if col not in self._BASE_SUMMARY_COLUMNS]
@property
def pymc_time_unit(self) -> str:
"""The pymc-marketing ``time_unit`` matching this summary's ``period`` (``"D"``/``"W"``/``"M"``).
Pass as the ``time_unit`` of ``GammaGammaModel.expected_customer_lifetime_value``, whose ``future_t``
is in months and defaults to ``"D"``, silently wrong for a weekly summary (horizon off ~7x) or a
monthly one (~30x).
Returns:
str: ``"D"`` for ``period="day"``, ``"W"`` for ``"week"``, ``"M"`` for ``"month"``.
"""
return self._PYMC_TIME_UNIT[self.period]