 Run in Google Colab
|
 View source on GitHub
|
This is a walkthrough of full-funnel MMM in Meridian. It uses Branded Google Query Volume (bGQV) as an example of a brand equity variable. For more on when one might want to use a full-funnel model, see Full-funnel MMM: unifying upper-funnel and lower-funnel measurement.
Step 0: Install and Setup
Ensure you are running on a GPU-enabled runtime. We install the google-meridian package and import necessary modules.
# Install meridian: from PyPI @ latest release
!pip install --upgrade google-meridian[colab,and-cuda,scenarioplanner,schema]
import contextlib
import dataclasses
import time
from typing import Any, Mapping, Optional, Sequence
import altair as alt
import arviz as az
from google.colab import auth
from IPython.display import HTML
import jax
from meridian import backend
from meridian import constants
from meridian.analysis import analyzer
from meridian.analysis import optimizer
from meridian.analysis import tensors
from meridian.analysis import visualizer
from meridian.analysis.review import reviewer
from meridian.data import load
from meridian.model import equations
from meridian.model import model
from meridian.model import spec
from meridian.schema.processors import budget_optimization_processor
from meridian.schema.processors import marketing_processor
from meridian.schema.processors import model_fit_processor
from meridian.schema.processors import model_processor
from meridian.schema.utils import date_range_bucketing
from meridian.templates import formatter
import numpy as np
import pandas as pd
from scenarioplanner import mmm_ui_proto_generator as mmm_ui_gen
from scenarioplanner.converters import sheets
from scenarioplanner.converters.dataframe import dataframe_model_converter
from scenarioplanner.linkingapi import url_generator
# Verify GPU availability
print(
"Num GPUs Available: ",
len([d for d in jax.devices() if d.platform == "gpu"]),
)
Collecting google-meridian[and-cuda,colab,scenarioplanner,schema]
Downloading google_meridian-2.1.0-py3-none-any.whl.metadata (10 kB)
Collecting arviz<0.20.0 (from google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading arviz-0.19.0-py3-none-any.whl.metadata (8.9 kB)
Requirement already satisfied: altair>=5 in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (5.5.0)
Requirement already satisfied: babel in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.18.0)
Collecting bidict (from google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading bidict-0.24.1-py3-none-any.whl.metadata (8.7 kB)
Requirement already satisfied: immutabledict in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (4.3.1)
Requirement already satisfied: jax<1.0.0,>=0.7.2 in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.11.1)
Requirement already satisfied: jaxlib<1.0.0,>=0.7.2 in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.11.1)
Requirement already satisfied: joblib in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.6.0)
Requirement already satisfied: matplotlib<3.11.0 in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.10.0)
Collecting natsort<8,>=7.1.1 (from google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading natsort-7.1.1-py3-none-any.whl.metadata (22 kB)
Collecting numpy<2.4.0,>=2.0.2 (from google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading numpy-2.3.5-cp313-cp313-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl.metadata (62 kB)
Requirement already satisfied: pandas<3,>=2.2.2 in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.2.3)
Requirement already satisfied: scipy<2,>=1.13.1 in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.16.3)
Collecting statsmodels<0.15,>=0.14.5 (from google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading statsmodels-0.14.6-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl.metadata (9.5 kB)
Requirement already satisfied: tensorflow<2.22,>=2.21.0 in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.21.0)
Collecting tfp-nightly==0.26.0.dev20260130 (from tfp-nightly[substrates-jax]==0.26.0.dev20260130->google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading tfp_nightly-0.26.0.dev20260130-py2.py3-none-any.whl.metadata (13 kB)
Requirement already satisfied: tf-keras<2.22,>=2.21.0 in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.21.0)
Requirement already satisfied: xarray in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (2025.12.0)
Requirement already satisfied: psutil in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (5.9.5)
Collecting python-calamine (from google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading python_calamine-0.9.0-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (3.1 kB)
Collecting mmm-proto-schema>=1.4.0 (from google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading mmm_proto_schema-1.6.0-py3-none-any.whl.metadata (1.7 kB)
Collecting semver (from google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading semver-3.1.0-py3-none-any.whl.metadata (6.9 kB)
Requirement already satisfied: google-api-python-client in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.201.0)
Requirement already satisfied: google-auth in /usr/local/lib/python3.13/dist-packages (from google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.49.0)
Requirement already satisfied: absl-py in /usr/local/lib/python3.13/dist-packages (from tfp-nightly==0.26.0.dev20260130->tfp-nightly[substrates-jax]==0.26.0.dev20260130->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.0.0)
Requirement already satisfied: six>=1.10.0 in /usr/local/lib/python3.13/dist-packages (from tfp-nightly==0.26.0.dev20260130->tfp-nightly[substrates-jax]==0.26.0.dev20260130->google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.17.0)
Requirement already satisfied: decorator in /usr/local/lib/python3.13/dist-packages (from tfp-nightly==0.26.0.dev20260130->tfp-nightly[substrates-jax]==0.26.0.dev20260130->google-meridian[and-cuda,colab,scenarioplanner,schema]) (4.4.2)
Requirement already satisfied: cloudpickle>=1.3 in /usr/local/lib/python3.13/dist-packages (from tfp-nightly==0.26.0.dev20260130->tfp-nightly[substrates-jax]==0.26.0.dev20260130->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.1.2)
Requirement already satisfied: gast>=0.3.2 in /usr/local/lib/python3.13/dist-packages (from tfp-nightly==0.26.0.dev20260130->tfp-nightly[substrates-jax]==0.26.0.dev20260130->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.7.0)
Requirement already satisfied: dm-tree in /usr/local/lib/python3.13/dist-packages (from tfp-nightly==0.26.0.dev20260130->tfp-nightly[substrates-jax]==0.26.0.dev20260130->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.1.10)
WARNING: tfp-nightly 0.26.0.dev20260130 does not provide the extra 'substrates-jax'
Requirement already satisfied: setuptools>=60.0.0 in /usr/local/lib/python3.13/dist-packages (from arviz<0.20.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (80.10.2)
Requirement already satisfied: packaging in /usr/local/lib/python3.13/dist-packages (from arviz<0.20.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (26.3)
Requirement already satisfied: h5netcdf>=1.0.2 in /usr/local/lib/python3.13/dist-packages (from arviz<0.20.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.8.2)
Requirement already satisfied: typing-extensions>=4.1.0 in /usr/local/lib/python3.13/dist-packages (from arviz<0.20.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (4.16.0)
Requirement already satisfied: xarray-einstats>=0.3 in /usr/local/lib/python3.13/dist-packages (from arviz<0.20.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.11.0)
Requirement already satisfied: ml_dtypes>=0.5.0 in /usr/local/lib/python3.13/dist-packages (from jax<1.0.0,>=0.7.2->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.6.0)
Requirement already satisfied: opt_einsum in /usr/local/lib/python3.13/dist-packages (from jax<1.0.0,>=0.7.2->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.4.0)
Collecting jax-cuda12-plugin<=0.11.1,>=0.11.1 (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading jax_cuda12_plugin-0.11.1-cp313-cp313-manylinux_2_27_x86_64.whl.metadata (2.0 kB)
Collecting jax-cuda12-pjrt==0.11.1 (from jax-cuda12-plugin<=0.11.1,>=0.11.1->jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading jax_cuda12_pjrt-0.11.1-py3-none-manylinux_2_27_x86_64.whl.metadata (580 bytes)
Requirement already satisfied: nvidia-cublas-cu12>=12.1.3.1 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (12.9.2.10)
Requirement already satisfied: nvidia-cuda-cupti-cu12>=12.1.105 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (12.9.79)
Requirement already satisfied: nvidia-cuda-nvcc-cu12>=12.6.85 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (12.9.86)
Requirement already satisfied: nvidia-cuda-runtime-cu12>=12.1.105 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (12.9.79)
Requirement already satisfied: nvidia-cudnn-cu12<10.0,>=9.8 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (9.27.0.42)
Requirement already satisfied: nvidia-cufft-cu12>=11.0.2.54 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (11.4.1.4)
Requirement already satisfied: nvidia-cusolver-cu12>=11.4.5.107 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (11.7.5.82)
Requirement already satisfied: nvidia-cusparse-cu12>=12.1.0.106 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (12.5.10.65)
Requirement already satisfied: nvidia-nccl-cu12>=2.18.1 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.32.3)
Requirement already satisfied: nvidia-nvjitlink-cu12>=12.1.105 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (12.9.86)
Requirement already satisfied: nvidia-cuda-nvrtc-cu12>=12.1.55 in /usr/local/lib/python3.13/dist-packages (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (12.9.86)
Collecting nvidia-nvshmem-cu12>=3.2.5 (from jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading nvidia_nvshmem_cu12-3.8.0-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl.metadata (2.2 kB)
Requirement already satisfied: contourpy>=1.0.1 in /usr/local/lib/python3.13/dist-packages (from matplotlib<3.11.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.4.0)
Requirement already satisfied: cycler>=0.10 in /usr/local/lib/python3.13/dist-packages (from matplotlib<3.11.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.12.1)
Requirement already satisfied: fonttools>=4.22.0 in /usr/local/lib/python3.13/dist-packages (from matplotlib<3.11.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (4.68.0)
Requirement already satisfied: kiwisolver>=1.3.1 in /usr/local/lib/python3.13/dist-packages (from matplotlib<3.11.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.5.1)
Requirement already satisfied: pillow>=8 in /usr/local/lib/python3.13/dist-packages (from matplotlib<3.11.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (11.3.0)
Requirement already satisfied: pyparsing>=2.3.1 in /usr/local/lib/python3.13/dist-packages (from matplotlib<3.11.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.3.3)
Requirement already satisfied: python-dateutil>=2.7 in /usr/local/lib/python3.13/dist-packages (from matplotlib<3.11.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.9.0.post0)
Requirement already satisfied: pytz>=2020.1 in /usr/local/lib/python3.13/dist-packages (from pandas<3,>=2.2.2->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2025.2)
Requirement already satisfied: tzdata>=2022.7 in /usr/local/lib/python3.13/dist-packages (from pandas<3,>=2.2.2->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2026.6)
Requirement already satisfied: patsy>=0.5.6 in /usr/local/lib/python3.13/dist-packages (from statsmodels<0.15,>=0.14.5->google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.0.3)
Requirement already satisfied: astunparse>=1.6.0 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.6.3)
Requirement already satisfied: flatbuffers>=25.9.23 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (25.12.19)
Requirement already satisfied: google_pasta>=0.1.1 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.2.0)
Requirement already satisfied: libclang>=13.0.0 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (18.1.1)
Requirement already satisfied: protobuf<8.0.0,>=6.31.1 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (6.33.6)
Requirement already satisfied: requests<3,>=2.21.0 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.32.4)
Requirement already satisfied: termcolor>=1.1.0 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.3.0)
Requirement already satisfied: wrapt>=1.11.0 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.5.1)
Requirement already satisfied: grpcio<2.0,>=1.24.3 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.85.0)
Requirement already satisfied: keras>=3.12.0 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.13.2)
Requirement already satisfied: h5py<3.15.0,>=3.11.0 in /usr/local/lib/python3.13/dist-packages (from tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.14.0)
Requirement already satisfied: charset_normalizer<4,>=2 in /usr/local/lib/python3.13/dist-packages (from requests<3,>=2.21.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.4.9)
Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.13/dist-packages (from requests<3,>=2.21.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.20)
Requirement already satisfied: urllib3<3,>=1.21.1 in /usr/local/lib/python3.13/dist-packages (from requests<3,>=2.21.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.5.0)
Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.13/dist-packages (from requests<3,>=2.21.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2026.10.23)
Requirement already satisfied: nvidia-curand-cu12<11.0,>=10.3.6.82 in /usr/local/lib/python3.13/dist-packages (from tensorflow[and-cuda]<2.22,>=2.21.0; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema]) (10.3.10.19)
Requirement already satisfied: jinja2 in /usr/local/lib/python3.13/dist-packages (from altair>=5->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.1.6)
Requirement already satisfied: jsonschema>=3.0 in /usr/local/lib/python3.13/dist-packages (from altair>=5->google-meridian[and-cuda,colab,scenarioplanner,schema]) (4.26.0)
Requirement already satisfied: narwhals>=1.14.2 in /usr/local/lib/python3.13/dist-packages (from altair>=5->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.28.0)
Requirement already satisfied: wheel<1.0,>=0.23.0 in /usr/local/lib/python3.13/dist-packages (from astunparse>=1.6.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.48.0)
Requirement already satisfied: attrs>=18.2.0 in /usr/local/lib/python3.13/dist-packages (from dm-tree->tfp-nightly==0.26.0.dev20260130->tfp-nightly[substrates-jax]==0.26.0.dev20260130->google-meridian[and-cuda,colab,scenarioplanner,schema]) (26.1.0)
Requirement already satisfied: jsonschema-specifications>=2023.03.6 in /usr/local/lib/python3.13/dist-packages (from jsonschema>=3.0->altair>=5->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2025.9.1)
Requirement already satisfied: referencing>=0.28.4 in /usr/local/lib/python3.13/dist-packages (from jsonschema>=3.0->altair>=5->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.37.0)
Requirement already satisfied: rpds-py>=0.25.0 in /usr/local/lib/python3.13/dist-packages (from jsonschema>=3.0->altair>=5->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2026.9.1)
Requirement already satisfied: rich in /usr/local/lib/python3.13/dist-packages (from keras>=3.12.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (13.9.4)
Requirement already satisfied: namex in /usr/local/lib/python3.13/dist-packages (from keras>=3.12.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.1.0)
Requirement already satisfied: optree in /usr/local/lib/python3.13/dist-packages (from keras>=3.12.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.20.0)
Requirement already satisfied: googleapis-common-protos in /usr/local/lib/python3.13/dist-packages (from mmm-proto-schema>=1.4.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.76.0)
Collecting nvidia-cuda-cccl-cu12 (from nvidia-nvshmem-cu12>=3.2.5->jax-cuda12-plugin[with-cuda]<=0.11.1,>=0.11.1; extra == "cuda12"->jax[cuda12]<1.0.0,>=0.7.2; extra == "and-cuda"->google-meridian[and-cuda,colab,scenarioplanner,schema])
Downloading nvidia_cuda_cccl_cu12-12.9.27-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl.metadata (1.7 kB)
Requirement already satisfied: httplib2<1.0.0,>=0.19.0 in /usr/local/lib/python3.13/dist-packages (from google-api-python-client->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.32.1)
Requirement already satisfied: google-auth-httplib2<1.0.0,>=0.2.0 in /usr/local/lib/python3.13/dist-packages (from google-api-python-client->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.4.4)
Requirement already satisfied: google-api-core!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.0,<3.0.0,>=1.31.5 in /usr/local/lib/python3.13/dist-packages (from google-api-python-client->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.30.3)
Requirement already satisfied: uritemplate<5,>=3.0.1 in /usr/local/lib/python3.13/dist-packages (from google-api-python-client->google-meridian[and-cuda,colab,scenarioplanner,schema]) (4.2.0)
Requirement already satisfied: proto-plus<2.0.0,>=1.22.3 in /usr/local/lib/python3.13/dist-packages (from google-api-core!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.0,<3.0.0,>=1.31.5->google-api-python-client->google-meridian[and-cuda,colab,scenarioplanner,schema]) (1.29.0)
Requirement already satisfied: pyasn1-modules>=0.2.1 in /usr/local/lib/python3.13/dist-packages (from google-auth->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.4.2)
Requirement already satisfied: cryptography>=38.0.3 in /usr/local/lib/python3.13/dist-packages (from google-auth->google-meridian[and-cuda,colab,scenarioplanner,schema]) (50.0.2)
Requirement already satisfied: rsa<5,>=3.1.4 in /usr/local/lib/python3.13/dist-packages (from google-auth->google-meridian[and-cuda,colab,scenarioplanner,schema]) (4.9.1)
Requirement already satisfied: pyasn1>=0.1.3 in /usr/local/lib/python3.13/dist-packages (from rsa<5,>=3.1.4->google-auth->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.6.5)
Requirement already satisfied: cffi>=2.0.0 in /usr/local/lib/python3.13/dist-packages (from cryptography>=38.0.3->google-auth->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.1.1)
Requirement already satisfied: pycparser in /usr/local/lib/python3.13/dist-packages (from cffi>=2.0.0->cryptography>=38.0.3->google-auth->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.12)
Requirement already satisfied: MarkupSafe>=2.0 in /usr/local/lib/python3.13/dist-packages (from jinja2->altair>=5->google-meridian[and-cuda,colab,scenarioplanner,schema]) (3.0.4)
Requirement already satisfied: markdown-it-py>=2.2.0 in /usr/local/lib/python3.13/dist-packages (from rich->keras>=3.12.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (4.2.0)
Requirement already satisfied: pygments<3.0.0,>=2.13.0 in /usr/local/lib/python3.13/dist-packages (from rich->keras>=3.12.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (2.21.0)
Requirement already satisfied: mdurl~=0.1 in /usr/local/lib/python3.13/dist-packages (from markdown-it-py>=2.2.0->rich->keras>=3.12.0->tensorflow<2.22,>=2.21.0->google-meridian[and-cuda,colab,scenarioplanner,schema]) (0.1.2)
Downloading google_meridian-2.1.0-py3-none-any.whl (635 kB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 635.9/635.9 kB 8.2 MB/s 0:00:00
Downloading tfp_nightly-0.26.0.dev20260130-py2.py3-none-any.whl (7.0 MB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 7.0/7.0 MB 38.5 MB/s 0:00:00
Downloading arviz-0.19.0-py3-none-any.whl (1.7 MB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 1.7/1.7 MB 57.7 MB/s 0:00:00
Downloading jax_cuda12_plugin-0.11.1-cp313-cp313-manylinux_2_27_x86_64.whl (8.6 MB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 8.6/8.6 MB 28.7 MB/s 0:00:00
Downloading jax_cuda12_pjrt-0.11.1-py3-none-manylinux_2_27_x86_64.whl (184.9 MB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 184.9/184.9 MB 11.2 MB/s 0:00:16
Downloading natsort-7.1.1-py3-none-any.whl (35 kB)
Downloading numpy-2.3.5-cp313-cp313-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl (16.6 MB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 16.6/16.6 MB 1.9 MB/s 0:00:08
Downloading statsmodels-0.14.6-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl (10.3 MB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 10.3/10.3 MB 1.5 MB/s 0:00:06
Downloading mmm_proto_schema-1.6.0-py3-none-any.whl (76 kB)
Downloading nvidia_nvshmem_cu12-3.8.0-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl (280.8 MB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 280.8/280.8 MB 1.5 MB/s 0:02:22
Downloading bidict-0.24.1-py3-none-any.whl (36 kB)
Downloading nvidia_cuda_cccl_cu12-12.9.27-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl (3.2 MB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 3.2/3.2 MB 2.3 MB/s 0:00:01
Downloading python_calamine-0.9.0-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (934 kB)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 935.0/935.0 kB 2.2 MB/s 0:00:00
Downloading semver-3.1.0-py3-none-any.whl (18 kB)
Installing collected packages: jax-cuda12-pjrt, semver, python-calamine, nvidia-cuda-cccl-cu12, numpy, natsort, jax-cuda12-plugin, bidict, nvidia-nvshmem-cu12, tfp-nightly, statsmodels, mmm-proto-schema, arviz, google-meridian
Attempting uninstall: numpy
Found existing installation: numpy 2.4.6
Uninstalling numpy-2.4.6:
Successfully uninstalled numpy-2.4.6
Attempting uninstall: natsort
Found existing installation: natsort 8.4.0
Uninstalling natsort-8.4.0:
Successfully uninstalled natsort-8.4.0
Attempting uninstall: statsmodels
Found existing installation: statsmodels 0.15.0
Uninstalling statsmodels-0.15.0:
Successfully uninstalled statsmodels-0.15.0
Attempting uninstall: arviz
Found existing installation: arviz 0.22.0
Uninstalling arviz-0.22.0:
Successfully uninstalled arviz-0.22.0
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 14/14 [google-meridian]
Successfully installed arviz-0.19.0 bidict-0.24.1 google-meridian-2.1.0 jax-cuda12-pjrt-0.11.1 jax-cuda12-plugin-0.11.1 mmm-proto-schema-1.6.0 natsort-7.1.1 numpy-2.3.5 nvidia-cuda-cccl-cu12-12.9.27 nvidia-nvshmem-cu12-3.8.0 python-calamine-0.9.0 semver-3.1.0 statsmodels-0.14.6 tfp-nightly-0.26.0.dev20260130
ERROR:jax._src.xla_bridge:Jax plugin configuration error: Exception when calling jax_plugins.xla_cuda13.initialize()
Traceback (most recent call last):
File "/usr/local/lib/python3.13/dist-packages/jax/_src/xla_bridge.py", line 495, in discover_pjrt_plugins
plugin_module.initialize()
~~~~~~~~~~~~~~~~~~~~~~~~^^
File "/usr/local/lib/python3.13/dist-packages/jax_plugins/xla_cuda13/__init__.py", line 373, in initialize
c_api = xb.register_plugin(
'cuda', priority=500, library_path=str(path), options=options
)
File "/usr/local/lib/python3.13/dist-packages/jax/_src/xla_bridge.py", line 637, in register_plugin
c_api = xla_client.load_pjrt_plugin_dynamically(plugin_name, library_path)
File "/usr/local/lib/python3.13/dist-packages/jaxlib/xla_client.py", line 115, in load_pjrt_plugin_dynamically
return _xla.load_pjrt_plugin(plugin_name, library_path, c_api=None)
~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
jax.errors.JaxRuntimeError: ALREADY_EXISTS: PJRT_Api already exists for device type cuda
Num GPUs Available: 1
Step 1: Load and Inspect Simulated Full-Funnel Data
We load a simulated dataset (geo_full_funnel_data.csv) that models the full-funnel marketing environment:
- Geos and Time: DMA-level weekly data across geos, times, and population.
- Paid Media Channels:
AwarenessVideo(impressions and spend): Brand awareness channel.PerformanceNative(impressions and spend): Performance conversion channel.PerformanceDisplay(impressions and spend): Performance conversion channel.
- KPI:
conversionsandrevenue_per_conversion. - Controls & Mediators:
GenericGQV_control: Confounder affecting all media channels and conversions.BrandedGQV_control: Mediator driven byAwarenessVideoand confounding performance channels.
Causal Relationships in the Data:
The simulated data is generated with the following relationships:
We see that BrandedGQV is a mediator for AwarenessVideo, but a confounder for PerformanceNative and PerformanceDisplay. This DAG is a special case of the full-funnel causal graph, making full-funnel MMM appropriate.
# Load dataset
DATA_URL = "https://raw.githubusercontent.com/google/meridian/refs/heads/main/meridian/data/simulated_data/csv/geo_full_funnel_data.csv"
df = pd.read_csv(DATA_URL)
HTML(f'<div style="overflow-x: auto;">{df.head().to_html()}</div>')
Step 2: Fitting the Full-Funnel Models
We now fit the two-stage model:
- Stage 1 (
mmm1s): ModelsBrandedGQVas the response variable driven byAwarenessVideo. - Stage 2 (
mmm2s): Modelsconversionsas the final KPI driven by all media channels. It includesGenericGQVas a control, andBrandedGQVas anorganic_mediachannel withsaturation_spec={'BrandedGQV_control': 'none'}.
Each of the Stage 1 and Stage 2 models are Meridian models and instantiations of the Meridian class.
# Shared sampling configuration.
SAMPLING_KWARGS = dict(
n_chains=4,
n_adapt=1000,
n_burnin=1000,
n_keep=500,
)
Stage 1 MMM: Brand Marketing → Brand Equity
Just like a single-stage Meridian model, each Meridian model in this two-stage setup needs all confounders between its treatments and the response variable.
In the Stage 1 model, we model the causal effect of channels expected to cause brand equity ("Brand marketing") on the brand equity variable itself. To model this causal effect, we must include confounding variables between Brand marketing and the Brand equity variable. In this illustrative example, we do not simulate such a variable for simplicity.
stage1_media = ["AwarenessVideo_impression"]
stage1_media_spend = ["AwarenessVideo_spend"]
loader1s = load.DataFrameDataLoader(
df,
kpi_type="non_revenue",
# Notice that `revenue_per_kpi` argument is not used.
coord_to_columns=load.CoordToColumns(
time="time",
geo="geo",
population="population",
kpi="BrandedGQV_control",
media=stage1_media,
media_spend=stage1_media_spend,
),
media_to_channel={c: c.replace("_impression", "") for c in stage1_media},
media_spend_to_channel={
c: c.replace("_spend", "") for c in stage1_media_spend
},
)
mmm1s = model.Meridian(
input_data=loader1s.load(),
model_spec=spec.ModelSpec(
# One knot per month
knots=36,
),
)
mmm1s.sample_prior(SAMPLING_KWARGS["n_keep"])
mmm1s.sample_posterior(**SAMPLING_KWARGS)
/usr/local/lib/python3.13/dist-packages/meridian/data/input_data.py:605: UserWarning: Consider setting custom priors, as kpi_type was specified as `non_revenue` with no `revenue_per_kpi` being set. Otherwise, the total media contribution prior will be used with `p_mean=0.4` and `p_sd=0.2`. Further documentation available at https://developers.google.com/meridian/docs/advanced-modeling/unknown-revenue-kpi-custom#set-total-paid-media-contribution-prior warnings.warn( /usr/local/lib/python3.13/dist-packages/meridian/model/context.py:1447: UserWarning: Consider setting custom ROI priors, as kpi_type was specified as `non_revenue` with no `revenue_per_kpi` being set. Otherwise, the total media contribution prior will be used with `p_mean=0.4` and `p_sd=0.2`. Further documentation available at https://developers.google.com/meridian/docs/advanced-modeling/unknown-revenue-kpi-custom#set-total-paid-media-contribution-prior warnings.warn( /usr/local/lib/python3.13/dist-packages/arviz/data/inference_data.py:157: UserWarning: trace group is not defined in the InferenceData scheme warnings.warn( /usr/local/lib/python3.13/dist-packages/arviz/data/inference_data.py:1647: UserWarning: trace group is not defined in the InferenceData scheme warnings.warn(
Stage 2 MMM: Marketing → KPI
In the Stage 2 model, we model the causal effect of all marketing treatment variables and the brand equity variable on the KPI. In this second stage, the brand equity variable is also a treatment variable (modeled as organic media). This allows us to measure the incremental effect of brand marketing on the KPI via brand equity. Like a single-stage MMM, we must include any confounding variables between our treatment variables and the KPI.
stage2_media = [
"PerformanceNative_impression",
"PerformanceDisplay_impression",
"AwarenessVideo_impression",
]
stage2_media_spend = [
"PerformanceNative_spend",
"PerformanceDisplay_spend",
"AwarenessVideo_spend",
]
loader2s = load.DataFrameDataLoader(
df,
kpi_type="non_revenue",
coord_to_columns=load.CoordToColumns(
time="time",
geo="geo",
population="population",
kpi="conversions",
revenue_per_kpi="revenue_per_conversion",
media=stage2_media,
media_spend=stage2_media_spend,
controls=["GenericGQV_control"],
organic_media=["BrandedGQV_control"],
),
media_to_channel={c: c.replace("_impression", "") for c in stage2_media},
media_spend_to_channel={
c: c.replace("_spend", "") for c in stage2_media_spend
},
)
mmm2s = model.Meridian(
input_data=loader2s.load(),
model_spec=spec.ModelSpec(
# One knot per month
knots=36,
# Required for the brand equity variable
saturation_spec={"BrandedGQV_control": "none"},
# Controls are not population-scaled by default. Meridian recommends
# population scaling GQV.
population_scaled_controls=["GenericGQV_control"],
),
)
mmm2s.sample_prior(SAMPLING_KWARGS["n_keep"])
mmm2s.sample_posterior(**SAMPLING_KWARGS)
/usr/local/lib/python3.13/dist-packages/arviz/data/inference_data.py:157: UserWarning: trace group is not defined in the InferenceData scheme warnings.warn( /usr/local/lib/python3.13/dist-packages/arviz/data/inference_data.py:1647: UserWarning: trace group is not defined in the InferenceData scheme warnings.warn(
Model Health Diagnostics
Before proceeding to combined post-modeling analysis, we verify that both MCMC models are converged and healthy (health scores \(> 70\)). A comprehensive negative baseline check that accounts for the expected outcome of the full-funnel model will be presented in Step 5.
# Diagnostic checks for Stage 1 Model
reviewer.ModelReviewer(mmm1s).run()
/tmp/ipykernel_2365/2495845249.py:2: DeprecationWarning: The `meridian` argument is deprecated. Please use `model_context` and `inference_data` instead. reviewer.ModelReviewer(mmm1s).run() /usr/local/lib/python3.13/dist-packages/meridian/analysis/analyzer.py:375: UserWarning: Revenue analysis is not available when `revenue_per_kpi` is unknown. Defaulting to KPI analysis. warnings.warn( ======================================== Model Quality Checks ======================================== Overall Status: PASS Summary: Passed: No major quality issues were identified. Health Score: 96.0 Check Results: ---------------------------------------- Convergence Check: Status: PASS Recommendation: The model has likely converged, as all parameters have R-hat values < 1.2. ---------------------------------------- Baseline Check: Status: PASS Recommendation: The posterior probability that the baseline is negative is 0.00. We recommend visually inspecting the baseline time series in the Model Fit charts to confirm this. ---------------------------------------- BayesianPPP Check: Status: PASS Recommendation: The Bayesian posterior predictive p-value is 0.97. The observed total outcome is consistent with the model's posterior predictive distribution. ---------------------------------------- GoodnessOfFit Check: Status: PASS Recommendation: R-squared = 0.5435, MAPE = 0.0499, and wMAPE = 0.0497. These goodness-of-fit metrics are intended for guidance and relative comparison. ---------------------------------------- PriorPosteriorShift Check: Status: PASS Recommendation: The model has successfully learned from the data. This is a positive sign that your data was informative.
reviewer.ModelReviewer(mmm2s).run()
/tmp/ipykernel_2365/4250511027.py:1: DeprecationWarning: The `meridian` argument is deprecated. Please use `model_context` and `inference_data` instead.
reviewer.ModelReviewer(mmm2s).run()
/usr/local/lib/python3.13/dist-packages/altair/utils/core.py:264: UserWarning: I don't know how to infer vegalite type from 'empty'. Defaulting to nominal.
warnings.warn(
/usr/local/lib/python3.13/dist-packages/altair/utils/core.py:264: UserWarning: I don't know how to infer vegalite type from 'empty'. Defaulting to nominal.
warnings.warn(
/usr/local/lib/python3.13/dist-packages/altair/utils/core.py:264: UserWarning: I don't know how to infer vegalite type from 'empty'. Defaulting to nominal.
warnings.warn(
/usr/local/lib/python3.13/dist-packages/altair/utils/core.py:264: UserWarning: I don't know how to infer vegalite type from 'empty'. Defaulting to nominal.
warnings.warn(
/usr/local/lib/python3.13/dist-packages/meridian/analysis/review/plots.py:588: FutureWarning: Downcasting object dtype arrays on .fillna, .ffill, .bfill is deprecated and will change in a future version. Call result.infer_objects(copy=False) instead. To opt-in to the future behavior, set `pd.set_option('future.no_silent_downcasting', True)`
df[constants.IS_MAX] = df[constants.IS_MAX].fillna(False)
========================================
Model Quality Checks
========================================
Overall Status: PASS
Summary: Passed: No major quality issues were identified.
Health Score: 92.8
Check Results:
----------------------------------------
Convergence Check:
Status: PASS
Recommendation: The model has likely converged, as all parameters have R-hat values < 1.2.
----------------------------------------
Baseline Check:
Status: PASS
Recommendation: The posterior probability that the baseline is negative is 0.00. We recommend visually inspecting the baseline time series in the Model Fit charts to confirm this.
----------------------------------------
BayesianPPP Check:
Status: PASS
Recommendation: The Bayesian posterior predictive p-value is 0.98. The observed total outcome is consistent with the model's posterior predictive distribution.
----------------------------------------
GoodnessOfFit Check:
Status: PASS
Recommendation: R-squared = 0.4693, MAPE = 0.0563, and wMAPE = 0.0555. These goodness-of-fit metrics are intended for guidance and relative comparison.
----------------------------------------
PriorPosteriorShift Check:
Status: PASS
Recommendation: The model has successfully learned from the data. This is a positive sign that your data was informative.
===================================================================================================================
Channel Calibration Recommendation
===================================================================================================================
Channel | Calibration Score | High ROI | Low ROI | High Variance ROI | Potential Bias
-------------------------------------------------------------------------------------------------------------------
PerformanceNative | 91.2 | Non-Driver | Non-Driver | Non-Driver | Non-Driver
PerformanceDisplay | 84.9 | Non-Driver | Non-Driver | Non-Driver | Non-Driver
AwarenessVideo | 90.2 | Non-Driver | Non-Driver | Non-Driver | Non-Driver
-------------------------------------------------------------------------------------------------------------------
Step 3: Upper-funnel (Stage 1) media analysis: quantify impact on brand search
Before analyzing total downstream conversions, evaluate the efficiency of brand advertising (AwarenessVideo) in driving brand search query volume (BrandedGQV_control) in the Stage 1 model (mmm1s).
Because the Stage 1 outcome (BrandedGQV_control) is a non-revenue metric, evaluate physical query units by passing use_kpi=True. This section examines the following:
- Response curves: diminishing marginal returns and Hill saturation of brand ad spend on query volume.
- Cost per incremental query (CPIK): cost-efficiency of incremental query generation with credible intervals.
- Search volume decomposition: contribution of organic baseline demand versus advertising lift.
# 1. Response Curves on Search Volume
visualizer.MediaEffects(meridian=mmm1s).plot_response_curves()
/usr/local/lib/python3.13/dist-packages/meridian/analysis/analyzer.py:375: UserWarning: Revenue analysis is not available when `revenue_per_kpi` is unknown. Defaulting to KPI analysis. warnings.warn(
# 2. Cost Per Incremental Query (CPIK)
visualizer.MediaSummary(meridian=mmm1s).plot_cpik(include_ci=True)
/usr/local/lib/python3.13/dist-packages/meridian/analysis/analyzer.py:375: UserWarning: Revenue analysis is not available when `revenue_per_kpi` is unknown. Defaulting to KPI analysis. warnings.warn(
# 3. Search Volume Decomposition Waterfall Chart
visualizer.MediaSummary(
meridian=mmm1s
).plot_contribution_waterfall_chart()
/usr/local/lib/python3.13/dist-packages/meridian/analysis/analyzer.py:375: UserWarning: Revenue analysis is not available when `revenue_per_kpi` is unknown. Defaulting to KPI analysis. warnings.warn(
Step 4: Custom AnalyzerFullFunnel Definition
This step defines class a AnalyzerFullFunnel class which extends Meridian's Analyzer class for full-funnel inference. It combines the stage 1 and stage 2 models allowing for the stage 1 model to correct for the bias induced in stage 2 by included the mediating brand equity variable. It does the following,
- Incremental Outcome: Combines the effect estimated in the second model with the indirect effect through the brand equity variable.
- Expected Outcome: Computes expected outcomes across both stages.
- Model Compatibility Verification: Enforces matching geos, timepoints, MCMC trace lengths, and verifies that the brand equity variable is specified in Stage 2 as an
organic_mediavariable withsaturation_spec='none'.
Since downstream causal estimates, including ROI, marginal ROI, response curves, baseline, and budget optimization, rely on these, they are all updated for full-funnel MMM.
# @title {display-mode: "form"}
def _map_new_data_to_mediator(
new_data: tensors.DataTensors | None,
model_context_2s: Any,
model_context_1s: Any,
) -> tensors.DataTensors | None:
"""Map new_data variables from stage 2 so they can be passed to stage 1."""
if new_data is None:
return None
mapped_kwargs = {}
def map_tensor(tensor_name, channels_2s, channels_1s, historical_tensor_1s):
tensor_2s = getattr(new_data, tensor_name)
if tensor_2s is None:
return None
tensors_list = []
is_backend_tensor = isinstance(tensor_2s, backend.Tensor)
for ch in channels_1s:
if ch in channels_2s:
idx = list(channels_2s).index(ch)
tensors_list.append(tensor_2s[..., idx : idx + 1])
else:
idx = list(channels_1s).index(ch)
historical_slice = historical_tensor_1s[..., idx : idx + 1]
if is_backend_tensor:
historical_slice = backend.to_tensor(
historical_slice, dtype=backend.float_dtype
)
# Broadcast historical_slice shape to match tensor_2s batch dims if
# optimizer uses batch evaluation
if (
hasattr(tensor_2s, "shape")
and tensor_2s.shape[:-1] != historical_slice.shape[:-1]
):
if is_backend_tensor:
historical_slice = backend.broadcast_to(
historical_slice, tensor_2s.shape[:-1] + (1,)
)
else:
historical_slice = np.broadcast_to(
historical_slice, tensor_2s.shape[:-1] + (1,)
)
tensors_list.append(historical_slice)
if is_backend_tensor:
return backend.concatenate(tensors_list, axis=-1)
else:
return np.concatenate(tensors_list, axis=-1)
# Defensively extract channel names only when they exist
def get_channels(channel_coord):
return channel_coord.values if channel_coord is not None else []
mapped_kwargs["media"] = map_tensor(
"media",
get_channels(model_context_2s.input_data.media_channel),
get_channels(model_context_1s.input_data.media_channel),
model_context_1s.input_data.media,
)
mapped_kwargs["media_spend"] = map_tensor(
"media_spend",
get_channels(model_context_2s.input_data.media_channel),
get_channels(model_context_1s.input_data.media_channel),
model_context_1s.input_data.media_spend,
)
mapped_kwargs["reach"] = map_tensor(
"reach",
get_channels(model_context_2s.input_data.rf_channel),
get_channels(model_context_1s.input_data.rf_channel),
model_context_1s.input_data.reach,
)
mapped_kwargs["frequency"] = map_tensor(
"frequency",
get_channels(model_context_2s.input_data.rf_channel),
get_channels(model_context_1s.input_data.rf_channel),
model_context_1s.input_data.frequency,
)
mapped_kwargs["rf_spend"] = map_tensor(
"rf_spend",
get_channels(model_context_2s.input_data.rf_channel),
get_channels(model_context_1s.input_data.rf_channel),
model_context_1s.input_data.rf_spend,
)
mapped_kwargs["time"] = new_data.time
return tensors.DataTensors(**mapped_kwargs)
class AnalyzerFullFunnel(analyzer.Analyzer):
"""Analyzer for full-funnel Meridian models.
This class extends the `Analyzer` class for full-funnel inference, allowing
the computation of total incremental outcome, which includes both
indirect effects through the mediator variables and remaining
effects from the primary Stage 2 model.
"""
def __init__(
self,
meridian: model.Meridian,
mediator_models: Mapping[str, model.Meridian],
inference_data: Optional[az.InferenceData] = None,
inference_data_mediators: Optional[Mapping[str, az.InferenceData]] = None,
):
super().__init__(meridian=meridian, inference_data=inference_data)
self._mediator_models = mediator_models
self._mediators = list(mediator_models.keys())
if inference_data_mediators is None:
inference_data_mediators = {}
self._analyzer_mediators = {
name: analyzer.Analyzer(
meridian=mediator_model,
inference_data=inference_data_mediators.get(name),
)
for name, mediator_model in mediator_models.items()
}
self._validate_two_stage_requirements()
def _validate_two_stage_requirements(self):
"""Validates the specific requirements for AnalyzerFullFunnel."""
input_data_2s = self.model_context.input_data
saturation_spec = self.model_context.model_spec.saturation_spec
inf_data_2s = self.inference_data
media_channels_2s = set(input_data_2s.get_all_paid_channels())
for mediator in self._mediators:
mediator_model = self._mediator_models[mediator]
analyzer_mediator = self._analyzer_mediators[mediator]
if (
input_data_2s.organic_media_channel is None
or mediator not in input_data_2s.organic_media_channel.values
):
raise ValueError(
f"Mediator '{mediator}' must be included as an `organic_media` "
"variable in the Stage 2 `meridian` model."
)
if isinstance(saturation_spec, str):
if saturation_spec != "none":
raise ValueError(
f"Saturation for mediator '{mediator}' in Stage 2 must be 'none'."
)
elif saturation_spec.get(mediator) != "none":
raise ValueError(
f"Saturation for mediator '{mediator}' in Stage 2 must be 'none'."
)
if not np.array_equal(
input_data_2s.geo.values,
mediator_model.model_context.input_data.geo.values,
):
raise ValueError(
f"Geos mismatch. Stage 2 has {len(input_data_2s.geo.values)} geos, "
f"but Stage 1 ('{mediator}') has "
f"{len(mediator_model.model_context.input_data.geo.values)}."
)
if not np.array_equal(
input_data_2s.time.values,
mediator_model.model_context.input_data.time.values,
):
raise ValueError(
f"Timepoints mismatch between Stage 2 and Stage 1 ('{mediator}')."
)
media_channels_1s = set(
mediator_model.model_context.input_data.get_all_paid_channels()
)
if not media_channels_1s.issubset(media_channels_2s):
raise ValueError(
"All paid media channels in the mediator models must be present in"
" the second stage model `meridian`."
)
inf_data_1s = analyzer_mediator.inference_data
for attr in ["posterior", "prior"]:
if hasattr(inf_data_2s, attr) and hasattr(inf_data_1s, attr):
group_2s = getattr(inf_data_2s, attr)
group_1s = getattr(inf_data_1s, attr)
if (
group_2s.chain.size != group_1s.chain.size
or group_2s.draw.size != group_1s.draw.size
):
raise ValueError(
f"MCMC traces mismatch. Stage 2 {attr} has "
f"({group_2s.chain.size} chains, {group_2s.draw.size} draws). "
f"Stage 1 {attr} has "
f"({group_1s.chain.size} chains, {group_1s.draw.size} draws)."
)
def _get_mediator_metadata(
self,
) -> tuple[list[str], list[backend.Tensor], list[Any]]:
"""Helper to extract mediator names, scale factors, and decay functions."""
mediator_names = []
scale_factors = []
decay_functions_list = []
for mediator in self._mediators:
mediator_names.append(mediator)
scale_factors.append(
self.model_context.get_media_scaling_factor(mediator)
)
channel_params = self.model_context.get_channel_parameters(mediator)
decay_functions_list.append(channel_params.decay_spec)
return mediator_names, scale_factors, decay_functions_list
def _get_remaining_incremental_kpi(
self,
data_tensors: tensors.DataTensors,
dist_tensors: tensors.DistributionTensors,
non_media_treatments_baseline_normalized: Optional[
Sequence[float]
] = None,
) -> backend.Tensor:
"""Computes remaining incremental KPI. The incremental KPI from the 2nd
stage model (`meridian`)."""
return self.get_incremental_kpi(
data_tensors=data_tensors,
dist_tensors=dist_tensors,
non_media_treatments_baseline_normalized=non_media_treatments_baseline_normalized,
)
def _get_indirect_incremental_kpi(
self,
analyzer_mediator: analyzer.Analyzer,
data_tensors_mediator: tensors.DataTensors,
dist_tensors_mediator: tensors.DistributionTensors,
dist_tensors: tensors.DistributionTensors,
scale_factors: backend.Tensor,
decay_functions: Any,
has_mapping: Sequence[bool],
channel_indices_1s: Sequence[int],
mediator_name: str,
include_non_paid_channels: bool,
non_media_treatments_baseline_normalized_mediator: Optional[
Sequence[float]
] = None,
) -> backend.Tensor:
"""Computes indirect incremental KPI through the mediator."""
# Step 1: Stage 1 KPI contribution on original mediator scale.
mediator_kpi = analyzer_mediator.get_incremental_kpi(
data_tensors=data_tensors_mediator,
dist_tensors=dist_tensors_mediator,
non_media_treatments_baseline_normalized=non_media_treatments_baseline_normalized_mediator,
)
mediator_original = analyzer_mediator.inverse_outcome(
mediator_kpi,
use_kpi=True,
revenue_per_kpi=data_tensors_mediator.revenue_per_kpi,
)
# Map Stage 1 media channels to Stage 2 media channels.
mediator_incr_list = []
for i, mapped in enumerate(has_mapping):
if mapped:
mediator_incr_list.append(mediator_original[..., channel_indices_1s[i]])
else:
mediator_incr_list.append(backend.zeros_like(mediator_original[..., 0]))
indirect_kpi = backend.stack(mediator_incr_list, axis=-1)
# Step 2: Rescale mediator effects for Stage 2
indirect_kpi /= scale_factors[None, None, :, None, None]
# Step 3: Apply Mediator Adstock transformation (from Stage 2 model)
alpha_mediator = self.model_context.get_channel_parameter_tensor(
dist_tensors,
param_base_name="alpha",
channel_name=mediator_name,
)
beta_g_mediator = self.model_context.get_channel_parameter_tensor(
dist_tensors,
param_base_name=constants.BETA_G,
channel_name=mediator_name,
)
n_chains, n_draws_batch, n_geos, n_times, n_paid_2s = indirect_kpi.shape
alpha_broadcast = backend.repeat(
alpha_mediator[:, :, None], n_paid_2s, axis=-1
)
batch_size_total = n_chains * n_draws_batch
indirect_kpi_reshaped = backend.reshape(
indirect_kpi, (batch_size_total, n_geos, n_times, n_paid_2s)
)
alpha_broadcast_reshaped = backend.reshape(
alpha_broadcast, (batch_size_total, n_paid_2s)
)
model_eqs = equations.ModelEquations(self.model_context)
indirect_kpi_transformed = model_eqs.adstock_hill_media(
media=indirect_kpi_reshaped,
alpha=alpha_broadcast_reshaped,
ec=backend.zeros_like(alpha_broadcast_reshaped),
slope=backend.ones_like(alpha_broadcast_reshaped),
decay_functions=decay_functions,
saturation_spec="none",
n_times_output=n_times,
)
indirect_kpi_transformed = backend.reshape(
indirect_kpi_transformed,
(n_chains, n_draws_batch, n_geos, n_times, n_paid_2s),
)
# Step 4: Multiply by Stage 2 mediator coefficient
combined_media_kpi_indirect = backend.einsum(
"...gtm,...g->...gtm", indirect_kpi_transformed, beta_g_mediator
)
# Pad indirect KPI with zeros for non-paid channels
if include_non_paid_channels:
n_non_paid_2s = (
self.model_context.n_organic_media_channels
+ self.model_context.n_organic_rf_channels
+ self.model_context.n_non_media_channels
)
if n_non_paid_2s > 0:
padding = backend.zeros(
(n_chains, n_draws_batch, n_geos, n_times, n_non_paid_2s),
dtype=backend.float_dtype,
)
combined_media_kpi_indirect = backend.concatenate(
[combined_media_kpi_indirect, padding], axis=-1
)
return combined_media_kpi_indirect
@backend.function(
jit_compile=True,
static_argnames=[
"effect_type",
"inverse_transform_outcome",
"use_kpi",
"selected_geos",
"selected_times",
"aggregate_geos",
"aggregate_times",
"include_non_paid_channels",
"has_mappings",
"channel_indices_1s_list",
"mediator_names",
"decay_functions_list",
"analyzer_mediators_list",
],
)
def _incremental_outcome_impl(
self,
data_tensors: tensors.DataTensors,
dist_tensors: tensors.DistributionTensors,
data_tensors_mediators: Sequence[tensors.DataTensors],
dist_tensors_mediators: Sequence[tensors.DistributionTensors],
scale_factors: Sequence[backend.Tensor],
decay_functions_list: Sequence[Any],
non_media_treatments_baseline_normalized: Optional[
Sequence[float]
] = None,
non_media_treatments_baseline_normalized_mediators: Optional[
Sequence[Optional[Sequence[float]]]
] = None,
effect_type: str = "total",
inverse_transform_outcome: bool = True,
use_kpi: bool = False,
selected_geos: Optional[Sequence[str]] = None,
selected_times: Optional[Sequence[str]] = None,
aggregate_geos: bool = True,
aggregate_times: bool = True,
include_non_paid_channels: bool = True,
has_mappings: Sequence[Sequence[bool]] = (),
channel_indices_1s_list: Sequence[Sequence[int]] = (),
mediator_names: Sequence[str] = (),
analyzer_mediators_list: Sequence[analyzer.Analyzer] = (),
) -> backend.Tensor:
"""Computes incremental outcome (revenue or KPI) on a batch of data."""
# 1. Remaining effect from the primary model (Stage 2)
remaining_kpi = self._get_remaining_incremental_kpi(
data_tensors=data_tensors,
dist_tensors=dist_tensors,
non_media_treatments_baseline_normalized=non_media_treatments_baseline_normalized,
)
# 2. Indirect effects through mediators
total_indirect_kpi = backend.zeros_like(remaining_kpi)
if effect_type in ("total", "indirect"):
for i in range(len(mediator_names)):
non_media_baseline_med = (
non_media_treatments_baseline_normalized_mediators[i]
if non_media_treatments_baseline_normalized_mediators is not None
else None
)
indirect_kpi = self._get_indirect_incremental_kpi(
analyzer_mediator=analyzer_mediators_list[i],
data_tensors_mediator=data_tensors_mediators[i],
dist_tensors_mediator=dist_tensors_mediators[i],
dist_tensors=dist_tensors,
scale_factors=scale_factors[i],
decay_functions=decay_functions_list[i],
has_mapping=has_mappings[i],
channel_indices_1s=channel_indices_1s_list[i],
mediator_name=mediator_names[i],
include_non_paid_channels=include_non_paid_channels,
non_media_treatments_baseline_normalized_mediator=(
non_media_baseline_med
),
)
total_indirect_kpi += indirect_kpi
# 3. Branch based on effect_type
if effect_type == "indirect":
total_transformed_outcome = total_indirect_kpi
elif effect_type == "remaining":
total_transformed_outcome = remaining_kpi
elif effect_type == "total":
total_transformed_outcome = total_indirect_kpi + remaining_kpi
else:
raise ValueError(
f"Invalid effect_type '{effect_type}'. Expected 'total',"
" 'indirect', or 'remaining'."
)
# 4. Final Inverse Transformation and Aggregation
if inverse_transform_outcome:
incremental_outcome = self.inverse_outcome(
total_transformed_outcome,
use_kpi=use_kpi,
revenue_per_kpi=data_tensors.revenue_per_kpi,
)
else:
incremental_outcome = total_transformed_outcome
# Resolve actual indices using the builder for final aggregation
inputs_for_indices = tensors.DataTensorsBuilder(
self.model_context
).build_unscaled_inputs(
selected_geos=selected_geos, selected_times=selected_times
)
return self.filter_and_aggregate_by_indices(
tensor=incremental_outcome,
geo_indices=inputs_for_indices.geo_indices,
time_indices=inputs_for_indices.time_indices,
aggregate_geos=aggregate_geos,
aggregate_times=aggregate_times,
flexible_time_dim=True,
has_media_dim=True,
)
def incremental_outcome(
self,
use_posterior: bool = True,
new_data: Optional[tensors.DataTensors] = None,
non_media_baseline_values: Optional[Sequence[float]] = None,
scaling_factor0: float = 0.0,
scaling_factor1: float = 1.0,
selected_geos: Optional[Sequence[str]] = None,
selected_times: Optional[Sequence[str]] = None,
media_selected_times: Optional[Sequence[str]] = None,
aggregate_geos: bool = True,
aggregate_times: bool = True,
inverse_transform_outcome: bool = True,
use_kpi: bool = False,
by_reach: bool = True,
include_non_paid_channels: bool = True,
batch_size: int = constants.DEFAULT_BATCH_SIZE,
effect_type: str = "total",
*,
non_media_baseline_values_mediator: Optional[
Sequence[float] | Mapping[str, Sequence[float]]
] = None,
) -> backend.Tensor:
"""Calculates either the posterior or prior total incremental outcome."""
if effect_type not in ("total", "indirect", "remaining"):
raise ValueError(
f"Invalid effect_type '{effect_type}'. Expected 'total',"
" 'indirect', or 'remaining'."
)
m_context = self.model_context
use_kpi = self._use_kpi(use_kpi)
self._check_kpi_transformation(inverse_transform_outcome, use_kpi)
if m_context.is_national:
analyzer._warn_if_geo_arg_in_kwargs(
aggregate_geos=aggregate_geos,
selected_geos=selected_geos,
)
dist_type = constants.POSTERIOR if use_posterior else constants.PRIOR
if dist_type not in self.inference_data.groups():
raise analyzer.errors.NotFittedModelError(
f"sample_{dist_type}() must be called prior to calling this method."
)
for mediator, analyzer_mediator in self._analyzer_mediators.items():
if dist_type not in analyzer_mediator.inference_data.groups():
raise analyzer.errors.NotFittedModelError(
f"sample_{dist_type}() must be called for the mediator model"
f" '{mediator}'."
)
if scaling_factor1 <= scaling_factor0 or scaling_factor0 < 0:
raise ValueError(
"Invalid scaling factors. Ensure 0 <= scaling_factor0 <"
" scaling_factor1."
)
# 1. Build inputs for Stage 2
builder = tensors.DataTensorsBuilder(self.model_context)
inputs0 = builder.build_counterfactual_inputs(
new_data=new_data,
scaling_factor=scaling_factor0,
non_media_baseline_values=non_media_baseline_values,
selected_geos=selected_geos,
selected_times=selected_times,
media_selected_times=media_selected_times,
by_reach=by_reach,
include_non_paid_channels=include_non_paid_channels,
is_baseline=True,
)
inputs1 = builder.build_counterfactual_inputs(
new_data=new_data,
scaling_factor=scaling_factor1,
non_media_baseline_values=non_media_baseline_values,
selected_geos=selected_geos,
selected_times=selected_times,
media_selected_times=media_selected_times,
by_reach=by_reach,
include_non_paid_channels=include_non_paid_channels,
is_baseline=False,
)
data_tensors0 = dataclasses.replace(inputs0.tensors, time=None)
data_tensors1 = dataclasses.replace(inputs1.tensors, time=None)
media_channels_2s = list(
self.model_context.input_data.get_all_paid_channels()
)
n_total_media_2s = len(media_channels_2s)
# Pre-fetch unified mediator metadata
mediator_names, scale_factors, decay_functions_list = (
self._get_mediator_metadata()
)
has_mappings = []
channel_indices_1s_list = []
analyzer_mediators_list = []
data_tensors0_mediators = []
data_tensors1_mediators = []
non_media_treatments_baseline_normalized_mediators = []
# 2. Build inputs and channel maps for all Stage 1 Mediators
for mediator in self._mediators:
mediator_model = self._mediator_models[mediator]
analyzer_mediator = self._analyzer_mediators[mediator]
analyzer_mediators_list.append(analyzer_mediator)
media_channels_1s = list(
mediator_model.model_context.input_data.get_all_paid_channels()
)
media_channel_map = {
i: media_channels_1s.index(ch)
for i, ch in enumerate(media_channels_2s)
if ch in media_channels_1s
}
channel_indices_1s = []
has_mapping = []
for i in range(n_total_media_2s):
if i in media_channel_map:
channel_indices_1s.append(media_channel_map[i])
has_mapping.append(True)
else:
channel_indices_1s.append(0)
has_mapping.append(False)
has_mappings.append(tuple(has_mapping))
channel_indices_1s_list.append(tuple(channel_indices_1s))
baseline_vals_med = None
if non_media_baseline_values_mediator is not None:
if isinstance(non_media_baseline_values_mediator, dict):
baseline_vals_med = non_media_baseline_values_mediator.get(mediator)
elif isinstance(non_media_baseline_values_mediator, (list, tuple)):
if len(non_media_baseline_values_mediator) == len(self._mediators):
baseline_vals_med = [
non_media_baseline_values_mediator[
self._mediators.index(mediator)
]
]
else:
baseline_vals_med = non_media_baseline_values_mediator
builder_med = tensors.DataTensorsBuilder(mediator_model.model_context)
mapped_new_data = _map_new_data_to_mediator(
new_data, self.model_context, mediator_model.model_context
)
inputs0_med = builder_med.build_counterfactual_inputs(
new_data=mapped_new_data,
scaling_factor=scaling_factor0,
non_media_baseline_values=baseline_vals_med,
selected_geos=selected_geos,
selected_times=selected_times,
media_selected_times=media_selected_times,
by_reach=by_reach,
include_non_paid_channels=include_non_paid_channels,
is_baseline=True,
)
inputs1_med = builder_med.build_counterfactual_inputs(
new_data=mapped_new_data,
scaling_factor=scaling_factor1,
non_media_baseline_values=baseline_vals_med,
selected_geos=selected_geos,
selected_times=selected_times,
media_selected_times=media_selected_times,
by_reach=by_reach,
include_non_paid_channels=include_non_paid_channels,
is_baseline=False,
)
data_tensors0_mediators.append(
dataclasses.replace(inputs0_med.tensors, time=None)
)
data_tensors1_mediators.append(
dataclasses.replace(inputs1_med.tensors, time=None)
)
non_media_treatments_baseline_normalized_mediators.append(
inputs1_med.non_media_baseline_normalized
)
# We must always include non-paid channels for Stage 2 to ensure organic
# mediator parameters (alpha_om, beta_gom) are available for XLA indirect
# calcs.
param_list = self._get_causal_param_names(include_non_paid_channels=True)
dim_kwargs = {
"selected_geos": (
tuple(selected_geos) if selected_geos is not None else None
),
"selected_times": (
tuple(selected_times) if selected_times is not None else None
),
"aggregate_geos": aggregate_geos,
"aggregate_times": aggregate_times,
}
mapping_kwargs = {
"mediator_names": tuple(mediator_names),
"has_mappings": tuple(has_mappings),
"channel_indices_1s_list": tuple(channel_indices_1s_list),
"decay_functions_list": tuple(decay_functions_list),
"analyzer_mediators_list": tuple(analyzer_mediators_list),
}
incremental_outcome_kwargs = {
"effect_type": effect_type,
"inverse_transform_outcome": inverse_transform_outcome,
"use_kpi": use_kpi,
"include_non_paid_channels": include_non_paid_channels,
"non_media_treatments_baseline_normalized": (
inputs1.non_media_baseline_normalized
),
"non_media_treatments_baseline_normalized_mediators": tuple(
non_media_treatments_baseline_normalized_mediators
),
**dim_kwargs,
**mapping_kwargs,
}
# 3. Calculate metrics sequentially in batches
stage2_gen = self.yield_batched_distribution_tensors(
param_list, use_posterior=use_posterior, batch_size=batch_size
)
mediator_gens = []
for mediator in self._mediators:
analyzer_med = self._analyzer_mediators[mediator]
param_list_med = analyzer_med._get_causal_param_names(
include_non_paid_channels=include_non_paid_channels
)
mediator_gens.append(
analyzer_med.yield_batched_distribution_tensors(
param_list_med, use_posterior=use_posterior, batch_size=batch_size
)
)
incremental_outcome_temps = []
for dist_tensors, *dist_tensors_mediators in zip(
stage2_gen, *mediator_gens
):
batch_incr = self._incremental_outcome_impl(
data_tensors=data_tensors1,
dist_tensors=dist_tensors,
data_tensors_mediators=tuple(data_tensors1_mediators),
dist_tensors_mediators=tuple(dist_tensors_mediators),
scale_factors=tuple(scale_factors),
**incremental_outcome_kwargs,
)
if scaling_factor0 != 0 or (
inputs0.media_selected_times_mask is not None
and not all(inputs0.media_selected_times_mask)
):
kwargs_0 = incremental_outcome_kwargs.copy()
kwargs_0["non_media_treatments_baseline_normalized"] = (
inputs0.non_media_baseline_normalized
)
batch_incr -= self._incremental_outcome_impl(
data_tensors=data_tensors0,
dist_tensors=dist_tensors,
data_tensors_mediators=tuple(data_tensors0_mediators),
dist_tensors_mediators=tuple(dist_tensors_mediators),
scale_factors=tuple(scale_factors),
**kwargs_0,
)
incremental_outcome_temps.append(batch_incr)
return backend.concatenate(incremental_outcome_temps, axis=1)
@backend.function(
jit_compile=True,
static_argnames=[
"mediator_names",
"decay_functions_list",
],
)
def _expected_outcome_impl(
self,
data_tensors_no_mediators: tensors.DataTensors,
dist_tensors: tensors.DistributionTensors,
mediator_expected_batches: Sequence[backend.Tensor],
scale_factors: Sequence[backend.Tensor],
mediator_names: Sequence[str],
decay_functions_list: Sequence[Any],
) -> backend.Tensor:
"""Computes expected outcome for a batch of draws."""
# 1. Base KPI means (excluding mediator)
kpi_means_base = self.get_kpi_means(
data_tensors=data_tensors_no_mediators,
dist_tensors=dist_tensors,
)
total_mediator_effect = backend.zeros_like(kpi_means_base)
# 2. Loop over mediators to compute their effects
for i, mediator_name in enumerate(mediator_names):
mediator_expected_batch = mediator_expected_batches[i]
scale_factor = scale_factors[i]
decay_functions = decay_functions_list[i]
# Scale mediator draws to Stage 2 scale.
mediator_scaled = (
mediator_expected_batch / scale_factor[None, None, :, None]
)
n_chains, n_draws_batch, n_geos, n_times = mediator_scaled.shape
batch_size_total = n_chains * n_draws_batch
mediator_reshaped = backend.reshape(
mediator_scaled, (batch_size_total, n_geos, n_times, 1)
)
alpha_mediator = self.model_context.get_channel_parameter_tensor(
dist_tensors,
param_base_name="alpha",
channel_name=mediator_name,
)
alpha_reshaped = backend.reshape(alpha_mediator, (batch_size_total, 1))
model_eqs = equations.ModelEquations(self.model_context)
transformed_mediator = model_eqs.adstock_hill_media(
media=mediator_reshaped,
alpha=alpha_reshaped,
ec=backend.zeros_like(alpha_reshaped),
slope=backend.ones_like(alpha_reshaped),
decay_functions=decay_functions,
saturation_spec="none",
n_times_output=n_times,
)
transformed_mediator = backend.reshape(
transformed_mediator,
(n_chains, n_draws_batch, n_geos, n_times),
)
beta_g_mediator = self.model_context.get_channel_parameter_tensor(
dist_tensors,
param_base_name=constants.BETA_G,
channel_name=mediator_name,
)
mediator_effect = transformed_mediator * beta_g_mediator[..., None]
total_mediator_effect += mediator_effect
return kpi_means_base + total_mediator_effect
def expected_outcome(
self,
use_posterior: bool = True,
new_data: Optional[tensors.DataTensors] = None,
selected_geos: Optional[Sequence[str]] = None,
selected_times: Optional[Sequence[str]] = None,
aggregate_geos: bool = True,
aggregate_times: bool = True,
inverse_transform_outcome: bool = True,
use_kpi: bool = False,
batch_size: int = constants.DEFAULT_BATCH_SIZE,
) -> backend.Tensor:
"""Calculates either prior or posterior expected outcome."""
use_kpi = self._use_kpi(use_kpi)
self._check_kpi_transformation(inverse_transform_outcome, use_kpi)
if self.model_context.is_national:
analyzer._warn_if_geo_arg_in_kwargs(
aggregate_geos=aggregate_geos,
selected_geos=selected_geos,
)
dist_type = constants.POSTERIOR if use_posterior else constants.PRIOR
if dist_type not in self.inference_data.groups():
raise analyzer.errors.NotFittedModelError(
f"sample_{dist_type}() must be called prior to calling this method."
)
for mediator, analyzer_mediator in self._analyzer_mediators.items():
if dist_type not in analyzer_mediator.inference_data.groups():
raise analyzer.errors.NotFittedModelError(
f"sample_{dist_type}() must be called for the mediator model"
f" '{mediator}'."
)
builder = tensors.DataTensorsBuilder(self.model_context)
inputs = builder.build_scaled_inputs(
new_data=new_data,
include_non_paid_channels=True,
selected_geos=selected_geos,
selected_times=selected_times,
)
data_tensors = dataclasses.replace(inputs.tensors, time=None)
mediator_names, scale_factors, decay_functions_list = (
self._get_mediator_metadata()
)
organic_media = data_tensors.organic_media
assert organic_media is not None
# Zero out the mediators in the Stage 2 organic media tensor
mask_np = np.ones(organic_media.shape[-1], dtype=np.float32)
for mediator in self._mediators:
idx = self.model_context.get_channel_parameters(mediator).index
mask_np[idx] = 0.0
mask = backend.to_tensor(mask_np, dtype=backend.float_dtype)
organic_media_no_mediators = organic_media * mask
data_tensors_no_mediators = dataclasses.replace(
data_tensors, organic_media=organic_media_no_mediators
)
# Pre-compute Stage 1 expected outcomes on original scale
mediator_expecteds = []
for mediator in self._mediators:
analyzer_mediator = self._analyzer_mediators[mediator]
mapped_new_data = _map_new_data_to_mediator(
new_data, self.model_context, analyzer_mediator.model_context
)
mediator_expected = analyzer_mediator.expected_outcome(
use_posterior=use_posterior,
new_data=mapped_new_data,
inverse_transform_outcome=True,
use_kpi=True,
batch_size=batch_size,
aggregate_geos=False,
aggregate_times=False,
)
mediator_expecteds.append(mediator_expected)
param_list = (
[constants.MU_T, constants.TAU_G]
+ ([constants.GAMMA_GC] if self.model_context.n_controls else [])
+ self._get_causal_param_names(include_non_paid_channels=True)
)
stage2_gen = self.yield_batched_distribution_tensors(
param_list, use_posterior=use_posterior, batch_size=batch_size
)
outcome_means_temps = []
params = (
self.inference_data.posterior # pyrefly: ignore[missing-attribute]
if use_posterior
else self.inference_data.prior # pyrefly: ignore[missing-attribute]
)
n_draws = params.draw.size
batch_starting_indices = np.arange(n_draws, step=batch_size)
for start_index, dist_tensors in zip(batch_starting_indices, stage2_gen):
stop_index = np.min([n_draws, start_index + batch_size])
mediator_expected_batches = [
med_exp[:, start_index:stop_index, ...]
for med_exp in mediator_expecteds
]
outcome_means_temps.append(
self._expected_outcome_impl(
data_tensors_no_mediators=data_tensors_no_mediators,
dist_tensors=dist_tensors,
mediator_expected_batches=tuple(mediator_expected_batches),
scale_factors=tuple(scale_factors),
mediator_names=tuple(mediator_names),
decay_functions_list=tuple(decay_functions_list),
)
)
outcome_means = backend.concatenate(outcome_means_temps, axis=1)
if inverse_transform_outcome:
outcome_means = self.model_context.kpi_transformer.inverse(outcome_means)
if not use_kpi:
revenue_per_kpi = (
inputs.tensors.revenue_per_kpi
if inputs.tensors.revenue_per_kpi is not None
else self.model_context.revenue_per_kpi
)
outcome_means *= revenue_per_kpi
return self.filter_and_aggregate_by_indices(
outcome_means,
geo_indices=inputs.geo_indices,
time_indices=inputs.time_indices,
aggregate_geos=aggregate_geos,
aggregate_times=aggregate_times,
)
Step 5: Post-Modeling Analysis with AnalyzerFullFunnel
We instantiate the AnalyzerFullFunnel class which is fully compatible with Meridian's downstream optimization, scenario planning, and visualization modules. It enables lots of post-modeling analysis and it can be passed to the MediaEffects and BudgetOptimizer classes. Here are some examples of the analysis you can do while accounting for the full effect according to the full-funnel model:
- ROI estimation
- Marginal ROI estimation
- Response curve estimation and visualization
- Budget optimization and visualization
- Negative baseline checks
We will also plot the following full-funnel specific visualizations:
- ROI decomposed into indirect and remaining effects
- Compounded adstock decay curves
# Initialize AnalyzerFullFunnel
analyzer_full_funnel = AnalyzerFullFunnel(
# Second stage goes here
meridian=mmm2s,
# First stage model goes here
mediator_models={"BrandedGQV_control": mmm1s},
)
# Check the negative baseline probability
prob = analyzer_full_funnel.negative_baseline_probability()
print(f"Negative baseline probability: {prob:.4f}")
/tmp/ipykernel_2365/1719713641.py:112: DeprecationWarning: The `meridian` argument is deprecated and will be removed in a future version. Use `model_context` instead. super().__init__(meridian=meridian, inference_data=inference_data) /tmp/ipykernel_2365/1719713641.py:120: DeprecationWarning: The `meridian` argument is deprecated and will be removed in a future version. Use `model_context` instead. name: analyzer.Analyzer( Negative baseline probability: 0.0000
Full-Funnel MMM Results:
AwarenessVideo: Includes its indirect effect viaBrandedGQV, estimating the full ROI.PerformanceNative&PerformanceDisplay: Protected from confounding bias by properly conditioning onBrandedGQVin the second stage model.
# Plot Media Response Curves
media_effects = visualizer.MediaEffects(analyzer=analyzer_full_funnel)
media_effects.plot_response_curves()
# Run Fixed Budget Optimization
budget_optimizer = optimizer.BudgetOptimizer(analyzer=analyzer_full_funnel)
fixed_optimizer = budget_optimizer.optimize()
# Plot Incremental Outcome Delta
fixed_optimizer.plot_spend_delta()
fixed_optimizer.plot_incremental_outcome_delta()
Custom functions for ROI decomposed into indirect and remaining effects
\[\text{Total Effect} = \text{Indirect Effect} + \text{Remaining Effect}\]
- Indirect effect: The impact of brand marketing in Stage 1 propagated
through the brand equity mediator (
BrandedGQV) into final conversions. - Remaining effect: The effect estimated directly in the primary Stage 2 model after controlling for the brand equity mediator.
These custom functions let you compute and visualize how total ROI estimates decompose into each component.
# @title Estimating Decomposed ROI
def compute_roi_decomposition(
analyzer_full_funnel: AnalyzerFullFunnel,
confidence_level: float = 0.8,
) -> pd.DataFrame:
"""Computes draw-level indirect, remaining, and total ROI distributions and
summaries.
Args:
analyzer_full_funnel: Fitted full-funnel analyzer.
confidence_level: Credible interval coverage level (default 0.8).
Returns:
A DataFrame with columns: 'channel', 'indirect_roi_mean',
'indirect_roi_ci_lo', 'indirect_roi_ci_hi', 'remaining_roi_mean',
'remaining_roi_ci_lo', 'remaining_roi_ci_hi', 'total_roi_mean',
'total_roi_ci_lo', 'total_roi_ci_hi'.
"""
channels = list(
analyzer_full_funnel.model_context.input_data.get_all_paid_channels()
)
spend = analyzer_full_funnel.model_context.total_spend
if len(spend.shape) > 1:
channel_spend = backend.reduce_sum(
spend,
axis=range(len(spend.shape) - 1),
)
else:
channel_spend = spend
indirect_outcome = analyzer_full_funnel.incremental_outcome(
effect_type="indirect",
include_non_paid_channels=False,
)
remaining_outcome = analyzer_full_funnel.incremental_outcome(
effect_type="remaining",
include_non_paid_channels=False,
)
# Draw-level ROI tensors
indirect_roi = indirect_outcome / channel_spend
remaining_roi = remaining_outcome / channel_spend
total_roi = indirect_roi + remaining_roi
indirect_stats = analyzer.get_central_tendency_and_ci(
indirect_roi, confidence_level=confidence_level
)
remaining_stats = analyzer.get_central_tendency_and_ci(
remaining_roi, confidence_level=confidence_level
)
total_stats = analyzer.get_central_tendency_and_ci(
total_roi, confidence_level=confidence_level
)
return pd.DataFrame({
"channel": channels,
"indirect_roi_mean": indirect_stats[:, 0],
"indirect_roi_ci_lo": indirect_stats[:, 1],
"indirect_roi_ci_hi": indirect_stats[:, 2],
"remaining_roi_mean": remaining_stats[:, 0],
"remaining_roi_ci_lo": remaining_stats[:, 1],
"remaining_roi_ci_hi": remaining_stats[:, 2],
"total_roi_mean": total_stats[:, 0],
"total_roi_ci_lo": total_stats[:, 1],
"total_roi_ci_hi": total_stats[:, 2],
})
roi_df = compute_roi_decomposition(analyzer_full_funnel)
comparison_df = pd.DataFrame({
"Indirect ROI": roi_df.set_index("channel")["indirect_roi_mean"],
"Remaining ROI": roi_df.set_index("channel")["remaining_roi_mean"],
"Total ROI": roi_df.set_index("channel")["total_roi_mean"],
}).round(2)
comparison_df
# @title Plotting Decomposed ROI
def plot_roi_decomposition(
analyzer_full_funnel: AnalyzerFullFunnel,
confidence_level: float = 0.8,
) -> alt.Chart:
"""Plots a stacked ROI bar chart (indirect vs remaining) per channel."""
ci = int(confidence_level * 100)
title = f"ROI by channel with {ci}% credible interval"
roi_df = compute_roi_decomposition(analyzer_full_funnel, confidence_level)
bars_data = []
for _, row in roi_df.iterrows():
bars_data.append({
"channel": row["channel"],
"component": "Indirect ROI",
"roi": row["indirect_roi_mean"],
"order": 1,
})
bars_data.append({
"channel": row["channel"],
"component": "Remaining ROI",
"roi": row["remaining_roi_mean"],
"order": 2,
})
bars_df = pd.DataFrame(bars_data)
color_scale = alt.Scale(
domain=["Total ROI", "Indirect ROI", "Remaining ROI"],
range=[constants.GREY_800, constants.YELLOW_600, constants.BLUE_700],
)
bars = (
alt.Chart(bars_df)
.mark_bar(
size=constants.BAR_SIZE,
cornerRadiusEnd=constants.CORNER_RADIUS,
)
.encode(
x=alt.X(
"channel:N",
title="Channel",
axis=alt.Axis(labelAngle=-45, **formatter.AXIS_CONFIG),
),
y=alt.Y(
"roi:Q",
title="ROI",
stack="zero",
axis=alt.Axis(gridDash=[3, 2], **formatter.Y_AXIS_TITLE_CONFIG),
),
color=alt.Color(
"component:N",
scale=color_scale,
legend=alt.Legend(
title=None,
orient="bottom",
direction="horizontal",
),
),
order=alt.Order("order:Q"),
tooltip=[
alt.Tooltip("channel:N", title="Channel"),
alt.Tooltip("component:N", title="Component"),
alt.Tooltip("roi:Q", title="Component Mean ROI", format=".2f"),
],
)
.properties(width=alt.Step(80))
)
bar_width = 2
error_bar = (
alt.Chart(roi_df)
.mark_errorbar(
ticks={"size": 12}, color=constants.GREY_800
)
.encode(
x=alt.X("channel:N"),
y=alt.Y("total_roi_ci_hi:Q", title="ROI"),
y2=alt.Y2("total_roi_ci_lo:Q"),
strokeWidth=alt.value(bar_width),
)
)
mean_dot = (
alt.Chart(roi_df)
.transform_calculate(component="'Total ROI'")
.mark_point(
filled=True,
size=constants.POINT_SIZE,
color=constants.GREY_800,
tooltip=True,
)
.encode(
x=alt.X("channel:N"),
y=alt.Y("total_roi_mean:Q"),
color=alt.Color("component:N", scale=color_scale),
tooltip=[
alt.Tooltip("channel:N", title="Channel"),
alt.Tooltip(
"total_roi_mean:Q", title="Total ROI Mean", format=".2f"
),
alt.Tooltip(
"total_roi_ci_lo:Q",
title=f"Total ROI {ci}% CI Lower",
format=".2f",
),
alt.Tooltip(
"total_roi_ci_hi:Q",
title=f"Total ROI {ci}% CI Upper",
format=".2f",
),
],
)
)
plot = alt.layer(bars, error_bar, mean_dot)
return (
plot.configure_tick(bandSize=10, thickness=bar_width)
.properties(title=formatter.custom_title_params(title))
.configure_axis(titlePadding=constants.PADDING_10, **formatter.TEXT_CONFIG)
.configure_legend(**formatter.TEXT_CONFIG)
)
plot_roi_decomposition(analyzer_full_funnel)
Custom functions to visualize compounded adstock decay
In a full-funnel model, upper-funnel channels are affected by adstock decay in both models (for more information, see Compounded adstock decay in full-funnel MMM).
To visualize the effect of compounded adstock decay, simulate incremental KPI for a counterfactual scenario with non-zero paid and organic media at a single time point (time 0). This lets you understand how the incremental outcome decays over time. To minimize the impact of saturation, media at time 0 is set to 1% of the average weekly value.
# @title Plotting Compounding Adstock Decay
def _create_media_pulse_data(
analyzer_full_funnel: AnalyzerFullFunnel,
pulse_fraction: float = 0.01,
) -> tensors.DataTensors:
"""Creates a DataTensors container simulating an impulse at time unit 0."""
input_data = analyzer_full_funnel.model_context.input_data
pulse_kwargs = {}
if input_data.media is not None:
hist_media = input_data.media.values
weekly_avg = np.mean(hist_media, axis=1)
pulse_media = np.zeros_like(hist_media)
pulse_media[:, 0, :] = pulse_fraction * weekly_avg
pulse_kwargs["media"] = pulse_media
if input_data.organic_media is not None:
hist_organic = input_data.organic_media.values
weekly_avg_om = np.mean(hist_organic, axis=1)
pulse_organic = np.zeros_like(hist_organic)
pulse_organic[:, 0, :] = pulse_fraction * weekly_avg_om
pulse_kwargs["organic_media"] = pulse_organic
if input_data.reach is not None and input_data.frequency is not None:
hist_reach = input_data.reach.values
weekly_avg_reach = np.mean(hist_reach, axis=1)
pulse_reach = np.zeros_like(hist_reach)
pulse_reach[:, 0, :] = pulse_fraction * weekly_avg_reach
pulse_kwargs["reach"] = pulse_reach
avg_freq = np.mean(input_data.frequency.values, axis=1, keepdims=True)
pulse_kwargs["frequency"] = np.broadcast_to(
avg_freq, input_data.frequency.shape
)
if (
input_data.organic_reach is not None
and input_data.organic_frequency is not None
):
hist_orf = input_data.organic_reach.values
weekly_avg_orf = np.mean(hist_orf, axis=1)
pulse_orf = np.zeros_like(hist_orf)
pulse_orf[:, 0, :] = pulse_fraction * weekly_avg_orf
pulse_kwargs["organic_reach"] = pulse_orf
avg_orf_freq = np.mean(
input_data.organic_frequency.values, axis=1, keepdims=True
)
pulse_kwargs["organic_frequency"] = np.broadcast_to(
avg_orf_freq, input_data.organic_frequency.shape
)
return tensors.DataTensors(**pulse_kwargs)
def plot_compounding_adstock_decay(
analyzer_full_funnel: AnalyzerFullFunnel,
confidence_level: float = 0.8,
use_posterior: bool = True,
) -> alt.Chart:
"""Plots compounding adstock decay curves across indirect, remaining, and
total pathways."""
# 1. Dynamic lag horizon
remaining_max_lag = analyzer_full_funnel.model_context.model_spec.max_lag
compounding_max_lag = remaining_max_lag
mediator_model_lags = [
m.model_context.model_spec.max_lag
for m in analyzer_full_funnel._mediator_models.values()
]
compounding_max_lag += max(mediator_model_lags)
# 2. Media pulse simulation
pulse_data = _create_media_pulse_data(analyzer_full_funnel)
# 3. Counterfactual evaluation
indirect_outcome = analyzer_full_funnel.incremental_outcome(
new_data=pulse_data,
aggregate_geos=True,
aggregate_times=False,
use_kpi=True,
effect_type="indirect",
include_non_paid_channels=True,
use_posterior=use_posterior,
)
remaining_outcome = analyzer_full_funnel.incremental_outcome(
new_data=pulse_data,
aggregate_geos=True,
aggregate_times=False,
use_kpi=True,
effect_type="remaining",
include_non_paid_channels=True,
use_posterior=use_posterior,
)
# Filter to within horizon
indirect_horizon = indirect_outcome[..., : compounding_max_lag + 1, :]
remaining_horizon = remaining_outcome[..., : compounding_max_lag + 1, :]
total_horizon = indirect_horizon + remaining_horizon
# 4. Draw-wise cumulative sum normalization
total_sum = np.sum(total_horizon, axis=2, keepdims=True)
total_sum_safe = np.where(total_sum == 0, 1.0, total_sum)
norm_indirect = indirect_horizon / total_sum_safe
norm_remaining = remaining_horizon / total_sum_safe
norm_total = total_horizon / total_sum_safe
# 5. Extract statistics
indirect_stats = analyzer.get_central_tendency_and_ci(
norm_indirect, confidence_level=confidence_level, axis=(0, 1)
)
remaining_stats = analyzer.get_central_tendency_and_ci(
norm_remaining, confidence_level=confidence_level, axis=(0, 1)
)
total_stats = analyzer.get_central_tendency_and_ci(
norm_total, confidence_level=confidence_level, axis=(0, 1)
)
# 6. Channels to display
all_channels = list(
analyzer_full_funnel.model_context.input_data.get_all_channels()
)
mediators = set(analyzer_full_funnel._mediators)
mediated_channels = set()
for m in analyzer_full_funnel._mediator_models.values():
mediated_channels.update(m.model_context.input_data.get_all_paid_channels())
channels_to_plot = [ch for ch in all_channels if ch not in mediators]
# 7. Construct plotting DataFrame
records = []
for ch in channels_to_plot:
ch_idx = all_channels.index(ch)
is_mediated = ch in mediated_channels
channel_label = (
f"{ch} (Stage 1 and 2)" if is_mediated else f"{ch} (Stage 2 only)"
)
max_lag = compounding_max_lag if is_mediated else remaining_max_lag
for t in range(max_lag + 1):
if is_mediated:
records.append({
"channel": channel_label,
"time_units": t,
"component": "Indirect Adstock",
"mean": float(indirect_stats[t, ch_idx, 0]),
"ci_lo": float(indirect_stats[t, ch_idx, 1]),
"ci_hi": float(indirect_stats[t, ch_idx, 2]),
})
if t <= remaining_max_lag:
records.append({
"channel": channel_label,
"time_units": t,
"component": "Remaining Adstock",
"mean": float(remaining_stats[t, ch_idx, 0]),
"ci_lo": float(remaining_stats[t, ch_idx, 1]),
"ci_hi": float(remaining_stats[t, ch_idx, 2]),
})
records.append({
"channel": channel_label,
"time_units": t,
"component": "Total Adstock",
"mean": float(total_stats[t, ch_idx, 0]),
"ci_lo": float(total_stats[t, ch_idx, 1]),
"ci_hi": float(total_stats[t, ch_idx, 2]),
})
df = pd.DataFrame(records)
base = alt.Chart(df, width=constants.VEGALITE_FACET_DEFAULT_WIDTH)
color_scale = alt.Scale(
domain=[
"Total Adstock",
"Indirect Adstock",
"Remaining Adstock",
],
range=[constants.GREY_800, constants.YELLOW_600, constants.BLUE_700],
)
dash_scale = alt.Scale(
domain=[
"Indirect Adstock",
"Remaining Adstock",
],
range=[[4, 3], [0, 0]],
)
legend_cfg = alt.Legend(
title=None,
orient="bottom",
direction="horizontal",
symbolOpacity=1.0,
symbolStrokeWidth=3,
)
# 1. Shaded CI band for Total Adstock (bottom layer)
band = (
base.transform_filter(alt.datum.component == "Total Adstock")
.mark_area(opacity=0.15, color=constants.GREY_800)
.encode(
x=alt.X("time_units:Q"),
y=alt.Y(
"ci_lo:Q",
title="Percentage of Total Effect",
axis=alt.Axis(format=".1%", **formatter.Y_AXIS_TITLE_CONFIG),
),
y2=alt.Y2("ci_hi:Q"),
)
)
# 2. Component lines (Indirect & Remaining) rendered underneath Total line
component_lines = (
base.transform_filter(alt.datum.component != "Total Adstock")
.mark_line(strokeWidth=1.5)
.encode(
x=alt.X("time_units:Q", title="Time Units", axis=alt.Axis(**formatter.AXIS_CONFIG)),
y=alt.Y(
"mean:Q",
title="Percentage of Total Effect",
axis=alt.Axis(format=".1%", **formatter.Y_AXIS_TITLE_CONFIG),
),
color=alt.Color(
"component:N",
scale=color_scale,
legend=legend_cfg,
),
strokeDash=alt.StrokeDash(
"component:N",
scale=dash_scale,
legend=None,
),
)
)
# 3. Total Adstock line rendered ON TOP of component lines
total_line = (
base.transform_filter(alt.datum.component == "Total Adstock")
.mark_line(strokeWidth=2.5)
.encode(
x=alt.X("time_units:Q", title="Time Units", axis=alt.Axis(**formatter.AXIS_CONFIG)),
y=alt.Y(
"mean:Q",
title="Percentage of Total Effect",
axis=alt.Axis(format=".1%", **formatter.Y_AXIS_TITLE_CONFIG),
),
color=alt.Color(
"component:N",
scale=color_scale,
legend=legend_cfg,
),
)
)
# 4. Observation points with hover tooltips
points = (
base.mark_circle(filled=True, size=constants.POINT_SIZE, opacity=1)
.encode(
tooltip=[
"channel:N",
"component:N",
"time_units:Q",
alt.Tooltip("mean:Q", title="Mean Effect", format=".2%"),
alt.Tooltip(
"ci_lo:Q", title="Credible Interval Lower", format=".2%"
),
alt.Tooltip(
"ci_hi:Q", title="Credible Interval Upper", format=".2%"
),
],
x=alt.X("time_units:Q"),
y=alt.Y("mean:Q"),
color=alt.Color(
"component:N",
scale=color_scale,
legend=legend_cfg,
),
)
)
plot = alt.layer(band, component_lines, total_line, points)
return (
plot.facet("channel:N", columns=3)
.properties(
title=formatter.custom_title_params("Compounded Adstock Decay")
)
.configure_axis(titlePadding=constants.PADDING_10, **formatter.TEXT_CONFIG)
.configure_header(**formatter.TEXT_CONFIG)
.configure_legend(**formatter.TEXT_CONFIG)
.resolve_scale(x=constants.INDEPENDENT, y=constants.INDEPENDENT)
)
plot_compounding_adstock_decay(analyzer_full_funnel)
Step 6: Meridian Scenario Planner (MSP)
Export the full-funnel model and scenario planning optimization results to Google Sheets to populate the interactive Looker Studio Meridian Scenario Planner dashboard.
# @title
# # @title {display-mode: "form"}
# @contextlib.contextmanager
# def userland_two_stage_msp_context(custom_analyzer, custom_optimizer):
# """Temporarily overrides TrainedModel properties in Colab userland."""
# orig_analyzer = model_processor.TrainedModel.internal_analyzer
# orig_optimizer = model_processor.TrainedModel.internal_optimizer
# try:
# model_processor.TrainedModel.internal_analyzer = property(
# lambda self: custom_analyzer
# )
# model_processor.TrainedModel.internal_optimizer = property(
# lambda self: custom_optimizer
# )
# yield
# finally:
# model_processor.TrainedModel.internal_analyzer = orig_analyzer
# model_processor.TrainedModel.internal_optimizer = orig_optimizer
# print(
# f"[{time.strftime('%X')}] Starting Multi-Stage MSP Looker Studio Export..."
# )
# t0 = time.time()
# # -----------------------------------------------------------------------------
# # 1. MSP CONFIGURATIONS
# # -----------------------------------------------------------------------------
# optimization_name = "MSP FullFunnel"
# grid_name_prefix = "msp-fullfunnel"
# include_non_paid_channels = True
# start_date = None
# end_date = None
# # Time breakdowns (Default: Quarterly=True)
# yearly = False
# quarterly = True
# monthly = False
# time_breakdown_generators = []
# if yearly:
# time_breakdown_generators.append(
# date_range_bucketing.YearlyDateRangeGenerator
# )
# if quarterly:
# time_breakdown_generators.append(
# date_range_bucketing.QuarterlyDateRangeGenerator
# )
# if monthly:
# time_breakdown_generators.append(
# date_range_bucketing.MonthlyDateRangeGenerator
# )
# # Spend shift constraints (Default: ±30% / 0.3)
# min_spend_shift_ratio = 0.3
# max_spend_shift_ratio = 0.3
# # Match frequency settings from optimization grid
# use_optimal_frequency = fixed_optimizer.optimization_grid.use_optimal_frequency
# max_frequency = fixed_optimizer.optimization_grid.max_frequency
# # -----------------------------------------------------------------------------
# # 2. BUILD SPECS
# # -----------------------------------------------------------------------------
# summary_spec = marketing_processor.MediaSummarySpec(
# include_non_paid_channels=include_non_paid_channels
# )
# channel_constraints = [
# budget_optimization_processor.ChannelConstraintRel(
# channel_name=channel,
# spend_constraint_lower=min_spend_shift_ratio,
# spend_constraint_upper=max_spend_shift_ratio,
# )
# for channel in mmm2s.input_data.get_all_paid_channels()
# ]
# budget_opt_spec = budget_optimization_processor.BudgetOptimizationSpec(
# start_date=start_date,
# end_date=end_date,
# optimization_name=optimization_name,
# grid_name=grid_name_prefix,
# grid=fixed_optimizer.optimization_grid,
# constraints=channel_constraints,
# use_optimal_frequency=use_optimal_frequency,
# max_frequency=max_frequency,
# include_response_curves=False,
# )
# # -----------------------------------------------------------------------------
# # 3. GENERATE MMM PROTO
# # -----------------------------------------------------------------------------
# print(
# f"[{time.strftime('%X')}] Step 1/4: Executing create_mmm_ui_data_proto"
# " (Quarterly breakdowns enabled)..."
# )
# t1 = time.time()
# with userland_two_stage_msp_context(analyzer_full_funnel, budget_optimizer):
# mmm_proto = mmm_ui_gen.create_mmm_ui_data_proto(
# mmm=mmm2s,
# specs=[
# marketing_processor.MarketingAnalysisSpec(
# media_summary_spec=summary_spec,
# ),
# budget_opt_spec,
# ],
# time_breakdown_generators=time_breakdown_generators,
# )
# t2 = time.time()
# print(f"[{time.strftime('%X')}] ✓ Proto generation completed in {t2-t1:.1f}s")
# @title
# # @title {display-mode: "form"}
# # -----------------------------------------------------------------------------
# # 4. CONVERT PROTO TO DATAFRAMES
# # -----------------------------------------------------------------------------
# print(
# f"[{time.strftime('%X')}] Step 2/4: Converting existing mmm_proto to"
# " DataFrames..."
# )
# converter = dataframe_model_converter.DataFrameModelConverter(mmm_proto)
# dataframes = converter()
# # -----------------------------------------------------------------------------
# # 5. MULTI-STAGE INJECTION (Stage 1 & Stage 2 Model Diagnostics & Fit)
# # -----------------------------------------------------------------------------
# print(
# f"[{time.strftime('%X')}] injecting multi-stage Expected vs. Actual &"
# " Diagnostics..."
# )
# stage_models = [
# ("Stage 1: Branded Search (Mediator)", mmm1s),
# ("Stage 2: KPI (Conversions)", mmm2s),
# ]
# multi_diag_rows = []
# multi_fit_rows = []
# for stage_name, model in stage_models:
# if stage_name == "Stage 2: KPI (Conversions)":
# with userland_two_stage_msp_context(analyzer_full_funnel, budget_optimizer):
# stage_proto = mmm_ui_gen.create_mmm_ui_data_proto(
# mmm=model,
# specs=[model_fit_processor.ModelFitSpec()],
# )
# else:
# stage_proto = mmm_ui_gen.create_mmm_ui_data_proto(
# mmm=model,
# specs=[model_fit_processor.ModelFitSpec()],
# )
# stage_dfs = dataframe_model_converter.DataFrameModelConverter(stage_proto)()
# if "ModelDiagnostics" in stage_dfs:
# df_diag = stage_dfs["ModelDiagnostics"].copy()
# df_diag["Stage"] = stage_name
# multi_diag_rows.append(df_diag)
# if "ModelFit" in stage_dfs:
# df_fit = stage_dfs["ModelFit"].copy()
# df_fit["Stage"] = stage_name
# multi_fit_rows.append(df_fit)
# if multi_diag_rows:
# dataframes["ModelDiagnostics"] = pd.concat(
# multi_diag_rows, ignore_index=True
# )
# if multi_fit_rows:
# dataframes["ModelFit"] = pd.concat(multi_fit_rows, ignore_index=True)
# t3 = time.time()
# print(
# f"[{time.strftime('%X')}] ✓ Multi-Stage DataFrames completed in"
# f" {t3-t2:.1f}s"
# )
# # -----------------------------------------------------------------------------
# # 6. AUTHENTICATE & UPLOAD TO GOOGLE SHEETS
# # -----------------------------------------------------------------------------
# print(
# f"[{time.strftime('%X')}] Step 3/4: Authenticating and uploading to Google"
# " Sheets..."
# )
# auth.authenticate_user()
# spreadsheet = sheets.upload_to_gsheet(
# dataframes,
# spreadsheet_name="Meridian_Full_Funnel_MSP_Report",
# )
# t4 = time.time()
# print(
# f"[{time.strftime('%X')}] ✓ Spreadsheet upload completed in {t4-t3:.1f}s"
# )
# # -----------------------------------------------------------------------------
# # 7. GENERATE CLICKABLE LOOKER STUDIO DASHBOARD LINK
# # -----------------------------------------------------------------------------
# print(f"[{time.strftime('%X')}] Step 4/4: Generating Looker Studio MSP URL...")
# url = url_generator.create_report_url(spreadsheet)
# t_end = time.time()
# print(f"[{time.strftime('%X')}] ✓ URL generation completed.")
# print(f"[{time.strftime('%X')}] 🏁 TOTAL EXPORT TIME: {t_end-t0:.1f} seconds")
# print(f"\n✅ Full-Funnel Spreadsheet created: {spreadsheet.url}")
# print(f"✅ Clickable Looker Studio MSP Dashboard URL generated:\n{url}")
# HTML(
# f'<a href="{url}" target="_blank" style="font-size: 16px; font-weight:'
# ' bold;">👉 Click here to open your Multi-Stage MSP Dashboard in Looker'
# " Studio</a>"
# )