"""Assemble Marketing Mix Model (MMM) calibration inputs from experiment results.
MMM practitioners calibrate their models against experimental evidence. The two
dominant Python MMM frameworks consume that evidence in different shapes:
- **PyMC-Marketing** (and prophetverse, which mirrors the same schema) ingests a
*lift-test* DataFrame with columns ``channel``, optional model dims (e.g. ``geo``),
``x`` (baseline channel spend), ``delta_x`` (spend change during the test),
``delta_y`` (measured incremental outcome), and ``sigma`` (the experiment's standard
error), which it scores against the model's saturation curve via
``MMM.add_lift_test_measurements``. See
https://www.pymc-marketing.io/en/stable/notebooks/mmm/mmm_lift_test.html.
- **Google Meridian** has no experiment-ingestion API; calibration means setting a
lognormal prior per channel - ``roi_m`` for the return of a channel's full spend
(a full-holdout/zero-spend estimand), ``mroi_m`` for a marginal return. Google's
documented workflow maps the experiment's ROI point estimate to the prior mean and
its standard error to the prior standard deviation, converted to lognormal
``(mu, sigma)`` via the closed form used by
``meridian.model.prior_distribution.lognormal_dist_from_mean_std``:
``mu = ln(m) - ln((s/m)^2 + 1)/2``, ``sigma = sqrt(ln((s/m)^2 + 1))``. See
https://developers.google.com/meridian/docs/advanced-modeling/set-custom-priors-past-experiments.
**Design: explicit in, validated out.** These functions do NOT rescale a result's
headline ATT. Reconciling an experiment's estimate to a calibration input requires
context diff-diff cannot see - the target MMM's row granularity (per-geo vs national),
its time window, and the outcome's scale (additive levels vs a log/rate/share) - so
the CALLER supplies the already-scoped incremental outcome and its standard error (the
numbers they read off a fitted result's ``summary()``, aggregated to the population and
window their MMM row represents). diff-diff does only what it can verify: assemble the
exact target schema, enforce the sign/positivity/monotonicity guards each consumer
requires, convert to the lognormal parameterization (with parity to Google's closed
form), pool multiple experiments, and emit ready-to-paste snippets. It is pure
numpy/pandas and never imports an MMM package.
"""
from __future__ import annotations
import ast
import math
import warnings
from dataclasses import dataclass, field
from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple, Union
import numpy as np
import pandas as pd
__all__ = ["MeridianROIPrior", "to_meridian_roi_prior", "to_pymc_marketing_lift_test"]
_WRONG_SIGN_POLICIES = ("raise", "drop", "keep")
_LIFT_TEST_RESERVED = frozenset({"channel", "x", "delta_x", "delta_y", "sigma"})
# Meridian prior parameters this exporter can target, with each one's Meridian
# default LogNormal(mu, sigma) per channel (verified against
# meridian/model/prior_distribution.py at 1.7.0): roi_m is the return on a
# channel's full spend (zero-spend counterfactual), mroi_m the marginal return.
# Channels without an experiment keep the default in the vector snippet.
_MERIDIAN_PARAM_DEFAULTS = {"roi_m": (0.2, 0.9), "mroi_m": (0.0, 0.5)}
# The .to_code() templates are pinned against google-meridian 1.7.0 (2026-06); they
# are convenience snippets, not a programmatic contract. Meridian's roi_m/mroi_m have
# batch shape n_media_channels; a scalar LogNormal broadcasts to EVERY channel, so the
# scalar template is gated behind an explicit single_channel opt-in.
_MERIDIAN_SINGLE_CHANNEL_TEMPLATE = """\
# Generated by diff_diff.mmm.to_meridian_roi_prior (template pinned to Meridian 1.7.0)
# TensorFlow-substrate snippet; for JAX-backed Meridian use
# `import tensorflow_probability.substrates.jax as tfp` instead.
# SINGLE-CHANNEL MODEL ONLY: a scalar {param} prior broadcasts to every media channel.
# For a multi-channel model, regenerate with to_code(channel=..., media_channels=[...]).
import tensorflow_probability as tfp
from meridian.model import prior_distribution, spec
roi_prior = tfp.distributions.LogNormal({mu!r}, {sigma!r}, name="{param}")
prior = prior_distribution.PriorDistribution({param}=roi_prior)
model_spec = spec.ModelSpec(
prior=prior,
media_prior_type="{prior_type}",
# {window_note}
roi_calibration_period={calibration_period},
)
"""
_MERIDIAN_MULTI_CHANNEL_TEMPLATE = """\
# Generated by diff_diff.mmm.to_meridian_roi_prior (template pinned to Meridian 1.7.0)
# TensorFlow-substrate snippet; for JAX-backed Meridian use
# `import tensorflow_probability.substrates.jax as tfp` instead.
# Channel order MUST match your Meridian InputData media channel order exactly:
# {channels}
# {channel!r} carries the experiment-informed prior; the other channels keep
# Meridian's default {param} prior LogNormal({default_mu}, {default_sigma}).
import tensorflow_probability as tfp
from meridian.model import prior_distribution, spec
mu = {mu_vector!r}
sigma = {sigma_vector!r}
roi_prior = tfp.distributions.LogNormal(mu, sigma, name="{param}")
prior = prior_distribution.PriorDistribution({param}=roi_prior)
model_spec = spec.ModelSpec(
prior=prior,
media_prior_type="{prior_type}",
# {window_note}
roi_calibration_period={calibration_period},
)
"""
def _is_sequence(value: Any) -> bool:
"""True for list-like containers (not strings, not mappings)."""
return isinstance(value, (list, tuple, np.ndarray, pd.Series))
def _broadcast(name: str, value: Any, n: int) -> List[Any]:
"""Broadcast a scalar to n rows, or validate a sequence's length."""
if _is_sequence(value):
values = list(value)
if len(values) != n:
raise ValueError(
f"{name} has length {len(values)} but {n} experiment(s) were given; "
f"pass one value per experiment or a single scalar"
)
return values
return [value] * n
def _seq_len(*values: Any) -> int:
"""Number of experiments implied by the first sequence argument (else 1)."""
for v in values:
if _is_sequence(v):
n = len(list(v))
if n == 0:
raise ValueError("empty sequence given; provide at least one experiment")
return n
return 1
def _finite_positive(name: str, value: Any, index: int) -> float:
v = float(value)
if not math.isfinite(v) or v <= 0:
raise ValueError(f"{name} must be finite and > 0; got {value!r} for experiment[{index}]")
return v
def _normalize_dims(
dims: Optional[Union[Mapping[str, str], Sequence[Mapping[str, str]]]], n: int
) -> Tuple[List[str], List[Mapping[str, str]]]:
"""Broadcast dims mappings and pin a single shared key set / column order."""
if dims is None:
return [], []
if isinstance(dims, Mapping):
rows: List[Mapping[str, str]] = [dims] * n
else:
rows = list(_broadcast("dims", dims, n))
for i, row in enumerate(rows):
if not isinstance(row, Mapping):
raise TypeError(
f"dims must be a mapping or a sequence of mappings (e.g. "
f"{{'geo': 'US-CA'}}); dims[{i}] is {type(row).__name__}"
)
dim_cols = list(rows[0].keys())
key_set = set(dim_cols)
reserved = key_set & _LIFT_TEST_RESERVED
if reserved:
raise ValueError(
f"dims keys {sorted(reserved)} collide with reserved lift-test columns "
f"{sorted(_LIFT_TEST_RESERVED)}; rename the model dimension"
)
for i, row in enumerate(rows):
if set(row.keys()) != key_set:
raise ValueError(
f"dims mappings must share one key set across rows; dims[{i}] has keys "
f"{sorted(row.keys())}, expected {sorted(key_set)}"
)
return dim_cols, rows
[docs]
def to_pymc_marketing_lift_test(
*,
channel: Union[str, Sequence[str]],
x: Union[float, Sequence[float]],
delta_x: Union[float, Sequence[float]],
delta_y: Union[float, Sequence[float]],
sigma: Union[float, Sequence[float]],
dims: Optional[Union[Mapping[str, str], Sequence[Mapping[str, str]]]] = None,
on_wrong_sign: str = "raise",
) -> pd.DataFrame:
"""Assemble a PyMC-Marketing lift-test DataFrame from scoped experiment results.
Produces one row per experiment with columns ``channel``, any ``dims`` columns,
``x``, ``delta_x``, ``delta_y``, ``sigma`` - the exact schema consumed by
``pymc_marketing.mmm.MMM.add_lift_test_measurements`` (prophetverse's lift-test
API accepts the same shape). All values stay in original data units
(PyMC-Marketing rescales internally).
**The caller supplies the scoped effect.** ``delta_y`` and ``sigma`` are the
measured incremental outcome and its standard error, already aggregated to the
population and time window ONE target-MMM row represents. This function does not
rescale an ATT, because the reconciliation needs the MMM's row granularity and
outcome scale, which it cannot see. Read the effect and SE off your fitted
result's ``summary()`` and scope them yourself: PyMC-Marketing scores one row's
``delta_y`` against ``saturation(x + delta_x) - saturation(x)``, so ``x``,
``delta_x``, ``delta_y``, and ``sigma`` must all describe the SAME observation
(same channel, same population, same period span, additive-level outcome).
Parameters
----------
channel : str or sequence of str
MMM channel name(s); must match the target model's ``channel_columns``.
x : float or sequence of float
Baseline channel spend for the row's observation, in original spend units.
delta_x : float or sequence of float
Spend change during the experiment (nonzero; negative for go-dark/holdout
tests, with ``x + delta_x >= 0``), in the same units as ``x``.
delta_y : float or sequence of float
Measured incremental outcome for the SAME observation as ``x``/``delta_x``,
in original outcome units. Finite; must be nonzero and share ``delta_x``'s
sign (see ``on_wrong_sign``).
sigma : float or sequence of float
Standard error of ``delta_y`` (finite, positive).
dims : mapping or sequence of mappings, optional
Extra model-dimension columns, e.g. ``{"geo": "US-CA"}`` for a geo-level MMM
built with ``MMM(dims=("geo",))``. Values must match the target model's
coordinate values exactly and identify the population ``delta_y`` describes
(do not label a multi-geo average with one specific geo). All rows must share
one key set; column order follows the first mapping.
on_wrong_sign : {"raise", "drop", "keep"}, default "raise"
Policy for rows PyMC-Marketing cannot use: ``sign(delta_y)`` contradicting
``sign(delta_x)`` (rejected upstream with ``NonMonotonicError``), or
``delta_y == 0`` (degenerate for its strictly-positive Gamma lift likelihood,
which its own monotonicity check does not catch). ``"raise"`` (default) errors
with guidance; ``"drop"`` warns and removes such rows (raises if that would
empty the frame); ``"keep"`` warns and emits them anyway (for non-PyMC
consumers - the frame is not valid PyMC-Marketing input).
Returns
-------
pd.DataFrame
Columns ``[channel, *dims, x, delta_x, delta_y, sigma]``, one row per
experiment.
Raises
------
ValueError
On invalid inputs (non-positive ``sigma``, non-finite values, ``delta_x == 0``,
``x + delta_x < 0``, broadcasting length mismatches, empty inputs,
heterogeneous ``dims`` key sets) or wrong-signed/zero rows under the default
policy.
"""
if on_wrong_sign not in _WRONG_SIGN_POLICIES:
raise ValueError(
f"on_wrong_sign must be one of {_WRONG_SIGN_POLICIES}; got {on_wrong_sign!r}"
)
n = _seq_len(channel, x, delta_x, delta_y, sigma, dims)
channels = _broadcast("channel", channel, n)
xs = _broadcast("x", x, n)
delta_xs = _broadcast("delta_x", delta_x, n)
delta_ys = _broadcast("delta_y", delta_y, n)
sigmas = _broadcast("sigma", sigma, n)
dim_cols, dim_rows = _normalize_dims(dims, n)
rows: List[Dict[str, Any]] = []
wrong_sign_rows: List[int] = []
zero_lift_rows: List[int] = []
for i in range(n):
x_i = float(xs[i])
dx_i = float(delta_xs[i])
dy_i = float(delta_ys[i])
sig_i = _finite_positive("sigma", sigmas[i], i)
if not math.isfinite(x_i) or x_i < 0:
raise ValueError(f"x must be finite and >= 0; got {xs[i]!r} for experiment[{i}]")
if not math.isfinite(dx_i) or dx_i == 0:
raise ValueError(
f"delta_x must be finite and nonzero (the experiment changed spend); "
f"got {delta_xs[i]!r} for experiment[{i}]"
)
post_spend = x_i + dx_i
if not math.isfinite(post_spend) or post_spend < 0:
raise ValueError(
f"x + delta_x must be finite and >= 0 (post-test spend cannot be "
f"negative or overflow; the saturation curve is evaluated at "
f"x + delta_x); got {x_i!r} + {dx_i!r} = {post_spend!r} for "
f"experiment[{i}]"
)
if not math.isfinite(dy_i):
raise ValueError(f"delta_y must be finite; got {delta_ys[i]!r} for experiment[{i}]")
if dy_i == 0:
zero_lift_rows.append(i)
elif (dx_i < 0) != (dy_i < 0):
# Compare signs directly: dx_i * dy_i can underflow to -0.0 for tiny
# magnitudes (e.g. 1e-200 * -1e-200), so a product < 0 check misses
# wrong-signed rows. Both are strictly nonzero here (delta_x validated
# above, delta_y != 0 handled just above), so the comparison is exact.
wrong_sign_rows.append(i)
row: Dict[str, Any] = {"channel": channels[i]}
if dim_cols:
row.update({k: dim_rows[i][k] for k in dim_cols})
row.update({"x": x_i, "delta_x": dx_i, "delta_y": dy_i, "sigma": sig_i})
rows.append(row)
# Two invalid-row classes, one shared disposition. Wrong sign is what
# PyMC-Marketing rejects with NonMonotonicError; zero lift passes its
# monotonicity check but is degenerate for the strictly-positive Gamma lift
# likelihood (an insignificant experiment that cannot calibrate saturation).
if wrong_sign_rows or zero_lift_rows:
parts = []
if wrong_sign_rows:
parts.append(
f"row(s) {wrong_sign_rows} have sign(delta_y) contradicting "
f"sign(delta_x) (PyMC-Marketing rejects these with NonMonotonicError)"
)
if zero_lift_rows:
parts.append(
f"row(s) {zero_lift_rows} have delta_y == 0 (degenerate for "
f"PyMC-Marketing's strictly-positive Gamma lift likelihood, which its "
f"monotonicity check does not catch)"
)
detail = "; ".join(parts)
if on_wrong_sign == "raise":
raise ValueError(
f"{detail}. Pool experiments, re-scope, or exclude the offending "
f"experiment; or pass on_wrong_sign='drop'/'keep' to handle these rows "
f"explicitly."
)
dropped = set(wrong_sign_rows) | set(zero_lift_rows)
if on_wrong_sign == "drop":
if len(dropped) == len(rows):
raise ValueError(f"on_wrong_sign='drop' would remove every row: {detail}.")
warnings.warn(
f"Dropping invalid lift-test row(s): {detail}.", UserWarning, stacklevel=2
)
rows = [row for i, row in enumerate(rows) if i not in dropped]
else: # "keep"
warnings.warn(
f"Keeping invalid lift-test row(s): {detail}. The frame is NOT valid "
f"input for PyMC-Marketing's lift likelihood.",
UserWarning,
stacklevel=2,
)
columns = ["channel", *dim_cols, "x", "delta_x", "delta_y", "sigma"]
return pd.DataFrame(rows, columns=columns)
@dataclass(frozen=True)
class ExperimentROI:
"""Per-experiment ROI contribution inside a :class:`MeridianROIPrior`."""
roi: float
roi_sd: float
spend: float
weight: float
def to_dict(self) -> Dict[str, float]:
return {
"roi": self.roi,
"roi_sd": self.roi_sd,
"spend": self.spend,
"weight": self.weight,
}
[docs]
@dataclass(frozen=True)
class MeridianROIPrior:
"""Lognormal prior parameters for Meridian's ``roi_m``/``mroi_m`` calibration.
``mu``/``sigma`` reproduce ``meridian.model.prior_distribution.
lognormal_dist_from_mean_std(roi_mean, roi_sd)`` exactly (closed form; no Meridian
dependency). ``parameter`` records which Meridian prior the caller is informing
(``"roi_m"`` or ``"mroi_m"``). ``per_experiment`` records each pooled experiment's
ROI, widened sd, spend, and spend weight.
"""
roi_mean: float
roi_sd: float
mu: float
sigma: float
parameter: str = "roi_m"
per_experiment: Tuple[ExperimentROI, ...] = field(default=())
[docs]
def to_dict(self) -> Dict[str, Any]:
return {
"distribution": "LogNormal",
"parameter": self.parameter,
"roi_mean": self.roi_mean,
"roi_sd": self.roi_sd,
"mu": self.mu,
"sigma": self.sigma,
"per_experiment": [e.to_dict() for e in self.per_experiment],
}
[docs]
def to_code(
self,
*,
channel: Optional[str] = None,
media_channels: Optional[Sequence[str]] = None,
single_channel: bool = False,
roi_calibration_period: Optional[str] = None,
full_model_window: bool = False,
) -> str:
"""Ready-to-paste Meridian snippet (channel- and time-scoped; 1.7.0 pinned).
Meridian's ``roi_m``/``mroi_m`` prior has batch shape ``n_media_channels`` and
a scalar distribution broadcasts to EVERY media channel - a TV experiment's
prior must not silently calibrate search, social, etc. Channel scope is
therefore always explicit:
- ``to_code(channel="tv", media_channels=["search", "tv"])`` emits vector
``mu``/``sigma`` in exactly the ``media_channels`` order (which must match
the Meridian ``InputData`` channel order); the experiment channel carries
this prior and every other channel keeps Meridian's default.
- ``to_code(single_channel=True)`` emits the scalar snippet for
single-channel models, marked as such in the generated code.
- Calling with neither raises ``ValueError``.
The prior's TIME scope is also required: Meridian's default
``roi_calibration_period=None`` applies the prior over all model times, but
the prior was estimated on the experiment window. Pass
``roi_calibration_period="<expression>"`` (the boolean
``(n_media_times, n_media_channels)`` mask for your window) or
``full_model_window=True`` to acknowledge that the two coincide.
Snippets use the TensorFlow substrate of TensorFlow Probability; JAX-backed
Meridian users should swap the import for
``tensorflow_probability.substrates.jax`` (noted in the generated code).
"""
if roi_calibration_period is None and not full_model_window:
raise ValueError(
"to_code() needs the prior's time scope: Meridian's default "
"roi_calibration_period=None applies the prior over ALL model times, "
"but this prior was estimated on the EXPERIMENT window, and ROI "
"differs across windows under varying spend and saturation. Pass "
"roi_calibration_period=<expression building the boolean "
"(n_media_times, n_media_channels) mask for your experiment window>, "
"or full_model_window=True to acknowledge that the MMM window and the "
"experiment window coincide."
)
if roi_calibration_period is not None and full_model_window:
raise ValueError(
"pass either roi_calibration_period or full_model_window=True, not both"
)
if roi_calibration_period is not None:
try:
ast.parse(roi_calibration_period, mode="eval")
except SyntaxError as exc:
raise ValueError(
f"roi_calibration_period must be a valid Python expression building "
f"the boolean (n_media_times, n_media_channels) mask (it is "
f"interpolated verbatim into the snippet); got "
f"{roi_calibration_period!r}, which does not parse: {exc.msg}"
) from exc
window_note = (
"Experiment window == full model window (acknowledged via full_model_window=True)."
if full_model_window
else "Mask restricting the prior to the experiment window."
)
calibration_period = "None" if full_model_window else roi_calibration_period
prior_type = "roi" if self.parameter == "roi_m" else "mroi"
if media_channels is not None:
if single_channel:
raise ValueError("pass either media_channels or single_channel=True, not both")
channels = list(media_channels)
if not channels:
raise ValueError("media_channels must be non-empty")
if channel is None or channel not in channels:
raise ValueError(
f"channel must name the experiment channel within media_channels; "
f"got channel={channel!r}, media_channels={channels!r}"
)
default_mu, default_sigma = _MERIDIAN_PARAM_DEFAULTS[self.parameter]
mu_vector = [self.mu if c == channel else default_mu for c in channels]
sigma_vector = [self.sigma if c == channel else default_sigma for c in channels]
return _MERIDIAN_MULTI_CHANNEL_TEMPLATE.format(
channels=channels,
channel=channel,
mu_vector=mu_vector,
sigma_vector=sigma_vector,
param=self.parameter,
default_mu=default_mu,
default_sigma=default_sigma,
window_note=window_note,
calibration_period=calibration_period,
prior_type=prior_type,
)
if single_channel:
return _MERIDIAN_SINGLE_CHANNEL_TEMPLATE.format(
mu=self.mu,
sigma=self.sigma,
param=self.parameter,
window_note=window_note,
calibration_period=calibration_period,
prior_type=prior_type,
)
raise ValueError(
"to_code() needs explicit channel scope: a scalar prior broadcasts to "
"every media channel in a multi-channel Meridian model. Pass channel=... "
"with media_channels=[...] (vector prior in model channel order), or "
"single_channel=True for a single-channel model."
)
[docs]
def to_meridian_roi_prior(
*,
incremental_outcome: Union[float, Sequence[float]],
incremental_outcome_se: Union[float, Sequence[float]],
spend: Union[float, Sequence[float]],
parameter: str = "roi_m",
se_widening: float = 1.0,
) -> MeridianROIPrior:
"""Build a Meridian lognormal ROI/mROI prior from scoped experiment result(s).
Per experiment ``roi = incremental_outcome / spend`` with standard deviation
``incremental_outcome_se / spend * se_widening``. Multiple experiments for the same
channel are pooled with spend weights - the spend-weighted average ROI the Meridian
FAQ suggests for multiple experiments (citing section 3.4 of Google's MMM
calibration whitepaper): ``roi_mean = sum(w_i * roi_i)`` with
``w_i = spend_i / sum(spend)``. The pooled ``roi_sd = sqrt(sum((w_i * sd_i)^2))``
is this library's uncertainty propagation for that average (treats experiments as
independent - widen via ``se_widening`` when experiments share controls or windows).
The pooled mean/sd map to lognormal ``(mu, sigma)`` via Google's closed form.
**The caller supplies the scoped estimand.** ``incremental_outcome`` is the total
incremental outcome the experiment measured, matching the estimand of the target
prior: for ``parameter="roi_m"`` that is the outcome attributable to the channel's
spend against a zero-spend counterfactual (a full holdout), divided here by
``spend`` = the channel's total spend over the window; for ``parameter="mroi_m"``
it is the marginal outcome of the spend change, divided by that spend change. Sign,
aggregation, and population are the caller's responsibility - diff-diff does not
rescale a headline ATT.
Parameters
----------
incremental_outcome : float or sequence of float
Total incremental outcome per experiment (the estimand matching ``parameter``),
in the same currency/units as the MMM. Must be finite; the pooled ROI must be
positive (lognormal support).
incremental_outcome_se : float or sequence of float
Standard error of ``incremental_outcome`` (finite, positive), computed for the
caller's aggregation.
spend : float or sequence of float
Spend the incremental outcome is divided by (finite, positive): the channel's
total spend over the window for ``roi_m``, or the spend change for ``mroi_m``.
parameter : {"roi_m", "mroi_m"}, default "roi_m"
Which Meridian prior the estimate informs. ``roi_m`` is the return on the
channel's full spend (a full-holdout / zero-spend estimand); ``mroi_m`` is the
marginal return of a spend change. Under saturation these differ, so the caller
declares which their ``incremental_outcome`` measures; the returned prior and
``.to_code()`` target that parameter with its own Meridian default for
non-experiment channels.
se_widening : float, default 1.0
Multiplier on each experiment's ROI standard deviation (finite, positive).
Values above 1 encode skepticism about experiment-to-MMM transferability.
Returns
-------
MeridianROIPrior
Pooled ``roi_mean``/``roi_sd``, the lognormal ``mu``/``sigma``, the target
``parameter``, per-experiment detail, plus ``.to_dict()`` and the
channel-scoped ``.to_code()`` snippet helper.
Raises
------
ValueError
On an invalid ``parameter``, non-positive ``spend``/``se``/``se_widening``,
non-finite inputs, broadcasting mismatches, empty inputs, a non-positive pooled
ROI mean (lognormal support), or non-finite scaled/pooled results.
"""
if parameter not in _MERIDIAN_PARAM_DEFAULTS:
raise ValueError(
f"parameter must be one of {sorted(_MERIDIAN_PARAM_DEFAULTS)}; got "
f"{parameter!r}. roi_m is the return on the channel's full spend (a "
f"full-holdout / zero-spend estimand); mroi_m is the marginal return of a "
f"spend change. Under saturation they differ - pass the one your "
f"incremental_outcome measures."
)
se_w = float(se_widening)
if not math.isfinite(se_w) or se_w <= 0:
raise ValueError(f"se_widening must be finite and > 0; got {se_widening!r}")
n = _seq_len(incremental_outcome, incremental_outcome_se, spend)
outcomes = _broadcast("incremental_outcome", incremental_outcome, n)
outcome_ses = _broadcast("incremental_outcome_se", incremental_outcome_se, n)
spends = _broadcast("spend", spend, n)
rois: List[float] = []
sds: List[float] = []
spend_vals: List[float] = []
for i in range(n):
total = float(outcomes[i])
if not math.isfinite(total):
raise ValueError(
f"incremental_outcome must be finite; got {outcomes[i]!r} for experiment[{i}]"
)
total_se = _finite_positive("incremental_outcome_se", outcome_ses[i], i)
spend_i = _finite_positive("spend", spends[i], i)
roi_i = total / spend_i
sd_i = total_se / spend_i * se_w
if not (math.isfinite(roi_i) and math.isfinite(sd_i) and sd_i > 0):
raise ValueError(
f"experiment[{i}] ROI is not finite-positive (roi={roi_i!r}, "
f"roi_sd={sd_i!r}); the magnitudes overflowed or underflowed the float "
f"range - rescale the outcome or spend units"
)
rois.append(roi_i)
sds.append(sd_i)
spend_vals.append(spend_i)
total_spend = sum(spend_vals)
weights = [s / total_spend for s in spend_vals]
roi_mean = sum(w * r for w, r in zip(weights, rois))
# math.hypot is a scaled Euclidean norm: it avoids the intermediate squaring
# that would raise a raw OverflowError before the finiteness guard below.
roi_sd = math.hypot(*(w * s for w, s in zip(weights, sds)))
if not (math.isfinite(roi_mean) and math.isfinite(roi_sd) and roi_sd > 0):
raise ValueError(
f"pooled ROI moments are not finite-positive (roi_mean={roi_mean!r}, "
f"roi_sd={roi_sd!r}); the per-experiment magnitudes overflowed or "
f"underflowed - rescale the units"
)
if roi_mean <= 0:
raise ValueError(
f"Pooled ROI mean is {roi_mean:.6g} <= 0, which cannot map to Meridian's "
f"lognormal {parameter} prior (strictly positive support). Options: pool "
f"with additional experiments, use a wider prior from a defensible positive "
f"range, or calibrate via Meridian's contribution/coefficient "
f"parameterizations manually."
)
# log_term = log1p((roi_sd / roi_mean)**2), computed in the log domain so an
# extreme coefficient of variation cannot overflow the squaring before the
# finiteness guard: logaddexp(0, 2*ln(sd/mean)) == log(1 + (sd/mean)**2), and
# it also keeps a tiny relative SE from rounding sigma to 0.
log_term = float(np.logaddexp(0.0, 2.0 * (math.log(roi_sd) - math.log(roi_mean))))
sigma = math.sqrt(log_term)
mu = math.log(roi_mean) - 0.5 * log_term
if not (math.isfinite(mu) and math.isfinite(sigma) and sigma > 0):
raise ValueError(
f"lognormal parameters are not finite (mu={mu!r}, sigma={sigma!r}) - the "
f"ROI mean/sd ratio is outside the representable range; rescale the outcome "
f"or spend units and re-export"
)
per_experiment = tuple(
ExperimentROI(roi=r, roi_sd=s, spend=sp, weight=w)
for r, s, sp, w in zip(rois, sds, spend_vals, weights)
)
return MeridianROIPrior(
roi_mean=roi_mean,
roi_sd=roi_sd,
mu=mu,
sigma=sigma,
parameter=parameter,
per_experiment=per_experiment,
)