src.core.lift_test

Lift test integration functionality for ammm.

Module Contents

exception src.core.lift_test.UnalignedValuesError(unaligned_values: Dict[str, List[int]])

Bases: Exception

Raised when some values are not aligned.

src.core.lift_test.exact_row_indices(df: pandas.DataFrame, model: pymc.Model) → Indices

Get indices in the model for each row in the DataFrame.

Assumes any column in the DataFrame is a coordinate in the model with the same name.

Parameters:

  • df (pd.DataFrame) – DataFrame with coordinates combinations.

  • model (pm.Model) – PyMC model with all the coordinates in the DataFrame.

Returns:

Dictionary of indices for the lift test results in the model.

Return type:

dict[str, np.ndarray]

Raises:

UnalignedValuesError, KeyError –

src.core.lift_test.create_variable_indexer(model: pymc.Model, indices: Indices) → VariableIndexer

Create a function to index variables in the model.

exception src.core.lift_test.MissingValueError(missing_values: List[str], required_values: List[str])

Bases: KeyError

Error when values are missing from a required set.

src.core.lift_test.assert_is_subset(required: set[str], available: set[str]) → None

Check if the available set is a subset of the required set.

exception src.core.lift_test.NonMonotonicError

Bases: ValueError

Raised when lift test data does not satisfy the monotonic assumption (delta_x * delta_y >= 0).

src.core.lift_test.assert_monotonic(delta_x: pandas.Series, delta_y: pandas.Series) → None

Check if the lift test results satisfy the increasing assumption.

src.core.lift_test.add_saturation_observations(df_lift_test: pandas.DataFrame, variable_mapping: VariableMapping, saturation_function: SaturationFunc, model: pymc.Model | None = None, dist: Type[pymc.Distribution] = pm.Gamma, name: str = 'lift_measurements', get_indices: collections.abc.Callable[[pandas.DataFrame, pymc.Model], Indices] = exact_row_indices, variable_indexer_factory: collections.abc.Callable[[pymc.Model, Indices], VariableIndexer] = create_variable_indexer) → None

Adds lift test observations to the likelihood of a PyMC model.

This function constrains the model by adding observed lift (change in outcome due to change in spend) based on a provided saturation function. It assumes that the difference in outcome predicted by the saturation curve for a change in spend (delta_x) should match the observed lift (delta_y), within some uncertainty (sigma).

Parameters:

  • df_lift_test – DataFrame containing lift test data. Must include columns: ‘x’ (spend before), ‘delta_x’ (change in spend), ‘delta_y’ (observed change in outcome), ‘sigma’ (uncertainty of delta_y), and any columns required for indexing model coordinates (dims).

  • variable_mapping – A dictionary mapping parameter names in the saturation_function to variable names in the PyMC model.

  • saturation_function – A callable that takes an input spend x and model parameters (obtained via variable_mapping) and returns the saturated response.

  • model – The PyMC model to which observations will be added. If None, uses the model from the current PyMC context. Defaults to None.

  • dist – The PyMC distribution to model the observed lift. Defaults to pm.Gamma.

  • name – Name for the likelihood term in the PyMC model. Defaults to “lift_measurements”.

  • get_indices – Function to obtain model coordinate indices from the lift test DataFrame. Defaults to exact_row_indices.

  • variable_indexer_factory – Factory function to create a variable indexer. Defaults to create_variable_indexer.

Returns:

None. Modifies the PyMC model in place by adding a likelihood term.

Raises:

  • MissingValueError – If df_lift_test is missing required columns or coordinate columns.

  • KeyError – If a variable in variable_mapping is not found in the model, or if a dimension required by a mapped variable is not in df_lift_test.

  • UnalignedValuesError – If coordinate values in df_lift_test do not align with model coordinates.

  • NonMonotonicError – If the lift test data (delta_x, delta_y) is not monotonic.

src.core.lift_test.create_time_varying_saturation(saturation: src.core.transformers.SaturationTransformation, time_varying_var_name: str) → tuple[SaturationFunc, VariableMapping]

Return function and variable mapping that use a time-varying variable.

src.core.lift_test.add_lift_measurements_to_likelihood_from_saturation(df_lift_test: pandas.DataFrame, saturation: src.core.transformers.SaturationTransformation, time_varying_var_name: str | None = None, model: pymc.Model | None = None, dist: Type[pymc.Distribution] = pm.Gamma, name: str = 'lift_measurements', get_indices: collections.abc.Callable[[pandas.DataFrame, pymc.Model], Indices] = exact_row_indices, variable_indexer_factory: collections.abc.Callable[[pymc.Model, Indices], VariableIndexer] = create_variable_indexer) → None

Add lift measurements to the likelihood from a saturation transformation.

src.core.lift_test.scale_channel_lift_measurements(df_lift_test: pandas.DataFrame, channel_col: str, channel_columns: List[str], transform: collections.abc.Callable[[numpy.ndarray], numpy.ndarray]) → pandas.DataFrame

Scale the lift measurements for a specific channel.

src.core.lift_test.scale_target_for_lift_measurements(target: pandas.Series, transform: collections.abc.Callable[[numpy.ndarray], numpy.ndarray]) → pandas.Series

Scale the target for the lift measurements.

src.core.lift_test.scale_lift_measurements(df_lift_test: pandas.DataFrame, channel_col: str, channel_columns: List[str | int], channel_transform: collections.abc.Callable[[numpy.ndarray], numpy.ndarray], target_transform: collections.abc.Callable[[numpy.ndarray], numpy.ndarray]) → pandas.DataFrame

Scale the DataFrame with lift test results to be used in the model.