src.core.opt

Provides functions for optimizing marketing budget allocation across channels.

This module includes utilities to calculate expected contributions based on different response curves (Michaelis-Menten, sigmoid) and to find the optimal budget distribution that maximizes total contribution given constraints.

Module Contents

src.core.opt.parse_and_validate_ramp_constraints(channels: List[str], ramp_constraints: float | Dict | None, strict: bool = False) → Tuple[Dict[str, float], Dict[str, float], List[str]]

Parse and validate ramp constraints, returning normalized dictionaries.

This function validates ramp constraint values, normalizes percentages, detects unknown channels, and checks for contradictory constraints.

Parameters:

  • channels – List of valid channel names.

  • ramp_constraints – Ramp constraint specification (see optimize_multiperiod_budget_distribution).

  • strict – If True, raise error on unknown channels. If False, warn and skip.

Returns:

  • ramp_abs_dict: Absolute ramp limits per channel {channel: limit}

  • ramp_pct_dict: Percentage ramp limits per channel {channel: limit}

  • warnings_list: List of warning messages for user display

Return type:

Tuple of (ramp_abs_dict, ramp_pct_dict, warnings_list)

Raises:

ValueError – If constraints are invalid (negative values, bad percentages, unknown channels in strict mode).

Example

>>> ramp_abs, ramp_pct, warns = parse_and_validate_ramp_constraints(
...     channels=['google', 'facebook'],
...     ramp_constraints={'absolute': {'google': 15000}, 'percentage': 25}
... )
>>> ramp_pct['facebook']  # Normalized from 25
0.25
src.core.opt.calculate_expected_contribution(method: str, parameters: Dict[str, Tuple[float, float]], budget: Dict[str, float]) → Dict[str, float]

Calculate expected contributions using the specified model.

This function calculates the expected contributions for each channel based on the chosen model. The selected model can be either the Michaelis-Menten model or the sigmoid model, each described by specific parameters. As the allocated budget varies, the expected contribution is computed according to the chosen model.

Parameters:

  • method – The model to use for contribution estimation. Choose from ‘michaelis-menten’ or ‘sigmoid’.

  • parameters –

    Model-specific parameters for each channel. For ‘michaelis-menten’, each entry is a tuple (L, k) where: - L is the maximum potential contribution. - k is the budget at which the contribution is half of its maximum.

    For ‘sigmoid’, each entry is a tuple (alpha, lam) where: - alpha controls the slope of the curve. - lam is the budget at which the curve transitions.

  • budget – A dictionary where keys are channel names and values are the allocated budgets for those channels.

Returns:

A dictionary with channels as keys and their respective contributions as values. The key ‘total’ contains the total expected contribution across all channels.

Raises:

ValueError – If the specified method is not recognized.

src.core.opt.objective_distribution(x: List[float], method: str, channels: List[str], parameters: Dict[str, Tuple[float, float]]) → float

Compute the total contribution for a given budget distribution.

This function calculates the negative sum of contributions for a proposed budget distribution using the Michaelis-Menten model. This value will be minimized in the optimization process to maximize the total expected contribution.

Parameters:

  • x – The proposed budget distribution across channels.

  • method – The response curve method (‘michaelis-menten’ or ‘sigmoid’).

  • channels – The list of channels for which the budget is being optimized.

  • parameters – Model-specific parameters for each channel, as described in calculate_expected_contribution.

Returns:

Negative of the total expected contribution for the given budget distribution.

src.core.opt.optimize_budget_distribution(method: str, total_budget: int | float, budget_ranges: Dict[str, Tuple[float, float]] | None, parameters: Dict[str, Tuple[float, float]], channels: List[str]) → Dict[str, float]

Optimize the budget allocation across channels to maximize total contribution.

Using the Michaelis-Menten or Sigmoid function, this function seeks the best budget distribution across channels that maximizes the total expected contribution.

This function leverages the Sequential Least Squares Quadratic Programming (SLSQP) optimization algorithm to find the best budget distribution across channels that maximizes the total expected contribution based on the Michaelis-Menten or Sigmoid functions.

The optimization is constrained such that: 1. The sum of budgets across all channels equals the total available budget. 2. The budget allocated to each individual channel lies within its specified range.

The SLSQP method is particularly suited for this kind of problem as it can handle both equality and inequality constraints. But beware that when tested on real data SLSQP optimizer couldn’t handle 100+ variables due to numerical gradient instability.

Parameters:

  • method – The response curve method (‘michaelis-menten’ or ‘sigmoid’).

  • total_budget – The total budget to be distributed across channels.

  • budget_ranges – An optional dictionary defining the minimum and maximum budget for each channel. If None, ranges are inferred (0 to min(total_budget, L_value) for Michaelis-Menten).

  • parameters – Model-specific parameters for each channel, as described in calculate_expected_contribution.

  • channels – The list of channels for which the budget is being optimized.

Returns:

A dictionary with channels as keys and the optimal budget for each channel as values.

src.core.opt.budget_allocator(method: str, total_budget: int | float, channels: List[str], parameters: Dict[str, Tuple[float, float]], budget_ranges: Dict[str, Tuple[float, float]] | None) → pandas.DataFrame

Allocate budget based on the specified method and constraints.

Parameters:

  • method – The method used for budget allocation (‘michaelis-menten’ or ‘sigmoid’).

  • total_budget – The total budget available for allocation.

  • channels – The list of channels to allocate the budget to.

  • parameters – Model-specific parameters for each channel.

  • budget_ranges – The budget ranges for each channel (min, max bounds).

Returns:

A DataFrame containing the estimated contribution and optimal budget allocation.

src.core.opt.calculate_seasonal_effectiveness(periods: List[int], seasonality_patterns: Dict[str, numpy.ndarray], channels: List[str], baseline_effectiveness: float = 1.0) → Dict[int, Dict[str, float]]

Calculate channel effectiveness multipliers for each period based on seasonality.

Parameters:

  • periods – List of time period indices (e.g., week numbers, 0-indexed).

  • seasonality_patterns – Dictionary mapping channel names to arrays of seasonal multipliers. Each array should have length >= max(periods) + 1.

  • channels – List of channel names to calculate effectiveness for.

  • baseline_effectiveness – Base effectiveness value (default 1.0 = no adjustment).

Returns:

Dictionary mapping period index to channel effectiveness multipliers. Format: {period_idx: {channel: effectiveness_multiplier}}

Example

>>> seasonality = {
...     'google': np.array([0.9, 0.95, 1.05, 1.1]),
...     'facebook': np.array([0.85, 0.9, 1.1, 1.15])
... }
>>> result = calculate_seasonal_effectiveness(
...     periods=[0, 1, 2],
...     seasonality_patterns=seasonality,
...     channels=['google', 'facebook']
... )
>>> result[0]['google']  # Period 0, Google effectiveness
0.9
src.core.opt.objective_multiperiod_distribution(x: numpy.ndarray, method: str, channels: List[str], parameters: Dict[str, Tuple[float, float]], n_periods: int, seasonal_effects: Dict[int, Dict[str, float]] | None = None) → float

Compute total contribution across multiple periods (negative for minimization).

This function calculates the negative sum of contributions for a proposed budget distribution across multiple periods. The budget array x is structured as a flat array: [period0_channel0, period0_channel1, …, period1_channel0, …].

Parameters:

  • x – Flat array of budget allocations across all periods and channels. Length = n_periods * len(channels).

  • method – Response curve method (‘michaelis-menten’ or ‘sigmoid’).

  • channels – List of channel names.

  • parameters – Model-specific parameters for each channel.

  • n_periods – Number of time periods to optimize over.

  • seasonal_effects – Optional seasonal effectiveness multipliers. Format: {period_idx: {channel: multiplier}}. If None, no seasonal adjustment is applied (multiplier = 1.0).

Returns:

Negative of total expected contribution across all periods (for minimization).

src.core.opt.objective_multiperiod_with_adstock(x: numpy.ndarray, method: str, channels: List[str], parameters: Dict[str, Tuple[float, float]], n_periods: int, adstock_params: Dict[str, float], adstock_max_lag: int, seasonal_effects: Dict[int, Dict[str, float]] | None = None, initial_state: Dict[str, float] | None = None) → float

Compute total contribution with adstock state dynamics (negative for minimization).

This objective function accounts for carryover effects by maintaining an adstock state for each channel across periods. The state evolves using geometric adstock:

s[0,c] = alpha[c] * initial_state[c] + spend[0,c] s[p,c] = alpha[c] * s[p-1,c] + spend[p,c] for p > 0

Where initial_state[c] represents the prior-period adstock state (s[-1,c]) for continuation scenarios. If initial_state is None, it defaults to zero (cold start).

Response curves are then evaluated on the adstocked state s[p,c], not raw spend. This matches the model’s transformation pipeline: Adstock → Saturation → Beta.

Parameters:

  • x – Flat array of budget allocations across all periods and channels. Length = n_periods * len(channels).

  • method – Response curve method (‘michaelis-menten’ or ‘sigmoid’).

  • channels – List of channel names.

  • parameters – Model-specific parameters for each channel. For sigmoid: (alpha_saturation, lam_saturation). Note: This alpha is for saturation, NOT adstock decay.

  • n_periods – Number of time periods to optimize over.

  • adstock_params – Adstock decay rates per channel. Format: {channel: alpha_decay} where alpha_decay in [0, 1].

  • adstock_max_lag – Maximum lag for adstock effect (for reference/validation).

  • seasonal_effects – Optional seasonal effectiveness multipliers. Applied AFTER adstock and saturation. Format: {period_idx: {channel: multiplier}}.

  • initial_state – Optional initial adstock state per channel. Format: {channel: initial_state_value}. If None, state initialized to zero (cold start).

Returns:

Negative of total expected contribution across all periods (for minimization).

Example

>>> # Channel with strong adstock (decay=0.7) benefits from early spending
>>> adstock_params = {'google': 0.7, 'facebook': 0.3}
>>> parameters = {'google': (0.5, 10000), 'facebook': (0.6, 15000)}
>>> budget = np.array([50000, 40000, 60000, 50000])  # 2 periods × 2 channels
>>> result = objective_multiperiod_with_adstock(
...     budget, 'sigmoid', ['google', 'facebook'], parameters,
...     n_periods=2, adstock_params=adstock_params, adstock_max_lag=12
... )
src.core.opt.validate_multiperiod_feasibility(total_budget: float, budget_ranges: Dict[str, Tuple[float, float]], channels: List[str], n_periods: int, period_budget_limits: Tuple[float, float] | None = None) → Tuple[bool, str]

Check if multi-period optimization constraints are feasible.

Performs basic arithmetic feasibility checks before optimization to catch obvious constraint conflicts early with clear error messages.

Parameters:

  • total_budget – Total budget across all periods and channels.

  • budget_ranges – Min/max bounds for each channel.

  • channels – List of channel names.

  • n_periods – Number of time periods.

  • period_budget_limits – Optional (min, max) per-period budget limits.

Returns:

Tuple of (is_feasible, error_message). If feasible: (True, “”) If infeasible: (False, “descriptive error message”)

Example

>>> is_ok, msg = validate_multiperiod_feasibility(
...     total_budget=100000,
...     budget_ranges={'google': (10000, 50000), 'fb': (5000, 30000)},
...     channels=['google', 'fb'],
...     n_periods=13
... )
>>> if not is_ok:
...     print(msg)
src.core.opt.validate_ramp_feasibility(budget_ranges: Dict[str, Tuple[float, float]], channels: List[str], n_periods: int, ramp_constraints: float | Dict[str, Any] | None, total_budget: float) → Tuple[bool, str, Dict[str, Any] | None]

Check if ramp constraints are compatible with budget bounds.

Performs envelope analysis to detect infeasible ramp scenarios before optimization. Computes the min/max reachable budget envelope for each channel across periods, then checks if total budget is achievable.

Parameters:

  • budget_ranges – Min/max bounds for each channel per period.

  • channels – List of channel names.

  • n_periods – Number of time periods.

  • ramp_constraints – Ramp constraint specification (see optimize_multiperiod_budget_distribution).

  • total_budget – Total budget across all periods and channels.

Returns:

Tuple of (is_feasible, error_message, suggestions). If feasible: (True, “”, None) If infeasible: (False, “error message”, {“suggested_changes”: …})

Example

>>> is_ok, msg, sug = validate_ramp_feasibility(
...     budget_ranges={'google': (20000, 60000), 'fb': (10000, 40000)},
...     channels=['google', 'fb'],
...     n_periods=13,
...     ramp_constraints={'absolute': 10000},
...     total_budget=3000000
... )
src.core.opt.round_and_repair_budget(budget_allocation: Dict[int, Dict[str, float]], budget_ranges: Dict[str, Tuple[float, float]], channels: List[str], n_periods: int, total_budget: float, round_increment: float = 1000.0, period_budget_limits: Tuple[float, float] | None = None, ramp_constraints: float | Dict[str, Any] | None = None, ramp_eps: float = 1e-06) → Tuple[Dict[int, Dict[str, float]], Dict[str, Any]]

Round budget values to increment and repair to satisfy constraints.

Rounds budget allocations to clean increments (e.g., £1,000) for operational convenience, then intelligently repairs the total to match the exact budget while preserving all constraints (bounds, ramps, period limits).

Parameters:

  • budget_allocation – Original optimization result. Format: {period: {channel: budget}}.

  • budget_ranges – Min/max bounds for each channel.

  • channels – List of channel names.

  • n_periods – Number of time periods.

  • total_budget – Target total budget (must match exactly after repair).

  • round_increment – Rounding increment (e.g., 1000 for £1K). Set to 0 to disable.

  • period_budget_limits – Optional (min, max) per-period budget limits.

  • ramp_constraints – Optional ramp constraint specification.

  • ramp_eps – Epsilon for safe percentage ramp denominators.

Returns:

  • rounded_budget: Rounded and repaired allocation {period: {channel: budget}}

  • diagnostics: Dict with repair statistics

Return type:

Tuple of (rounded_budget, diagnostics)

Example

>>> rounded, diag = round_and_repair_budget(
...     budget_allocation={0: {'google': 24567.89, 'fb': 18432.10}},
...     budget_ranges={'google': (10000, 50000), 'fb': (5000, 30000)},
...     channels=['google', 'fb'],
...     n_periods=1,
...     total_budget=43000,
...     round_increment=1000
... )
>>> rounded[0]['google']  # Should be multiple of 1000
25000.0
src.core.opt.optimize_multiperiod_budget_distribution(method: str, total_budget: int | float, budget_ranges: Dict[str, Tuple[float, float]], parameters: Dict[str, Tuple[float, float]], channels: List[str], n_periods: int, seasonal_effects: Dict[int, Dict[str, float]] | None = None, period_budget_limits: Tuple[float, float] | None = None, ramp_constraints: float | Dict[str, Any] | None = None, ramp_eps: float = 1e-06, ramp_to_baseline: Dict[str, float] | None = None, use_adstock: bool = False, adstock_params: Dict[str, float] | None = None, adstock_max_lag: int | None = None, initial_adstock_state: Dict[str, float] | None = None, optimization_method: str = 'SLSQP', max_iterations: int = 5000) → Dict[int, Dict[str, float]]

Optimize budget allocation across multiple periods and channels.

This function finds the optimal distribution of a total budget across multiple time periods and marketing channels to maximize total expected contribution. It can optionally account for seasonal variations in channel effectiveness and impose per-period budget constraints.

Parameters:

  • method – Response curve method (‘michaelis-menten’ or ‘sigmoid’).

  • total_budget – Total budget to distribute across all periods and channels.

  • budget_ranges – Budget ranges (min, max) for each channel per period. Format: {channel: (min_budget, max_budget)}.

  • parameters – Model-specific parameters for each channel.

  • channels – List of channel names.

  • n_periods – Number of time periods to optimize over.

  • seasonal_effects – Optional seasonal effectiveness multipliers. Format: {period_idx: {channel: multiplier}}. Defaults to None.

  • period_budget_limits – Optional (min, max) budget limits per period. If specified, each period’s total budget must fall within these limits. Defaults to None (no per-period limits).

  • ramp_constraints –

    Optional constraints on period-to-period budget changes. Prevents wild budget swings between periods. Formats: - float: Uniform absolute limit for all channels (e.g., 10000) - dict with ‘absolute’ key: Per-channel absolute limits

    {‘absolute’: {‘google’: 5000, ‘facebook’: 8000}}

    • dict with ‘percentage’ key: Per-channel percentage limits {‘percentage’: {‘google’: 0.15, ‘facebook’: 0.20}}

    • dict with both: Mixed absolute and percentage limits {‘absolute’: {‘google’: 5000}, ‘percentage’: {‘facebook’: 0.20}}

    • dict (direct): Per-channel absolute limits (legacy format) {‘google’: 5000, ‘facebook’: 8000}

    Both absolute and percentage can apply to same channel (intersection). Defaults to None (no ramp constraints).

  • ramp_eps – Epsilon for safe denominator in percentage ramp constraints. Prevents division by zero when previous period spend is near zero. Defaults to 1e-6.

  • ramp_to_baseline – Optional anchor constraint for period 0 vs historical baseline. Format: {channel: max_change_from_baseline}. Not yet implemented. Defaults to None.

  • use_adstock – Whether to use adstock-aware optimization. When True, the optimizer accounts for carryover effects between periods using state dynamics. The model’s adstock parameters must be provided. Defaults to False.

  • adstock_params – Adstock decay rates per channel (required if use_adstock=True). Format: {channel: alpha_decay} where alpha_decay in [0, 1]. Example: {‘google’: 0.7, ‘facebook’: 0.3} means Google has strong carryover. Defaults to None.

  • adstock_max_lag – Maximum lag for adstock effect (required if use_adstock=True). This parameter is used for validation and model consistency checking. It is not used in the objective calculation itself but ensures the optimizer’s adstock assumptions match the model’s configuration. Should match the model’s adstock_max_lag parameter (typically 12 weeks). Defaults to None.

  • initial_adstock_state – Optional initial adstock state per channel. Useful when continuing from a previous optimization period. Format: {channel: initial_state_value}. If None, state is initialized to zero (cold start). Defaults to None.

  • optimization_method – Scipy optimization method to use. Defaults to ‘SLSQP’.

  • max_iterations – Maximum number of optimization iterations. Defaults to 1000.

Returns:

Dictionary mapping period index to channel budgets. Format: {period_idx: {channel: budget}}.

Example

>>> result = optimize_multiperiod_budget_distribution(
...     method='sigmoid',
...     total_budget=1000000,
...     budget_ranges={'google': (0, 50000), 'facebook': (0, 80000)},
...     parameters={'google': (0.5, 10000), 'facebook': (0.6, 15000)},
...     channels=['google', 'facebook'],
...     n_periods=13
... )
>>> result[0]['google']  # Budget for Google in period 0
25000.0