CLV (BTYD Model Prep)

Customer-lifetime-value (BTYD) model input preparation.

This module adapts OpenRetailScience transaction data into the per-customer summary frame consumed by the "buy-till-you-die" models in pymc-marketing <https://www.pymc-marketing.io/>_ — ParetoNBDModel (churn and purchase frequency) and GammaGammaModel (per-transaction spend).

The summary follows the standard BTYD convention, where each customer is "born" at their first purchase and observed until observation_period_end:

• frequency — number of repeat purchase occasions (total occasions minus one). An occasion is a distinct calendar day with positive net spend (returns-only days excluded).
• recency — time from the first purchase to the last purchase.
• T — the customer's "age": time from the first purchase to observation_period_end.
• monetary_value — mean spend across the repeat purchases (excludes the first, per the Gamma-Gamma assumption). NaN for one-time buyers (excluded by repeat_buyers).

recency and T are whole calendar days converted to fractional period units ("day", "week", or "month"; a month is a fixed 365.25/12 days), so 10 days is 1.43 weeks.

The first purchase is always the earliest transaction in the supplied data, exactly as pymc_marketing.clv.rfm_summary treats it. A customer who was already active before the data window is therefore treated as born at the window's start — a known limitation of a truncated observation period, not something this adapter corrects.

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); :meth:sample narrows the summary to a subset of customers to fit on. See the module docstring for column definitions.

Construction runs a few small validation queries; the per-customer summary stays lazy until df is first read, then is cached. A :meth:sample result runs no queries until its own df is read.

Parameters:

Name	Type	Description	Default
data	DataFrame | 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.	required
period	str	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".	'week'
observation_period_end	str | date | None	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.	None
customer_attributes	DataFrame | Table | None	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.	None
one_hot_col	str | list[str] | None	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.	None

Raises:

Type	Description
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.

Source code in openretailscience/experimental/clv.py

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); :meth:`sample` narrows the summary to a subset of customers to
    fit on. See the module docstring for column definitions.

    Construction runs a few small validation queries; the per-customer summary stays lazy until
    ``df`` is first read, then is cached. A :meth:`sample` result runs no queries until its own
    ``df`` is read.

    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 per-customer sampling hash is folded into this many buckets, setting ``frac``'s resolution.
    _SAMPLE_BUCKETS: ClassVar[int] = 1_000_000
    #: 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 | dt.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: dt.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",
        )

    def sample(self, *, n: int | None = None, frac: float | None = None, random_state: int = 42) -> Self:
        """Draws a random sample of customers, as a new :class:`CLVStats` over the same summary.

        Fit on ``clv.sample(n=50_000)``, score on ``clv.df``. Customers are selected by a deterministic
        hash of ``customer_id`` salted with ``random_state``.

        - A draw repeats only on the same engine at the same version: the key is the backend's own hash,
          and DuckDB's ``HASH`` changes between releases. Record the drawn ids, not the seed, to
          reproduce a fit. Backends with no Ibis ``hash`` rule (SQLite, MySQL, Trino, Athena) raise
          ``OperationNotDefinedError`` when :attr:`df` is read.
        - Draws nest rather than partition: at one ``random_state``, ``sample(frac=0.2)`` is a subset of
          ``sample(frac=0.8)`` and re-sampling a sample returns it unchanged. A different
          ``random_state`` re-draws independently but still overlaps, so neither gives a disjoint
          train/holdout split.

        Args:
            n (int | None, optional): Number of customers to draw; every customer if it exceeds the
                population. Mutually exclusive with ``frac``. Defaults to ``None``.
            frac (float | None, optional): Share of customers to draw, in [1e-6, 1]. Decided per
                customer, so the count lands near ``frac`` * population rather than on it; costs a
                single filter instead of ``n``'s sort. Mutually exclusive with ``n``. Defaults to ``None``.
            random_state (int, optional): Salt for the selection hash. Defaults to 42.

        Returns:
            Self: A ``CLVStats`` over the sampled customers, with the same ``period`` and covariates.

        Raises:
            TypeError: If ``n`` or ``random_state`` is not an integer, or ``frac`` is not a number.
            ValueError: If both or neither of ``n`` and ``frac`` are given, if ``n`` is not positive,
                or if ``frac`` is outside [1e-6, 1].
        """
        if (n is None) == (frac is None):
            msg = "sample requires exactly one of n or frac."
            raise ValueError(msg)
        ensure_integer(random_state, "random_state")
        # The id column is the literal "customer_id" here (__init__ renames it); cast so the salt
        # concatenates onto any id type.
        key = (self.table["customer_id"].cast("string") + ibis.literal(f"::{random_state}")).hash()
        # abs() after the modulo, not before: abs(-2**63) overflows int64.
        # Select on the bucket, never the raw hash: SQL Server's CHECKSUM keeps the id's leading
        # characters in its high bits, so ordering on the raw key returns a run of sequential ids. The
        # low bits the modulo keeps are well mixed.
        bucket = (key % self._SAMPLE_BUCKETS).abs()
        if n is not None:
            ensure_integer(n, "n")
            ensure_positive(n, "n")
            # The n smallest buckets: the frac predicate with the cut solved for n. customer_id breaks
            # ties in the boundary bucket; without it ORDER BY ... LIMIT can differ per execution.
            sampled = self.table.order_by([bucket, self.table["customer_id"]]).limit(n)
        else:
            ensure_positive(frac, "frac")
            ensure_unit_interval(frac, "frac")
            cut = int(frac * self._SAMPLE_BUCKETS)
            if cut == 0:
                msg = (
                    f"frac must be at least {1 / self._SAMPLE_BUCKETS} (the sampling hash resolution); "
                    f"{frac} rounds to zero buckets, which would draw no customers. Use n for a smaller share."
                )
                raise ValueError(msg)
            sampled = self.table.filter(bucket < cut)
        # Bypass __init__, which would re-run the observation-window and attribute queries against a
        # table that no longer holds transactions. (period, table) is the whole state.
        sample = object.__new__(type(self))
        sample.period = self.period
        sample.table = sampled
        return sample

    @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, sorted by ``customer_id``, 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.
        """
        # Sorted here rather than on ``table``: a grouped or filtered expression has no row order, so two
        # reads of one summary could otherwise interleave differently.
        result = self.table.order_by("customer_id").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 0.15, 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``: its ``future_t``
        is in months, but ``time_unit`` (the unit of ``recency`` and ``T``) defaults to ``"D"``, so omitting
        it for a weekly or monthly summary silently misreads the horizon by ~7x or ~30x.

        Returns:
            str: ``"D"`` for ``period="day"``, ``"W"`` for ``"week"``, ``"M"`` for ``"month"``.
        """
        return self._PYMC_TIME_UNIT[self.period]

covariate_cols property

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:

Type	Description
list[str]	list[str]: The covariate column names in summary order (empty if none were requested).

df cached property

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:

Type	Description
DataFrame	pd.DataFrame: One row per customer, sorted by customer_id, with columns customer_id, frequency, recency, T, and monetary_value, followed by any customer_attributes and one-hot columns requested at construction.

Raises:

Type	Description
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.

pymc_time_unit property

The pymc-marketing time_unit matching this summary's period ("D"/"W"/"M").

Pass as the time_unit of GammaGammaModel.expected_customer_lifetime_value: its future_t is in months, but time_unit (the unit of recency and T) defaults to "D", so omitting it for a weekly or monthly summary silently misreads the horizon by ~7x or ~30x.

Returns:

Name	Type	Description
str	str	"D" for period="day", "W" for "week", "M" for "month".

repeat_buyers property

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 0.15, breaking GammaGamma's independence assumption.

Returns:

Type	Description
DataFrame	pd.DataFrame: The frequency > 0 rows of :attr:df, index reset.

__init__(data, *, period='week', observation_period_end=None, customer_attributes=None, one_hot_col=None)

Initializes and computes the BTYD summary.

Source code in openretailscience/experimental/clv.py

def __init__(
    self,
    data: pd.DataFrame | ibis.Table,
    *,
    period: str = "week",
    observation_period_end: str | dt.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

sample(*, n=None, frac=None, random_state=42)

Draws a random sample of customers, as a new :class:CLVStats over the same summary.

Fit on clv.sample(n=50_000), score on clv.df. Customers are selected by a deterministic hash of customer_id salted with random_state.

• A draw repeats only on the same engine at the same version: the key is the backend's own hash, and DuckDB's HASH changes between releases. Record the drawn ids, not the seed, to reproduce a fit. Backends with no Ibis hash rule (SQLite, MySQL, Trino, Athena) raise OperationNotDefinedError when :attr:df is read.
• Draws nest rather than partition: at one random_state, sample(frac=0.2) is a subset of sample(frac=0.8) and re-sampling a sample returns it unchanged. A different random_state re-draws independently but still overlaps, so neither gives a disjoint train/holdout split.

Parameters:

Name	Type	Description	Default
n	int | None	Number of customers to draw; every customer if it exceeds the population. Mutually exclusive with frac. Defaults to None.	None
frac	float | None	Share of customers to draw, in [1e-6, 1]. Decided per customer, so the count lands near frac * population rather than on it; costs a single filter instead of n's sort. Mutually exclusive with n. Defaults to None.	None
random_state	int	Salt for the selection hash. Defaults to 42.	42

Returns:

Name	Type	Description
Self	Self	A CLVStats over the sampled customers, with the same period and covariates.

Raises:

Type	Description
TypeError	If n or random_state is not an integer, or frac is not a number.
ValueError	If both or neither of n and frac are given, if n is not positive, or if frac is outside [1e-6, 1].

Source code in openretailscience/experimental/clv.py

def sample(self, *, n: int | None = None, frac: float | None = None, random_state: int = 42) -> Self:
    """Draws a random sample of customers, as a new :class:`CLVStats` over the same summary.

    Fit on ``clv.sample(n=50_000)``, score on ``clv.df``. Customers are selected by a deterministic
    hash of ``customer_id`` salted with ``random_state``.

    - A draw repeats only on the same engine at the same version: the key is the backend's own hash,
      and DuckDB's ``HASH`` changes between releases. Record the drawn ids, not the seed, to
      reproduce a fit. Backends with no Ibis ``hash`` rule (SQLite, MySQL, Trino, Athena) raise
      ``OperationNotDefinedError`` when :attr:`df` is read.
    - Draws nest rather than partition: at one ``random_state``, ``sample(frac=0.2)`` is a subset of
      ``sample(frac=0.8)`` and re-sampling a sample returns it unchanged. A different
      ``random_state`` re-draws independently but still overlaps, so neither gives a disjoint
      train/holdout split.

    Args:
        n (int | None, optional): Number of customers to draw; every customer if it exceeds the
            population. Mutually exclusive with ``frac``. Defaults to ``None``.
        frac (float | None, optional): Share of customers to draw, in [1e-6, 1]. Decided per
            customer, so the count lands near ``frac`` * population rather than on it; costs a
            single filter instead of ``n``'s sort. Mutually exclusive with ``n``. Defaults to ``None``.
        random_state (int, optional): Salt for the selection hash. Defaults to 42.

    Returns:
        Self: A ``CLVStats`` over the sampled customers, with the same ``period`` and covariates.

    Raises:
        TypeError: If ``n`` or ``random_state`` is not an integer, or ``frac`` is not a number.
        ValueError: If both or neither of ``n`` and ``frac`` are given, if ``n`` is not positive,
            or if ``frac`` is outside [1e-6, 1].
    """
    if (n is None) == (frac is None):
        msg = "sample requires exactly one of n or frac."
        raise ValueError(msg)
    ensure_integer(random_state, "random_state")
    # The id column is the literal "customer_id" here (__init__ renames it); cast so the salt
    # concatenates onto any id type.
    key = (self.table["customer_id"].cast("string") + ibis.literal(f"::{random_state}")).hash()
    # abs() after the modulo, not before: abs(-2**63) overflows int64.
    # Select on the bucket, never the raw hash: SQL Server's CHECKSUM keeps the id's leading
    # characters in its high bits, so ordering on the raw key returns a run of sequential ids. The
    # low bits the modulo keeps are well mixed.
    bucket = (key % self._SAMPLE_BUCKETS).abs()
    if n is not None:
        ensure_integer(n, "n")
        ensure_positive(n, "n")
        # The n smallest buckets: the frac predicate with the cut solved for n. customer_id breaks
        # ties in the boundary bucket; without it ORDER BY ... LIMIT can differ per execution.
        sampled = self.table.order_by([bucket, self.table["customer_id"]]).limit(n)
    else:
        ensure_positive(frac, "frac")
        ensure_unit_interval(frac, "frac")
        cut = int(frac * self._SAMPLE_BUCKETS)
        if cut == 0:
            msg = (
                f"frac must be at least {1 / self._SAMPLE_BUCKETS} (the sampling hash resolution); "
                f"{frac} rounds to zero buckets, which would draw no customers. Use n for a smaller share."
            )
            raise ValueError(msg)
        sampled = self.table.filter(bucket < cut)
    # Bypass __init__, which would re-run the observation-window and attribute queries against a
    # table that no longer holds transactions. (period, table) is the whole state.
    sample = object.__new__(type(self))
    sample.period = self.period
    sample.table = sampled
    return sample