"""Model-level configuration: hierarchy, seasonality, control selection, and ModelConfig."""
from __future__ import annotations
import warnings
from typing import Literal
from pydantic import BaseModel, Field, model_validator
from .enums import FitMethod, InferenceMethod, ModelSpecification, PriorType
from .events import EventsConfig
from .interactions import ChannelInteraction
from .levers import PriceConfig, PromoConfig
from .reach_frequency import ReachFrequencyConfig
from .likelihood import LikelihoodConfig
from .priors import PriorConfig
[docs]
class HierarchicalConfig(BaseModel):
"""Configuration for hierarchical/panel model structure."""
enabled: bool = True
# Pooling dimensions
pool_across_geo: bool = True
pool_across_product: bool = True
# Per-geo CHANNEL COEFFICIENTS (effectiveness), not just intercept offsets.
# Off by default (the model keeps one shared beta per channel + per-geo
# intercepts). When on (with geo data), each channel's beta is partial-pooled
# across geos so geo-heterogeneous effectiveness is estimable — the pooled
# model otherwise lands near a spend-weighted average and scrambles regional
# ROI. See model/base._build_channel_betas_geo (V3).
vary_media_by_geo: bool = False
# Prior SD of per-geo log-effectiveness deviations (between-geo spread).
media_geo_sigma: float = 0.3
# Parameterization
use_non_centered: bool = True # Better for sparse groups
non_centered_threshold: int = 20 # Min obs for centered
# Hyperprior settings
mu_prior: PriorConfig = Field(
default_factory=lambda: PriorConfig(
distribution=PriorType.NORMAL, params={"mu": 0, "sigma": 1}
)
)
sigma_prior: PriorConfig = Field(
default_factory=lambda: PriorConfig.half_normal(sigma=0.5)
)
model_config = {"extra": "forbid"}
[docs]
class SeasonalityConfig(BaseModel):
"""Configuration for seasonality components.
Amplitude scale note: the Fourier coefficients get ``Normal(0, prior_sigma)``
priors on **standardized** ``y``, so ``prior_sigma`` bounds how much of a
standard deviation each harmonic may swing the KPI — it is the seasonal
amplitude prior. The historic default (0.3) suits mild seasonality; raise it
(e.g. 0.5–1.0) for strongly seasonal categories, or the seasonal signal gets
squeezed into trend/media. Per-component overrides win over ``prior_sigma``.
"""
yearly: int | None = 2 # Fourier order for yearly seasonality
monthly: int | None = None
weekly: int | None = None
# Amplitude prior (sigma of the Normal prior on Fourier coefficients)
prior_sigma: float = 0.3
yearly_prior_sigma: float | None = None
monthly_prior_sigma: float | None = None
weekly_prior_sigma: float | None = None
model_config = {"extra": "forbid"}
[docs]
def prior_sigma_for(self, component: str) -> float:
"""Amplitude prior sigma for ``component`` ('yearly'/'monthly'/'weekly'),
falling back to the shared ``prior_sigma``. getattr defaults keep
configs pickled before these fields existed loadable."""
override = getattr(self, f"{component}_prior_sigma", None)
return getattr(self, "prior_sigma", 0.3) if override is None else override
[docs]
class ControlSelectionConfig(BaseModel):
"""Configuration for control variable selection."""
method: Literal["none", "horseshoe", "spike_slab", "lasso"] = "none"
# Horseshoe settings
expected_nonzero: int = 3
# Regularization strength (for lasso-like)
regularization: float = 1.0
model_config = {"extra": "forbid"}
#: Which ``pm.sample(nuts_sampler=...)`` backend each Bayesian inference
#: method selects. Kept here rather than on the enum so the enum stays a plain
#: serialization contract (its VALUES are frozen — see tests/test_api_contracts.py).
_NUTS_SAMPLER_BY_METHOD = {
InferenceMethod.BAYESIAN_PYMC: "pymc",
InferenceMethod.BAYESIAN_NUMPYRO: "numpyro",
InferenceMethod.BAYESIAN_NUTPIE: "nutpie",
}
#: Which frequentist estimator each non-Bayesian inference method selects.
#: ``"ridge"`` is the closed-form penalized solve; ``"constrained"`` is the
#: convex program, which needs the optional ``[frequentist]`` extra (``cvxpy``).
_FREQUENTIST_ESTIMATOR_BY_METHOD = {
InferenceMethod.FREQUENTIST_RIDGE: "ridge",
InferenceMethod.FREQUENTIST_CVXPY: "constrained",
}
#: Inference methods that are declared but have NO estimator behind them.
#: **Empty since #188** — both frequentist methods now dispatch to
#: :mod:`mmm_framework.frequentist`. Kept (rather than deleted along with the
#: machinery) because it is the mechanism for declaring a future enum value
#: before its estimator lands, which is what stopped ``frequentist_ridge`` from
#: silently fitting NUTS between 1.2.0 and now.
_UNIMPLEMENTED_METHODS: frozenset[InferenceMethod] = frozenset()
_FREQUENTIST_TRACKING_URL = "https://github.com/redam94/mmm-framework/issues/180"
def unimplemented_inference_message(method: InferenceMethod) -> str:
"""The single message every surface uses to refuse an unimplemented method.
Names the supported alternative rather than only refusing. Currently
unreachable — :data:`_UNIMPLEMENTED_METHODS` is empty — and kept so the next
declared-before-implemented method has somewhere to refuse from.
"""
return (
f"Inference method {method.value!r} is not implemented: the enum value "
"exists but no estimator backs it, so this config cannot be fitted. "
"For a fast penalized point estimate use fit(method='map'). For full "
"inference select a Bayesian method, e.g. "
"ModelConfigBuilder().bayesian_pymc(). "
f"Tracking: {_FREQUENTIST_TRACKING_URL}"
)
[docs]
class ModelConfig(BaseModel):
"""Complete model configuration."""
# Functional form
specification: ModelSpecification = ModelSpecification.ADDITIVE
# Observation model (likelihood family + link + family params). Default
# normal/identity reproduces the historical hard-coded pm.Normal likelihood
# byte-for-byte. The built-in additive model fits only the Gaussian families
# (normal/student_t) directly; non-Gaussian families (e.g. binomial for an
# awareness model) are read by models that define their own observation block.
likelihood: LikelihoodConfig = Field(default_factory=LikelihoodConfig)
# Intercept prior: Normal(mu, sigma) on standardized y, so mu is measured in
# KPI standard deviations from the mean (values beyond ±2 are extreme).
intercept_prior_mu: float = 0.0
intercept_prior_sigma: float = Field(default=0.5, gt=0)
# DEFAULT media-effect prior parameterization (applies only to channels with
# no experiment-calibrated ``roi_prior`` and no explicit
# ``coefficient_prior``):
#
# * ``"coefficient"`` (default) — the historical ``beta_<ch> ~
# Gamma(mu=1.5, sigma=1)`` on the standardized-coefficient scale. Its
# *implied* prior ROI depends on each channel's spend and the KPI scale,
# so two channels get very different (and arbitrary) prior ROIs.
# * ``"roi"`` — sample the channel's prior ROI directly
# (``roi_<ch> ~ LogNormal(media_roi_prior_mu, media_roi_prior_sigma)``,
# default median 1.0 = break-even) and derive ``beta_<ch>`` in-graph as
# ``roi * spend_total / (y_std * Σ saturation)`` — the prior lives on the
# decision scale and is comparable across channels regardless of
# spend/KPI units. Channels whose ROI divisor is non-monetary
# (efficiency basis) or zero-spend fall back to the coefficient default.
# The agent's spec-built models default to this mode (agents/fitting).
media_prior_mode: Literal["coefficient", "roi"] = "coefficient"
# LogNormal hyper-params of the ROI-mode prior (mu is log-scale: 0 → median
# ROI 1.0; sigma 1.0 → 90% prior interval ≈ [0.19, 5.2]).
media_roi_prior_mu: float = 0.0
media_roi_prior_sigma: float = Field(default=1.0, gt=0)
# DF-2: partial-pool the coefficients of channels sharing a `parent_channel`
# group toward a shared (log-normal) group mean, so genuinely EXCHANGEABLE
# collinear channels (e.g. several social platforms) borrow strength and
# their split is regularized. OFF BY DEFAULT (R0.1): the media block is
# byte-identical when disabled. Channels with a calibrated `roi_prior` or an
# explicit `coefficient_prior` are excluded from the pool (their prior wins).
# Collinearity ≠ "should be pooled" — only pool a priori exchangeable
# channels; see technical-docs/future_implementation.md (DF-2).
use_grouped_media_priors: bool = False
# Inference settings
inference_method: InferenceMethod = InferenceMethod.BAYESIAN_NUMPYRO
# MCMC settings (for Bayesian)
n_chains: int = 4
n_draws: int = 1000
n_tune: int = 1000
target_accept: float = 0.9
# Default fit method when ``fit()`` is called without an explicit ``method``.
# NUTS (full MCMC) is the default; the approximate methods (MAP / ADVI /
# full-rank ADVI / Pathfinder) trade calibrated uncertainty for speed and
# are meant for fast model checks, not final inference.
#
# `None` after a FREQUENTIST fit, and deliberately so: `FitMethod` has no
# frequentist member, this field selects among the *Bayesian* estimators,
# and a stray "nuts" left here is what makes downstream surfaces (the
# interactive report's inference card, `saved_model_settings`) announce a
# full MCMC posterior for a ridge fit. Read `inference_method` /
# `is_frequentist` to learn the paradigm.
fit_method: FitMethod | None = FitMethod.NUTS
# Hierarchical structure
hierarchical: HierarchicalConfig = Field(default_factory=HierarchicalConfig)
# Seasonality
seasonality: SeasonalityConfig = Field(default_factory=SeasonalityConfig)
# Holiday / event effects (#143). None ⇒ no event block (default). When set,
# sharp date-specific effects enter as an additive `event_component`, distinct
# from the smooth Fourier seasonality.
events: EventsConfig | None = None
# Cross-channel synergy / interaction terms (#142). Empty ⇒ strictly additive
# (default). Each entry adds a `beta_ij * sat_i * sat_j` term so a channel pair
# can do more (synergy) or less (cannibalization) than the sum of its parts.
channel_interactions: list[ChannelInteraction] = Field(default_factory=list)
# Price & promotion levers (#138). None / empty ⇒ these enter (if present) as
# ordinary linear controls. When set, the named control column is promoted to
# a first-class lever (log-price elasticity / promo lift with carryover) and
# removed from the linear control block.
price: PriceConfig | None = None
promotions: list[PromoConfig] = Field(default_factory=list)
# Reach & frequency channels (#141). Empty ⇒ channels are ordinary volume
# channels (byte-identical). When set, the named channel's column is treated
# as reach and modulated by a frequency-saturation curve built from a
# frequency column pulled out of the control block.
reach_frequency: list[ReachFrequencyConfig] = Field(default_factory=list)
# Control selection
control_selection: ControlSelectionConfig = Field(
default_factory=ControlSelectionConfig
)
# Adstock estimation strategy.
# True (default since the 0.1.0 development line, 2026-06): estimate a
# continuous adstock kernel in-graph per channel, honoring each
# MediaChannelConfig.adstock (type/l_max/normalize and priors), which
# enables geometric, delayed, and Weibull carryover shapes. Made the
# default after the pressure-testing series measured the legacy blend at
# ~28% attribution error vs ~7% parametric on carryover-sensitive worlds.
# False (legacy): fast two-point interpolation between fixed low/high
# geometric adstock. Set explicitly to reproduce pre-change fits;
# models pickled before this change keep their original behavior.
use_parametric_adstock: bool = True
# Frequentist settings. INERT — no code reads these; they are the reserved
# config surface for the frequentist path (#180). LIVE since #188 — read by
# `model/base.py::run_frequentist_fit` when `inference_method` is one of the
# frequentist values, and ignored entirely by the Bayesian methods.
#
# `ridge_alpha` is the L2 strength on the STANDARDIZED design, so it means
# the same thing across channels and datasets. It is only a fallback: the
# transform search selects the penalty by the same out-of-sample criterion it
# uses for adstock and saturation, and an in-sample rule (AIC, GCV) would
# reintroduce exactly the bias that criterion exists to avoid.
ridge_alpha: float = 1.0
bootstrap_samples: int = 1000
# Optimization settings (for transformation search). `optim_maxiter` is the
# number of (adstock, saturation) candidate points evaluated; `optim_seed`
# is the random-seed fallback for every method, frequentist included.
optim_maxiter: int = 500
optim_seed: int | None = 42
model_config = {"extra": "forbid"}
@property
def is_bayesian(self) -> bool:
return self.inference_method in [
InferenceMethod.BAYESIAN_PYMC,
InferenceMethod.BAYESIAN_NUMPYRO,
InferenceMethod.BAYESIAN_NUTPIE,
]
@property
def is_frequentist(self) -> bool:
"""True for the ridge / constrained estimation path (epic #180).
The complement of :attr:`is_bayesian` over the *implemented* methods, but
written as its own membership test rather than ``not is_bayesian`` so a
future third paradigm cannot silently inherit the frequentist branch.
"""
return self.inference_method in _FREQUENTIST_ESTIMATOR_BY_METHOD
@property
def frequentist_estimator(self) -> str | None:
"""``"ridge"`` / ``"constrained"``, or ``None`` for a Bayesian config."""
return _FREQUENTIST_ESTIMATOR_BY_METHOD.get(self.inference_method)
@property
def use_numpyro(self) -> bool:
return self.inference_method == InferenceMethod.BAYESIAN_NUMPYRO
@property
def nuts_sampler(self) -> str:
"""The ``pm.sample(nuts_sampler=...)`` backend this config selects.
Frequentist methods have no NUTS backend; they report ``"pymc"`` so a
caller that reads this on a non-Bayesian config still gets the
historical default rather than an error. That fall-through is why
selecting one used to silently fit NUTS — ``fit()`` now branches on
:attr:`is_frequentist` *before* reading this, so the value is never
consumed on a frequentist config.
"""
return _NUTS_SAMPLER_BY_METHOD.get(self.inference_method, "pymc")
@property
def is_implemented(self) -> bool:
"""False for declared-but-unbacked inference methods (see #180)."""
return self.inference_method not in _UNIMPLEMENTED_METHODS
@model_validator(mode="after")
def _clear_fit_method_for_frequentist(self) -> "ModelConfig":
"""A frequentist config never carries a Bayesian ``fit_method``.
Enforced here rather than in each construction path because there are
four of them — the builder, the Excel template parser, a direct ctor
call, and deserializing a stored config — and the invariant has to hold
for all of them. ``FitMethod`` has no frequentist member, so the field's
``"nuts"`` default would otherwise make an **unfitted** frequentist model
announce a full MCMC posterior everywhere the prefit surfaces read it
(the prefit readout's inference-plan row, saved-model settings), right
up until the moment it runs something else entirely.
A stored config carrying both is corrected on load, which is the point:
the pair was never meaningful.
"""
if self.inference_method in _FREQUENTIST_ESTIMATOR_BY_METHOD:
object.__setattr__(self, "fit_method", None)
return self
@model_validator(mode="after")
def _warn_unimplemented_inference(self) -> "ModelConfig":
"""Flag an unimplemented method at construction, not after a long fit.
Deliberately a warning rather than a hard error here: a *stored* config
carrying one of these values must still deserialize (the enum values are
frozen), and code that only reads a config should not blow up. ``fit()``
is where it becomes an error.
"""
if self.inference_method in _UNIMPLEMENTED_METHODS:
warnings.warn(
unimplemented_inference_message(self.inference_method),
DeprecationWarning,
stacklevel=2,
)
return self