Planning

The measurement-loop planning engine: EIG/EVOI experiment priorities, experiment design, economics, the Pareto-front optimizer, and run history.

Decision layer on top of a fitted MMM: budget optimization, EIG/EVOI experiment prioritization, and experiment-design recommendation.

Import-light by design (numpy/pandas only) so it can run inside the agent’s session kernels via model_ops.

mmm_framework.planning.design_experiment(dataset_path, kpi, channel, *, design_key=None, method=None, **kwargs)

Dispatch to the right design family (auto-recommended when omitted).

method names the analysis methodology (synthetic_control / tbr / gbr / did_mmt). It selects which estimator the read-out and economics use and, when design_key is not given, infers the design family. The chosen method’s metadata is attached to the returned design dict.

Return type:

dict[str, Any]

mmm_framework.planning.design_options(dataset_path, kpi, channel)

What designs the data supports: geo designs need >= 4 geographies.

Return type:

dict[str, Any]

mmm_framework.planning.geo_lift_design(dataset_path, kpi, channel, *, design='holdout', intensity_pct=50.0, n_pairs=None, duration=8, durations=(4, 6, 8, 10, 12, 16), randomize=True, seed=42)

Design a geo lift study (or a matched-market DiD when randomize=False).

Pipeline — robust by construction, on the model’s own view of history: :rtype: dict[str, Any]

1. Residualize each geo’s KPI on the MMM’s baseline structure (trend + yearly seasonality + the channel’s spend response).

2. Match on residual co-movement + standardized model-feature distance; solve the pairing globally (minimum-weight perfect matching).

3. Randomize treatment within each pair (or largest-geo-treated for the observational matched-market variant).

4. Power: the analytic DiD curve is CALIBRATED against a placebo simulation — the actual estimator run over historical windows — so the reported SE/MDE reflects how this design behaves on this data, not just a formula. Covariate balance on the model’s features is reported.

mmm_framework.planning.flighting_design(dataset_path, kpi, channel, *, amplitude_pct=50.0, block_weeks=2, duration=12, budget_neutral=True, levels=None, x0=None, seed=42)

Block-randomized flighting schedule for a national series.

Historical spend is usually smooth and co-moves with demand — the model cannot tell the channel from the season. A budget-neutral pulse schedule (randomized block order) manufactures exogenous variance: identification the data finally pays for. Block length should be >= the channel’s adstock memory so carryover doesn’t smear the contrast.

levels (spend multipliers, e.g. (0.5, 1.0, 1.5, 2.0)) gives a MULTI-LEVEL schedule — ≥3 distinct levels let the experiment trace the response CURVE (so the marginal ROAS / saturation is identified), not just a single on/off contrast. Default (None) is the binary high/low pulse at ±amplitude_pct.

Return type:

dict[str, Any]

mmm_framework.planning.matched_pairs(kpi_wide, n_pairs=None, spend_wide=None)

Model-structured matched pairs, best-match first.

Each pair carries correlation (raw KPI — what naive matching would have scored), residual_correlation (co-movement AFTER removing trend/ seasonality/spend — what the DiD’s precision actually depends on), size_ratio, and the matching distance.

Return type:

list[dict]

class mmm_framework.planning.BudgetOptimizationResult(table, total_budget, expected_uplift, uplift_hdi, prob_positive_uplift, n_draws, notes=<factory>, per_draw_alloc=None, optimal_alloc=None, expected_regret=0.0, n_extrapolated=0, objective='mean', objective_label='expected KPI (posterior mean)', mode='fixed', shadow_price=None, marginal_roas=None)

Bases: object

Optimal allocation plus the decision-uncertainty diagnostics.

table: DataFrame

total_budget: float

expected_uplift: float

uplift_hdi: tuple[float, float]

prob_positive_uplift: float

n_draws: int

notes: list[str]

per_draw_alloc: ndarray | None = None

optimal_alloc: ndarray | None = None

expected_regret: float = 0.0

n_extrapolated: int = 0

objective: str = 'mean'

objective_label: str = 'expected KPI (posterior mean)'

mode: str = 'fixed'

shadow_price: float | None = None

marginal_roas: dict[str, float] | None = None

__init__(table, total_budget, expected_uplift, uplift_hdi, prob_positive_uplift, n_draws, notes=<factory>, per_draw_alloc=None, optimal_alloc=None, expected_regret=0.0, n_extrapolated=0, objective='mean', objective_label='expected KPI (posterior mean)', mode='fixed', shadow_price=None, marginal_roas=None)

class mmm_framework.planning.ResponseCurves(channel_names, multipliers, base_spend, contributions, obs_max_spend=None, n_obs=None)

Bases: object

Per-channel contribution as a function of spend, with posterior draws.

contributions[d, c, g] = total (window-summed) original-scale contribution of channel c at spend multiplier multipliers[g], under posterior draw d.

channel_names: list[str]

multipliers: ndarray

base_spend: ndarray

contributions: ndarray

obs_max_spend: ndarray | None = None

n_obs: int | ndarray | None = None

property spend_grid: ndarray

(C, G) actual spend per channel at each multiplier.

property max_obs_multiplier: ndarray | None

(C,) the largest CURRENT-spend multiple that stays within observed support: max_per_obs_spend / mean_per_obs_spend. Scaling current spend past this pushes the average period beyond anything observed, so the response there is the saturation FORM extrapolating. None when the observed range was not captured.

mean_curves()

(C, G) posterior-mean contribution curves.

Return type:

ndarray

__init__(channel_names, multipliers, base_spend, contributions, obs_max_spend=None, n_obs=None)

mmm_framework.planning.compute_response_curves(mmm, multipliers=None, max_draws=200, random_seed=None)

Sample per-channel spend-response curves from the fitted model.

One posterior-predictive evaluation per multiplier (NOT per channel × multiplier): all channels are scaled together and read off the additive channel_contributions deterministic.

Return type:

ResponseCurves

mmm_framework.planning.optimize_budget(mmm=None, *, curves=None, total_budget=None, budget_change_pct=None, min_multiplier=0.0, max_multiplier=2.0, bounds=None, abs_bounds=None, groups=None, min_channel_spend=None, objective='mean', mode='fixed', value_per_kpi=None, n_steps=400, max_draws=200, random_seed=None)

Find the budget allocation that maximizes the chosen KPI objective.

Parameters:

• curves (ResponseCurves | None) – precomputed response curves (else sampled from mmm).

• total_budget (float | None) – budget to allocate. Default: current total spend (pure reallocation). budget_change_pct (e.g. -10 or 15) scales the current total instead. Ignored in mode="free".

• max_multiplier (min_multiplier /) – per-channel spend bounds as multiples of CURRENT channel spend (floors/caps keep recommendations inside the range the model has evidence for; extrapolating far beyond observed spend is curve-fiction).

• bounds (dict[str, tuple[float, float]] | None) – per-channel {name: (lo_mult, hi_mult)} multiplier overrides.

• abs_bounds (dict[str, tuple[float, float]] | None) – per-channel {name: (lo_$, hi_$)} ABSOLUTE dollar bounds (override the multiplier bounds for that channel) — #139.

• groups (list[dict] | None) – portfolio constraints, a list of {"name", "channels": [...], "min_share"/"max_share": frac, "min_spend"/"max_spend": $} — e.g. “digital ≥ 40%” — #139.

• min_channel_spend (float | dict[str, float] | None) – keep-every-channel-on floor (a scalar applied to all channels, or per-channel {name: $}) — #139.

• objective (str) – the risk objective the allocator maximizes — ‘mean’ (posterior mean), ‘p<q>’ (downside q-th percentile, e.g. ‘p10’), or ‘cvar<a>’ (mean of the worst a% of draws, e.g. ‘cvar5’) (#139).

• mode (str) – "fixed" spends the budget; "free" funds each channel to breakeven (marginal value = $1), the total becomes an output (#139).

• value_per_kpi (float | None) – $ value of one KPI unit (for breakeven / marginal ROAS when the KPI is not already revenue). Default 1.0.

• n_steps (int) – greedy increments (granularity of the default allocation).

• max_draws (int) – posterior draws used for curves and stability analysis.

Return type:

BudgetOptimizationResult

mmm_framework.planning.objective_curves(curves, objective='mean')

(C, G) response curve for a risk objective — the curve the allocator maximizes. 'mean' = posterior mean (risk-neutral); 'p<q>' = the per-channel q-th percentile across draws (a conservative downside curve); 'cvar<a>' = the per-channel mean of the worst a% of draws (expected shortfall). Per-channel quantiles are a conservative proxy for the joint portfolio quantile (they ignore cross-channel draw correlation), documented in technical-docs/budget-optimizer-v2.md.

Return type:

ndarray

mmm_framework.planning.objective_label(objective='mean')

Human-readable label for an objective, for reporting which one was used.

Return type:

str

mmm_framework.planning.budget_frontier(mmm=None, *, curves=None, budget_min=None, budget_max=None, n_points=12, objective='mean', min_multiplier=0.0, max_multiplier=2.0, bounds=None, abs_bounds=None, groups=None, min_channel_spend=None, value_per_kpi=None, max_draws=120, random_seed=42)

Sweep a budget range → the efficient frontier (optimized return vs budget).

At each budget the constrained allocator maximizes the objective curve; the resulting plan is evaluated across all posterior draws for a return band, and the discrete slope gives the marginal ROI (the next-dollar return at that budget). By default the sweep runs from 40% to 200% of current spend.

Return type:

FrontierResult

mmm_framework.planning.goal_seek(mmm=None, *, curves=None, target_kpi, objective='mean', min_multiplier=0.0, max_multiplier=2.0, bounds=None, abs_bounds=None, groups=None, min_channel_spend=None, value_per_kpi=None, budget_max=None, tol=0.001, max_iter=40, max_draws=120, random_seed=42)

Inverse solve: the minimum budget (and its mix) whose optimized plan reaches target_kpi total media contribution.

The frontier return is monotone increasing and concave in budget, so a bisection on the optimized return finds the required budget. Reports the plan and prob_hit_target — the posterior probability that the plan’s contribution actually clears the target (the honest answer to “will this budget hit my number?”).

Return type:

GoalSeekResult

class mmm_framework.planning.FrontierResult(objective, objective_label, channel_names, current_total, current_return, points=<factory>, notes=<factory>)

Bases: object

The efficient frontier plus the current operating point.

objective: str

objective_label: str

channel_names: list[str]

current_total: float

current_return: float

points: list[FrontierPoint]

notes: list[str]

to_dict()

Return type:

dict[str, Any]

__init__(objective, objective_label, channel_names, current_total, current_return, points=<factory>, notes=<factory>)

class mmm_framework.planning.GoalSeekResult(target_kpi, objective, objective_label, channel_names, feasible, required_budget, allocation, expected_return, prob_hit_target, notes=<factory>)

Bases: object

The minimum budget (and mix) that reaches a target KPI contribution.

target_kpi: float

objective: str

objective_label: str

channel_names: list[str]

feasible: bool

required_budget: float | None

allocation: dict[str, float] | None

expected_return: float | None

prob_hit_target: float | None

notes: list[str]

to_dict()

Return type:

dict[str, Any]

__init__(target_kpi, objective, objective_label, channel_names, feasible, required_budget, allocation, expected_return, prob_hit_target, notes=<factory>)

mmm_framework.planning.default_reallocation(mmm, *, deviation=0.2, max_draws=150, random_seed=42)

A conservative, report-ready default reallocation of the current budget.

This is the plan surfaced in a generated client report when the user has not run the Planner studio. It reallocates the current total spend (no budget change) across channels, but constrains every channel to within ±deviation of its current spend — default ±20% — so that:

• no channel is turned off (the floor is 1 - deviation of current spend, never zero), and

• no recommendation extrapolates beyond the spend range the model has evidence for — the response curves are sampled on a multiplier grid that stays inside [1 - deviation, 1 + deviation] (plus the 1.0 anchor), so the greedy allocator only ever interpolates within observed support.

Parameters:

• mmm (Any) – a fitted model exposing X_media_raw / channel_names / sample_channel_contributions (the compute_response_curves() contract).

• deviation (float) – max fractional move per channel (0.20 ⇒ ±20%).

• max_draws (int) – posterior draws for the curves and decision-uncertainty.

• random_seed (int | None) – seed for reproducible curve sampling.

Return type:

dict[str, Any]

Returns:

The report-ready allocation_results dict (see _result_to_report_dict()), tagged with deviation_cap.

mmm_framework.planning.compute_response_curves_per_geo(mmm, multipliers=None, max_draws=200, random_seed=None)

Per-geography spend-response curves from a fitted geo panel.

One posterior-predictive evaluation per multiplier (all geos scaled together, as in compute_response_curves()); contributions are then grouped by mmm.geo_idx so each geography gets its own ResponseCurves over that geography’s current spend. Requires mmm.has_geo.

Return type:

dict[str, ResponseCurves]

mmm_framework.planning.combine_geo_curves(geo_curves)

Flatten {geo: ResponseCurves} into a single ResponseCurves whose “channels” are "{geo_index} │ {channel}" arms, so the national greedy optimizer can allocate one budget jointly across every geo×channel arm. All inputs must share the same multipliers and channel ordering.

Return type:

ResponseCurves

mmm_framework.planning.optimize_budget_by_geo(mmm, *, total_budget=None, budget_change_pct=None, min_multiplier=0.0, max_multiplier=2.0, bounds=None, n_steps=400, max_draws=200, random_seed=None)

Allocate one national budget jointly across every (geo, channel) arm.

Builds per-geo curves, flattens them to arms, and reuses optimize_budget() so the greedy marginal allocator, per-draw stability, and uplift diagnostics are identical to the national path. The returned table gains a geo column and the channel column holds the bare channel name. Per-channel bounds apply to that channel in every geography.

Return type:

BudgetOptimizationResult

class mmm_framework.planning.PacingChannel(channel, planned, actual, divergence_pct, status, planned_series=<factory>, actual_series=<factory>)

Bases: object

One channel’s pacing status.

channel: str

planned: float

actual: float

divergence_pct: float

status: str

planned_series: list[float]

actual_series: list[float]

to_dict()

Return type:

dict[str, Any]

__init__(channel, planned, actual, divergence_pct, status, planned_series=<factory>, actual_series=<factory>)

class mmm_framework.planning.PacingResult(channels, periods, threshold, planned_total, actual_total, divergence_pct, flagged, outcome_delta=None, join='positional')

Bases: object

Portfolio + per-channel pacing, with the flagged (off-pace) channels.

channels: list[PacingChannel]

periods: list[str]

threshold: float

planned_total: float

actual_total: float

divergence_pct: float

flagged: list[str]

outcome_delta: dict[str, float] | None = None

join: str = 'positional'

‘label’ when plan and actual were joined on a shared period vocabulary, ‘positional’ when either side had none. Recorded rather than assumed: a positional join on labelled data was the pre-v1.4 defect.

to_dict()

Return type:

dict[str, Any]

__init__(channels, periods, threshold, planned_total, actual_total, divergence_pct, flagged, outcome_delta=None, join='positional')

mmm_framework.planning.compute_pacing(planned, actual, *, threshold=0.1)

Compare a plan against actual delivery, per channel and overall.

Both inputs normalize to per-channel per-period arrays. When both sides carry real period labels the comparison joins by label, so a mid-flight upload covering weeks 5-8 is compared against the plan’s weeks 5-8. Inputs without labels fall back to positional truncation. PacingResult.join records which happened.

Return type:

PacingResult

mmm_framework.planning.expected_outcome_delta(curves, planned_totals, actual_totals, *, hdi_prob=0.9)

Expected KPI delta from the divergence, off the fitted response curves.

curves is a ResponseCurves (or any object exposing channel_names, spend_grid (C,G) and contributions (D,C,G)). For each posterior draw we read the channel’s contribution at the ACTUAL total vs the PLANNED total and sum the difference — turning parameter uncertainty into a credible interval on the outcome impact. Returns {"mean","lower","upper","planned_kpi","actual_kpi"} or None.

Return type:

dict[str, float] | None

mmm_framework.planning.pacing_report(planned, actual, *, curves=None, threshold=0.1, hdi_prob=0.9)

Full pacing assessment: divergence + (when curves given) the expected KPI impact of that divergence.

Return type:

PacingResult

mmm_framework.planning.build_flighting_schedule(channel_budgets, n_periods, *, pattern='even', period_labels=None, calendar=None, front_load=0.65, pulse_on=1, pulse_off=1, seasonal=None, per_channel_pattern=None)

Distribute each channel’s total budget across n_periods future periods.

Returns a JSON-safe dict:

{
  "pattern", "n_periods", "total_budget",
  "periods": [labels...],
  "channels": [names...],
  "schedule": [{"period": label, "<channel>": spend, ..., "total": tot}],
  "by_channel": {channel: [spend per period]},
}

per_channel_pattern overrides the global pattern for named channels (e.g. an always-on retainer line vs a pulsed promo line).

Return type:

dict[str, Any]

mmm_framework.planning.recommend_experiments(mmm, *, curves=None, optimization=None, top_k=3, max_draws=200, random_seed=None, method='eig_evoi', evidence=None)

Rank channels by experiment value and propose concrete designs.

Returns (table, designs): a per-channel scoring table (all channels, sorted by priority) and a list of design dicts for the top_k channels.

method=”eig_evoi” (default): priority is the normalized geometric mean of EIG (what the experiment teaches) and EVOI (what that learning is worth to the budget decision) — see planning.priority. The table gains eig / evoi / quadrant columns and designs carry the priority snapshot.

Return type:

tuple[DataFrame, list[dict]]

method=”heuristic”: the transparent legacy score

spend_share × roas_cv × (1 + allocation_instability)

— money at stake, times relative ROAS uncertainty, amplified when the uncertainty visibly destabilizes the optimal allocation.

class mmm_framework.planning.ChannelPriority(channel, spend, spend_share, roi_mean, roi_sd, roi_hdi_low, roi_hdi_high, sigma_exp, design_type, eig, eig_method, evoi, evoi_pct_budget, evpi_share, priority, quadrant, weeks_since_evidence=None, eig_decayed=None, retest_due=False)

Bases: object

One row of the priority grid (JSON-safe via .to_dict()).

channel: str

spend: float

spend_share: float

roi_mean: float

roi_sd: float

roi_hdi_low: float

roi_hdi_high: float

sigma_exp: float

design_type: str

eig: float

eig_method: str

evoi: float

evoi_pct_budget: float

evpi_share: float

priority: float

quadrant: str

weeks_since_evidence: float | None = None

eig_decayed: float | None = None

retest_due: bool = False

to_dict()

Return type:

dict[str, Any]

__init__(channel, spend, spend_share, roi_mean, roi_sd, roi_hdi_low, roi_hdi_high, sigma_exp, design_type, eig, eig_method, evoi, evoi_pct_budget, evpi_share, priority, quadrant, weeks_since_evidence=None, eig_decayed=None, retest_due=False)

mmm_framework.planning.compute_experiment_priorities(mmm=None, *, curves=None, optimization=None, evidence=None, as_of=None, design_type=None, sigma_exp_overrides=None, roi_draws_overrides=None, half_life_overrides=None, eig_threshold=None, evoi_threshold=None, retest_threshold_nats=0.15, n_outcomes=48, n_steps=400, max_draws=200, random_seed=42)

Per-channel EIG/EVOI priority grid + portfolio summary.

Parameters:

• evidence (dict[str, dict] | None) – {channel: {"end_date": ISO, ...}} — the newest calibrated experiment per channel (sessions.latest_calibrated_evidence). Drives information decay and the re-test trigger.

• as_of (str | None) – ISO date for the decay clock (default: today).

• design_type (str | None) – assumed experiment design for sigma_exp; default geo_holdout when the model has geo structure, else national_pulse.

• evoi_threshold (eig_threshold /) – absolute quadrant splits; default is the median of each axis across channels.

Return type:

tuple[list[ChannelPriority], dict[str, Any]]

Returns (grid, portfolio) where portfolio carries v_current, EVPI, total budget, draws, and the thresholds used.

class mmm_framework.planning.EvoiResult(v_current, evpi, n_draws)

Bases: object

Portfolio-level value-of-information summary.

v_current: float

evpi: float

n_draws: int

__init__(v_current, evpi, n_draws)

mmm_framework.planning.compute_evoi_for_channel(curves, channel_idx, roi_draws, sigma_exp, *, optimal_alloc=None, n_outcomes=48, n_steps=400, total_budget=None, lo_spend=None, hi_spend=None, outcome_draws=None, rng=None)

EVOI of an experiment on one channel, by paired preposterior Monte Carlo.

For each simulated outcome y_j = roi[d_j] + sigma_exp * z_j: reweight the posterior draws by w_d ∝ N(y_j | roi_d, sigma_exp), re-optimize the allocation on the REWEIGHTED mean curves, and score the per-outcome gain :rtype: float

gain_j = V_j(a_j*) - V_j(a*) with V_j(a) = E_{theta|y_j}[f(a, theta)]

against the no-experiment allocation a* under the SAME reweighted posterior (the paired estimator: identical estimand by the tower property, but the common-mode outcome noise cancels, so EVOI cannot go negative from sampling luck — each gain_j >= 0 up to allocator discretization, enforced by a per-term floor).

outcome_draws = (d_idx, z) lets the caller share common random numbers across channels so near-ties don’t rank-flip on MC noise. The caller should cap the result at EVPI.

mmm_framework.planning.cpa_interval(lift, se_lift, cost, *, alpha=0.05)

CPA point + interval by inverting the lift interval — and, for contrast, the naive symmetric delta-method interval that clients are usually shown.

Returns status: :rtype: dict[str, Any]

• "bounded" — the lift interval is strictly positive; the CPA interval is [cost/lift_hi, cost/lift_lo] (asymmetric: the upper arm is longer).

• "upper_unbounded" — the lift interval touches zero; only a lower CPA bound exists. The data cannot rule out an arbitrarily bad CPA.

• "undefined" — the whole lift interval is ≤ 0; no positive-lift evidence, no CPA statement at all.

mmm_framework.planning.cpa_power(cost, se_lift, true_lift, *, target_cpa=None, alpha=0.05)

Power on the CPA scale.

With target_cpa=None: the probability the test comes away with a bounded CPA interval at all (the lift interval clears zero) — Φ(true_lift/se − z). With a target: the probability the CPA interval’s upper end lands below target_cpa, i.e. the test certifies CPA ≤ target — equivalent to the estimated lift clearing cost/target + z·se.

Return type:

float

mmm_framework.planning.max_detectable_cpa(cost, lift_mde)

The WORST (highest) cost per conversion the design can still certify: cost / MDE. A true CPA above this is indistinguishable from “the media did nothing” — the honest planning summary on the CPA scale.

Return type:

float

mmm_framework.planning.simulate_cpa_distribution(true_lift, se_lift, cost, *, alpha=0.05, n_sims=20000, seed=0)

Monte-Carlo the CPA readout a design of this precision would produce — the demonstration that the reciprocal is NOT well-behaved even when the lift is.

Simulates lift̂ ~ N(true_lift, se_lift), forms cpâ = cost/lift̂ where the estimate is positive (non-positive draws are “no measurable conversions” — CPA undefined/infinite), and scores both interval constructions against the true CPA. Returns the draws (for plotting) plus skew diagnostics and empirical coverage.

Return type:

dict[str, Any]

class mmm_framework.planning.EvoiSurrogate(k, delta, tau, sigma_anchors=())

Bases: object

Calibrated Gaussian EVOI surrogate sigma → k·s(sigma)·Psi(delta/s).

tau is the channel’s prior (pre-experiment) ROI sd; k and delta are fitted to the MC anchors. Call it with any design precision to price that design’s EVOI without another preposterior MC. Accurate BETWEEN the anchor sigmas; extrapolation beyond the weak anchor under-estimates.

k: float

delta: float

tau: float

sigma_anchors: tuple[float, ...] = ()

__init__(k, delta, tau, sigma_anchors=())

mmm_framework.planning.fit_evoi_surrogate(tau, anchors)

Fit (k, delta) of EvoiSurrogate to MC anchors [(sigma_exp, evoi_mc), ...].

Two usable anchors (distinct sigmas, both positive EVOI) pin both parameters by bisection on the anchor ratio; one usable anchor degrades to delta = 0 (the surrogate_evoi() scaling). Returns None when tau is degenerate or no anchor has positive EVOI.

Return type:

EvoiSurrogate | None

mmm_framework.planning.preposterior_sd_ratio(tau, sigma_exp)

sd of the preposterior mean as a FRACTION of the prior sd tau: sqrt(tau^2 / (tau^2 + sigma_exp^2)) — 1 for a perfect experiment (sigma_exp → 0), → 0 for an uninformative one.

Return type:

float

mmm_framework.planning.surrogate_evoi(evoi_ref, sigma_ref, sigma_new, tau, *, evpi=None)

Single-anchor fallback: EVOI scaled by the preposterior-sd ratio s(sigma_new)/s(sigma_ref) (the delta = 0 special case of the fitted surrogate — decision boundary at the prior mean). Prefer fit_evoi_surrogate() with two anchors when you can afford the second MC evaluation; the single-anchor form decays too slowly for weak designs. Capped at evpi when given; 0 when the anchor or prior is degenerate.

Return type:

float

mmm_framework.planning.compute_evpi(curves, *, total_budget=None, lo_spend=None, hi_spend=None, n_steps=400, per_draw_alloc=None, optimal_alloc=None)

V_current and EVPI from the response curves.

Pass per_draw_alloc/optimal_alloc from a BudgetOptimizationResult to reuse its D greedy optimizations instead of recomputing them.

Return type:

EvoiResult

mmm_framework.planning.eig_gaussian(sigma_k, sigma_exp)

Closed-form EIG (nats) for a Gaussian prior and Gaussian experiment.

Return type:

float

mmm_framework.planning.eig_monte_carlo(roi_draws, sigma_exp, *, n_outcomes=64, rng=None)

Monte Carlo EIG for non-Gaussian ROI posteriors.

Prior-predictive outcomes y_j = roi[d_j] + sigma_exp * z_j; for each, the posterior over draws is the importance reweighting w_d ∝ N(y_j | roi_d, sigma_exp) and entropy is approximated by the Gaussian proxy 0.5 * ln(2*pi*e * Var) on the (weighted) draws — exact in the Gaussian limit, robust to skew where the closed form overstates.

Return type:

float

mmm_framework.planning.sigma_exp_for_design(design_type, roi_median, *, floor=0.05)

Design-grounded experiment sd on the ROI scale.

sigma_exp = precision(design) * max(|roi_median|, floor) — proportional to the channel’s ROI scale (a relative-precision design) but NEVER to its posterior uncertainty.

Return type:

float

mmm_framework.planning.decayed_sigma(sigma_post, weeks_elapsed, half_life_weeks)

Effective sd after weeks_elapsed: sigma_eff^2(t) = sigma_post^2 * exp(lambda * t), lambda = ln2/half-life — so sigma_eff(t) = sigma_post * exp(0.5 * lambda * t).

Return type:

float

mmm_framework.planning.channel_half_life(channel, overrides=None)

Deprecated alias for information_half_life().

The old name did not say which half-life it meant, and planning sat one import away from the model-derived carryover half-life.

Return type:

float

mmm_framework.planning.information_half_life(channel, overrides=None)

How fast evidence about a channel goes stale, in weeks.

This is an information-decay half-life, not a media-carryover one. It reads nothing from the model — it is a keyword table over the channel name (see DEFAULT_HALF_LIVES_WEEKS) answering “when is this channel’s last read old enough to re-test?”.

Do not confuse it with mmm_framework.transforms.carryover.carryover_half_life(), which is per-draw, read from the fitted kernel, and answers “how long does a exposure keep working?”. The two legitimately disagree by ~10x, which is why this one is no longer called channel_half_life: planning exported two similarly-named half-lives for the same channel meaning different things.

Return type:

float

mmm_framework.planning.reexperiment_due(sigma_post, weeks_elapsed, half_life_weeks, sigma_exp, *, threshold_nats=0.15, min_age_weeks=13.0)

(due, current_eig): the EIG of a fresh experiment given the decayed effective uncertainty, and whether it crosses the operational threshold. Evidence younger than min_age_weeks is never due (freshness floor).

Return type:

tuple[bool, float]

class mmm_framework.planning.OpportunityCostResult(channel, kpi, design_key, design_type, duration_requested, duration_effective, n_treated_cells, n_test_rows, n_draws, carryover_basis, expected_kpi_delta, kpi_delta_median, kpi_delta_p5, kpi_delta_p95, kpi_delta_with_carryover_median, forgone_kpi_median, forgone_kpi_p95, prob_kpi_loss, pct_of_window_kpi, spend_delta, abs_spend_change, spend_at_risk, margin_per_kpi, margin_source, kpi_kind, net_profit_impact_median, net_profit_impact_p5, net_profit_impact_p95, opportunity_cost_dollar_median, opportunity_cost_dollar_p95, prob_net_loss, prob_loss_over_threshold, loss_threshold, evoi_kpi_units, evoi_per_week, cost_per_week, learning_to_cost_ratio, learning_to_cost_basis, response_horizon_weeks, low_information, extrapolation_warning, warnings=<factory>, notes=<factory>, draws=None)

Bases: object

Per-draw short-term risk of a design, summarized. JSON-safe via to_dict.

channel: str

kpi: str

design_key: str

design_type: str

duration_requested: int

duration_effective: int

n_treated_cells: int

n_test_rows: int

n_draws: int

carryover_basis: str

expected_kpi_delta: float

kpi_delta_median: float

kpi_delta_p5: float

kpi_delta_p95: float

kpi_delta_with_carryover_median: float

forgone_kpi_median: float

forgone_kpi_p95: float

prob_kpi_loss: float

pct_of_window_kpi: float | None

spend_delta: float

abs_spend_change: float

spend_at_risk: float

margin_per_kpi: float | None

margin_source: str

kpi_kind: str

net_profit_impact_median: float | None

net_profit_impact_p5: float | None

net_profit_impact_p95: float | None

opportunity_cost_dollar_median: float | None

opportunity_cost_dollar_p95: float | None

prob_net_loss: float | None

prob_loss_over_threshold: float | None

loss_threshold: float | None

evoi_kpi_units: float | None

evoi_per_week: float | None

cost_per_week: float | None

learning_to_cost_ratio: float | None

learning_to_cost_basis: str

response_horizon_weeks: int | None

low_information: bool

extrapolation_warning: bool

warnings: list[str]

notes: list[str]

draws: dict | None = None

to_dict()

Return type:

dict[str, Any]

__init__(channel, kpi, design_key, design_type, duration_requested, duration_effective, n_treated_cells, n_test_rows, n_draws, carryover_basis, expected_kpi_delta, kpi_delta_median, kpi_delta_p5, kpi_delta_p95, kpi_delta_with_carryover_median, forgone_kpi_median, forgone_kpi_p95, prob_kpi_loss, pct_of_window_kpi, spend_delta, abs_spend_change, spend_at_risk, margin_per_kpi, margin_source, kpi_kind, net_profit_impact_median, net_profit_impact_p5, net_profit_impact_p95, opportunity_cost_dollar_median, opportunity_cost_dollar_p95, prob_net_loss, prob_loss_over_threshold, loss_threshold, evoi_kpi_units, evoi_per_week, cost_per_week, learning_to_cost_ratio, learning_to_cost_basis, response_horizon_weeks, low_information, extrapolation_warning, warnings=<factory>, notes=<factory>, draws=None)

mmm_framework.planning.compute_opportunity_cost(mmm, design, *, margin_per_kpi=None, kpi_kind='revenue', price=None, preferences=None, branding=None, loss_threshold=None, evoi_kpi_units=None, response_horizon_weeks=None, max_draws=200, random_seed=42, contrib_bau=None, contrib_exp=None, return_draws=False)

Short-term risk of running design on the fitted model mmm.

Parameters:

• design (dict) – a planning.design.design_experiment payload (carries channel, kpi, design_key, intensity_pct / schedule, treatment_geos, duration).

• kpi_kind (margin_per_kpi / price /) – net-$ valuation inputs. revenue multiplies KPI by margin; units by margin*price; other reports KPI units only. Margin also auto-resolves from branding / preferences economics blocks.

• evoi_kpi_units (float | None) – the experiment’s EVOI (planning.evoi, KPI-contribution units) for the learning-vs-cost ratio. Pass the EVPI-capped value.

• response_horizon_weeks (int | None) – denominator for evoi_per_week (default mmm.n_periods).

Return type:

OpportunityCostResult

class mmm_framework.planning.ExperimentNetValue(channel, unit, basis, test_loss, test_loss_p5, test_loss_p95, net_profit_during_test, evoi_raw, evpi_cap, decay_factor, reallocation_gain, net_value, net_value_p5, net_value_p95, prob_net_positive, breakeven_horizon_weeks, horizon_weeks, half_life_weeks, margin_per_kpi, assumptions=<factory>, warnings=<factory>)

Bases: object

The netted decision figure. unit is ‘$’ when a margin resolved, else ‘KPI units’ (spend cannot be netted against KPI — noted in assumptions).

channel: str

unit: str

basis: str

test_loss: float | None

test_loss_p5: float | None

test_loss_p95: float | None

net_profit_during_test: float | None

evoi_raw: float | None

evpi_cap: float | None

decay_factor: float | None

reallocation_gain: float | None

net_value: float | None

net_value_p5: float | None

net_value_p95: float | None

prob_net_positive: float | None

breakeven_horizon_weeks: float | None

horizon_weeks: int

half_life_weeks: float | None

margin_per_kpi: float | None

assumptions: dict

warnings: list[str]

to_dict()

Return type:

dict[str, Any]

__init__(channel, unit, basis, test_loss, test_loss_p5, test_loss_p95, net_profit_during_test, evoi_raw, evpi_cap, decay_factor, reallocation_gain, net_value, net_value_p5, net_value_p95, prob_net_positive, breakeven_horizon_weeks, horizon_weeks, half_life_weeks, margin_per_kpi, assumptions=<factory>, warnings=<factory>)

mmm_framework.planning.compute_experiment_net_value(*, channel, evoi_kpi_units, evpi_kpi_units=None, kpi_delta_draws=None, spend_delta=0.0, margin_per_kpi=None, response_horizon_weeks=26, half_life_weeks=None, discount_rate_annual=0.0, model_anchored=False, opportunity_cost_result=None)

Net the reallocation gain against the test’s short-term loss.

Parameters:

• evoi_kpi_units (float | None) – the experiment’s EVOI (KPI-contribution units) — pass the realized-precision value from the anchor loopback when available and set model_anchored=True.

• evpi_kpi_units (float | None) – the EVPI ceiling; the gain is capped at it.

• kpi_delta_draws (ndarray | None) – per-draw signed window KPI delta of running the test (from compute_opportunity_cost(..., return_draws=True)). Gives the net-value distribution + P(net > 0); omit for point-only.

• spend_delta (float) – signed $ spend change over the window (negative = holdout savings).

• margin_per_kpi (float | None) – $ per KPI unit (the opportunity-cost module’s resolved value). None → KPI-units basis, spend excluded.

• half_life_weeks (float | None) – information half-life (eig.information_half_life); None disables the decay haircut.

• opportunity_cost_result (Any) – an OpportunityCostResult to pull draws / spend / margin / half-life defaults from (fields explicitly passed win).

Return type:

ExperimentNetValue

mmm_framework.planning.model_anchored_effect(mmm, design, *, max_draws=100, random_seed=42, hdi_prob=0.9, contrib_bau=None, contrib_exp=None)

The model-implied effect the design would produce, as a ROAS posterior.

Returns a JSON-safe dict with the incremental-ROAS draws (the estimand the experiment measures), its median/HDI, the expected incremental KPI, the average ROAS at current spend (context), and an extrapolation flag.

contrib_bau/contrib_exp let a caller pass the (shared) posterior passes so the same BAU/experiment evaluation is not recomputed here and in compute_opportunity_cost.

Return type:

dict[str, Any]

mmm_framework.planning.powered_to_detect(effect, power_curve, duration, se_roas, *, overpowered_ratio=2.0)

Is the design powered to detect the model’s expected effect?

assurance is the SIGNED two-sided power averaged over the effect posterior — mean_d[Phi(eff_d/se - z) + Phi(-eff_d/se - z)] — so a null channel scores ~alpha rather than being rewarded for posterior width (ANCHOR-3). prob_detectable (the MDE indicator) leads the verdict.

Return type:

dict[str, Any]

mmm_framework.planning.realized_sigma_exp_for_anchor(incremental_roas_draws, se_roas, *, roi_floor=0.05, floor=1e-06, rel_lower=0.05, rel_upper=50.0)

(sigma_exp, incremental_roas_draws) for the EIG/EVOI loopback.

sigma_exp is the design’s realized ROAS SE, clamped so a flat-placebo se_roas -> 0 can’t return an absurd EIG and an exploded SE can’t silently zero a strong design’s EIG (ANCHOR-4). It pairs with incremental_roas_draws — the SAME estimand the experiment measures, not the average-ROAS posterior the priority grid uses by default (ANCHOR-1).

Return type:

tuple[float, ndarray]

mmm_framework.planning.build_sim_panel(dataset_path, kpi, channel)

Return type:

SimPanel

mmm_framework.planning.methodology_leaderboard(dataset_path, kpi, channel, *, mmm=None, design=None, duration=8, alpha=0.05, target_mde_roas=None, spend_delta_window=0.0, expected_effect_total=None, fixed_pct_of_baseline=0.1, max_aa_windows=200, max_ab_windows=120, seed=42)

Run A/A + A/B for every estimator the data supports on ONE fixed assignment, and rank them by validity → power → cost.

Return type:

dict[str, Any]

mmm_framework.planning.run_aa_simulation(panel, estimator, assignment, *, duration, alpha=0.05, max_windows=200, seed=42, name=None)

Return type:

AAResult

mmm_framework.planning.run_ab_simulation(panel, estimator, assignment, injector, *, duration, aa_result, effect_grid=(0.25, 0.5, 0.75, 1.0, 1.5, 2.0), spend_delta_window=0.0, estimand_scale=1.0, max_windows=120, seed=42, name=None)

Return type:

ABResult

mmm_framework.planning.cooldown_weeks(mmm, channel, *, threshold=0.05, min_weeks=1, max_weeks=26, default_weeks=4)

Weeks for the channel’s carryover to wash out below threshold of the impulse, from the fitted geometric-adstock decay alpha.

After a holdout (or between flights) the channel’s effect decays as alpha**k; the cool-down is the smallest k with alpha**k < threshold — the time before the treated cells are back to true BAU and a post-test (or next-flight) reading is clean. It also sets the minimum flighting block length (a block shorter than the memory smears the contrast).

Return type:

dict[str, Any]

mmm_framework.planning.evaluate_experiment_grid(mmm, dataset_path, kpi, channel, *, duration_min=4, duration_max=12, intensity_min=50.0, intensity_max=100.0, durations=None, scaling_intensities=None, include_holdout=True, footprints=('full', 'half'), amplitudes=None, margin=None, price=None, kpi_kind='revenue', power_target=0.8, net_value_axis=True, response_horizon_weeks=26, max_draws=80, random_seed=42)

Evaluate a grid of candidate designs on FOUR objectives, all lower-better: MDE, power shortfall below power_target (default 80%), short-term cost, and duration.

When a margin is known and net_value_axis (default), the cost objective is upgraded to the net value of testing: each candidate’s EVOI (priced by the Gaussian surrogate at its own design precision, from ONE anchored preposterior-MC EVOI per grid) is decayed over response_horizon_weeks, capped at EVPI, converted to $, and netted against the candidate’s signed short-term test loss — the Pareto axis becomes −net_value and every candidate carries net_value/evoi_kpi/reallocation_gain. If the anchor cannot compute (no draw pairing, degenerate posterior), the axis degrades to the existing net-$ downside.

The design space is bounded by RANGES the caller controls: durations in [duration_min, duration_max] weeks and signed spend variations in [intensity_min, intensity_max] % (e.g. -100 go dark … +150 scale up; intensities at -100% render as a go-dark holdout). The optimizer auto-samples a few points within each range; pass explicit durations / scaling_intensities to override the sampling.

Reuses one pure-pandas power-curve call per (footprint, intensity) for the MDE of every duration, and one shared BAU posterior pass for the whole grid’s opportunity cost. Power is the model-anchored probability each design detects the expected effect (no extra posterior passes).

Return type:

dict[str, Any]

mmm_framework.planning.pareto_front(cands)

Indices of the non-dominated designs over (MDE, power shortfall, cost, duration). Designs with a non-finite objective can never be on the front.

Return type:

list[int]

mmm_framework.planning.suggest_experiment(mmm, dataset_path, kpi, channel, *, margin=None, price=None, kpi_kind='revenue', duration_min=4, duration_max=12, intensity_min=50.0, intensity_max=100.0, durations=None, scaling_intensities=None, include_holdout=True, footprints=('full', 'half'), power_target=0.8, net_value_axis=True, response_horizon_weeks=26, max_draws=80, random_seed=42)

Recommend a runnable experiment setup for channel and return the full Pareto front (MDE × power shortfall below power_target × short-term cost × duration), over a design space bounded by the duration and spend-variation ranges. The recommended design carries its test/control groups (or flighting schedule), duration, intensity, and the adstock-derived cool-down.

With a known margin the cost axis is the net value of testing (see evaluate_experiment_grid()) — gain from the learning minus the short-term loss — so the front trades precision and power against the dollars the test actually creates or destroys.

Return type:

dict[str, Any]

Eig

Expected information gain (EIG) for channel experiments, plus the information-decay model that schedules re-experimentation.

EIG measures how much an experiment of a given precision would shrink the posterior over a channel’s ROI. The Gaussian closed form follows from the conjugate update: with prior N(mu, sigma_k^2) and an experiment observing y ~ N(roi, sigma_exp^2), the posterior variance is (1/sigma_k^2 + 1/sigma_exp^2)^-1 and the entropy reduction is

EIG = 0.5 * ln(1 + sigma_k^2 / sigma_exp^2) [nats]

sigma_exp must be DESIGN-GROUNDED (what precision the experiment can actually achieve, given geo footprint / design type) — if it were derived from sigma_k itself, every channel would have the same EIG and the ranking would collapse.

Information decay: experimental evidence has a shelf life. The effective uncertainty grows as sigma_eff^2(t) = sigma_post^2 * exp(lambda * t) with lambda = ln(2) / half_life; when the EIG of a fresh experiment (computed from the decayed sigma) crosses an operational threshold, the channel is due for a re-test.

Import-light (numpy only) so it can run inside the session kernels.

mmm_framework.planning.eig.sigma_exp_for_design(design_type, roi_median, *, floor=0.05)

Design-grounded experiment sd on the ROI scale.

sigma_exp = precision(design) * max(|roi_median|, floor) — proportional to the channel’s ROI scale (a relative-precision design) but NEVER to its posterior uncertainty.

Return type:

float

mmm_framework.planning.eig.eig_gaussian(sigma_k, sigma_exp)

Closed-form EIG (nats) for a Gaussian prior and Gaussian experiment.

Return type:

float

mmm_framework.planning.eig.use_gaussian(roi_draws)

Cheap normality gate: closed form unless the posterior is visibly skewed or heavy-tailed (|skew| >= 0.5 or excess kurtosis >= 1).

Return type:

bool

mmm_framework.planning.eig.eig_monte_carlo(roi_draws, sigma_exp, *, n_outcomes=64, rng=None)

Monte Carlo EIG for non-Gaussian ROI posteriors.

Prior-predictive outcomes y_j = roi[d_j] + sigma_exp * z_j; for each, the posterior over draws is the importance reweighting w_d ∝ N(y_j | roi_d, sigma_exp) and entropy is approximated by the Gaussian proxy 0.5 * ln(2*pi*e * Var) on the (weighted) draws — exact in the Gaussian limit, robust to skew where the closed form overstates.

Return type:

float

mmm_framework.planning.eig.information_half_life(channel, overrides=None)

How fast evidence about a channel goes stale, in weeks.

This is an information-decay half-life, not a media-carryover one. It reads nothing from the model — it is a keyword table over the channel name (see DEFAULT_HALF_LIVES_WEEKS) answering “when is this channel’s last read old enough to re-test?”.

Do not confuse it with mmm_framework.transforms.carryover.carryover_half_life(), which is per-draw, read from the fitted kernel, and answers “how long does a exposure keep working?”. The two legitimately disagree by ~10x, which is why this one is no longer called channel_half_life: planning exported two similarly-named half-lives for the same channel meaning different things.

Return type:

float

mmm_framework.planning.eig.channel_half_life(channel, overrides=None)

Deprecated alias for information_half_life().

The old name did not say which half-life it meant, and planning sat one import away from the model-derived carryover half-life.

Return type:

float

mmm_framework.planning.eig.decayed_sigma(sigma_post, weeks_elapsed, half_life_weeks)

Effective sd after weeks_elapsed: sigma_eff^2(t) = sigma_post^2 * exp(lambda * t), lambda = ln2/half-life — so sigma_eff(t) = sigma_post * exp(0.5 * lambda * t).

Return type:

float

mmm_framework.planning.eig.reexperiment_due(sigma_post, weeks_elapsed, half_life_weeks, sigma_exp, *, threshold_nats=0.15, min_age_weeks=13.0)

(due, current_eig): the EIG of a fresh experiment given the decayed effective uncertainty, and whether it crosses the operational threshold. Evidence younger than min_age_weeks is never due (freshness floor).

Return type:

tuple[bool, float]

Evoi

Expected value of information (EVOI) for the budget decision.

EIG says how much an experiment teaches; EVOI says how much that learning is WORTH — the expected improvement in the budget allocation’s outcome from deciding with the experiment’s information instead of without it (preposterior analysis / EVSI for the allocation decision):

EVOI_k = E_y[ max_a E_{theta|y} f(a, theta) ] - max_a E_theta f(a, theta)

with the outer expectation over the prior predictive of the experiment outcome y ~ N(roi_k, sigma_exp^2). EVPI (perfect information about everything) is the upper bound: EVPI = E_d[f_d(a_d*)] - E_d[f_d(a*)].

Units: KPI contribution over the response-curve window — the same units as optimize_budget’s expected_uplift.

Reuses the greedy allocator and per-draw machinery from planning.budget. Import-light (numpy only) so it can run inside the session kernels.

class mmm_framework.planning.evoi.EvoiResult(v_current, evpi, n_draws)

Bases: object

Portfolio-level value-of-information summary.

v_current: float

evpi: float

n_draws: int

__init__(v_current, evpi, n_draws)

mmm_framework.planning.evoi.compute_evpi(curves, *, total_budget=None, lo_spend=None, hi_spend=None, n_steps=400, per_draw_alloc=None, optimal_alloc=None)

V_current and EVPI from the response curves.

Pass per_draw_alloc/optimal_alloc from a BudgetOptimizationResult to reuse its D greedy optimizations instead of recomputing them.

Return type:

EvoiResult

mmm_framework.planning.evoi.compute_evoi_for_channel(curves, channel_idx, roi_draws, sigma_exp, *, optimal_alloc=None, n_outcomes=48, n_steps=400, total_budget=None, lo_spend=None, hi_spend=None, outcome_draws=None, rng=None)

EVOI of an experiment on one channel, by paired preposterior Monte Carlo.

For each simulated outcome y_j = roi[d_j] + sigma_exp * z_j: reweight the posterior draws by w_d ∝ N(y_j | roi_d, sigma_exp), re-optimize the allocation on the REWEIGHTED mean curves, and score the per-outcome gain :rtype: float

gain_j = V_j(a_j*) - V_j(a*) with V_j(a) = E_{theta|y_j}[f(a, theta)]

against the no-experiment allocation a* under the SAME reweighted posterior (the paired estimator: identical estimand by the tower property, but the common-mode outcome noise cancels, so EVOI cannot go negative from sampling luck — each gain_j >= 0 up to allocator discretization, enforced by a per-term floor).

outcome_draws = (d_idx, z) lets the caller share common random numbers across channels so near-ties don’t rank-flip on MC noise. The caller should cap the result at EVPI.

mmm_framework.planning.evoi.preposterior_sd_ratio(tau, sigma_exp)

sd of the preposterior mean as a FRACTION of the prior sd tau: sqrt(tau^2 / (tau^2 + sigma_exp^2)) — 1 for a perfect experiment (sigma_exp → 0), → 0 for an uninformative one.

Return type:

float

mmm_framework.planning.evoi.surrogate_evoi(evoi_ref, sigma_ref, sigma_new, tau, *, evpi=None)

Single-anchor fallback: EVOI scaled by the preposterior-sd ratio s(sigma_new)/s(sigma_ref) (the delta = 0 special case of the fitted surrogate — decision boundary at the prior mean). Prefer fit_evoi_surrogate() with two anchors when you can afford the second MC evaluation; the single-anchor form decays too slowly for weak designs. Capped at evpi when given; 0 when the anchor or prior is degenerate.

Return type:

float

class mmm_framework.planning.evoi.EvoiSurrogate(k, delta, tau, sigma_anchors=())

Bases: object

Calibrated Gaussian EVOI surrogate sigma → k·s(sigma)·Psi(delta/s).

tau is the channel’s prior (pre-experiment) ROI sd; k and delta are fitted to the MC anchors. Call it with any design precision to price that design’s EVOI without another preposterior MC. Accurate BETWEEN the anchor sigmas; extrapolation beyond the weak anchor under-estimates.

k: float

delta: float

tau: float

sigma_anchors: tuple[float, ...] = ()

__init__(k, delta, tau, sigma_anchors=())

mmm_framework.planning.evoi.fit_evoi_surrogate(tau, anchors)

Fit (k, delta) of EvoiSurrogate to MC anchors [(sigma_exp, evoi_mc), ...].

Two usable anchors (distinct sigmas, both positive EVOI) pin both parameters by bisection on the anchor ratio; one usable anchor degrades to delta = 0 (the surrogate_evoi() scaling). Returns None when tau is degenerate or no anchor has positive EVOI.

Return type:

EvoiSurrogate | None

Priority

The EIG/EVOI experiment-priority grid.

Combines EIG (what an experiment would teach — planning.eig) with EVOI (what that learning is worth to the budget decision — planning.evoi) into a per-channel priority grid with the 2×2 quadrant classification:

high EIG, high EVOI → test_now (decision-critical uncertainty) high EIG, low EVOI → learn_cheaply (informative, but the decision is robust) low EIG, high EVOI → monitor (high stakes, already precise — watch drift) low EIG, low EVOI → deprioritize

Composite score: priority = sqrt(eig_norm * evoi_norm) — a normalized geometric mean, so a channel cannot win on one axis alone.

When the experiment registry supplies per-channel evidence dates, information decay (planning.eig) yields the decayed EIG and the re-test trigger.

Import-light (numpy/pandas only) so it can run inside the session kernels.

class mmm_framework.planning.priority.ChannelPriority(channel, spend, spend_share, roi_mean, roi_sd, roi_hdi_low, roi_hdi_high, sigma_exp, design_type, eig, eig_method, evoi, evoi_pct_budget, evpi_share, priority, quadrant, weeks_since_evidence=None, eig_decayed=None, retest_due=False)

Bases: object

One row of the priority grid (JSON-safe via .to_dict()).

channel: str

spend: float

spend_share: float

roi_mean: float

roi_sd: float

roi_hdi_low: float

roi_hdi_high: float

sigma_exp: float

design_type: str

eig: float

eig_method: str

evoi: float

evoi_pct_budget: float

evpi_share: float

priority: float

quadrant: str

weeks_since_evidence: float | None = None

eig_decayed: float | None = None

retest_due: bool = False

to_dict()

Return type:

dict[str, Any]

__init__(channel, spend, spend_share, roi_mean, roi_sd, roi_hdi_low, roi_hdi_high, sigma_exp, design_type, eig, eig_method, evoi, evoi_pct_budget, evpi_share, priority, quadrant, weeks_since_evidence=None, eig_decayed=None, retest_due=False)

mmm_framework.planning.priority.compute_experiment_priorities(mmm=None, *, curves=None, optimization=None, evidence=None, as_of=None, design_type=None, sigma_exp_overrides=None, roi_draws_overrides=None, half_life_overrides=None, eig_threshold=None, evoi_threshold=None, retest_threshold_nats=0.15, n_outcomes=48, n_steps=400, max_draws=200, random_seed=42)

Per-channel EIG/EVOI priority grid + portfolio summary.

Parameters:

• evidence (dict[str, dict] | None) – {channel: {"end_date": ISO, ...}} — the newest calibrated experiment per channel (sessions.latest_calibrated_evidence). Drives information decay and the re-test trigger.

• as_of (str | None) – ISO date for the decay clock (default: today).

• design_type (str | None) – assumed experiment design for sigma_exp; default geo_holdout when the model has geo structure, else national_pulse.

• evoi_threshold (eig_threshold /) – absolute quadrant splits; default is the median of each axis across channels.

Return type:

tuple[list[ChannelPriority], dict[str, Any]]

Returns (grid, portfolio) where portfolio carries v_current, EVPI, total budget, draws, and the thresholds used.

Design

Experiment design engine: turn “test this channel” into a runnable design.

Three design families, chosen by the data’s granularity:

• Randomized geo lift (DMA/geo panels): model-structured matched pairs — each geo’s KPI is residualized on the MMM’s own baseline structure (trend + yearly seasonality + the channel’s spend response) and pairs are matched on RESIDUAL co-movement plus a standardized covariate distance on the model’s terms, solved as an optimal (blossom) matching. Treatment is RANDOMIZED within pairs (the randomization is what buys unconfoundedness). Power is a DiD analysis whose level is CALIBRATED against a placebo simulation (the estimator run over every historical window), with a covariate balance table.

• Matched-market DiD (pseudo-experimental): the same matching + power machinery without randomization, for when treatment geos are dictated by the business — the placebo distribution calibrates how big a “lift” the pre-period produces by chance, and the caveat is explicit.

• Randomized flighting (national data, or any single series): a budget-neutral, block-randomized on/off spend schedule that manufactures the exogenous variance a collinear history never provides — with the induced identification gain and an on/off-contrast power estimate.

All functions are pandas/numpy only (kernel-safe) and work straight off an MFF csv — no fitted model required, so the endpoint stays fast.

mmm_framework.planning.design.load_design_frame(dataset_path, kpi, channel)

Pivot an MFF csv into the matrices the designs need.

Returns {kpi_wide, spend_wide, kpi_national, spend_national, periods, geos} where the *_wide frames are weeks × geos (None for national data: no Geography column, or a single geography).

Return type:

dict[str, Any]

mmm_framework.planning.design.residualize_geo_panel(kpi_wide, spend_wide=None)

Remove the model-explained structure from each geo’s KPI.

Geo-by-geo OLS of KPI on [intercept, trend, yearly Fourier(2), channel spend (standardized, when available)] — the deterministic skeleton of the MMM applied per geo. Returns {"residuals": DataFrame weeks×geos, "features": DataFrame geos×feature} where features live on the model’s terms: level (log mean), trend_slope, seasonal_amplitude (Fourier coefficient norm), residual_sd, spend_share.

Return type:

dict[str, Any]

mmm_framework.planning.design.matched_pairs(kpi_wide, n_pairs=None, spend_wide=None)

Model-structured matched pairs, best-match first.

Each pair carries correlation (raw KPI — what naive matching would have scored), residual_correlation (co-movement AFTER removing trend/ seasonality/spend — what the DiD’s precision actually depends on), size_ratio, and the matching distance.

Return type:

list[dict]

mmm_framework.planning.design.geo_lift_design(dataset_path, kpi, channel, *, design='holdout', intensity_pct=50.0, n_pairs=None, duration=8, durations=(4, 6, 8, 10, 12, 16), randomize=True, seed=42)

Design a geo lift study (or a matched-market DiD when randomize=False).

Pipeline — robust by construction, on the model’s own view of history: :rtype: dict[str, Any]

1. Residualize each geo’s KPI on the MMM’s baseline structure (trend + yearly seasonality + the channel’s spend response).

2. Match on residual co-movement + standardized model-feature distance; solve the pairing globally (minimum-weight perfect matching).

3. Randomize treatment within each pair (or largest-geo-treated for the observational matched-market variant).

4. Power: the analytic DiD curve is CALIBRATED against a placebo simulation — the actual estimator run over historical windows — so the reported SE/MDE reflects how this design behaves on this data, not just a formula. Covariate balance on the model’s features is reported.

mmm_framework.planning.design.flighting_estimand_ses(mults, mean_spend, sigma_y, *, x0=None)

SEs of a flighting design’s three estimands from its spend levels.

Models the local channel response as y_t = g(x_t) + noise and fits g with the design’s spend regressors (a quadratic [1, x, x²] when ≥3 distinct levels are present, else a line). From the OLS covariance σ²(XᵀX)⁻¹ it reads off the SE of: :rtype: dict[str, Any] | None

• the LEVEL g(x₀) at the operating point → the contribution SE (and, divided by x₀, the average ROAS SE — the same estimate rescaled);

• the SLOPE g'(x₀) → the mROAS SE.

The marginal (tangent) g'(x₀) is only identifiable with ≥3 distinct levels; a binary on/off yields a secant slope (se_mroas is still returned but mroas_identified=False). Returns None for a degenerate design (too few / near-collinear levels, non-finite inputs, ill-conditioned fit) so a meaningless SE never reaches the power math. x0 overrides the operating point (pass the MODEL’s per-week spend so the ROAS/contribution posteriors and these SEs share one basis).

mmm_framework.planning.design.flighting_design(dataset_path, kpi, channel, *, amplitude_pct=50.0, block_weeks=2, duration=12, budget_neutral=True, levels=None, x0=None, seed=42)

Block-randomized flighting schedule for a national series.

Historical spend is usually smooth and co-moves with demand — the model cannot tell the channel from the season. A budget-neutral pulse schedule (randomized block order) manufactures exogenous variance: identification the data finally pays for. Block length should be >= the channel’s adstock memory so carryover doesn’t smear the contrast.

levels (spend multipliers, e.g. (0.5, 1.0, 1.5, 2.0)) gives a MULTI-LEVEL schedule — ≥3 distinct levels let the experiment trace the response CURVE (so the marginal ROAS / saturation is identified), not just a single on/off contrast. Default (None) is the binary high/low pulse at ±amplitude_pct.

Return type:

dict[str, Any]

mmm_framework.planning.design.design_options(dataset_path, kpi, channel)

What designs the data supports: geo designs need >= 4 geographies.

Return type:

dict[str, Any]

mmm_framework.planning.design.design_experiment(dataset_path, kpi, channel, *, design_key=None, method=None, **kwargs)

Dispatch to the right design family (auto-recommended when omitted).

method names the analysis methodology (synthetic_control / tbr / gbr / did_mmt). It selects which estimator the read-out and economics use and, when design_key is not given, infers the design family. The chosen method’s metadata is attached to the returned design dict.

Return type:

dict[str, Any]

Design Anchor

Anchor a design’s power analysis to the fitted model’s expected effect.

The pure-data designer (planning.design) reports an MDE — the smallest ROAS effect a test could detect — but says nothing about whether the effect we actually EXPECT clears that bar. This module closes that gap using the model:

• model_anchored_effect perturbs ONLY the treated geo × test-window rows at the design’s full intensity and reads the incremental KPI straight off the model’s deterministic channel_contributions (paired draws, F1). It does NOT interpolate the global response curve — that scales every geo uniformly and mis-states a treated subset under heterogeneous saturation (F8/ANCHOR-2).

• powered_to_detect compares the design’s MDE to that model-implied effect posterior → a powered / underpowered / overpowered / inconclusive verdict, a signed two-sided assurance, and the duration that would reach power.

• realized_sigma_exp_for_anchor turns the design’s realized ROAS SE into the sigma_exp the EIG/EVOI engine expects, so the priority grid can use the ACTUAL design precision instead of a generic table.

numpy/pandas only — kernel-safe. Reuses the perturbation mechanics in opportunity_cost so the anchor and the opportunity cost see the same rows.

mmm_framework.planning.design_anchor.model_anchored_effect(mmm, design, *, max_draws=100, random_seed=42, hdi_prob=0.9, contrib_bau=None, contrib_exp=None)

The model-implied effect the design would produce, as a ROAS posterior.

Returns a JSON-safe dict with the incremental-ROAS draws (the estimand the experiment measures), its median/HDI, the expected incremental KPI, the average ROAS at current spend (context), and an extrapolation flag.

contrib_bau/contrib_exp let a caller pass the (shared) posterior passes so the same BAU/experiment evaluation is not recomputed here and in compute_opportunity_cost.

Return type:

dict[str, Any]

mmm_framework.planning.design_anchor.powered_to_detect(effect, power_curve, duration, se_roas, *, overpowered_ratio=2.0)

Is the design powered to detect the model’s expected effect?

assurance is the SIGNED two-sided power averaged over the effect posterior — mean_d[Phi(eff_d/se - z) + Phi(-eff_d/se - z)] — so a null channel scores ~alpha rather than being rewarded for posterior width (ANCHOR-3). prob_detectable (the MDE indicator) leads the verdict.

Return type:

dict[str, Any]

mmm_framework.planning.design_anchor.realized_sigma_exp_for_anchor(incremental_roas_draws, se_roas, *, roi_floor=0.05, floor=1e-06, rel_lower=0.05, rel_upper=50.0)

(sigma_exp, incremental_roas_draws) for the EIG/EVOI loopback.

sigma_exp is the design’s realized ROAS SE, clamped so a flat-placebo se_roas -> 0 can’t return an absurd EIG and an exploded SE can’t silently zero a strong design’s EIG (ANCHOR-4). It pairs with incremental_roas_draws — the SAME estimand the experiment measures, not the average-ROAS posterior the priority grid uses by default (ANCHOR-1).

Return type:

tuple[float, ndarray]

Opportunity Cost

Short-term risk / opportunity cost of running an experiment, from the model.

A design perturbs the tested channel’s spend away from business-as-usual (BAU) over the test window — a holdout goes dark (-100%) in the treated cells, a scaling test lifts spend by +X%, a national flighting schedule pulses on/off. That perturbation changes the KPI during the test, and clients want to know the short-term cost of buying the measurement: how much KPI (and money) do we forgo by deviating from BAU?

We answer it from the fitted model, with posterior uncertainty:

1. Build X_experiment — a copy of the training media matrix with ONLY the treated geo × test-window rows perturbed (the rest is byte-identical BAU).

2. Read the counterfactual per-channel contributions for BOTH the BAU and the experiment media via mmm.sample_channel_contributions. That node is a PyMC Deterministic (no observation noise), and identical max_draws gives identical posterior thinning, so the two passes are draw-paired exactly — the per-draw KPI delta is exact, not a noisy difference (F1).

3. Summarize the per-draw KPI delta into expected / worst-case loss, the probability of a loss, the deterministic spend delta, an optional net-$ business impact (when a margin/price is supplied), and a learning-vs-cost ratio against the experiment’s EVOI.

Sign conventions are load-bearing and unit-tested:

• kpi_delta is SIGNED (negative = lost KPI). forgone_kpi is the non-negative downside max(0, -kpi_delta).

• spend_delta is SIGNED and computed INTERNALLY from the perturbed vs BAU matrices (negative for a holdout = money saved). We NEVER read design['weekly_spend_delta'] — it is an absolute magnitude and would invert the holdout’s net (F6).

• net_profit_impact = margin * kpi_delta - spend_delta is SIGNED; opportunity_cost_dollar = max(0, -net_profit_impact) is the non-negative cost the client asked for.

numpy/pandas only — kernel-safe (runs inside the sandboxed session kernels).

class mmm_framework.planning.opportunity_cost.OpportunityCostResult(channel, kpi, design_key, design_type, duration_requested, duration_effective, n_treated_cells, n_test_rows, n_draws, carryover_basis, expected_kpi_delta, kpi_delta_median, kpi_delta_p5, kpi_delta_p95, kpi_delta_with_carryover_median, forgone_kpi_median, forgone_kpi_p95, prob_kpi_loss, pct_of_window_kpi, spend_delta, abs_spend_change, spend_at_risk, margin_per_kpi, margin_source, kpi_kind, net_profit_impact_median, net_profit_impact_p5, net_profit_impact_p95, opportunity_cost_dollar_median, opportunity_cost_dollar_p95, prob_net_loss, prob_loss_over_threshold, loss_threshold, evoi_kpi_units, evoi_per_week, cost_per_week, learning_to_cost_ratio, learning_to_cost_basis, response_horizon_weeks, low_information, extrapolation_warning, warnings=<factory>, notes=<factory>, draws=None)

Bases: object

Per-draw short-term risk of a design, summarized. JSON-safe via to_dict.

channel: str

kpi: str

design_key: str

design_type: str

duration_requested: int

duration_effective: int

n_treated_cells: int

n_test_rows: int

n_draws: int

carryover_basis: str

expected_kpi_delta: float

kpi_delta_median: float

kpi_delta_p5: float

kpi_delta_p95: float

kpi_delta_with_carryover_median: float

forgone_kpi_median: float

forgone_kpi_p95: float

prob_kpi_loss: float

pct_of_window_kpi: float | None

spend_delta: float

abs_spend_change: float

spend_at_risk: float

margin_per_kpi: float | None

margin_source: str

kpi_kind: str

net_profit_impact_median: float | None

net_profit_impact_p5: float | None

net_profit_impact_p95: float | None

opportunity_cost_dollar_median: float | None

opportunity_cost_dollar_p95: float | None

prob_net_loss: float | None

prob_loss_over_threshold: float | None

loss_threshold: float | None

evoi_kpi_units: float | None

evoi_per_week: float | None

cost_per_week: float | None

learning_to_cost_ratio: float | None

learning_to_cost_basis: str

response_horizon_weeks: int | None

low_information: bool

extrapolation_warning: bool

warnings: list[str]

notes: list[str]

draws: dict | None = None

to_dict()

Return type:

dict[str, Any]

__init__(channel, kpi, design_key, design_type, duration_requested, duration_effective, n_treated_cells, n_test_rows, n_draws, carryover_basis, expected_kpi_delta, kpi_delta_median, kpi_delta_p5, kpi_delta_p95, kpi_delta_with_carryover_median, forgone_kpi_median, forgone_kpi_p95, prob_kpi_loss, pct_of_window_kpi, spend_delta, abs_spend_change, spend_at_risk, margin_per_kpi, margin_source, kpi_kind, net_profit_impact_median, net_profit_impact_p5, net_profit_impact_p95, opportunity_cost_dollar_median, opportunity_cost_dollar_p95, prob_net_loss, prob_loss_over_threshold, loss_threshold, evoi_kpi_units, evoi_per_week, cost_per_week, learning_to_cost_ratio, learning_to_cost_basis, response_horizon_weeks, low_information, extrapolation_warning, warnings=<factory>, notes=<factory>, draws=None)

mmm_framework.planning.opportunity_cost.build_experiment_media(mmm, design, *, treated_mask, window_codes)

(X_experiment, channel_idx, n_test_rows).

A float64 copy of mmm.X_media_raw with the tested channel column scaled on treated_mask rows: holdout → max(0, 1+intensity/100) (=0 at -100), scaling → 1+intensity/100, national_flighting → the per-period multiplier from design['schedule'] mapped through the window’s period codes. BAU everywhere else.

Return type:

tuple[ndarray, int, int]

mmm_framework.planning.opportunity_cost.compute_opportunity_cost(mmm, design, *, margin_per_kpi=None, kpi_kind='revenue', price=None, preferences=None, branding=None, loss_threshold=None, evoi_kpi_units=None, response_horizon_weeks=None, max_draws=200, random_seed=42, contrib_bau=None, contrib_exp=None, return_draws=False)

Short-term risk of running design on the fitted model mmm.

Parameters:

• design (dict) – a planning.design.design_experiment payload (carries channel, kpi, design_key, intensity_pct / schedule, treatment_geos, duration).

• kpi_kind (margin_per_kpi / price /) – net-$ valuation inputs. revenue multiplies KPI by margin; units by margin*price; other reports KPI units only. Margin also auto-resolves from branding / preferences economics blocks.

• evoi_kpi_units (float | None) – the experiment’s EVOI (planning.evoi, KPI-contribution units) for the learning-vs-cost ratio. Pass the EVPI-capped value.

• response_horizon_weeks (int | None) – denominator for evoi_per_week (default mmm.n_periods).

Return type:

OpportunityCostResult

Simulation

A/A and A/B simulation on historical data — to compare experiment methodologies on their REAL false-positive rate, power, MDE, and cost.

The pure-data designer reports an analytic MDE from a first-difference noise model. But real KPI series are autocorrelated, so that analytic standard error understates the spread of the estimator and the nominal false-positive rate inflates — sometimes several-fold. The only honest way to know how a methodology behaves on THIS data is to run it on the data:

• A/A — slide the design’s estimator over historical windows where no treatment was applied. The estimator should return ~0; the spread of its estimates is the true null distribution. We report the empirical false-positive rate at the analytic decision rule (the inflation diagnostic) AND a design-calibrated critical value — the empirical 1-alpha quantile of |estimate| — that restores the nominal size. An estimator whose A/A FPR far exceeds alpha is INVALID for this data regardless of its nominal power.

• A/B — inject a KNOWN lift (the model’s predicted incremental KPI when a model is available, otherwise a fixed magnitude) onto the treatment arm of real historical windows and re-run the estimator. Sweeping a grid of effect sizes gives an EMPIRICAL power curve and MDE that respect the real noise structure. Power is scored at the calibrated critical value, so it is power at the correct size — not a number inflated by the same SE bug.

methodology_leaderboard runs A/A + A/B for several estimators on the SAME fixed assignment (so the comparison isolates the estimator, not the randomization draw) and ranks them: valid (FPR within tolerance), powered (empirical MDE below the expected effect), and cheap (opportunity cost).

numpy/pandas only — kernel-safe. Reuses planning.design for the panel, matching, and SE helpers.

class mmm_framework.planning.simulation.Window(pre_slice, test_slice, t_pre, t_test)

Bases: object

pre_slice: slice

test_slice: slice

t_pre: int

t_test: int

__init__(pre_slice, test_slice, t_pre, t_test)

class mmm_framework.planning.simulation.Assignment(kind, treatment_geos=(), control_geos=(), pairs=(), schedule_mult=(), seed=42)

Bases: object

A FROZEN treatment design, reused across every estimator so the methodology comparison isolates the estimator (SIM-6).

kind: str

treatment_geos: tuple[str, ...] = ()

control_geos: tuple[str, ...] = ()

pairs: tuple[tuple[str, str], ...] = ()

schedule_mult: tuple[float, ...] = ()

seed: int = 42

__init__(kind, treatment_geos=(), control_geos=(), pairs=(), schedule_mult=(), seed=42)

class mmm_framework.planning.simulation.EstimatorResult(estimate, se, spend_delta, n_eff)

Bases: object

estimate: float

se: float | None

spend_delta: float

n_eff: int | None

__init__(estimate, se, spend_delta, n_eff)

class mmm_framework.planning.simulation.AAResult(estimator, n_windows, n_eff_windows, null_mean, null_sd, crit_value, fpr_at_nominal, fpr_at_crit, fpr_ci, fpr_inflated, fpr_tolerance, null_method, status)

Bases: object

estimator: str

n_windows: int

n_eff_windows: int

null_mean: float

null_sd: float

crit_value: float

fpr_at_nominal: float

fpr_at_crit: float

fpr_ci: tuple[float, float]

fpr_inflated: bool

fpr_tolerance: float

null_method: str

status: str

__init__(estimator, n_windows, n_eff_windows, null_mean, null_sd, crit_value, fpr_at_nominal, fpr_at_crit, fpr_ci, fpr_inflated, fpr_tolerance, null_method, status)

class mmm_framework.planning.simulation.ABResult(estimator, per_effect, empirical_mde, empirical_mde_roas, mde_method, mde_ci, expected_effect, power_at_expected, null_sd, status)

Bases: object

estimator: str

per_effect: tuple[dict, ...]

empirical_mde: float | None

empirical_mde_roas: float | None

mde_method: str

mde_ci: tuple[float, float] | None

expected_effect: float

power_at_expected: float

null_sd: float

status: str

__init__(estimator, per_effect, empirical_mde, empirical_mde_roas, mde_method, mde_ci, expected_effect, power_at_expected, null_sd, status)

class mmm_framework.planning.simulation.SimPanel(kpi_wide, spend_wide, kpi_national, spend_national, residuals, periods, geos)

Bases: object

kpi_wide: DataFrame | None

spend_wide: DataFrame | None

kpi_national: Series

spend_national: Series

residuals: DataFrame | None

periods: list

geos: list

__init__(kpi_wide, spend_wide, kpi_national, spend_national, residuals, periods, geos)

mmm_framework.planning.simulation.build_sim_panel(dataset_path, kpi, channel)

Return type:

SimPanel

mmm_framework.planning.simulation.pooled_did_estimator(panel, assignment, window)

Return type:

EstimatorResult

mmm_framework.planning.simulation.per_pair_did_estimator(panel, assignment, window)

Return type:

EstimatorResult

mmm_framework.planning.simulation.regadj_geo_estimator(panel, assignment, window, *, ridge=1.0)

Synthetic-control style: fit control weights on the pre-period, score the test-window gap. Robust to between-geo scale differences the pooled DiD ignores.

Return type:

EstimatorResult

mmm_framework.planning.simulation.national_onoff_estimator(panel, assignment, window)

High-block vs low-block KPI contrast over the test window, using the design’s on/off labels. On un-injected history the contrast is null.

Return type:

EstimatorResult

mmm_framework.planning.simulation.build_geo_assignment(panel, *, n_pairs=None, seed=42)

Return type:

Assignment

mmm_framework.planning.simulation.build_national_assignment(*, duration, block_weeks=2, amplitude_pct=50.0, seed=42)

Return type:

Assignment

mmm_framework.planning.simulation.run_aa_simulation(panel, estimator, assignment, *, duration, alpha=0.05, max_windows=200, seed=42, name=None)

Return type:

AAResult

class mmm_framework.planning.simulation.LiftInjector(kind, expected_total, per_draw_total=None, spend_share=None, per_draw_total_list=<factory>)

Bases: object

Adds a known lift to the treatment arm of a window. expected_total is the lift magnitude at scale=1 (model-predicted when anchored). __call__ returns (true_total, injected_kpi_wide) for the geo path or (true_total, injected_kpi_national) for national.

kind: str

expected_total: float

per_draw_total: ndarray | None = None

spend_share: dict[str, float] | None = None

per_draw_total_list: list

inject_geo(panel, assignment, window, scale)

Return type:

tuple[float, DataFrame]

inject_national(panel, assignment, window, scale)

Return type:

tuple[float, Series]

__init__(kind, expected_total, per_draw_total=None, spend_share=None, per_draw_total_list=<factory>)

mmm_framework.planning.simulation.fixed_lift_injector(panel, assignment, *, duration, pct_of_baseline=None, kpi_units=None)

Model-free injector: lift = kpi_units or pct_of_baseline of the treatment arm’s mean window KPI. Distributed by spend share (clipped).

Return type:

LiftInjector

mmm_framework.planning.simulation.run_ab_simulation(panel, estimator, assignment, injector, *, duration, aa_result, effect_grid=(0.25, 0.5, 0.75, 1.0, 1.5, 2.0), spend_delta_window=0.0, estimand_scale=1.0, max_windows=120, seed=42, name=None)

Return type:

ABResult

mmm_framework.planning.simulation.methodology_leaderboard(dataset_path, kpi, channel, *, mmm=None, design=None, duration=8, alpha=0.05, target_mde_roas=None, spend_delta_window=0.0, expected_effect_total=None, fixed_pct_of_baseline=0.1, max_aa_windows=200, max_ab_windows=120, seed=42)

Run A/A + A/B for every estimator the data supports on ONE fixed assignment, and rank them by validity → power → cost.

Return type:

dict[str, Any]

Experiment Optimizer

Suggest an experiment setup, and Pareto-optimize the design space.

Given a fitted MMM and a channel, this turns “test this channel” into a recommended, runnable setup — design family, intensity, duration, test/control groups (or a flighting schedule), and a cool-down period derived from the channel’s fitted adstock — by exploring a grid of candidate designs and ranking them on FOUR objectives the client cares about:

• lowest MDE — the smallest ROAS effect the test can detect (precision),

• highest statistical power (as a shortfall below an 80% target) to detect the model’s expected effect — for flighting designs the power is computed for the ROAS, contribution AND marginal-ROAS estimands separately (a multi-level spend schedule is what makes the marginal / saturation curve identifiable),

• smallest short-term tradeoff — the opportunity cost of deviating from business-as-usual (forgone KPI, or net-$ downside when a margin is known),

• shortest duration — weeks in market.

These trade off against each other (a longer, bigger test detects more but costs more), so there is no single best design — there is a Pareto front of non-dominated designs. We compute that front, then recommend the design at its “knee” that is also powered to detect the model’s expected effect.

Efficiency: the MDE for every duration comes from one pure-pandas power-curve call per (footprint, intensity); the opportunity cost reuses a single shared BAU posterior pass across the whole grid (only the perturbed pass is per-config).

numpy/pandas only — kernel-safe. Reuses planning.design (designs + power), planning.opportunity_cost (short-term risk), and the fitted adstock.

mmm_framework.planning.experiment_optimizer.cooldown_weeks(mmm, channel, *, threshold=0.05, min_weeks=1, max_weeks=26, default_weeks=4)

Weeks for the channel’s carryover to wash out below threshold of the impulse, from the fitted geometric-adstock decay alpha.

After a holdout (or between flights) the channel’s effect decays as alpha**k; the cool-down is the smallest k with alpha**k < threshold — the time before the treated cells are back to true BAU and a post-test (or next-flight) reading is clean. It also sets the minimum flighting block length (a block shorter than the memory smears the contrast).

Return type:

dict[str, Any]

class mmm_framework.planning.experiment_optimizer.CandidateEval(index, design_key, mode, footprint, n_pairs, intensity_pct, duration, mde_roas, power_shortfall, tradeoff, tradeoff_basis, forgone_kpi_median, opportunity_cost_dollar_median, net_profit_impact_median, spend_at_risk, pct_of_window_kpi, duration_effective, powered, power=None, power_lower=None, power_target=0.8, power_breakdown=None, sigma_exp=None, evoi_kpi=None, reallocation_gain=None, net_value=None, net_value_basis=None, on_pareto=False, is_recommended=False, treatment_geos=<factory>, control_geos=<factory>, schedule=None, block_weeks=None, duration_requested=None, warnings=<factory>)

Bases: object

One evaluated experiment design with its three Pareto objectives + the runnable setup. JSON-safe via to_dict.

index: int

design_key: str

mode: str

footprint: str

n_pairs: int | None

intensity_pct: float

duration: int

mde_roas: float

power_shortfall: float

tradeoff: float

tradeoff_basis: str

forgone_kpi_median: float

opportunity_cost_dollar_median: float | None

net_profit_impact_median: float | None

spend_at_risk: float

pct_of_window_kpi: float | None

duration_effective: int

powered: bool

power: float | None = None

power_lower: float | None = None

power_target: float = 0.8

power_breakdown: dict | None = None

sigma_exp: float | None = None

evoi_kpi: float | None = None

reallocation_gain: float | None = None

net_value: float | None = None

net_value_basis: str | None = None

on_pareto: bool = False

is_recommended: bool = False

treatment_geos: list[str]

control_geos: list[str]

schedule: list[dict] | None = None

block_weeks: int | None = None

duration_requested: int | None = None

warnings: list[str]

objectives()

Return type:

tuple[float, float, float, float]

to_dict()

Return type:

dict[str, Any]

__init__(index, design_key, mode, footprint, n_pairs, intensity_pct, duration, mde_roas, power_shortfall, tradeoff, tradeoff_basis, forgone_kpi_median, opportunity_cost_dollar_median, net_profit_impact_median, spend_at_risk, pct_of_window_kpi, duration_effective, powered, power=None, power_lower=None, power_target=0.8, power_breakdown=None, sigma_exp=None, evoi_kpi=None, reallocation_gain=None, net_value=None, net_value_basis=None, on_pareto=False, is_recommended=False, treatment_geos=<factory>, control_geos=<factory>, schedule=None, block_weeks=None, duration_requested=None, warnings=<factory>)

mmm_framework.planning.experiment_optimizer.evaluate_experiment_grid(mmm, dataset_path, kpi, channel, *, duration_min=4, duration_max=12, intensity_min=50.0, intensity_max=100.0, durations=None, scaling_intensities=None, include_holdout=True, footprints=('full', 'half'), amplitudes=None, margin=None, price=None, kpi_kind='revenue', power_target=0.8, net_value_axis=True, response_horizon_weeks=26, max_draws=80, random_seed=42)

Evaluate a grid of candidate designs on FOUR objectives, all lower-better: MDE, power shortfall below power_target (default 80%), short-term cost, and duration.

When a margin is known and net_value_axis (default), the cost objective is upgraded to the net value of testing: each candidate’s EVOI (priced by the Gaussian surrogate at its own design precision, from ONE anchored preposterior-MC EVOI per grid) is decayed over response_horizon_weeks, capped at EVPI, converted to $, and netted against the candidate’s signed short-term test loss — the Pareto axis becomes −net_value and every candidate carries net_value/evoi_kpi/reallocation_gain. If the anchor cannot compute (no draw pairing, degenerate posterior), the axis degrades to the existing net-$ downside.

The design space is bounded by RANGES the caller controls: durations in [duration_min, duration_max] weeks and signed spend variations in [intensity_min, intensity_max] % (e.g. -100 go dark … +150 scale up; intensities at -100% render as a go-dark holdout). The optimizer auto-samples a few points within each range; pass explicit durations / scaling_intensities to override the sampling.

Reuses one pure-pandas power-curve call per (footprint, intensity) for the MDE of every duration, and one shared BAU posterior pass for the whole grid’s opportunity cost. Power is the model-anchored probability each design detects the expected effect (no extra posterior passes).

Return type:

dict[str, Any]

mmm_framework.planning.experiment_optimizer.pareto_front(cands)

Indices of the non-dominated designs over (MDE, power shortfall, cost, duration). Designs with a non-finite objective can never be on the front.

Return type:

list[int]

mmm_framework.planning.experiment_optimizer.recommend(cands, front)

The recommended design: among the powered designs on the front (or all front designs if none are powered), the ‘knee’ — the one closest to the ideal point in objectives normalized over the front.

Return type:

int | None

mmm_framework.planning.experiment_optimizer.suggest_experiment(mmm, dataset_path, kpi, channel, *, margin=None, price=None, kpi_kind='revenue', duration_min=4, duration_max=12, intensity_min=50.0, intensity_max=100.0, durations=None, scaling_intensities=None, include_holdout=True, footprints=('full', 'half'), power_target=0.8, net_value_axis=True, response_horizon_weeks=26, max_draws=80, random_seed=42)

Recommend a runnable experiment setup for channel and return the full Pareto front (MDE × power shortfall below power_target × short-term cost × duration), over a design space bounded by the duration and spend-variation ranges. The recommended design carries its test/control groups (or flighting schedule), duration, intensity, and the adstock-derived cool-down.

With a known margin the cost axis is the net value of testing (see evaluate_experiment_grid()) — gain from the learning minus the short-term loss — so the front trades precision and power against the dollars the test actually creates or destroys.

Return type:

dict[str, Any]

History

Per-run history metrics: the JSON snapshot each fit leaves behind so the Performance page can plot cycle-over-cycle trajectories (ROI CI contraction, budget-share migration, misallocation cost, portfolio marginal ROI) WITHOUT ever unpickling old models.

compute_run_metrics runs at fit time, kernel-side: it must touch only the fitted model (no registry / DB access — that enrichment happens host-side in mmm_framework.platform.history) and never fail a fit (callers wrap it in try/except). One compute_response_curves pass is shared by the ROI draws, the budget optimization, and the EIG/EVOI grid.

mmm_framework.planning.history.compute_run_metrics(mmm, *, max_draws=200, n_outcomes=48, random_seed=42)

JSON-safe per-run metrics snapshot (schema v2).

Per channel: spend & share, ROI posterior (mean/sd/5–95% interval/width), marginal ROI at current spend, optimal-vs-current share gap, allocation instability, and the EIG/EVOI priority fields. Portfolio: total spend, portfolio marginal ROI, the misallocation proxy (median uplift of the optimal vs current allocation), v_current/EVPI, and the mean CI width (the contraction series). Schema v2 adds response_curves: the compact per-channel saturation curve (spend grid × mean/5–95% contribution) the UI needs to show why average and marginal ROAS diverge.

Return type:

dict[str, Any]

Identification

Structural-parameter identification design for multi-level flighting.

The reduced-form flighting power (planning.design.flighting_estimand_ses) answers can the test measure the contribution / average-ROAS / marginal-ROAS. This module answers a different, deeper question the client asked for: will the next MMM refit, fed the manufactured exogenous spend variance, actually move the channel’s STRUCTURAL parameters — the response-curve shape psi (saturation), the carryover alpha (adstock), and the coefficient beta — off their priors? Per transforms.adstock §3.6 those three are entangled on an equifinality ridge and only weakly identified from smooth observational spend; a schedule that spans spend levels (curvature → psi) and pulses sharply (temporal contrast → alpha) is what breaks the ridge.

Method (a LOCAL Laplace design, kernel-safe — numpy + transforms only, no model/pymc import; the fitted-model anchor is extracted by the caller and passed in as plain arrays):

• Reconstruct the model’s exact forward op on the NORMALIZED basis x_norm = mult * op_spend / raw_max → normalized-FIR adstock → logistic saturation 1 - exp(-clip(lam*a, 0, 20)) → beta * s * y_std (original KPI scale). This byte-mirrors BayesianMMM.sample_channel_contributions for the v1-gated geometric + logistic parametric path.

• Build the per-week Jacobian J_t = dc_t/dtheta (analytic dc/dbeta; central finite differences for dc/dalpha and dc/dpsi THROUGH the real adstock_weights``+``apply_adstock, never the infinite-recursion analytic derivative). Take it over the EXPERIMENT-WINDOW INCREMENTAL rows only — the flighting contrast vs business-as-usual (J_t minus the BAU row) — so a flat schedule contributes nothing and the historical data is never double-counted.

• PROFILE OUT the nuisance baseline (intercept + linear trend + yearly Fourier) by residualizing every Jacobian column against it, so a schedule collinear with trend/season can’t earn phantom identification.

• Experiment Fisher F = (1/sigma^2) J_res^T J_res; posterior precision F_post = F + diag(1/prior_var) (a DIAGONAL prior-precision ridge from the robust draw-based widths — NEVER inv(cov) of the skewed, banana-shaped structural posterior). Per-parameter posterior SD = sqrt(diag(inv(F_post))).

Honesty rails (all from the locked design review): identification is driven by the smallest eigenvalue / per-parameter contraction (E-optimality), NEVER the determinant; a parameter is only “claimed” when the design actually excites it (alpha only with temporal contrast, psi only with >=3 in-support levels); the result is an OPTIMISTIC UPPER BOUND on what a refit achieves and must never solely drive a Pareto objective; degenerate / near-zero-gradient / over-condition designs return None rather than a falsely-tight CI.

mmm_framework.planning.identification.ar1_design_effect(residual, window, *, rho_max=0.95)

AR(1) variance design-effect for a length-window average/contrast.

Autocorrelated residuals inflate the variance of a windowed estimator beyond the i.i.d. sigma^2/n rule by deff = 1 + 2*sum_{k=1}^{L-1}(1 - k/L)*rho^k (Newey-West / batch-means form). rho is the lag-1 autocorrelation of the baseline-residualized series, clipped to [0, rho_max]; deff is capped at the window length L (the fully-correlated bound). Multiply sigma by sqrt(deff) BEFORE the OLS covariance so an autocorrelated window cannot be labelled powered on an i.i.d. SE.

Return type:

dict[str, Any]

mmm_framework.planning.identification.cross_channel_vif(target_window_spend, other_window_spend)

Variance-inflation from co-spending channels over the scheduled window.

Regresses the test channel’s SCHEDULED window spend on the other channels’ actual window spend and returns R2 and the inflation 1/(1-R2). The scheduled variance is randomized, but if other channels happen to co-move with it over the window the design’s effective contrast shrinks. Surface it and degrade power when R2 is high; other_window_spend is (L, n_other) or None (single-channel → no inflation).

Return type:

dict[str, Any]

mmm_framework.planning.identification.structural_jacobian(mults, op_spend, raw_max, y_std, theta, *, l_max=8, normalize=True, prior_sd=None, fd_rel=0.01, residualize=True)

Residualized, BAU-incremental per-week Jacobian of channel contribution w.r.t. theta = {beta, alpha, lam} for a multiplier schedule.

Columns: analytic dc/dbeta = s*y_std; central finite differences for dc/dalpha and dc/dlam through the real normalized-FIR adstock + logistic saturation. Each column is taken as the CONTRAST vs the BAU (all-ones) schedule and residualized against the nuisance basis. Rows where the saturation exponent lam*a is clamped (>20) are zeroed (the model’s own gradient is exactly 0 there). Returns None on non-finite/degenerate input.

Return type:

dict[str, Any] | None

mmm_framework.planning.identification.structural_information(J, names, sigma, prior_sd)

Experiment + posterior Fisher information from a residualized Jacobian.

F = (1/sigma^2) J^T J; F_post = F + diag(1/prior_var) (a diagonal prior-precision ridge from the robust draw widths — never inv(cov) of the structural posterior). Returns per-parameter posterior SD, the smallest eigenvalue + its eigenvector (the equifinality direction the design fails to pin), the condition number, and a relative information gain. None if the inputs are degenerate.

Return type:

dict[str, Any] | None

mmm_framework.planning.identification.structural_identification(mults, op_spend, raw_max, y_std, theta_draws, *, sigma_lo, sigma_hi, l_max=8, normalize=True, in_support=True, power_target=0.8, fd_rel=0.01)

Per-parameter identification power / MDE / contraction for a multi-level flighting schedule — the structural (beta, alpha, psi=lam) block.

OPTIMISTIC UPPER BOUND on what the next refit achieves; never let it solely drive a Pareto objective. theta_draws are the channel’s flattened posterior draws (beta, alpha, lam); theta_hat is the median. sigma_lo (model sigma*y_std, optimistic floor) and sigma_hi (regression residual, pessimistic) bracket the noise; the headline power uses the conservative max(sigma_lo, sigma_hi). A parameter is “claimed” only when the design is structurally eligible AND the experiment actually contracts it (an absolute floor): beta/alpha need real contrast, lam needs >=3 in-support distinct levels. When nothing is claimed (a flat / near-flat / nuisance-collinear schedule) binding_power is None and identifies_anything is False — never a prior-driven false confidence. Returns None only on degenerate INPUTS.

Return type:

dict[str, Any] | None

Pacing

In-flight pacing — actual-vs-plan loop (issue #107).

MMM is retrospective and the measurement loop re-tests on a decay cadence, but nothing connects a recommended plan to what’s actually being spent in-flight. Pacing — planned vs actual delivery — is where media planners live day-to-day, and between fits it is a blind spot: the recommendation is delivered and then goes dark until the next model run.

This module closes that loop with pure, testable numpy:

• compute_pacing() — aligns a plan and actual delivery per channel/period, computes divergence, and flags channels pacing outside a threshold.

• expected_outcome_delta() — the expected KPI impact of the divergence, read off the fitted response curves with uncertainty (contribution at actual spend vs at planned spend, per posterior draw).

• pacing_report() — combines both into the report/agent payload.

The plan and the actuals can be given as a flighting schedule ({"schedule": [{"period", "<channel>": spend, ...}]}), a list of period rows, a {channel: [per-period]} map, or a {channel: total} map — all normalize to per-channel per-period arrays.

Alignment: when both sides carry real period labels the join is by label; inputs without a usable vocabulary fall back to positional truncation. Before v1.4 the claim above read “aligned by period” and was false — the join was always positional, so a mid-flight upload was compared against the wrong weeks.

class mmm_framework.planning.pacing.PacingChannel(channel, planned, actual, divergence_pct, status, planned_series=<factory>, actual_series=<factory>)

Bases: object

One channel’s pacing status.

channel: str

planned: float

actual: float

divergence_pct: float

status: str

planned_series: list[float]

actual_series: list[float]

to_dict()

Return type:

dict[str, Any]

__init__(channel, planned, actual, divergence_pct, status, planned_series=<factory>, actual_series=<factory>)

class mmm_framework.planning.pacing.PacingResult(channels, periods, threshold, planned_total, actual_total, divergence_pct, flagged, outcome_delta=None, join='positional')

Bases: object

Portfolio + per-channel pacing, with the flagged (off-pace) channels.

channels: list[PacingChannel]

periods: list[str]

threshold: float

planned_total: float

actual_total: float

divergence_pct: float

flagged: list[str]

outcome_delta: dict[str, float] | None = None

join: str = 'positional'

‘label’ when plan and actual were joined on a shared period vocabulary, ‘positional’ when either side had none. Recorded rather than assumed: a positional join on labelled data was the pre-v1.4 defect.

to_dict()

Return type:

dict[str, Any]

__init__(channels, periods, threshold, planned_total, actual_total, divergence_pct, flagged, outcome_delta=None, join='positional')

mmm_framework.planning.pacing.compute_pacing(planned, actual, *, threshold=0.1)

Compare a plan against actual delivery, per channel and overall.

Both inputs normalize to per-channel per-period arrays. When both sides carry real period labels the comparison joins by label, so a mid-flight upload covering weeks 5-8 is compared against the plan’s weeks 5-8. Inputs without labels fall back to positional truncation. PacingResult.join records which happened.

Return type:

PacingResult

mmm_framework.planning.pacing.expected_outcome_delta(curves, planned_totals, actual_totals, *, hdi_prob=0.9)

Expected KPI delta from the divergence, off the fitted response curves.

curves is a ResponseCurves (or any object exposing channel_names, spend_grid (C,G) and contributions (D,C,G)). For each posterior draw we read the channel’s contribution at the ACTUAL total vs the PLANNED total and sum the difference — turning parameter uncertainty into a credible interval on the outcome impact. Returns {"mean","lower","upper","planned_kpi","actual_kpi"} or None.

Return type:

dict[str, float] | None

mmm_framework.planning.pacing.pacing_report(planned, actual, *, curves=None, threshold=0.1, hdi_prob=0.9)

Full pacing assessment: divergence + (when curves given) the expected KPI impact of that divergence.

Return type:

PacingResult

mmm_framework.planning.pacing.DEFAULT_PACING_THRESHOLD = 0.1

Default divergence threshold (fraction of planned spend) above which a channel is flagged as off-pace.

Cpa

Cost-per-conversion (CPA) power math — the client metric that CANNOT be naively inverted from the lift.

CPA = cost / incremental conversions. The lift estimate is approximately Gaussian, but its reciprocal is not: dividing a constant by a noisy quantity whose interval approaches zero produces an extremely right-skewed ratio distribution — the estimated CPA has a long expensive tail, its mean overshoots the truth, and the symmetric “delta-method” interval cpa ± z·(cost·se/lift²) under-covers (it can even go negative). Three consequences this module encodes:

• Maximum detectable CPA — the planning summary that DOES survive inversion. A design that can detect lift ≥ MDE can certify CPA ≤ cost/MDE; larger (worse) CPAs are indistinguishable from “no effect”. Invert the detectable-lift bound, never the point estimate.

• Interval by inverting the lift bound — the CPA interval is the image of the lift interval under x → cost/x (monotone on the positive half-line): asymmetric, and unbounded above whenever the lift interval touches zero (the data cannot rule out an infinite cost per conversion — reporting that honestly is the point).

• Power on the CPA scale — power to come away with a bounded CPA at all, or to certify CPA ≤ target (equivalent to detecting lift ≥ cost/target).

numpy only (kernel-safe). Cost and lift can be on any per-unit basis as long as they share it (e.g. ghost ads: cost per treated user vs per-user lift).

mmm_framework.planning.cpa.max_detectable_cpa(cost, lift_mde)

The WORST (highest) cost per conversion the design can still certify: cost / MDE. A true CPA above this is indistinguishable from “the media did nothing” — the honest planning summary on the CPA scale.

Return type:

float

mmm_framework.planning.cpa.cpa_interval(lift, se_lift, cost, *, alpha=0.05)

CPA point + interval by inverting the lift interval — and, for contrast, the naive symmetric delta-method interval that clients are usually shown.

Returns status: :rtype: dict[str, Any]

• "bounded" — the lift interval is strictly positive; the CPA interval is [cost/lift_hi, cost/lift_lo] (asymmetric: the upper arm is longer).

• "upper_unbounded" — the lift interval touches zero; only a lower CPA bound exists. The data cannot rule out an arbitrarily bad CPA.

• "undefined" — the whole lift interval is ≤ 0; no positive-lift evidence, no CPA statement at all.

mmm_framework.planning.cpa.cpa_power(cost, se_lift, true_lift, *, target_cpa=None, alpha=0.05)

Power on the CPA scale.

With target_cpa=None: the probability the test comes away with a bounded CPA interval at all (the lift interval clears zero) — Φ(true_lift/se − z). With a target: the probability the CPA interval’s upper end lands below target_cpa, i.e. the test certifies CPA ≤ target — equivalent to the estimated lift clearing cost/target + z·se.

Return type:

float

mmm_framework.planning.cpa.simulate_cpa_distribution(true_lift, se_lift, cost, *, alpha=0.05, n_sims=20000, seed=0)

Monte-Carlo the CPA readout a design of this precision would produce — the demonstration that the reciprocal is NOT well-behaved even when the lift is.

Simulates lift̂ ~ N(true_lift, se_lift), forms cpâ = cost/lift̂ where the estimate is positive (non-positive draws are “no measurable conversions” — CPA undefined/infinite), and scores both interval constructions against the true CPA. Returns the draws (for plotting) plus skew diagnostics and empirical coverage.

Return type:

dict[str, Any]