src.core.transformers¶

Provides various adstock and saturation transformation functions for MMM.

Module Contents¶

class src.core.transformers.SaturationTransformation¶

Bases: Protocol

Protocol defining the interface for saturation transformations.

property function: Callable[Ellipsis, pytensor.tensor.variable.TensorVariable]¶: The mathematical function implementing the saturation.

property variable_mapping: Dict[str, str]¶: Mapping from saturation function parameter names to model variable names.

class src.core.transformers.ConvMode¶

Bases: str, enum.Enum

Convolution mode enumeration.

Defines how convolution is applied at signal boundaries.

class src.core.transformers.WeibullType¶

Bases: str, enum.Enum

Weibull transformation type enumeration.

Specifies whether to use PDF or CDF for Weibull adstock.

src.core.transformers.batched_convolution(x: pytensor.tensor.TensorLike, w: pytensor.tensor.TensorLike, axis: int = 0, mode: ConvMode | str = ConvMode.After) → pytensor.tensor.variable.TensorVariable¶

Apply a 1D convolution in a vectorized way across multiple batch dimensions.

Parameters:

x – The array to convolve.
w – The weight of the convolution. The last axis of w determines the number of steps to use in the convolution.
axis – The axis of x along which to apply the convolution. Defaults to 0.
mode –
The convolution mode. Determines how the convolution is applied at the boundaries of the input signal. Defaults to ConvMode.After.
- ConvMode.After: Applies the convolution with the “Adstock” effect, resulting in a trailing decay effect.
- ConvMode.Before: Applies the convolution with the “Excitement” effect, creating a leading effect.
- ConvMode.Overlap: Applies the convolution with both “Pull-Forward” and “Pull-Backward” effects.

Returns:

The result of convolving x with w along the desired axis. The shape of the result will match the shape of x up to broadcasting with w.

src.core.transformers.geometric_adstock(x: pytensor.tensor.TensorLike, alpha: pytensor.tensor.TensorLike | float = 0.0, l_max: int = 12, normalize: bool = False, axis: int = 0) → pytensor.tensor.variable.TensorVariable¶

Geometric adstock transformation.

Adstock with geometric decay assumes advertising effect peaks at the same time period as ad exposure. The cumulative media effect is a weighted average of media spend in the current time-period (e.g. week) and previous l_max - 1 periods (e.g. weeks). l_max is the maximum duration of carryover effect.

Parameters:

x – Input tensor.
alpha – Retention rate of ad effect. Must be between 0 and 1. Defaults to 0.0.
l_max – Maximum duration of carryover effect. Defaults to 12.
normalize – Whether to normalize the weights. Defaults to False.
axis – The axis of x along which to apply the adstock. Defaults to 0.

Returns:

Transformed tensor.

src.core.transformers.delayed_adstock(x: pytensor.tensor.TensorLike, alpha: pytensor.tensor.TensorLike | float = 0.0, theta: pytensor.tensor.TensorLike | int = 0, l_max: int = 12, normalize: bool = False, axis: int = 0) → pytensor.tensor.variable.TensorVariable¶

Delayed adstock transformation.

This transformation is similar to geometric adstock transformation, but it allows for a delayed peak of the effect. The peak is assumed to occur at theta.

Parameters:

x – Input tensor.
alpha – Retention rate of ad effect. Must be between 0 and 1. Defaults to 0.0.
theta – Delay of the peak effect. Must be between 0 and l_max - 1. Defaults to 0.
l_max – Maximum duration of carryover effect. Defaults to 12.
normalize – Whether to normalize the weights. Defaults to False.
axis – The axis of x along which to apply the adstock. Defaults to 0.

Returns:

Transformed tensor.

src.core.transformers.weibull_adstock(x: pytensor.tensor.TensorLike, lam: pytensor.tensor.TensorLike | float = 1.0, k: pytensor.tensor.TensorLike | float = 1.0, l_max: int = 12, axis: int = 0, type: WeibullType | str = WeibullType.PDF) → pytensor.tensor.variable.TensorVariable¶

Weibull Adstocking Transformation.

This transformation is similar to geometric adstock transformation but has more degrees of freedom, adding more flexibility.

Parameters:

x – Input tensor.
lam – Scale parameter of the Weibull distribution. Must be positive. Defaults to 1.0.
k – Shape parameter of the Weibull distribution. Must be positive. Defaults to 1.0.
l_max – Maximum duration of carryover effect. Defaults to 12.
axis – The axis of x along which to apply the adstock. Defaults to 0.
type – Type of Weibull adstock transformation to be applied (WeibullType.PDF or WeibullType.CDF). Defaults to WeibullType.PDF.

Returns:

Transformed tensor based on Weibull adstock transformation.

src.core.transformers.logistic_saturation(x: pytensor.tensor.TensorLike, lam: pytensor.tensor.TensorLike | float = 0.5) → pytensor.tensor.variable.TensorVariable¶

Logistic saturation transformation.

\[f(x) = \frac{1 - e^{-\lambda x}}{1 + e^{-\lambda x}}\]

Parameters:

x – Input tensor.
lam – Saturation parameter. Defaults to 0.5.

Returns:

Transformed tensor.

class src.core.transformers.LogisticSaturation¶

Wrapper class for logistic_saturation to conform to SaturationTransformation protocol.

property function: Callable[[Any, Any], pytensor.tensor.variable.TensorVariable]¶: The mathematical function implementing logistic saturation.

property variable_mapping: Dict[str, str]¶: Mapping from saturation parameter to model variable.

class src.core.transformers.TanhSaturationParameters¶

Bases: NamedTuple

Parameters for tanh saturation transformation.

b¶: Saturation level (number of users at saturation).

c¶: Customer Acquisition Cost at zero spend.

baseline(x0: pytensor.tensor.TensorLike) → TanhSaturationBaselinedParameters¶

Change the parameterization to baselined at x_0.

Parameters:: x0 – Baseline spend point.
Returns:: Baselined parameters at x_0.

class src.core.transformers.TanhSaturationBaselinedParameters¶

Bases: NamedTuple

Baselined parameters for tanh saturation transformation.

x0¶: Baseline spend.

gain¶: ROAS at x_0.

r¶: Overspend fraction.

debaseline() → TanhSaturationParameters¶

Change the parameterization to classic saturation and cac.

Returns:: Classic tanh saturation parameters.

rebaseline(x1: pytensor.tensor.TensorLike) → TanhSaturationBaselinedParameters¶

Change the parameterization to baselined at x_1.

Parameters:: x1 – New baseline spend point.
Returns:: Baselined parameters at x_1.

src.core.transformers.tanh_saturation(x: pytensor.tensor.TensorLike, b: pytensor.tensor.TensorLike = 0.5, c: pytensor.tensor.TensorLike = 0.5) → pytensor.tensor.variable.TensorVariable¶

Tanh saturation transformation.

\[f(x) = b \tanh \left( \frac{x}{bc} \right)\]

Parameters:

x – Input tensor.
b – Number of users at saturation. Must be non-negative. Defaults to 0.5.
c – Initial cost per user. Must be non-zero. Defaults to 0.5.

Returns:

Transformed tensor.

Notes

Uses a hyperbolic tangent saturation form that captures diminishing returns as spend increases. Suitable for modelling concave response curves in media mix contexts.

src.core.transformers.tanh_saturation_baselined(x: pytensor.tensor.TensorLike, x0: pytensor.tensor.TensorLike, gain: pytensor.tensor.TensorLike = 0.5, r: pytensor.tensor.TensorLike = 0.5) → pytensor.tensor.variable.TensorVariable¶

Baselined Tanh Saturation.

This parameterization that is easier than tanh_saturation() to use for industry applications where domain knowledge is an essence.

In a nutshell, it is an alternative parameterization of the reach function is given by:

\[\begin{split}\begin{align} c_0 &= \frac{r}{g \cdot \arctan(r)} \\ \beta &= \frac{g \cdot x_0}{r} \\ \operatorname{saturation}(x, \beta, c_0) &= \beta \cdot \tanh \left( \frac{x}{c_0 \cdot \beta} \right) \end{align}\end{split}\]

where:

\(x_0\) is the “reference point”. This is a point chosen by the user (not given a prior) where they expect most of their data to lie. For example, if you’re spending between 50 and 150 dollars on a particular channel, you might choose \(x_0 = 100\). Suggested value is median channel spend: np.median(spend).
\(g\) is the “gain”, which is the value of the CAC (\(c_0\)) at the reference point. You have to set a prior on what you think the CAC is when you spend \(x_0 = 100\). Imagine you have four advertising channels, and you acquired 1000 new users. If each channel performed equally well, and advertising drove all sales, you might expect that you gained 250 users from each channel. Here, your “gain” would be \(250 / 100 = 2.5\). Suggested prior is pm.Exponential
\(r\), the overspend fraction is telling you where the reference point is.
- \(0\) - we can increase our budget by a lot to reach the saturated region, the diminishing returns are not visible yet.
- \(1\) - the reference point is already in the saturation region and additional dollar spend will not lead to any new users.
- \(0.8\), you can still increase acquired users by \(50\%\) as much you get in the reference point by increasing the budget. \(x_0\) effect is 20% away from saturation point
Suggested prior is pm.Beta

Note

The reference point \(x_0\) has to be set within the range of the actual spends. As in, you buy ads three times and spend \(5\), \(6\) and \(7\) dollars, \(x_0\) has to be set within \([5, 7]\), so not \(4\) not \(8\). Otherwise the posterior of r and gain becomes a skinny diagonal line. It could be very relevant if there is very little spend observations for a particular channel.

The original reach or saturation function used in an MMM is formulated as

\[\operatorname{saturation}(x, \beta, c_0) = \beta \cdot \tanh \left( \frac{x}{c_0 \cdot \beta} \right)\]

where:

\(\beta\) is the saturation, or the limit of the total number of new users obtained when an infinite number of dollars are spent on that channel.
\(c_0\) is the cost per acquisition (CAC0), so the initial cost per new user.
\(\frac{1}{c_0}\) is the inverse of the CAC0, so it’s the number of new users we might expect after spending our first dollar.

Examples

import pymc as pm
import numpy as np

x_in = np.exp(3+np.random.randn(100))
true_cac = 1
true_saturation = 100
y_out = abs(np.random.normal(tanh_saturation(x_in, true_saturation, true_cac).eval(), 0.1))

with pm.Model() as model_reparam:
    r = pm.Uniform("r")
    gain = pm.Exponential("gain", 1)
    input = pm.ConstantData("spent", x_in)
    response = pm.ConstantData("response", y_out)
    sigma = pm.HalfNormal("n")
    output = tanh_saturation_baselined(input, np.median(x_in), gain, r)
    pm.Normal("output", output, sigma, observed=response)
    trace = pm.sample()

Parameters:

x – Input tensor.
x0 – Baseline for saturation.
gain – ROAS at the baseline point, mathematically as \(gain = f(x0) / x0\). Defaults to 0.5.
r – The overspend fraction, mathematically as \(r = f(x0) / \text{saturation}\). Defaults to 0.5.

Returns:

Transformed tensor.