Validation¶

Rolling-origin backtests, specification-curve analysis, calibration checks and channel diagnostics.

Model Validation Package for MMM Framework.

Provides comprehensive tools for verifying the robustness of marketing mix models:

Posterior Predictive Checks (PPC)
Residual Diagnostics
Channel Diagnostics (VIF, convergence)
Model Comparison (LOO-CV, WAIC)
Cross-Validation (time-series CV)
Sensitivity Analysis
Stability Analysis
Experimental Calibration

Examples¶

Quick validation:

>>> from mmm_framework.validation import ModelValidator
>>> validator = ModelValidator(model, results)
>>> summary = validator.quick_check()
>>> print(summary.overall_quality)
>>> print(summary.recommendations)

Standard validation with model comparison:

>>> from mmm_framework.validation import ValidationConfig
>>> config = ValidationConfig.standard()
>>> summary = validator.validate(config)

Thorough validation with custom settings:

>>> from mmm_framework.validation import ValidationConfigBuilder
>>> config = (ValidationConfigBuilder()
...     .thorough()
...     .with_cross_validation(n_folds=5, strategy="expanding")
...     .with_residual_tests(("durbin_watson", "ljung_box"))
...     .build())
>>> summary = validator.validate(config)

Generate HTML report:

>>> html = summary.to_html_report()
>>> with open("validation_report.html", "w") as f:
...     f.write(html)

class mmm_framework.validation.ModelValidator(model, results=None)[source]

Bases: object

Main validation orchestrator.

Provides unified interface for all validation types.

Examples¶

>>> from mmm_framework.validation import ModelValidator, ValidationConfig
>>>
>>> # Quick validation
>>> validator = ModelValidator(model, results)
>>> summary = validator.quick_check()
>>>
>>> # Thorough validation with calibration
>>> config = (ValidationConfigBuilder()
...     .thorough()
...     .with_calibration(lift_tests)
...     .build())
>>> summary = validator.validate(config)
>>> summary.to_html_report()

__init__(model, results=None)[source]

Initialize model validator.

Parameters¶

modelAny: Fitted model (BayesianMMM, NestedMMM, MultivariateMMM, etc.).
resultsAny, optional: Model results container. If None, extracted from model.

validate(config=None)[source]

Run validation according to config.

Return type:: ValidationSummary

Parameters¶

configValidationConfig, optional: Validation configuration. Defaults to standard validation.

Returns¶

ValidationSummary: Comprehensive validation results.

quick_check()[source]

Run quick validation only.

Return type:: ValidationSummary

Returns¶

ValidationSummary: Quick validation results.

full_validation()[source]

Run thorough validation.

Return type:: ValidationSummary

Returns¶

ValidationSummary: Comprehensive validation results.

class mmm_framework.validation.ValidationConfigBuilder[source]

Bases: object

Fluent builder for ValidationConfig.

Examples¶

>>> # Quick validation
>>> config = ValidationConfigBuilder().quick().build()

>>> # Standard with custom residual tests
>>> config = (ValidationConfigBuilder()
...     .standard()
...     .with_residual_tests(("durbin_watson", "ljung_box"))
...     .build())

>>> # Thorough with calibration
>>> config = (ValidationConfigBuilder()
...     .thorough()
...     .with_calibration(lift_tests)
...     .build())

__init__()[source]: Initialize builder with default values.

quick()[source]

Configure for quick validation.

Return type:: Self

standard()[source]

Configure for standard validation.

Return type:: Self

thorough()[source]

Configure for thorough validation.

Return type:: Self

with_ppc(n_samples=500, checks=None, include_channel_checks=True)[source]

Configure posterior predictive checks.

Return type:: Self

Parameters¶

n_samplesint: Number of posterior samples to use.
checkstuple[str, …], optional: Which checks to run. Default: mean, variance, autocorrelation, skewness, extremes.
include_channel_checksbool: Whether to include channel-specific checks.

Returns¶

Self: Builder instance for chaining.

with_residual_tests(tests, max_lag=20, significance_level=0.05)[source]

Configure residual diagnostic tests.

Return type:: Self

Parameters¶

teststuple[str, …]: Which tests to run. Options: durbin_watson, ljung_box, breusch_pagan, shapiro_wilk, jarque_bera.
max_lagint: Maximum lag for autocorrelation tests.
significance_levelfloat: Significance level for hypothesis tests.

Returns¶

Self: Builder instance for chaining.

with_channel_diagnostics(vif_threshold=10.0, correlation_threshold=0.8, rhat_threshold=1.01, ess_threshold=400)[source]

Configure channel diagnostics.

Return type:: Self

Parameters¶

vif_thresholdfloat: VIF threshold for multicollinearity warning.
correlation_thresholdfloat: Correlation threshold for multicollinearity warning.
rhat_thresholdfloat: R-hat threshold for convergence.
ess_thresholdint: ESS threshold for convergence.

Returns¶

Self: Builder instance for chaining.

with_cross_validation(n_folds=5, strategy='expanding', min_train_size=52, gap=0, test_size=None)[source]

Enable cross-validation.

Return type:: Self

Parameters¶

n_foldsint: Number of CV folds.
strategystr: CV strategy: expanding, rolling, or blocked.
min_train_sizeint: Minimum training set size.
gapint: Gap between train and test (for blocked CV).
test_sizeint, optional: Fixed test size (for rolling CV).

Returns¶

Self: Builder instance for chaining.

with_model_comparison(method='loo', pointwise=True)[source]

Enable model comparison.

Return type:: Self

Parameters¶

methodstr: Comparison method: loo, waic, or both.
pointwisebool: Whether to compute pointwise values.

Returns¶

Self: Builder instance for chaining.

with_sensitivity_analysis(prior_multipliers=(0.5, 2.0), parameters_of_interest=None, include_specification_tests=True)[source]

Enable sensitivity analysis.

Return type:: Self

Parameters¶

prior_multiplierstuple[float, …]: Multipliers for prior variance in sensitivity tests.
parameters_of_interesttuple[str, …], optional: Specific parameters to analyze.
include_specification_testsbool: Whether to test specification variants.

Returns¶

Self: Builder instance for chaining.

with_stability_analysis(n_bootstrap=100, loo_subset_size=None, perturbation_level=0.1, n_perturbations=20)[source]

Enable stability analysis.

Return type:: Self

Parameters¶

n_bootstrapint: Number of bootstrap samples.
loo_subset_sizeint, optional: Subset size for LOO influence (None = all).
perturbation_levelfloat: Perturbation level for sensitivity.
n_perturbationsint: Number of perturbation runs.

Returns¶

Self: Builder instance for chaining.

with_calibration(lift_tests, ci_level=0.94, tolerance_multiplier=1.5)[source]

Enable calibration with external experiments.

Return type:: Self

Parameters¶

lift_testslist[LiftTestResult]: List of lift test results for calibration.
ci_levelfloat: Credible interval level for comparison.
tolerance_multiplierfloat: Tolerance multiplier for SE deviation.

Returns¶

Self: Builder instance for chaining.

with_unobserved_confounding(q=1.0)[source]

Enable per-channel robustness-value sensitivity to unobserved confounding.

Return type:: Self

Parameters¶

qfloat: Fraction of the effect a hidden confounder must explain away (1.0 -> nullify the effect entirely).

with_causal_refutation(*, placebo=True, negative_control=True, random_common_cause=True, data_subset=True, subset_fraction=0.8, draws=300, tune=300, chains=2)[source]

Enable the causal refutation suite (each enabled test refits once).

Expensive: every enabled test refits the model on perturbed data. Tests either expect the effect to vanish (placebo, negative control) or to stay stable (random common cause, data subset).

Return type:: Self

without_ppc()[source]

Disable posterior predictive checks.

Return type:: Self

without_residuals()[source]

Disable residual diagnostics.

Return type:: Self

without_channel_diagnostics()[source]

Disable channel diagnostics.

Return type:: Self

without_plots()[source]

Disable plot generation.

Return type:: Self

silent()[source]

Disable verbose output.

Return type:: Self

build()[source]

Build and return the ValidationConfig.

Return type:: ValidationConfig

Returns¶

ValidationConfig: The configured validation settings.

mmm_framework.validation.loo_pit_check(*, y, y_hat, log_weights=None, alpha=0.05)[source]

LOO-PIT calibration check from observed values + posterior-predictive draws.

Return type:: LooPitResult

Parameters¶

yarray (n_obs,): Observed outcomes.
y_hatarray (n_obs, n_samples) or (n_samples, n_obs): Posterior-predictive draws (orientation auto-detected).
log_weightsarray, optional: PSIS log-weights aligned to y_hat. If omitted, uniform weights are used (this reduces LOO-PIT to ordinary PIT — still a useful check).
alphafloat: Significance level for the KS uniformity test. calibrated is ks_pvalue > alpha.

class mmm_framework.validation.LooPitResult(pit, ks_stat, ks_pvalue, calibrated, n)[source]

Bases: object

pit: ndarray

ks_stat: float

ks_pvalue: float

calibrated: bool

n: int

summary()[source]

Return type:: str

__init__(pit, ks_stat, ks_pvalue, calibrated, n)

mmm_framework.validation.simulation_based_calibration(sample_prior, simulate, fit, *, n_sims=100, seed=0, alpha=0.05)[source]

Simulation-based calibration (Talts et al. 2018), model-agnostic.

For each of n_sims iterations: draw a “true” parameter set from the prior, simulate a dataset, fit it, and record where each true value falls within its posterior draws (the normalized rank). Under correctly-calibrated inference those normalized ranks are Uniform(0, 1); a per-parameter KS test flags miscalibration.

Callbacks (each receives the shared Generator for reproducibility): :rtype: SBCResult

sample_prior(rng) -> {param: scalar} — one draw of the true parameters.
simulate(theta, rng) -> data — synthetic data given the true params.
fit(data, rng) -> {param: 1-D posterior draws} — keys must match sample_prior.

Scalar parameters only (the common SBC case). Returns an SBCResult.

class mmm_framework.validation.SBCResult(param_names, n_sims, normalized_ranks=<factory>, ks_pvalue=<factory>, calibrated=<factory>)[source]

Bases: object

param_names: list[str]

n_sims: int

normalized_ranks: dict[str, ndarray]

ks_pvalue: dict[str, float]

calibrated: dict[str, bool]

property all_calibrated: bool

summary()[source]

Return type:: str

__init__(param_names, n_sims, normalized_ranks=<factory>, ks_pvalue=<factory>, calibrated=<factory>)

class mmm_framework.validation.BacktestConfig(min_train_size=104, horizon=13, step=None, max_origins=None, coverage_levels=(0.5, 0.8, 0.95), draws=500, tune=500, chains=4, include_noise=True, season_period=None, random_seed=42)[source]

Bases: object

Configuration for a rolling-origin backtest.

Attributes¶

min_train_sizeint: Periods in the first training window. Two seasonal cycles of weekly data (104) is a sensible floor; below ~78 the seasonality and trend posteriors are barely informed and forecast intervals balloon.
horizonint: Forecast length (periods) past each training cutoff.
stepint or None: Spacing between consecutive training cutoffs. None uses horizon (non-overlapping forecast windows).
max_originsint or None: Cap on the number of refits (None = as many as the data allows).
coverage_levelstuple[float, …]: Central prediction-interval levels to record and grade.
draws, tune, chainsint: MCMC budget per refit. Backtests refit the model once per origin, so these default lower than a production fit; convergence is still recorded per origin in BacktestResult.fits.
include_noisebool: Forecast the observable (mean + observation noise) rather than the latent mean. Keep True for honest interval coverage.
season_periodint or None: Seasonal lag for the seasonal-naive baseline and the MASE scale. None derives it from the data frequency (weekly -> 52).
random_seedint: Base seed; each origin offsets it deterministically.

min_train_size: int = 104

horizon: int = 13

step: int | None = None

max_origins: int | None = None

coverage_levels: tuple[float, ...] = (0.5, 0.8, 0.95)

draws: int = 500

tune: int = 500

chains: int = 4

include_noise: bool = True

season_period: int | None = None

random_seed: int = 42

__init__(min_train_size=104, horizon=13, step=None, max_origins=None, coverage_levels=(0.5, 0.8, 0.95), draws=500, tune=500, chains=4, include_noise=True, season_period=None, random_seed=42)

class mmm_framework.validation.BacktestResult(config, records, fits, season_period, mase_scales=<factory>)[source]

Bases: object

Rolling-origin backtest records plus summary views.

Attributes¶

recordspd.DataFrame: One row per (origin, horizon) forecast: origin, position, date, horizon, y_true, y_pred (posterior-mean), pred_naive, pred_snaive, per-level interval bounds (lo_80/hi_80, …) and coverage flags (cov_80, …).
fitspd.DataFrame: One row per refit origin: train size, wall-clock seconds, R-hat max, divergences. Check this before trusting the records.

config: BacktestConfig

records: DataFrame

fits: DataFrame

season_period: int

mase_scales: dict[int, float]

property n_origins: int

property mape: float: Headline out-of-time MAPE of the MMM over all records (the number the docs quote). Convenience for summary().loc["mmm", "mape"].

summary()[source]

Headline accuracy: the model vs naive baselines, all records.

Return type:: DataFrame

by_horizon()[source]

Accuracy and coverage as a function of forecast lead time.

Return type:: DataFrame

by_origin()[source]

Accuracy per refit origin (forecast-window heterogeneity).

Return type:: DataFrame

coverage()[source]

Interval calibration: nominal vs empirical coverage + sharpness.

Return type:: DataFrame

__init__(config, records, fits, season_period, mase_scales=<factory>)

class mmm_framework.validation.PosteriorForecaster(model, *, strict=True)[source]

Bases: object

Out-of-time forecasts from a fitted BayesianMMM’s posterior draws.

Replays the model’s structural forward pass (intercept, trend, seasonality, geo and product level offsets, adstock, saturation, controls, observation noise) in NumPy at arbitrary future period positions, using the full spend history for adstock carryover.

The replay is exact: over training positions, forecast(include_noise= False) equals the sum of the model’s registered component Deterministics to floating-point tolerance. Terms it cannot replay are refused here, in the constructor, rather than at forecast time — so a caller cannot hold a forecaster whose output it is not allowed to trust.

Parameters¶

modelBayesianMMM: A fitted model, possibly trained on a prefix of the full series.
strictbool, default True: When True, an unreplayable term raises ForecastUnsupportedError. False downgrades the refusal to a warning and records it on self.unsupported — for diagnostic use on a model you already know is incomplete. Never use it for planning: the resulting forecast omits a term the model estimated.

Attributes¶

trend_extrapolationTrendExtrapolation: The stated policy for continuing the trend past the training window.
unsupportedlist[tuple[str, str]]: Empty unless strict=False suppressed a refusal.

Raises¶

ForecastUnsupportedError: The model carries a term the forward pass cannot replay.

__init__(model, *, strict=True)[source]

property n_samples: int

forecast(X_media_full_raw, X_controls_full_raw, positions, *, include_noise=True, random_seed=None, train_offset=0)[source]

Posterior predictive draws at absolute period positions.

Return type:: ndarray

Parameters¶

X_media_full_rawnp.ndarray: Raw media, shape (n_full, n_channels) – the FULL history (training + forecast periods) so adstock carryover is correct.
X_controls_full_rawnp.ndarray or None: Raw controls over the full history (required if the model has controls; future control values are assumed known/planned).
positionsnp.ndarray: Absolute period positions (0-based on the full axis) to forecast.
train_offsetint: Absolute position of the trained model’s first period. 0 for prefix training (the backtest); the window start for rolling-window clones (validator cross-validation).

Returns¶

np.ndarray: Samples in original KPI scale, shape (n_samples, len(positions)).

mmm_framework.validation.rolling_origins(n_periods, *, min_train_size, horizon, step=None, max_origins=None)[source]

Training cutoffs for a rolling-origin (expanding-window) backtest.

Each returned cutoff T means: train on periods [0, T), forecast periods [T, min(T + horizon, n_periods)). Only full forecast windows are emitted so every origin is graded on the same horizons.

Return type:: list[int]

exception mmm_framework.validation.ForecastUnsupportedError(feature, reason, all_unsupported=None)[source]

Bases: NotImplementedError

The fitted model carries a term the forward pass cannot replay.

Raised instead of silently dropping the term, mirroring mmm_framework.frequentist.design.UnsupportedModelError. Carries feature so callers (the backtest harness, the agent, the REST job) can report which configuration blocked the forecast rather than a generic refusal, and all_unsupported so a UI can list every blocker at once instead of making the user fix them one at a time.

__init__(feature, reason, all_unsupported=None)[source]

class mmm_framework.validation.TrendExtrapolation(policy, trend_type, n_train_periods)[source]

Bases: object

How the forecaster continues the trend past the training window.

Recorded rather than assumed, because the three policies carry very different forecast semantics and only one of them is model-defined.

Attributes¶

policy{‘none’, ‘linear’, ‘held_flat’}: held_flat is a heuristic: a spline/GP/piecewise basis has no out-of-time forecast, so the last fitted level is carried forward and the interval consequently does not widen with horizon.
trend_typestr: The configured trend family.
n_train_periodsint: Training length. Load-bearing for linear: the fitted slope is on a t_scaled = pos / (n_train - 1) axis, so the same posterior implies a different slope per period depending on how long the training panel was. Reporting it makes that visible rather than surprising.

policy: str

trend_type: str

n_train_periods: int

property is_model_defined: bool: False when the continuation is a heuristic rather than the model’s.

describe()[source]

Return type:: str

__init__(policy, trend_type, n_train_periods)

mmm_framework.validation.audit_forward_pass(model)[source]

Every fitted-mean term PosteriorForecaster cannot replay.

Returns (feature, reason) pairs, most-likely-to-be-hit first, so the first message a user sees names the thing they actually configured. Empty means the forward pass reproduces the fitted mean exactly.

The reference for “every term” is the mu construction in model/base.py::_build_model: intercept + trend + seasonality + geo + product + media + controls plus the conditional event, interaction and lever blocks. Each conditional block below is one of those.

Return type:: list[tuple[str, str]]

mmm_framework.validation.audit_refit(model)[source]

What a rolling-origin refit would change about the model under test.

Distinct from audit_forward_pass(): these terms do not break the forward pass, they make the backtest grade a different model than the one the user holds. Reported by run_backtest(), not by the forecaster.

Return type:: list[tuple[str, str]]

mmm_framework.validation.rebuild_like(model, panel, **overrides)[source]

A fresh, unfitted model of type(model) on panel.

The single place that reconstructs a model from another one. Preserving the CLASS and model_params is load-bearing: hard-constructing BayesianMMM means a garden or custom model is fit and graded as a plain additive MMM and its results reported under the custom model’s name.

Return type:: Any

Raises¶

ForecastUnsupportedError: The class cannot be rebuilt from (panel, model_config, trend_config, adstock_alphas, model_params) — e.g. a model declaring REQUIRED_DATASET_CAPABILITIES the sliced panel no longer satisfies. Refusing beats falling back to a different model class.

mmm_framework.validation.run_backtest(model, config=None, **fit_kwargs)[source]

Rolling-origin backtest of a BayesianMMM specification.

Refits the model’s exact configuration on an expanding training window, forecasts config.horizon periods past each cutoff with PosteriorForecaster, and grades against held-out actuals and naive baselines.

Return type:: BacktestResult

Parameters¶

modelBayesianMMM: The full-data model (does not need to be fitted); supplies the panel, configuration, and the raw media/control history.
configBacktestConfig, optional: Backtest settings; defaults to BacktestConfig().
**fit_kwargs: Extra arguments forwarded to each refit’s fit() (e.g. progressbar=False).

Returns¶

BacktestResult

class mmm_framework.validation.SpecSet(**data)[source]

Bases: BaseModel

A pre-registered set of defensible specs, declared before results.

Serialize this (model_dump) into the design-readout / assumption log before fitting so the robustness report is auditable — the spec set was fixed in advance, not chosen after seeing which answer looked nicest.

variants: list[SpecVariant]

rationale: str

registered_at: str | None: ISO timestamp of pre-registration (set by the caller; kept as data, not computed here so the module stays deterministic / clock-free).

property primary_variant: SpecVariant | None

names()[source]

Return type:: list[str]

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class mmm_framework.validation.SpecVariant(**data)[source]

Bases: BaseModel

One defensible specification, expressed as overrides on a base spec.

The structured axes cover the four the review calls out (adstock form, saturation form, control set, pooling); overrides is an escape hatch for any other spec key (deep-merged last).

name: str

adstock: str | None: Adstock family applied to every media channel (“geometric”/”weibull”/”delayed”).

saturation: str | None: Saturation family applied to every media channel (“hill”/”logistic”/”michaelis_menten”/”tanh”).

controls: list[str] | None: Full replacement control set (channel-name list). None keeps the base’s.

kpi_level: str | None: Pooling scheme (“national” / “geo”).

media_prior_mode: str | None: Default media-prior parameterization (“roi” / “coefficient”).

trend: str | None: Trend family (“none”/”linear”/”piecewise”/”spline”/”gaussian_process”).

seasonality: dict[str, Any] | None: Seasonality override, e.g. {"yearly": 4} (or {} to disable).

overrides: dict[str, Any]: Arbitrary deep-merge overrides (applied last).

primary: bool: Whether this is the pre-registered PRIMARY specification.

description: str

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class mmm_framework.validation.SpecFit(name, primary, roi=<factory>, roi_draws=<factory>, loo=None, weight=0.0, error=None)[source]

Bases: object

One spec’s realized ROI + model-fit summary.

name: str

primary: bool

roi: dict[str, dict[str, float]]

roi_draws: dict[str, ndarray]

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

weight: float = 0.0

error: str | None = None

__init__(name, primary, roi=<factory>, roi_draws=<factory>, loo=None, weight=0.0, error=None)

class mmm_framework.validation.SpecCurveResult(channels, specs, primary, hdi_prob, fits, weights, bma, robustness, weighting='equal', predictive_weights=<factory>)[source]

Bases: object

Spec-curve outcome: per-spec ROI, applied weights, BMA, robustness.

channels: list[str]

specs: list[str]

primary: str | None

hdi_prob: float

fits: list[SpecFit]

weights: dict[str, float]

bma: dict[str, dict[str, float]]

robustness: dict[str, dict[str, Any]]

weighting: str = 'equal': Which weighting formed bma — "equal" (default) or "stacking".

predictive_weights: dict[str, float]: LOO-stacking weights, reported for diagnosis even when not applied. Empty when compute_loo=False or arviz could not run.

to_dict()[source]

JSON-friendly payload for the report bundle / agent tool (no draws).

Return type:: dict[str, Any]

__init__(channels, specs, primary, hdi_prob, fits, weights, bma, robustness, weighting='equal', predictive_weights=<factory>)

mmm_framework.validation.apply_variant(base_spec, variant)[source]

Return a new spec = base_spec with variant applied.

The structured axes are applied first (adstock/saturation forms are set on every media channel; controls/pooling/prior-mode/trend/seasonality set the corresponding top-level keys), then variant.overrides is deep-merged.

Return type:: dict

mmm_framework.validation.default_spec_variants(base_spec, *, adstock_forms=('geometric', 'weibull'), saturation_forms=('hill', 'logistic'), include_prior_mode=False)[source]

The standard defensible spec grid: adstock × saturation forms.

The base spec’s own (adstock, saturation) combination is marked primary. include_prior_mode adds a coefficient-scale-prior sibling of the primary (the other big modelling fork). Callers may of course hand-write their own SpecVariant list instead.

Return type:: list[SpecVariant]

mmm_framework.validation.run_spec_curve(base_spec, dataset_path, *, variants=None, hdi_prob=0.94, max_draws=400, random_seed=42, compute_loo=True, weighting='equal', fit_fn=None, roi_fn=None)[source]

Fit a spec set, collect per-channel ROI, and model-average across specs.

Return type:: SpecCurveResult

Parameters¶

base_spec, dataset_path:: The base spec and the dataset file each variant is fit against.
variants:: A SpecSet, a list of SpecVariant, or None (uses default_spec_variants()).
hdi_prob, max_draws, random_seed:: Credible-interval mass, ROI-draw thinning, and seed.
compute_loo:: Compute per-spec LOO and the diagnostic LOO-stacking weights. These are reported either way; whether they are applied is weighting.
weighting:: "equal" (default) weights every non-failing spec equally — the pre-registration-consistent choice for averaging a causal estimand. "stacking" applies LOO-stacking weights instead and emits a warning: stacking maximizes expected predictive utility, which is a different objective from validity for ROI. See the module docstring.
fit_fn, roi_fn:: Injection points for testing. fit_fn(spec, dataset_path) -> model (default: build + fit, no serialization); roi_fn(model, channels, max_draws=, random_seed=) -> {channel: draws} (default: channel_roi_draws()).

class mmm_framework.validation.PPCValidator(model, config=None)[source]

Bases: object

Posterior predictive check orchestrator.

Runs a suite of posterior predictive checks to assess model adequacy.

Examples¶

>>> validator = PPCValidator(results)
>>> ppc_results = validator.run()
>>> print(ppc_results.summary())

__init__(model, config=None)[source]

Initialize PPC validator.

Parameters¶

modelAny: Fitted model with trace and predict method.
configPPCConfig, optional: Configuration for PPC checks.

run(y_obs=None, y_rep=None, random_seed=None)[source]

Run posterior predictive checks.

Return type:: PPCResults

Parameters¶

y_obsnp.ndarray, optional: Observed data. If None, extracted from model.
y_repnp.ndarray, optional: Replicated data. If None, generated from posterior.
random_seedint, optional: Random seed for reproducibility.

Returns¶

PPCResults: Results of all PPC checks.

class mmm_framework.validation.ResidualDiagnostics(model, config=None)[source]

Bases: object

Comprehensive residual analysis.

Performs statistical tests on model residuals to check for autocorrelation, heteroscedasticity, and non-normality.

Examples¶

>>> diagnostics = ResidualDiagnostics(model, results)
>>> diag_results = diagnostics.run_all()
>>> print(diag_results.summary())

__init__(model, config=None)[source]

Initialize residual diagnostics.

Parameters¶

modelAny: Fitted model with trace.
configResidualConfig, optional: Configuration for residual tests.

run_all(residuals=None, fitted_values=None)[source]

Run all configured residual diagnostic tests.

Return type:: ResidualDiagnosticsResults

Parameters¶

residualsnp.ndarray, optional: Model residuals. If None, computed from model.
fitted_valuesnp.ndarray, optional: Fitted values. If None, computed from model.

Returns¶

ResidualDiagnosticsResults: Results of all diagnostic tests.

class mmm_framework.validation.ChannelDiagnostics(model, config=None)[source]

Bases: object

Comprehensive channel-level diagnostics.

Combines VIF analysis, correlation matrix, and per-channel convergence.

Examples¶

>>> diagnostics = ChannelDiagnostics(model)
>>> results = diagnostics.run_all()
>>> print(results.summary())

__init__(model, config=None)[source]

Initialize channel diagnostics.

Parameters¶

modelAny: Fitted model with media data and trace.
configChannelDiagnosticsConfig, optional: Configuration for diagnostics.

run_all()[source]

Run all channel diagnostics.

Return type:: ChannelDiagnosticsResults

Returns¶

ChannelDiagnosticsResults: Comprehensive channel diagnostics results.

vif_analysis()[source]

Get VIF analysis as DataFrame.

Return type:: DataFrame

Returns¶

pd.DataFrame: VIF scores with interpretation.

correlation_matrix()[source]

Get media channel correlation matrix.

Return type:: DataFrame

Returns¶

pd.DataFrame: Correlation matrix.

class mmm_framework.validation.FrozenPredictor(predict_fn, input_names, input_shapes, posterior_samples, sigma_samples, config)[source]

Bases: object

Compiled predictor using frozen posterior samples.

Reuses the model’s actual computation graph with: - Free RVs replaced by frozen posterior samples - Input data replaced by new symbolic inputs - SpecifyShape ops removed for dynamic sizing

This avoids manually reconstructing model logic and ensures predictions match exactly what the model would produce.

Parameters¶

predict_fncallable: Compiled PyTensor function for predictions.
input_nameslist[str]: Names of input variables (pm.Data nodes).
input_shapesdict[str, tuple]: Expected shapes for each input (for validation).
posterior_samplesdict[str, np.ndarray]: Flattened posterior samples for all needed RVs.
sigma_samplesnp.ndarray | None: Flattened sigma samples for observation noise.
configFrozenPredictorConfig: Configuration used to create this predictor.

__init__(predict_fn, input_names, input_shapes, posterior_samples, sigma_samples, config)[source]

property n_samples: int: Number of posterior samples available.

property input_names: list[str]: Names of required input variables.

predict(inputs, include_noise=None, seed=None)[source]

Generate predictions for new inputs.

Return type:: FrozenPredictorOutput

Parameters¶

inputsdict[str, np.ndarray]: New input data. Keys must match input_names. Values should have shape (n_points, …) matching original dims.
include_noisebool | None: Whether to add observation noise. If None, uses config setting.
seedint | None: Random seed for noise generation. If None, uses config seed.

Returns¶

FrozenPredictorOutput: Contains mu (deterministic) and optionally y_samples (with noise).

class mmm_framework.validation.FrozenPredictorConfig(output_var='y_obs', include_noise=True, seed=None, num_samples=None)[source]

Bases: object

Configuration for frozen predictor creation.

Parameters¶

output_varstr: Name of the output variable to predict (e.g., “y_obs”, “mu”). Default is “y_obs” which gives the full likelihood output.
include_noisebool: Whether to include observation noise (sigma) in predictions. If True, samples from Normal(mu, sigma). Default is True.
seedint | None: Random seed for reproducibility. Affects both posterior sample selection and observation noise generation.
num_samplesint | None: Number of posterior samples to use. If None, uses all available.

output_var: str = 'y_obs'

include_noise: bool = True

seed: int | None = None

num_samples: int | None = None

__init__(output_var='y_obs', include_noise=True, seed=None, num_samples=None)

exception mmm_framework.validation.FrozenPredictorError[source]

Bases: Exception

Raised when frozen predictor cannot be created or used.

class mmm_framework.validation.FrozenPredictorOutput(mu, y_samples=None, sigma=None)[source]

Bases: object

Output from frozen predictor evaluation.

Parameters¶

munp.ndarray: Deterministic predictions (mean), shape (n_samples, n_points).
y_samplesnp.ndarray | None: Predictions with observation noise (if include_noise=True), shape (n_samples, n_points).
sigmanp.ndarray | None: Sigma samples used for noise generation.

mu: ndarray[tuple[Any, ...], dtype[floating[Any]]]

y_samples: ndarray[tuple[Any, ...], dtype[floating[Any]]] | None = None

sigma: ndarray[tuple[Any, ...], dtype[floating[Any]]] | None = None

__init__(mu, y_samples=None, sigma=None)

mmm_framework.validation.create_frozen_predictor(model, trace, config=None)[source]

Create a frozen predictor from a fitted PyMC model.

This function creates a compiled PyTensor function that evaluates model expressions at new input values, vectorized over posterior samples. It reuses the actual model graph rather than reconstructing it manually.

Return type:: FrozenPredictor

Parameters¶

modelpm.Model: The fitted PyMC model.
traceaz.InferenceData: ArviZ InferenceData with posterior samples.
configFrozenPredictorConfig | None: Configuration options. If None, uses defaults.

Returns¶

FrozenPredictor: Compiled predictor ready for inference.

Raises¶

FrozenPredictorError: If the predictor cannot be created (e.g., missing output var, ICDF not available for observed RV).

Examples¶

>>> predictor = create_frozen_predictor(model.model, model._trace)
>>> output = predictor.predict({
...     "X_media_low": X_new_low,
...     "X_media_high": X_new_high,
...     "X_controls": X_controls_new,
...     "time_idx": time_idx_new,
...     "geo_idx": geo_idx_new,
...     "product_idx": product_idx_new,
... })
>>> y_pred = output.y_samples.mean(axis=0)

mmm_framework.validation.create_frozen_predictor_from_model(mmm_model, config=None)[source]

Create a frozen predictor from an MMM model object.

Convenience wrapper that extracts the PyMC model and trace from a BayesianMMM or extended model object.

Return type:: FrozenPredictor

Parameters¶

mmm_modelHasPyMCModel: MMM model with .model and ._trace attributes.
configFrozenPredictorConfig | None: Configuration options.

Returns¶

FrozenPredictor: Compiled predictor.

Raises¶

FrozenPredictorError: If model hasn’t been fitted or predictor creation fails.

class mmm_framework.validation.ValidationConfig(level=ValidationLevel.STANDARD, ppc=<factory>, residuals=<factory>, channel_diagnostics=<factory>, cross_validation=<factory>, model_comparison=<factory>, sensitivity=<factory>, stability=<factory>, calibration=<factory>, causal_refutation=<factory>, run_ppc=True, run_residuals=True, run_channel_diagnostics=True, run_model_comparison=False, run_cross_validation=False, run_sensitivity=False, run_stability=False, run_calibration=False, run_unobserved_confounding=False, run_causal_refutation=False, unobserved_confounding_q=1.0, lift_tests=None, generate_plots=True, verbose=True)[source]

Bases: object

Complete validation configuration.

Controls which validations to run and their parameters.

Examples¶

>>> # Quick validation
>>> config = ValidationConfig.quick()

>>> # Standard with custom residual tests
>>> config = ValidationConfig.standard()

>>> # Thorough with calibration data
>>> config = ValidationConfig.thorough()

level: ValidationLevel = 'standard'

ppc: PPCConfig

residuals: ResidualConfig

channel_diagnostics: ChannelDiagnosticsConfig

cross_validation: CrossValidationConfig

model_comparison: ModelComparisonConfig

sensitivity: SensitivityConfig

stability: StabilityConfig

calibration: CalibrationConfig

causal_refutation: CausalRefutationConfig

run_ppc: bool = True

run_residuals: bool = True

run_channel_diagnostics: bool = True

run_model_comparison: bool = False

run_cross_validation: bool = False

run_sensitivity: bool = False

run_stability: bool = False

run_calibration: bool = False

run_unobserved_confounding: bool = False

run_causal_refutation: bool = False

unobserved_confounding_q: float = 1.0

lift_tests: list[LiftTestResult] | None = None

generate_plots: bool = True

verbose: bool = True

classmethod quick()[source]

Quick validation (convergence, PPC, residuals, channel diagnostics).

Fast feedback on model quality, suitable for iterative development.

Return type:: ValidationConfig

classmethod standard()[source]

Standard validation (quick + LOO-CV, WAIC).

Good balance of thoroughness and compute time.

Return type:: ValidationConfig

classmethod thorough()[source]

Thorough validation (all checks).

Comprehensive validation for production models.

Return type:: ValidationConfig

__init__(level=ValidationLevel.STANDARD, ppc=<factory>, residuals=<factory>, channel_diagnostics=<factory>, cross_validation=<factory>, model_comparison=<factory>, sensitivity=<factory>, stability=<factory>, calibration=<factory>, causal_refutation=<factory>, run_ppc=True, run_residuals=True, run_channel_diagnostics=True, run_model_comparison=False, run_cross_validation=False, run_sensitivity=False, run_stability=False, run_calibration=False, run_unobserved_confounding=False, run_causal_refutation=False, unobserved_confounding_q=1.0, lift_tests=None, generate_plots=True, verbose=True)

class mmm_framework.validation.ValidationLevel(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

Validation thoroughness level.

QUICK = 'quick'

STANDARD = 'standard'

THOROUGH = 'thorough'

class mmm_framework.validation.PPCConfig(n_samples=500, checks=('mean', 'variance', 'autocorrelation', 'skewness', 'extremes'), include_channel_checks=True, significance_level=0.05)[source]

Bases: object

Configuration for posterior predictive checks.

n_samples: int = 500

checks: tuple[str, ...] = ('mean', 'variance', 'autocorrelation', 'skewness', 'extremes')

include_channel_checks: bool = True

significance_level: float = 0.05

__init__(n_samples=500, checks=('mean', 'variance', 'autocorrelation', 'skewness', 'extremes'), include_channel_checks=True, significance_level=0.05)

class mmm_framework.validation.ResidualConfig(max_lag=20, significance_level=0.05, tests=('durbin_watson', 'ljung_box', 'breusch_pagan', 'shapiro_wilk', 'jarque_bera'))[source]

Bases: object

Configuration for residual diagnostics.

max_lag: int = 20

significance_level: float = 0.05

tests: tuple[str, ...] = ('durbin_watson', 'ljung_box', 'breusch_pagan', 'shapiro_wilk', 'jarque_bera')

__init__(max_lag=20, significance_level=0.05, tests=('durbin_watson', 'ljung_box', 'breusch_pagan', 'shapiro_wilk', 'jarque_bera'))

class mmm_framework.validation.ChannelDiagnosticsConfig(vif_threshold=10.0, correlation_threshold=0.8, rhat_threshold=1.01, ess_threshold=400)[source]

Bases: object

Configuration for channel diagnostics.

vif_threshold: float = 10.0

correlation_threshold: float = 0.8

rhat_threshold: float = 1.01

ess_threshold: int = 400

__init__(vif_threshold=10.0, correlation_threshold=0.8, rhat_threshold=1.01, ess_threshold=400)

class mmm_framework.validation.CrossValidationConfig(strategy='expanding', n_folds=5, min_train_size=52, gap=0, test_size=None, draws_per_fold=500, tune_per_fold=250, chains_per_fold=2, use_frozen_predictor=True, frozen_predictor_seed=42)[source]

Bases: object

Configuration for cross-validation.

strategy: Literal['expanding', 'rolling', 'blocked'] = 'expanding'

n_folds: int = 5

min_train_size: int = 52

gap: int = 0

test_size: int | None = None

draws_per_fold: int = 500

tune_per_fold: int = 250

chains_per_fold: int = 2

use_frozen_predictor: bool = True

frozen_predictor_seed: int | None = 42

__init__(strategy='expanding', n_folds=5, min_train_size=52, gap=0, test_size=None, draws_per_fold=500, tune_per_fold=250, chains_per_fold=2, use_frozen_predictor=True, frozen_predictor_seed=42)

class mmm_framework.validation.ModelComparisonConfig(method='loo', pointwise=True, pareto_k_threshold=0.7)[source]

Bases: object

Configuration for model comparison.

method: Literal['loo', 'waic', 'both'] = 'loo'

pointwise: bool = True

pareto_k_threshold: float = 0.7

__init__(method='loo', pointwise=True, pareto_k_threshold=0.7)

class mmm_framework.validation.SensitivityConfig(prior_multipliers=(0.5, 2.0), parameters_of_interest=None, include_specification_tests=True, draws_per_variant=500, tune_per_variant=250, chains_per_variant=2)[source]

Bases: object

Configuration for sensitivity analysis.

prior_multipliers: tuple[float, ...] = (0.5, 2.0)

parameters_of_interest: tuple[str, ...] | None = None

include_specification_tests: bool = True

draws_per_variant: int = 500

tune_per_variant: int = 250

chains_per_variant: int = 2

__init__(prior_multipliers=(0.5, 2.0), parameters_of_interest=None, include_specification_tests=True, draws_per_variant=500, tune_per_variant=250, chains_per_variant=2)

class mmm_framework.validation.StabilityConfig(n_bootstrap=100, loo_subset_size=None, perturbation_level=0.1, n_perturbations=20)[source]

Bases: object

Configuration for stability analysis.

n_bootstrap: int = 100

loo_subset_size: int | None = None

perturbation_level: float = 0.1

n_perturbations: int = 20

__init__(n_bootstrap=100, loo_subset_size=None, perturbation_level=0.1, n_perturbations=20)

class mmm_framework.validation.CalibrationConfig(ci_level=0.94, tolerance_multiplier=1.5)[source]

Bases: object

Configuration for calibration checks.

ci_level: float = 0.94

tolerance_multiplier: float = 1.5

__init__(ci_level=0.94, tolerance_multiplier=1.5)

class mmm_framework.validation.CausalRefutationConfig(run_placebo=True, run_negative_control=True, run_random_common_cause=True, run_data_subset=True, subset_fraction=0.8, media_r2_threshold=0.05, negative_control_r2_threshold=0.3, move_tolerance=0.5, underpowered_se_ratio=1.0, draws=300, tune=300, chains=2, random_seed=1234)[source]

Bases: object

Configuration for the causal refutation suite (each test refits once).

Because the media coefficient prior is positive-mean, a placebo coefficient does NOT vanish to zero – so vanishing-effect tests are measured by fit, not by the coefficient:

negative_control_outcome permutes the KPI; a valid, regularized model should be unable to fit the scrambled outcome (refit R^2 below negative_control_r2_threshold).
placebo_treatment permutes media; scrambled media should add no incremental explanatory power over a media-free baseline (incremental media R^2 below media_r2_threshold).

Stability tests compare the per-channel coefficient before/after and pass when it moves less than move_tolerance (relative):

random_common_cause injects a random control.
data_subset refits on a random subset.

precision (refit coefficient SD) is reported on stability tests so an underpowered “pass” is not oversold.

run_placebo: bool = True

run_negative_control: bool = True

run_random_common_cause: bool = True

run_data_subset: bool = True

subset_fraction: float = 0.8

media_r2_threshold: float = 0.05

negative_control_r2_threshold: float = 0.3

move_tolerance: float = 0.5

underpowered_se_ratio: float = 1.0

draws: int = 300

tune: int = 300

chains: int = 2

random_seed: int = 1234

__init__(run_placebo=True, run_negative_control=True, run_random_common_cause=True, run_data_subset=True, subset_fraction=0.8, media_r2_threshold=0.05, negative_control_r2_threshold=0.3, move_tolerance=0.5, underpowered_se_ratio=1.0, draws=300, tune=300, chains=2, random_seed=1234)

class mmm_framework.validation.UnobservedConfoundingAnalysis(model)[source]

Bases: object

Per-channel sensitivity of media effects to unobserved confounding.

Parameters¶

modelBayesianMMM: A fitted model exposing _trace.posterior with beta_<channel> variables, channel_names and n_obs.

__init__(model)[source]

run(q=1.0)[source]

Compute robustness values for every channel coefficient.

Return type:: UnobservedConfoundingSensitivity

class mmm_framework.validation.UnobservedConfoundingSensitivity(channels, dof, q, caveat)[source]

Bases: object

Robustness of each channel effect to an unobserved confounder.

channels: list[ChannelRobustness]

dof: int

q: float

caveat: str

property fragile_channels: list[str]

property unassessable_channels: list[str]

Channels whose RV could not be computed (see ChannelRobustness.is_assessable).

These are NOT in fragile_channels — a non-finite RV fails the rv < threshold test — so any caller reading an empty fragile_channels as “all robust” must consult this list too.

summary()[source]

Return type:: DataFrame

to_dict()[source]

Return type:: dict[str, Any]

__init__(channels, dof, q, caveat)

class mmm_framework.validation.ChannelRobustness(channel, estimate, std_error, t_value, dof, partial_r2, robustness_value, robustness_value_half, fragile_threshold=0.1, prior_contraction=None, prior_dominated_threshold=0.2)[source]

Bases: object

Per-channel sensitivity of the media effect to unobserved confounding.

channel: str

estimate: float

std_error: float

t_value: float

dof: int

partial_r2: float

robustness_value: float

robustness_value_half: float

fragile_threshold: float = 0.1

prior_contraction: float | None = None: Prior->posterior contraction 1 - Var_post / Var_prior for this channel’s coefficient, or None when no prior group was available. The RV is strictly increasing in |t| = |mean| / sd, and a tight prior shrinks sd — so a prior-dominated coefficient reports a high RV without any supporting evidence. See mmm_framework.validation.sensitivity_unobserved.

prior_dominated_threshold: float = 0.2: Below this contraction the posterior is treated as prior-dominated and the RV is not quotable as evidence of robustness.

property is_fragile: bool: Effect could be overturned by a weak (< threshold) confounder.

property is_prior_dominated: bool

The posterior barely narrowed the prior, so the RV reflects the prior.

None contraction (no prior group sampled) reads as not flagged — absence of the check is reported separately rather than as a failure.

property is_assessable: bool

Whether a robustness value could be computed at all.

A non-finite RV means |t| = |mean| / sd was undefined — typically an approximate (MAP/ADVI) fit, whose degenerate posterior gives a zero or undefined sd. This is distinct from is_prior_dominated, where the RV exists but reflects the prior: here there is no value to quote.

Load-bearing because is_fragile is isfinite(rv) and rv < thr, so a non-finite RV is not fragile — and every consumer that reads “not fragile” as “robust” would otherwise report a passed sensitivity check that was never computed.

property rv_is_quotable: bool: Whether the robustness value can be quoted as evidence at all.

to_dict()[source]

Return type:: dict[str, Any]

__init__(channel, estimate, std_error, t_value, dof, partial_r2, robustness_value, robustness_value_half, fragile_threshold=0.1, prior_contraction=None, prior_dominated_threshold=0.2)

class mmm_framework.validation.CausalRefutationResults(tests, underpowered=False)[source]

Bases: object

Aggregate results of the causal refutation suite.

tests: list[RefutationTest]

underpowered: bool = False

n_passed: int

n_failed: int

all_passed: bool

summary()[source]

Return type:: DataFrame

to_dict()[source]

Return type:: dict[str, Any]

__init__(tests, underpowered=False)

class mmm_framework.validation.RefutationTest(name, kind, passed, description, original_effect=None, refuted_effect=None, refuted_ci_low=None, refuted_ci_high=None, precision=None, channel=None, details='')[source]

Bases: object

Result of a single causal refutation test.

For vanishing-effect tests (placebo treatment, negative-control outcome) a correct model should collapse the effect toward zero. For stability tests (random common cause, data subset) a correct model’s estimate should barely move. passed encodes the appropriate per-test criterion; precision is the refit standard error, reported so an underpowered “pass” is not oversold.

name: str

kind: Literal['vanish', 'stable']

passed: bool

description: str

original_effect: float | None = None

refuted_effect: float | None = None

refuted_ci_low: float | None = None

refuted_ci_high: float | None = None

precision: float | None = None

channel: str | None = None

details: str = ''

to_dict()[source]

Return type:: dict[str, Any]

__init__(name, kind, passed, description, original_effect=None, refuted_effect=None, refuted_ci_low=None, refuted_ci_high=None, precision=None, channel=None, details='')

class mmm_framework.validation.ValidationSummary(model_name, validation_date=<factory>, convergence=None, ppc=None, residuals=None, channel_diagnostics=None, model_comparison=None, cross_validation=None, sensitivity=None, stability=None, calibration=None, unobserved_confounding=None, causal_refutation=None, overall_quality='acceptable', critical_issues=<factory>, warnings=<factory>, recommendations=<factory>, _full_y_actual=None)[source]

Bases: object

Comprehensive validation summary.

Aggregates all validation results with overall assessment.

model_name: str

validation_date: str

convergence: ConvergenceSummary | None = None

ppc: PPCResults | None = None

residuals: ResidualDiagnosticsResults | None = None

channel_diagnostics: ChannelDiagnosticsResults | None = None

model_comparison: ModelComparisonResults | None = None

cross_validation: CrossValidationResults | None = None

sensitivity: SensitivityResults | None = None

stability: StabilityResults | None = None

calibration: CalibrationResults | None = None

unobserved_confounding: UnobservedConfoundingSensitivity | None = None

causal_refutation: CausalRefutationResults | None = None

overall_quality: Literal['excellent', 'good', 'acceptable', 'poor'] = 'acceptable'

critical_issues: list[str]

warnings: list[str]

recommendations: list[str]

summary()[source]

Get high-level summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

to_html_report(include_charts=True)[source]

Generate HTML validation report.

Return type:: str

Parameters¶

include_chartsbool: Whether to include interactive Plotly charts.

Returns¶

str: HTML report string.

__init__(model_name, validation_date=<factory>, convergence=None, ppc=None, residuals=None, channel_diagnostics=None, model_comparison=None, cross_validation=None, sensitivity=None, stability=None, calibration=None, unobserved_confounding=None, causal_refutation=None, overall_quality='acceptable', critical_issues=<factory>, warnings=<factory>, recommendations=<factory>, _full_y_actual=None)

class mmm_framework.validation.ConvergenceSummary(divergences, rhat_max, ess_bulk_min, ess_tail_min, converged)[source]

Bases: object

MCMC convergence summary.

divergences: int

rhat_max: float

ess_bulk_min: float

ess_tail_min: float

converged: bool

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

__init__(divergences, rhat_max, ess_bulk_min, ess_tail_min, converged)

class mmm_framework.validation.TestResult(test_name, statistic, p_value, passed, threshold, interpretation)[source]

Bases: object

Result of a single statistical test.

test_name: str

statistic: float

p_value: float

passed: bool

threshold: float

interpretation: str

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(test_name, statistic, p_value, passed, threshold, interpretation)

class mmm_framework.validation.PPCCheckResult(check_name, observed_statistic, replicated_mean, replicated_std, p_value, passed, description)[source]

Bases: object

Result of a single posterior predictive check.

check_name: str

observed_statistic: float

replicated_mean: float

replicated_std: float

p_value: float

passed: bool

description: str

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(check_name, observed_statistic, replicated_mean, replicated_std, p_value, passed, description)

class mmm_framework.validation.PPCResults(checks, y_obs, y_rep)[source]

Bases: object

Posterior predictive check results.

checks: list[PPCCheckResult]

y_obs: ndarray

y_rep: ndarray

overall_pass: bool

problematic_checks: list[str]

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(checks, y_obs, y_rep)

class mmm_framework.validation.ResidualDiagnosticsResults(test_results, residuals, fitted_values, acf_values, pacf_values, recommendations=<factory>)[source]

Bases: object

Results from residual diagnostics.

test_results: list[TestResult]

residuals: ndarray

fitted_values: ndarray

acf_values: ndarray

pacf_values: ndarray

overall_adequate: bool

recommendations: list[str]

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(test_results, residuals, fitted_values, acf_values, pacf_values, recommendations=<factory>)

class mmm_framework.validation.ChannelConvergenceResult(channel, rhat, ess_bulk, ess_tail, converged)[source]

Bases: object

Convergence diagnostics for a single channel.

channel: str

rhat: float

ess_bulk: float

ess_tail: float

converged: bool

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(channel, rhat, ess_bulk, ess_tail, converged)

class mmm_framework.validation.ChannelDiagnosticsResults(vif_scores, correlation_matrix, convergence_by_channel, identifiability_issues=<factory>, collinear_clusters=<factory>, condition_number=None, grouped_prior_recommendations=<factory>)[source]

Bases: object

Results from channel diagnostics.

vif_scores: dict[str, float]

correlation_matrix: DataFrame

convergence_by_channel: dict[str, ChannelConvergenceResult]

identifiability_issues: list[str]

collinear_clusters: list[CollinearCluster]

condition_number: float | None = None

grouped_prior_recommendations: list[str]

multicollinearity_warning: bool

convergence_warning: bool

weak_identification_warning: bool

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(vif_scores, correlation_matrix, convergence_by_channel, identifiability_issues=<factory>, collinear_clusters=<factory>, condition_number=None, grouped_prior_recommendations=<factory>)

class mmm_framework.validation.CollinearCluster(channels, max_correlation, explanation)[source]

Bases: object

A cluster of channels that are too collinear to identify separately.

When channels move together (spend is scaled up/down jointly), the data cannot attribute effect between them: their combined effect is identified but the per-channel split is not, regardless of confounding. The cluster’s per-channel ROIs should be read as a group, not individually.

channels: list[str]

max_correlation: float

explanation: str

to_dict()[source]

Return type:: dict[str, Any]

__init__(channels, max_correlation, explanation)

class mmm_framework.validation.GeoIdentificationDiagnostic(n_geos, channels, cv_threshold, caveat)[source]

Bases: object

Whether cross-geo spend variation can support geo-level identification.

n_geos: int

channels: list[GeoSpendVariation]

cv_threshold: float

caveat: str

property weak_channels: list[str]: Channels whose spend is too uniform across geos to inform inference.

property has_identifying_variation: bool

to_dict()[source]

Return type:: dict[str, Any]

__init__(n_geos, channels, cv_threshold, caveat)

class mmm_framework.validation.GeoSpendVariation(channel, cv_across_geos, sufficient)[source]

Bases: object

Cross-geo spend dispersion for a single channel.

channel: str

cv_across_geos: float

sufficient: bool

to_dict()[source]

Return type:: dict[str, Any]

__init__(channel, cv_across_geos, sufficient)

mmm_framework.validation.geo_spend_variation_diagnostic(model, cv_threshold=0.15)[source]

Report per-channel cross-geo spend variation for a fitted/built model.

Return type:: GeoIdentificationDiagnostic

Parameters¶

model: A model exposing has_geo, n_geos, geo_idx (obs -> geo code), X_media_raw (n_obs x n_channels) and channel_names.
cv_threshold: Minimum coefficient of variation of per-geo spend for a channel to be considered to carry geo-level identifying variation.

Raises¶

ValueError: If the model is national (no geo dimension); the diagnostic only applies to multi-geo panels.

class mmm_framework.validation.LOOResults(elpd_loo, se_elpd_loo, p_loo, pareto_k, n_bad_k, pointwise_elpd=None)[source]

Bases: object

Leave-one-out cross-validation results (PSIS-LOO).

elpd_loo: float

se_elpd_loo: float

p_loo: float

pareto_k: ndarray

n_bad_k: int

pointwise_elpd: ndarray | None = None

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(elpd_loo, se_elpd_loo, p_loo, pareto_k, n_bad_k, pointwise_elpd=None)

class mmm_framework.validation.WAICResults(waic, se_waic, p_waic, pointwise=None)[source]

Bases: object

WAIC results.

waic: float

se_waic: float

p_waic: float

pointwise: ndarray | None = None

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(waic, se_waic, p_waic, pointwise=None)

class mmm_framework.validation.ModelComparisonEntry(name, loo=None, waic=None)[source]

Bases: object

Single model entry for comparison.

name: str

loo: LOOResults | None = None

waic: WAICResults | None = None

__init__(name, loo=None, waic=None)

class mmm_framework.validation.ModelComparisonResults(models, loo_comparison=None, waic_comparison=None, stacking_weights=None)[source]

Bases: object

Results from comparing multiple models.

models: list[ModelComparisonEntry]

loo_comparison: DataFrame | None = None

waic_comparison: DataFrame | None = None

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

best_model: str

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(models, loo_comparison=None, waic_comparison=None, stacking_weights=None)

class mmm_framework.validation.CVFoldResult(fold_idx, train_size, test_size, rmse, mae, mape, r2, coverage, test_indices=None, y_true=None, y_pred_mean=None, y_pred_ci_low=None, y_pred_ci_high=None)[source]

Bases: object

Result for a single cross-validation fold.

fold_idx: int

train_size: int

test_size: int

rmse: float

mae: float

mape: float

r2: float

coverage: float

test_indices: ndarray | None = None

y_true: ndarray | None = None

y_pred_mean: ndarray | None = None

y_pred_ci_low: ndarray | None = None

y_pred_ci_high: ndarray | None = None

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(fold_idx, train_size, test_size, rmse, mae, mape, r2, coverage, test_indices=None, y_true=None, y_pred_mean=None, y_pred_ci_low=None, y_pred_ci_high=None)

class mmm_framework.validation.CrossValidationResults(strategy, n_folds, fold_results)[source]

Bases: object

Results from cross-validation.

strategy: str

n_folds: int

fold_results: list[CVFoldResult]

mean_rmse: float

std_rmse: float

mean_mae: float

mean_mape: float

mean_r2: float

mean_coverage: float

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

fold_summary()[source]

Get per-fold summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(strategy, n_folds, fold_results)

class mmm_framework.validation.SensitivityResults(base_estimates, variant_estimates, sensitivity_indices, robust_parameters=<factory>, sensitive_parameters=<factory>)[source]

Bases: object

Results from sensitivity analysis.

base_estimates: dict[str, float]

variant_estimates: dict[str, dict[str, float]]

sensitivity_indices: dict[str, float]

robust_parameters: list[str]

sensitive_parameters: list[str]

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(base_estimates, variant_estimates, sensitivity_indices, robust_parameters=<factory>, sensitive_parameters=<factory>)

class mmm_framework.validation.BootstrapResults(n_bootstrap, parameter_means, parameter_stds, parameter_ci_low, parameter_ci_high)[source]

Bases: object

Results from parametric bootstrap.

n_bootstrap: int

parameter_means: dict[str, float]

parameter_stds: dict[str, float]

parameter_ci_low: dict[str, float]

parameter_ci_high: dict[str, float]

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(n_bootstrap, parameter_means, parameter_stds, parameter_ci_low, parameter_ci_high)

class mmm_framework.validation.InfluenceResults(observation_influence, influential_indices, influence_threshold)[source]

Bases: object

Results from leave-one-out influence analysis.

observation_influence: ndarray

influential_indices: list[int]

influence_threshold: float

summary()[source]

Get summary DataFrame for influential observations.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(observation_influence, influential_indices, influence_threshold)

class mmm_framework.validation.StabilityResults(bootstrap_results=None, influence_results=None, influential_observations=<factory>, stability_score=1.0)[source]

Bases: object

Results from stability analysis.

bootstrap_results: BootstrapResults | None = None

influence_results: InfluenceResults | None = None

influential_observations: list[int]

stability_score: float = 1.0

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(bootstrap_results=None, influence_results=None, influential_observations=<factory>, stability_score=1.0)

class mmm_framework.validation.LiftTestResult(channel, test_period, measured_lift, lift_se, holdout_regions=None, confidence_level=0.95)[source]

Bases: object

External lift test result for calibration.

channel: str

test_period: tuple[str, str]

measured_lift: float

lift_se: float

holdout_regions: list[str] | None = None

confidence_level: float = 0.95

__init__(channel, test_period, measured_lift, lift_se, holdout_regions=None, confidence_level=0.95)

class mmm_framework.validation.LiftTestComparison(channel, model_estimate, model_ci_low, model_ci_high, experimental_estimate, experimental_se, within_ci, relative_error)[source]

Bases: object

Comparison of model estimate to lift test.

channel: str

model_estimate: float

model_ci_low: float

model_ci_high: float

experimental_estimate: float

experimental_se: float

within_ci: bool

relative_error: float

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(channel, model_estimate, model_ci_low, model_ci_high, experimental_estimate, experimental_se, within_ci, relative_error)

class mmm_framework.validation.CalibrationResults(lift_test_comparisons, coverage_rate, mean_absolute_calibration_error)[source]

Bases: object

Results from calibration check.

lift_test_comparisons: list[LiftTestComparison]

coverage_rate: float

mean_absolute_calibration_error: float

calibrated: bool

summary()[source]

Get summary DataFrame.

Return type:: DataFrame

to_dict()[source]

Convert to dictionary.

Return type:: dict[str, Any]

__init__(lift_test_comparisons, coverage_rate, mean_absolute_calibration_error)

class mmm_framework.validation.Validatable(*args, **kwargs)[source]

Bases: Protocol

Protocol for models that can be validated.

All model types (BayesianMMM, NestedMMM, MultivariateMMM, CombinedMMM) should satisfy this protocol to be used with the validation package.

property model: pm.Model: Access the underlying PyMC model.

property channel_names: list[str]: Get list of media channel names.

property n_obs: int: Number of observations in the dataset.

predict(**kwargs)[source]

Generate predictions from the fitted model.

Return type:: PredictionResults

__init__(*args, **kwargs)

class mmm_framework.validation.HasMediaData(*args, **kwargs)[source]

Bases: Protocol

Protocol for models with media spend data.

property X_media: ndarray: Media spend data matrix (n_obs, n_channels).

property y: ndarray: Target variable array.

__init__(*args, **kwargs)

class mmm_framework.validation.HasPanelData(*args, **kwargs)[source]

Bases: Protocol

Protocol for models with panel dataset.

property panel: Any: PanelDataset containing the model data.

__init__(*args, **kwargs)

class mmm_framework.validation.HasScalingParams(*args, **kwargs)[source]

Bases: Protocol

Protocol for models with scaling parameters.

property y_mean: float: Mean of target variable used for standardization.

property y_std: float: Standard deviation of target variable used for standardization.

__init__(*args, **kwargs)

class mmm_framework.validation.HasControlData(*args, **kwargs)[source]

Bases: Protocol

Protocol for models with control variables.

property X_control: ndarray | None: Control variable data matrix.

property control_names: list[str] | None: Control variable names.

__init__(*args, **kwargs)

Backtest¶

Rolling-origin backtesting and forecast-accuracy evaluation for BayesianMMM.

Answers the question every client asks and most MMM documentation dodges: “how accurate were the model’s forecasts on data it had not seen?”

The harness refits the model on an expanding training window, forecasts a fixed horizon past each training cutoff (a genuine out-of-time forecast, not an in-sample re-prediction), and grades the forecasts against the held-out actuals: point accuracy (MAPE, sMAPE, RMSE, MASE), interval calibration (empirical coverage of the central 50/80/95% prediction intervals), and skill relative to naive baselines (last-value and seasonal-naive).

Out-of-time prediction cannot go through BayesianMMM.predict: the model graph bakes the seasonality lookup and trend scale over the training periods, so future time indices are out of range. Instead PosteriorForecaster replays the model’s forward pass in NumPy from the posterior draws – the same approach as the cross-validation path in mmm_framework.validation.validator, generalized to both adstock parameterizations (the legacy fixed-alpha blend and the parametric geometric/delayed/Weibull kernels) and all configured saturation types. Adstock is convolved over the full spend history so carryover from the training period flows into the forecast window correctly.

Scope (enforced with explicit errors, not silent wrong answers).

The forward pass reproduces the fitted mean exactly for the terms it supports, and ForecastUnsupportedError refuses the rest at construction time, so a caller cannot obtain a forecaster it is not allowed to trust:

supported — national and geo/product panels, every trend family, seasonality, parametric and legacy adstock, all saturation families, controls, per-geo media coefficients (vary_media_by_geo), and the geo/product level offsets;
refused — price and promotion levers, event/holiday effects, cross-channel interactions, reach/frequency channels, time-varying coefficients, and the multiplicative specification. Each of these contributes to the fitted mean and the forward pass cannot replay it, so dropping it would shift the forecast level while leaving the interval width untouched;
trend extrapolation is a stated policy, not a silent default — read forecaster.trend_extrapolation. LINEAR extrapolates in closed form; spline/GP/piecewise hold the last fitted level flat, so their intervals do not widen with horizon;
experiment-calibration likelihood terms are dropped by the backtest refit (their estimands reference the full-period spend), so run_backtest() refuses a calibrated model rather than reporting an uncalibrated model’s accuracy under the calibrated model’s name.

Before v1.3.1 the forward pass summed five of the fitted mean’s ten terms and raised nothing, and the refit downcast any garden/custom model class to a plain additive BayesianMMM. See the CHANGELOG entry for v1.3.1.

A good backtest validates the predictive model, not the causal one: a model can forecast well while attributing wrongly (and vice versa). Use this alongside – never instead of – the pressure-testing and calibration machinery.

Examples¶

>>> from mmm_framework.validation import BacktestConfig, run_backtest
>>> config = BacktestConfig(min_train_size=104, horizon=13, step=13)
>>> result = run_backtest(mmm, config)          # mmm: an (unfitted) BayesianMMM
>>> result.summary()                            # model vs naive baselines
>>> result.by_horizon()                         # accuracy decay with lead time
>>> result.coverage()                           # interval calibration

class mmm_framework.validation.backtest.BacktestConfig(min_train_size=104, horizon=13, step=None, max_origins=None, coverage_levels=(0.5, 0.8, 0.95), draws=500, tune=500, chains=4, include_noise=True, season_period=None, random_seed=42)[source]¶

Bases: object

Configuration for a rolling-origin backtest.

Attributes¶

min_train_sizeint: Periods in the first training window. Two seasonal cycles of weekly data (104) is a sensible floor; below ~78 the seasonality and trend posteriors are barely informed and forecast intervals balloon.
horizonint: Forecast length (periods) past each training cutoff.
stepint or None: Spacing between consecutive training cutoffs. None uses horizon (non-overlapping forecast windows).
max_originsint or None: Cap on the number of refits (None = as many as the data allows).
coverage_levelstuple[float, …]: Central prediction-interval levels to record and grade.
draws, tune, chainsint: MCMC budget per refit. Backtests refit the model once per origin, so these default lower than a production fit; convergence is still recorded per origin in BacktestResult.fits.
include_noisebool: Forecast the observable (mean + observation noise) rather than the latent mean. Keep True for honest interval coverage.
season_periodint or None: Seasonal lag for the seasonal-naive baseline and the MASE scale. None derives it from the data frequency (weekly -> 52).
random_seedint: Base seed; each origin offsets it deterministically.

min_train_size: int = 104¶

horizon: int = 13¶

step: int | None = None¶

max_origins: int | None = None¶

coverage_levels: tuple[float, ...] = (0.5, 0.8, 0.95)¶

draws: int = 500¶

tune: int = 500¶

chains: int = 4¶

include_noise: bool = True¶

season_period: int | None = None¶

random_seed: int = 42¶

class mmm_framework.validation.backtest.BacktestResult(config, records, fits, season_period, mase_scales=<factory>)[source]¶

Bases: object

Rolling-origin backtest records plus summary views.

Attributes¶

recordspd.DataFrame: One row per (origin, horizon) forecast: origin, position, date, horizon, y_true, y_pred (posterior-mean), pred_naive, pred_snaive, per-level interval bounds (lo_80/hi_80, …) and coverage flags (cov_80, …).
fitspd.DataFrame: One row per refit origin: train size, wall-clock seconds, R-hat max, divergences. Check this before trusting the records.

config: BacktestConfig¶

records: DataFrame¶

fits: DataFrame¶

season_period: int¶

mase_scales: dict[int, float]¶

property n_origins: int¶

property mape: float¶: Headline out-of-time MAPE of the MMM over all records (the number the docs quote). Convenience for summary().loc["mmm", "mape"].

summary()[source]¶

Headline accuracy: the model vs naive baselines, all records.

Return type:: DataFrame

by_horizon()[source]¶

Accuracy and coverage as a function of forecast lead time.

Return type:: DataFrame

by_origin()[source]¶

Accuracy per refit origin (forecast-window heterogeneity).

Return type:: DataFrame

coverage()[source]¶

Interval calibration: nominal vs empirical coverage + sharpness.

Return type:: DataFrame

__init__(config, records, fits, season_period, mase_scales=<factory>)¶

exception mmm_framework.validation.backtest.ForecastUnsupportedError(feature, reason, all_unsupported=None)[source]¶

Bases: NotImplementedError

The fitted model carries a term the forward pass cannot replay.

__init__(feature, reason, all_unsupported=None)[source]¶

class mmm_framework.validation.backtest.PosteriorForecaster(model, *, strict=True)[source]¶

Bases: object

Out-of-time forecasts from a fitted BayesianMMM’s posterior draws.

Parameters¶

modelBayesianMMM: A fitted model, possibly trained on a prefix of the full series.
strictbool, default True: When True, an unreplayable term raises ForecastUnsupportedError. False downgrades the refusal to a warning and records it on self.unsupported — for diagnostic use on a model you already know is incomplete. Never use it for planning: the resulting forecast omits a term the model estimated.

Attributes¶

trend_extrapolationTrendExtrapolation: The stated policy for continuing the trend past the training window.
unsupportedlist[tuple[str, str]]: Empty unless strict=False suppressed a refusal.

Raises¶

ForecastUnsupportedError: The model carries a term the forward pass cannot replay.

__init__(model, *, strict=True)[source]¶

property n_samples: int¶

forecast(X_media_full_raw, X_controls_full_raw, positions, *, include_noise=True, random_seed=None, train_offset=0)[source]¶

Posterior predictive draws at absolute period positions.

Return type:: ndarray

Parameters¶

X_media_full_rawnp.ndarray: Raw media, shape (n_full, n_channels) – the FULL history (training + forecast periods) so adstock carryover is correct.
X_controls_full_rawnp.ndarray or None: Raw controls over the full history (required if the model has controls; future control values are assumed known/planned).
positionsnp.ndarray: Absolute period positions (0-based on the full axis) to forecast.
train_offsetint: Absolute position of the trained model’s first period. 0 for prefix training (the backtest); the window start for rolling-window clones (validator cross-validation).

Returns¶

np.ndarray: Samples in original KPI scale, shape (n_samples, len(positions)).

class mmm_framework.validation.backtest.TrendExtrapolation(policy, trend_type, n_train_periods)[source]¶

Bases: object

How the forecaster continues the trend past the training window.

Recorded rather than assumed, because the three policies carry very different forecast semantics and only one of them is model-defined.

Attributes¶

policy{‘none’, ‘linear’, ‘held_flat’}: held_flat is a heuristic: a spline/GP/piecewise basis has no out-of-time forecast, so the last fitted level is carried forward and the interval consequently does not widen with horizon.
trend_typestr: The configured trend family.
n_train_periodsint: Training length. Load-bearing for linear: the fitted slope is on a t_scaled = pos / (n_train - 1) axis, so the same posterior implies a different slope per period depending on how long the training panel was. Reporting it makes that visible rather than surprising.

policy: str¶

trend_type: str¶

n_train_periods: int¶

property is_model_defined: bool¶: False when the continuation is a heuristic rather than the model’s.

describe()[source]¶

Return type:: str

__init__(policy, trend_type, n_train_periods)¶

mmm_framework.validation.backtest.rebuild_like(model, panel, **overrides)[source]¶

A fresh, unfitted model of type(model) on panel.

Return type:: Any

Raises¶

ForecastUnsupportedError: The class cannot be rebuilt from (panel, model_config, trend_config, adstock_alphas, model_params) — e.g. a model declaring REQUIRED_DATASET_CAPABILITIES the sliced panel no longer satisfies. Refusing beats falling back to a different model class.

mmm_framework.validation.backtest.rolling_origins(n_periods, *, min_train_size, horizon, step=None, max_origins=None)[source]¶

Training cutoffs for a rolling-origin (expanding-window) backtest.

Each returned cutoff T means: train on periods [0, T), forecast periods [T, min(T + horizon, n_periods)). Only full forecast windows are emitted so every origin is graded on the same horizons.

Return type:: list[int]

mmm_framework.validation.backtest.run_backtest(model, config=None, **fit_kwargs)[source]¶

Rolling-origin backtest of a BayesianMMM specification.

Return type:: BacktestResult

Parameters¶

modelBayesianMMM: The full-data model (does not need to be fitted); supplies the panel, configuration, and the raw media/control history.
configBacktestConfig, optional: Backtest settings; defaults to BacktestConfig().
**fit_kwargs: Extra arguments forwarded to each refit’s fit() (e.g. progressbar=False).

Returns¶

BacktestResult

Spec Curve¶

Spec-curve / Bayesian model-averaging over a pre-registered spec set (#103).

There are dozens of defensible MMM specifications — adstock form, saturation form, control set, pooling scheme — and each can move a channel’s ROI. Landing on the single spec that gives the “nice” answer is the garden of forking paths. The honest alternative is to declare a set of defensible specs before seeing results, fit them all, and report how each channel’s ROI moves across the whole set. Robustness across specs is itself a headline; fragility across specs is a warning the single-spec number hides.

This module provides:

SpecVariant / SpecSet — a serializable, pre-registerable declaration of the spec set (store it in the design-readout / assumption log before the fit).
apply_variant() — deep-merge a variant onto a base spec (the same spec dict mmm_framework.agents.fitting.build_and_fit() consumes).
default_spec_variants() — the standard defensible grid (adstock × saturation forms, optionally pooling / prior mode) when the caller does not hand-pick one.
run_spec_curve() — fit/collect every spec’s per-channel ROI posterior and produce a model-averaged (BMA) estimate with propagated uncertainty.
SpecCurveResult — the per-spec ROI table, the applied weights, the BMA estimate, and a per-channel robustness summary; to_dict() feeds the report robustness section.

Weighting: predictive weights are not causal weights¶

The BMA step defaults to equal weights over the pre-registered set, not to LOO-stacking weights, and the distinction is methodological rather than cosmetic.

Stacking (Yao et al. 2018) chooses the mixture that maximizes expected predictive utility — it answers “which combination forecasts held-out y best?”. A spec curve averages a causal estimand (per-channel ROI), and the two objectives come apart precisely where MMM specs differ. Two specs can predict the KPI equally well while splitting that same fitted mean very differently between media and baseline; predictive skill is close to uninformative about which split is right, because the decomposition is the free parameter. Worse, the direction can invert: a spec that overfits the confounder block often predicts better while being less trustworthy for the causal contrast, so stacking can systematically upweight the specs a causal analyst should trust least.

Equal weights encode what pre-registration already asserted — every variant in the set was declared defensible before the fit, so there is no post-hoc predictive reason to promote one. Set weighting="stacking" to opt back in (it warns), which is defensible only when every spec in the set is identified for the same estimand and you genuinely want a predictive mixture. Stacking weights are still computed and reported as predictive_weights whenever compute_loo is on: divergence between them and equal weights is a useful diagnostic (it says predictive fit discriminates between your specs), it is just not a reason to follow them.

The ROI is computed with the framework’s canonical semantics (per-channel contribution draws over the full window ÷ the measurement-aware divisor), so a spec-curve number is directly comparable to the contribution_roi estimand. Fitting and ROI extraction are injectable (fit_fn / roi_fn) so the engine is unit-testable without MCMC.

class mmm_framework.validation.spec_curve.SpecVariant(**data)[source]¶

Bases: BaseModel

One defensible specification, expressed as overrides on a base spec.

The structured axes cover the four the review calls out (adstock form, saturation form, control set, pooling); overrides is an escape hatch for any other spec key (deep-merged last).

name: str¶

adstock: str | None¶: Adstock family applied to every media channel (“geometric”/”weibull”/”delayed”).

saturation: str | None¶: Saturation family applied to every media channel (“hill”/”logistic”/”michaelis_menten”/”tanh”).

controls: list[str] | None¶: Full replacement control set (channel-name list). None keeps the base’s.

kpi_level: str | None¶: Pooling scheme (“national” / “geo”).

media_prior_mode: str | None¶: Default media-prior parameterization (“roi” / “coefficient”).

trend: str | None¶: Trend family (“none”/”linear”/”piecewise”/”spline”/”gaussian_process”).

seasonality: dict[str, Any] | None¶: Seasonality override, e.g. {"yearly": 4} (or {} to disable).

overrides: dict[str, Any]¶: Arbitrary deep-merge overrides (applied last).

primary: bool¶: Whether this is the pre-registered PRIMARY specification.

description: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class mmm_framework.validation.spec_curve.SpecSet(**data)[source]¶

Bases: BaseModel

A pre-registered set of defensible specs, declared before results.

variants: list[SpecVariant]¶

rationale: str¶

registered_at: str | None¶: ISO timestamp of pre-registration (set by the caller; kept as data, not computed here so the module stays deterministic / clock-free).

property primary_variant: SpecVariant | None¶

names()[source]¶

Return type:: list[str]

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class mmm_framework.validation.spec_curve.SpecFit(name, primary, roi=<factory>, roi_draws=<factory>, loo=None, weight=0.0, error=None)[source]¶

Bases: object

One spec’s realized ROI + model-fit summary.

name: str¶

primary: bool¶

roi: dict[str, dict[str, float]]¶

roi_draws: dict[str, ndarray]¶

loo: dict[str, float] | None = None¶

weight: float = 0.0¶

error: str | None = None¶

__init__(name, primary, roi=<factory>, roi_draws=<factory>, loo=None, weight=0.0, error=None)¶

class mmm_framework.validation.spec_curve.SpecCurveResult(channels, specs, primary, hdi_prob, fits, weights, bma, robustness, weighting='equal', predictive_weights=<factory>)[source]¶

Bases: object

Spec-curve outcome: per-spec ROI, applied weights, BMA, robustness.

channels: list[str]¶

specs: list[str]¶

primary: str | None¶

hdi_prob: float¶

fits: list[SpecFit]¶

weights: dict[str, float]¶

bma: dict[str, dict[str, float]]¶

robustness: dict[str, dict[str, Any]]¶

weighting: str = 'equal'¶: Which weighting formed bma — "equal" (default) or "stacking".

predictive_weights: dict[str, float]¶: LOO-stacking weights, reported for diagnosis even when not applied. Empty when compute_loo=False or arviz could not run.

to_dict()[source]¶

JSON-friendly payload for the report bundle / agent tool (no draws).

Return type:: dict[str, Any]

__init__(channels, specs, primary, hdi_prob, fits, weights, bma, robustness, weighting='equal', predictive_weights=<factory>)¶

mmm_framework.validation.spec_curve.apply_variant(base_spec, variant)[source]¶

Return a new spec = base_spec with variant applied.

Return type:: dict

mmm_framework.validation.spec_curve.default_spec_variants(base_spec, *, adstock_forms=('geometric', 'weibull'), saturation_forms=('hill', 'logistic'), include_prior_mode=False)[source]¶

The standard defensible spec grid: adstock × saturation forms.

Return type:: list[SpecVariant]

mmm_framework.validation.spec_curve.run_spec_curve(base_spec, dataset_path, *, variants=None, hdi_prob=0.94, max_draws=400, random_seed=42, compute_loo=True, weighting='equal', fit_fn=None, roi_fn=None)[source]¶

Fit a spec set, collect per-channel ROI, and model-average across specs.

Return type:: SpecCurveResult

Parameters¶

base_spec, dataset_path:: The base spec and the dataset file each variant is fit against.
variants:: A SpecSet, a list of SpecVariant, or None (uses default_spec_variants()).
hdi_prob, max_draws, random_seed:: Credible-interval mass, ROI-draw thinning, and seed.
compute_loo:: Compute per-spec LOO and the diagnostic LOO-stacking weights. These are reported either way; whether they are applied is weighting.
weighting:: "equal" (default) weights every non-failing spec equally — the pre-registration-consistent choice for averaging a causal estimand. "stacking" applies LOO-stacking weights instead and emits a warning: stacking maximizes expected predictive utility, which is a different objective from validity for ROI. See the module docstring.
fit_fn, roi_fn:: Injection points for testing. fit_fn(spec, dataset_path) -> model (default: build + fit, no serialization); roi_fn(model, channels, max_draws=, random_seed=) -> {channel: draws} (default: channel_roi_draws()).

mmm_framework.validation.spec_curve.channel_roi_draws(model, channels, *, max_draws=400, random_seed=42)[source]¶

Per-channel ROI posterior draws over the full window.

Uses model.sample_channel_contributions (KPI-unit per-draw contributions) summed over observations, divided by the measurement-aware divisor (resolve_channel_divisor — spend $ or efficiency volume) — the same numerator/denominator as the contribution_roi estimand. Returns {channel: (n_draws,) ndarray}; a channel with a non-positive divisor is omitted.

Return type:: dict[str, ndarray]

mmm_framework.validation.spec_curve.WEIGHTING_CAVEAT = {'equal': 'Model-averaged with equal weights over the pre-registered spec set. Every variant was declared defensible before the fit, so none is promoted on post-hoc grounds. Predictive (LOO-stacking) weights are reported alongside for diagnosis but deliberately not applied: stacking optimizes held-out prediction, and two specs can predict the KPI equally well while splitting it very differently between media and baseline — which is the quantity being averaged.', 'stacking': 'Model-averaged with LOO-stacking weights, which optimize expected PREDICTIVE utility, not validity for a causal estimand. Two specs can predict the KPI equally well while disagreeing sharply on the media/baseline split, and a spec that overfits the confounder block can predict better while being less trustworthy for ROI. Treat this mixture as a forecast-weighted summary, not a causal estimate.'}¶: Surfaced next to any model-averaged ROI so a reader knows what the mixture weights mean. See the module docstring for the full argument.

Calibration¶

Inference-calibration verification: LOO-PIT (predictive) + SBC (structural).

The framework’s headline claim is genuine uncertainty quantification — that the posterior intervals it reports have nominal coverage. This module machine-verifies that claim instead of merely asserting it:

loo_pit_check() — leave-one-out probability-integral-transform. From an existing fit’s posterior-predictive draws, the LOO-PIT values should be Uniform(0, 1) if the predictive distribution is calibrated. Cheap.
simulation_based_calibration() — SBC (Talts et al. 2018). Draw “true” parameters from the prior, simulate data, fit, and record the rank of each true value within its posterior draws. Under a correctly-implemented inference engine those ranks are uniform. This validates the inference engine itself, on data generated from the model’s own prior — the strongest calibration check. It is EXPENSIVE (one fit per simulation), so it is a verification tool, not a per-fit step. The harness is model-agnostic (callbacks for prior/simulate/fit), so it can be unit-tested on a fast conjugate model and applied to the MMM offline.

class mmm_framework.validation.calibration.LooPitResult(pit, ks_stat, ks_pvalue, calibrated, n)[source]¶

Bases: object

pit: ndarray¶

ks_stat: float¶

ks_pvalue: float¶

calibrated: bool¶

n: int¶

summary()[source]¶

Return type:: str

__init__(pit, ks_stat, ks_pvalue, calibrated, n)¶

mmm_framework.validation.calibration.loo_pit_check(*, y, y_hat, log_weights=None, alpha=0.05)[source]¶

LOO-PIT calibration check from observed values + posterior-predictive draws.

Return type:: LooPitResult

Parameters¶

yarray (n_obs,): Observed outcomes.
y_hatarray (n_obs, n_samples) or (n_samples, n_obs): Posterior-predictive draws (orientation auto-detected).
log_weightsarray, optional: PSIS log-weights aligned to y_hat. If omitted, uniform weights are used (this reduces LOO-PIT to ordinary PIT — still a useful check).
alphafloat: Significance level for the KS uniformity test. calibrated is ks_pvalue > alpha.

class mmm_framework.validation.calibration.SBCResult(param_names, n_sims, normalized_ranks=<factory>, ks_pvalue=<factory>, calibrated=<factory>)[source]¶

Bases: object

param_names: list[str]¶

n_sims: int¶

normalized_ranks: dict[str, ndarray]¶

ks_pvalue: dict[str, float]¶

calibrated: dict[str, bool]¶

property all_calibrated: bool¶

summary()[source]¶

Return type:: str

__init__(param_names, n_sims, normalized_ranks=<factory>, ks_pvalue=<factory>, calibrated=<factory>)¶

mmm_framework.validation.calibration.simulation_based_calibration(sample_prior, simulate, fit, *, n_sims=100, seed=0, alpha=0.05)[source]¶

Simulation-based calibration (Talts et al. 2018), model-agnostic.

Callbacks (each receives the shared Generator for reproducibility): :rtype: SBCResult

sample_prior(rng) -> {param: scalar} — one draw of the true parameters.
simulate(theta, rng) -> data — synthetic data given the true params.
fit(data, rng) -> {param: 1-D posterior draws} — keys must match sample_prior.

Scalar parameters only (the common SBC case). Returns an SBCResult.

Channel Diagnostics¶

Channel-level diagnostics for model validation.

Provides multicollinearity detection (VIF), per-channel convergence checks, and identifiability analysis.

class mmm_framework.validation.channel_diagnostics.VIFCalculator[source]¶

Bases: object

Variance Inflation Factor calculator.

VIF measures multicollinearity between media channels. VIF > 10 indicates problematic multicollinearity.

compute(X_media, channel_names)[source]¶

Compute VIF for each media channel.

Return type:: dict[str, float]

Parameters¶

X_medianp.ndarray: Media spend matrix, shape (n_obs, n_channels).
channel_nameslist[str]: Names of media channels.

Returns¶

dict[str, float]: VIF score for each channel.

class mmm_framework.validation.channel_diagnostics.ChannelConvergenceChecker[source]¶

Bases: object

Per-channel MCMC convergence diagnostics.

Checks R-hat and ESS for channel-specific parameters.

check(trace, channel_names, config)[source]¶

Check convergence for each channel.

Return type:: dict[str, ChannelConvergenceResult]

Parameters¶

traceaz.InferenceData: ArviZ InferenceData with posterior samples.
channel_nameslist[str]: Names of media channels.
configChannelDiagnosticsConfig: Configuration with thresholds.

Returns¶

dict[str, ChannelConvergenceResult]: Convergence result for each channel.

class mmm_framework.validation.channel_diagnostics.ChannelDiagnostics(model, config=None)[source]¶

Bases: object

Comprehensive channel-level diagnostics.

Combines VIF analysis, correlation matrix, and per-channel convergence.

Examples¶

>>> diagnostics = ChannelDiagnostics(model)
>>> results = diagnostics.run_all()
>>> print(results.summary())

__init__(model, config=None)[source]¶

Initialize channel diagnostics.

Parameters¶

modelAny: Fitted model with media data and trace.
configChannelDiagnosticsConfig, optional: Configuration for diagnostics.

run_all()[source]¶

Run all channel diagnostics.

Return type:: ChannelDiagnosticsResults

Returns¶

ChannelDiagnosticsResults: Comprehensive channel diagnostics results.

vif_analysis()[source]¶

Get VIF analysis as DataFrame.

Return type:: DataFrame

Returns¶

pd.DataFrame: VIF scores with interpretation.

correlation_matrix()[source]¶

Get media channel correlation matrix.

Return type:: DataFrame

Returns¶

pd.DataFrame: Correlation matrix.

mmm_framework.validation.channel_diagnostics.detect_collinear_clusters(correlation_matrix, names, threshold)[source]¶

Group variables into clusters that cannot be separately identified.

Two variables are linked when their absolute correlation exceeds threshold; clusters are the connected components of that graph. A cluster of size >= 2 is weakly identified: the data sees their combined movement, not the per-variable split.

Return type:: list[CollinearCluster]