"""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

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,

}

@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=())

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],

}

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."

)

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,

)