Abacus Driver: Budget Optimisation (opt.py)

This module provides high-level functions to perform marketing budget optimisation using a fitted Abacus Marketing Mix Model (MMM). It leverages the model’s understanding of channel response curves to suggest an optimal allocation of a total budget across channels to maximise expected contribution.

Functions

optimize_marketing_budget

def optimize_marketing_budget(
    model,
    data: pd.DataFrame,
    config: dict,
    results_dir: str,
    total_budget: Optional[float] = None,
    frequency: str = 'M',
    n_time_periods: int = 1
) -> None:

Orchestrates the marketing budget optimisation process.

This function performs the following steps:

  1. Aggregates historical spending data (data) based on the specified frequency.

  2. Calculates the total_budget based on historical means if not explicitly provided.

  3. Determines dynamic lower and upper bounds for each channel’s budget based on historical spend patterns (e.g., +/- 20% of the mean spend over n_time_periods).

  4. Estimates the parameters of the channel saturation curves (currently using the ‘sigmoid’ method via model.compute_channel_curve_optimization_parameters_original_scale).

  5. Calls the model’s optimisation method (model.optimize_channel_budget_for_maximum_contribution) to find the budget allocation that maximises total expected contribution, subject to the total_budget and calculated budget_bounds.

  6. Calculates the expected contribution from the initial (historical mean) budget allocation for comparison.

  7. Generates and saves a plot comparing the initial and optimal budget allocations and their corresponding contributions using model.plot_budget_scenarios.

  8. Saves the numerical results (initial vs. optimal budget and contribution per channel, plus totals and estimated lift) to a CSV file (optimization_results.csv) in the results_dir.

Parameters:

  • model: An instance of a fitted Abacus MMM model (e.g., abacus.core.mmm_model.DelayedSaturatedMMM) possessing the required optimisation and plotting methods.

  • data (pd.DataFrame): DataFrame containing historical marketing data, including a date column (specified implicitly or via config) and spend columns for each channel (specified in config['media']).

  • config (dict): Configuration dictionary, expected to contain a 'media' key which is a list of dictionaries, each specifying a 'spend_col' for a channel.

  • results_dir (str): Path to the directory where the output plot (budget_optimisation.png) and CSV file (optimization_results.csv) will be saved.

  • total_budget (Optional[float], optional): The total budget to allocate across channels. If None (default), the total budget is calculated based on the mean historical spend in data aggregated by frequency over n_time_periods.

  • frequency (str, optional): The frequency for aggregating historical spend data (‘W’ for weekly, ‘M’ for monthly, ‘Q’ for quarterly). Defaults to 'M'.

  • n_time_periods (int, optional): The number of future time periods the optimisation pertains to (used for scaling the total budget and bounds). Defaults to 1.

Returns:

  • None: The function saves output files to disk and prints confirmation messages.

Notes:

  • This function relies heavily on methods assumed to be present in the passed model object, specifically those related to estimating saturation parameters (compute_channel_curve_optimization_parameters_original_scale), performing the budget optimisation (optimize_channel_budget_for_maximum_contribution), and plotting the results (plot_budget_scenarios). These methods are typically inherited via mixins like OptimizationMixin and ScenarioPlottingMixin.

  • The budget bounds are currently hardcoded to +/- 20% of the historical mean spend.

  • The saturation curve estimation currently defaults to the ‘sigmoid’ method.

get_lower_and_upper_bounds

def get_lower_and_upper_bounds(
    media: pd.DataFrame,
    n_time_periods: int,
    lower_pct: np.ndarray,
    upper_pct: np.ndarray
) -> tuple[np.ndarray, np.ndarray]:

Calculates lower and upper budget bounds for each channel based on historical spend and percentage deviations.

Parameters:

  • media (pd.DataFrame): DataFrame of historical media spend (channels as columns, time as index).

  • n_time_periods (int): Number of future periods to scale the bounds for.

  • lower_pct (np.ndarray): Array of percentages (0 to 1) defining the maximum decrease from the mean spend for the lower bound (per channel).

  • upper_pct (np.ndarray): Array of percentages (0 to 1) defining the maximum increase from the mean spend for the upper bound (per channel).

Returns:

  • tuple[np.ndarray, np.ndarray]: A tuple containing two NumPy arrays: (lower_bounds, upper_bounds) for each channel over the specified n_time_periods.

calculate_total_budget

def calculate_total_budget(
    mean_costs: Union[np.ndarray, pd.Series],
    n_time_periods: int
) -> float:

Calculates the total budget based on mean channel costs and the number of time periods.

Parameters:

  • mean_costs (np.ndarray | pd.Series): Array or Series containing the mean cost per time period for each channel.

  • n_time_periods (int): The number of time periods to calculate the total budget for.

Returns:

  • float: The calculated total budget.