DAG Model Builder¶

The dag_model_builder module provides a fluent builder API for constructing Marketing Mix Models from directed acyclic graphs (DAGs).

DAG Specification¶

DAG Specification Classes

Defines the core data structures for representing model DAGs: - DAGNode: A single node (variable) in the graph - DAGEdge: A directed edge (relationship) between nodes - DAGSpec: The complete DAG specification

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

Bases: str, Enum

Type of node in the DAG.

KPI = 'kpi'¶

MEDIA = 'media'¶

CONTROL = 'control'¶

MEDIATOR = 'mediator'¶

OUTCOME = 'outcome'¶

INSTRUMENT = 'instrument'¶

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

Bases: str, Enum

Type of edge in the DAG.

DIRECT = 'direct'¶

MEDIATED = 'mediated'¶

CROSS_EFFECT = 'cross_effect'¶

class mmm_framework.dag_model_builder.dag_spec.DAGNode(**data)[source]¶

Bases: BaseModel

A node in the DAG representing a variable.

Attributes¶

idstr: Unique identifier for the node.
variable_namestr: Name of the variable in the MFF dataset.
node_typeNodeType: Type of node (KPI, MEDIA, CONTROL, MEDIATOR, OUTCOME, INSTRUMENT). INSTRUMENT nodes drive IV identification checks only; they are not emitted as model regressors.
labelstr | None: Display label for the node (defaults to variable_name).
dimensionslist[str]: Dimensions this variable is defined over (e.g., [“Period”, “Geography”]).
configdict[str, Any]: Node-specific configuration (adstock, saturation, priors, etc.).

id: str¶

variable_name: str¶

node_type: NodeType¶

label: str | None¶

dimensions: list[str]¶

config: dict[str, Any]¶

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

property display_label: str¶: Get display label, defaulting to variable_name.

property is_target: bool¶: Check if this node is a target/outcome variable.

property is_input: bool¶: Check if this node is an input variable.

class mmm_framework.dag_model_builder.dag_spec.DAGEdge(**data)[source]¶

Bases: BaseModel

A directed edge in the DAG representing a relationship.

Attributes¶

sourcestr: ID of the source node.
targetstr: ID of the target node.
edge_typeEdgeType: Type of edge (DIRECT, MEDIATED, CROSS_EFFECT).

source: str¶

target: str¶

edge_type: EdgeType¶

metadata: dict[str, Any]¶

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

class mmm_framework.dag_model_builder.dag_spec.DAGSpec(**data)[source]¶

Bases: BaseModel

Complete DAG specification for an MMM model.

Attributes¶

nodeslist[DAGNode]: All nodes in the DAG.
edgeslist[DAGEdge]: All edges in the DAG.
metadatadict[str, Any]: Optional metadata (e.g., frontend layout info).

Examples¶

>>> dag = DAGSpec(
...     nodes=[
...         DAGNode(id="sales", variable_name="Sales", node_type=NodeType.KPI),
...         DAGNode(id="tv", variable_name="TV", node_type=NodeType.MEDIA),
...     ],
...     edges=[
...         DAGEdge(source="tv", target="sales"),
...     ]
... )

nodes: list[DAGNode]¶

edges: list[DAGEdge]¶

metadata: dict[str, Any]¶

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

get_node(node_id)[source]¶

Get a node by ID.

Return type:: DAGNode | None

get_node_by_variable(variable_name)[source]¶

Get a node by variable name.

Return type:: DAGNode | None

get_nodes_by_type(node_type)[source]¶

Get all nodes of a specific type.

Return type:: list[DAGNode]

get_incoming_edges(node_id)[source]¶

Get all edges pointing to a node.

Return type:: list[DAGEdge]

get_outgoing_edges(node_id)[source]¶

Get all edges originating from a node.

Return type:: list[DAGEdge]

get_parents(node_id)[source]¶

Get all parent nodes of a given node.

Return type:: list[DAGNode]

get_children(node_id)[source]¶

Get all child nodes of a given node.

Return type:: list[DAGNode]

property kpi_nodes: list[DAGNode]¶: Get all KPI nodes.

property media_nodes: list[DAGNode]¶: Get all media nodes.

property control_nodes: list[DAGNode]¶: Get all control nodes.

property mediator_nodes: list[DAGNode]¶: Get all mediator nodes.

property instrument_nodes: list[DAGNode]¶: Get all instrument nodes.

property has_instruments: bool¶: Check if DAG has any instrument nodes.

property outcome_nodes: list[DAGNode]¶: Get all outcome nodes (including KPI).

property has_mediators: bool¶: Check if DAG has any mediator nodes.

property has_multiple_outcomes: bool¶: Check if DAG has multiple outcome/KPI nodes.

property has_cross_effects: bool¶: Check if DAG has any cross-effect edges.

property node_ids: list[str]¶: Get all node IDs.

property variable_names: list[str]¶: Get all variable names.

to_adjacency_list()[source]¶

Convert DAG to adjacency list representation.

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

Builder¶

DAG Model Builder

Main builder class for constructing MMM models from DAG specifications.

exception mmm_framework.dag_model_builder.builder.DAGBuildError[source]¶

Bases: Exception

Exception raised when model building fails.

class mmm_framework.dag_model_builder.builder.DAGModelBuilder[source]¶

Bases: object

Builder for constructing MMM models from DAG specifications.

This is the primary interface for the DAG-based model building workflow. It supports fluent API for configuration and automatic model type selection.

Examples¶

Basic usage:

>>> from mmm_framework.dag_model_builder import (
...     DAGModelBuilder, DAGSpec, DAGNode, DAGEdge, NodeType
... )
>>> dag = DAGSpec(
...     nodes=[
...         DAGNode(id="sales", variable_name="Sales", node_type=NodeType.KPI),
...         DAGNode(id="tv", variable_name="TV", node_type=NodeType.MEDIA),
...     ],
...     edges=[DAGEdge(source="tv", target="sales")]
... )
>>> model = (
...     DAGModelBuilder()
...     .with_dag(dag)
...     .with_mff_data("data.csv")
...     .bayesian_numpyro()
...     .build()
... )

With mediation:

>>> dag = DAGSpec(
...     nodes=[
...         DAGNode(id="sales", variable_name="Sales", node_type=NodeType.KPI),
...         DAGNode(id="tv", variable_name="TV", node_type=NodeType.MEDIA),
...         DAGNode(id="awareness", variable_name="Awareness", node_type=NodeType.MEDIATOR),
...     ],
...     edges=[
...         DAGEdge(source="tv", target="awareness"),
...         DAGEdge(source="awareness", target="sales"),
...         DAGEdge(source="tv", target="sales"),
...     ]
... )
>>> # Automatically uses NestedMMM
>>> model = DAGModelBuilder().with_dag(dag).with_mff_data(df).build()

__init__()[source]¶

with_dag(dag)[source]¶

Set the DAG specification.

Return type:: Self

Parameters¶

dagDAGSpec: The DAG specification.

Returns¶

Self: The builder instance for chaining.

with_dag_dict(dag_dict)[source]¶

Set DAG from a dictionary.

Return type:: Self

Parameters¶

dag_dictdict: Dictionary with “nodes” and “edges” keys.

Returns¶

Self: The builder instance for chaining.

classmethod from_dag(dag)[source]¶

Create a builder with a DAG specification.

Return type:: Self

Parameters¶

dagDAGSpec: The DAG specification.

Returns¶

DAGModelBuilder: A new builder instance with the DAG set.

classmethod from_frontend_json(json_data, panel=None)[source]¶

Create builder from React Flow frontend JSON format.

Return type:: Self

Parameters¶

json_datadict | str: React Flow JSON data or JSON string.
panelPanelDataset | None: Optional panel dataset.

Returns¶

DAGModelBuilder: A new builder instance.

with_panel(panel)[source]¶

Set the panel dataset.

Return type:: Self

Parameters¶

panelPanelDataset: The panel dataset.

Returns¶

Self: The builder instance for chaining.

with_mff_data(data, mff_config=None)[source]¶

Load data from MFF format.

If mff_config is not provided, it will be auto-generated from the DAG.

Return type:: Self

Parameters¶

datapd.DataFrame | str: MFF data or path to CSV file.
mff_configMFFConfig | None: Optional MFF configuration. If None, generated from DAG.

Returns¶

Self: The builder instance for chaining.

with_date_format(date_format)[source]¶

Set the date format for parsing MFF data.

Return type:: Self

Parameters¶

date_formatstr: Date format string (e.g., “%Y-%m-%d”).

Returns¶

Self: The builder instance for chaining.

with_frequency(frequency)[source]¶

Set the data frequency.

Return type:: Self

Parameters¶

frequencystr: Data frequency (“W” for weekly, “D” for daily, “M” for monthly).

Returns¶

Self: The builder instance for chaining.

with_model_config(config)[source]¶

Set model-level configuration.

Return type:: Self

Parameters¶

configModelConfig: The model configuration.

Returns¶

Self: The builder instance for chaining.

with_trend_config(config)[source]¶

Set trend configuration.

Return type:: Self

Parameters¶

configTrendConfig: The trend configuration.

Returns¶

Self: The builder instance for chaining.

bayesian_pymc()[source]¶

Use PyMC backend for inference.

Return type:: Self

Returns¶

Self: The builder instance for chaining.

bayesian_numpyro()[source]¶

Use NumPyro backend for inference (faster).

Return type:: Self

Returns¶

Self: The builder instance for chaining.

bayesian_nutpie()[source]¶

Use the nutpie NUTS sampler (Rust backend).

Return type:: Self

Returns¶

Self: The builder instance for chaining.

with_draws(n_draws)[source]¶

Set number of posterior draws.

Return type:: Self

Parameters¶

n_drawsint: Number of draws per chain.

Returns¶

Self: The builder instance for chaining.

with_tune(n_tune)[source]¶

Set number of tuning samples.

Return type:: Self

Parameters¶

n_tuneint: Number of tuning samples per chain.

Returns¶

Self: The builder instance for chaining.

with_chains(n_chains)[source]¶

Set number of chains.

Return type:: Self

Parameters¶

n_chainsint: Number of MCMC chains.

Returns¶

Self: The builder instance for chaining.

configure_media(variable_name, adstock_lmax=None, adstock_type=None, saturation_type=None, coefficient_prior_sigma=None, parent_channel=None)[source]¶

Override configuration for a specific media channel.

Return type:: Self

Parameters¶

variable_namestr: Name of the media variable.
adstock_lmaxint | None: Maximum lag for adstock.
adstock_typestr | None: Type of adstock (“geometric”, “weibull”, etc.).
saturation_typestr | None: Type of saturation (“hill”, “logistic”, etc.).
coefficient_prior_sigmafloat | None: Sigma for coefficient prior.
parent_channelstr | None: Parent channel for hierarchical grouping.

Returns¶

Self: The builder instance for chaining.

configure_control(variable_name, allow_negative=None, coefficient_prior_mu=None, coefficient_prior_sigma=None, use_shrinkage=None)[source]¶

Override configuration for a specific control variable.

Return type:: Self

Parameters¶

variable_namestr: Name of the control variable.
allow_negativebool | None: Whether to allow negative coefficient.
coefficient_prior_mufloat | None: Mean for coefficient prior.
coefficient_prior_sigmafloat | None: Sigma for coefficient prior.
use_shrinkagebool | None: Whether to apply shrinkage prior.

Returns¶

Self: The builder instance for chaining.

configure_mediator(variable_name, mediator_type=None, observation_noise_sigma=None, allow_direct_effect=None, media_effect_constraint=None)[source]¶

Override configuration for a specific mediator.

Return type:: Self

Parameters¶

variable_namestr: Name of the mediator variable.
mediator_typestr | None: Type of mediator observation model.
observation_noise_sigmafloat | None: Observation noise sigma.
allow_direct_effectbool | None: Whether to allow direct media -> outcome effect.
media_effect_constraintstr | None: Constraint on media -> mediator effect.

Returns¶

Self: The builder instance for chaining.

validate()[source]¶

Validate DAG structure and data compatibility.

Return type:: ValidationResult

Returns¶

ValidationResult: Validation result with errors and warnings.

Raises¶

DAGBuildError: If no DAG has been set.

get_model_type()[source]¶

Get the resolved model type based on DAG structure.

Return type:: ModelType

Returns¶

ModelType: The model type that will be built.

Raises¶

DAGBuildError: If no DAG has been set.

build_mff_config()[source]¶

Build MFFConfig from DAG without building the full model.

Return type:: MFFConfig

Returns¶

MFFConfig: The generated MFF configuration.

Raises¶

DAGBuildError: If no DAG has been set.

build()[source]¶

Build the appropriate model based on DAG structure.

Return type:: BayesianMMM | NestedMMM | MultivariateMMM | CombinedMMM

Returns¶

BayesianMMM | NestedMMM | MultivariateMMM | CombinedMMM: The constructed model.

Raises¶

DAGBuildError: If DAG or data is not set, or validation fails.

Validation¶

DAG Validation

Validates DAG structure and compatibility with data.

class mmm_framework.dag_model_builder.validation.ValidationResult(valid, errors=<factory>, warnings=<factory>)[source]¶

Bases: object

Result of DAG validation.

Attributes¶

validbool: Whether the DAG passed all validation checks.
errorslist[str]: List of validation errors (fatal).
warningslist[str]: List of validation warnings (non-fatal).

valid: bool¶

errors: list[str]¶

warnings: list[str]¶

raise_if_invalid()[source]¶

Raise DAGValidationError if not valid.

Return type:: None

__init__(valid, errors=<factory>, warnings=<factory>)¶

exception mmm_framework.dag_model_builder.validation.DAGValidationError(errors, warnings=None)[source]¶

Bases: Exception

Exception raised when DAG validation fails.

__init__(errors, warnings=None)[source]¶

mmm_framework.dag_model_builder.validation.is_acyclic(dag)[source]¶

Check if the DAG is acyclic using topological sort (Kahn’s algorithm).

Return type:: bool

Parameters¶

dagDAGSpec: The DAG to check.

Returns¶

bool: True if the DAG is acyclic, False otherwise.

mmm_framework.dag_model_builder.validation.validate_dag(dag)[source]¶

Validate DAG structure.

Checks: - DAG is acyclic - Has at least one KPI/outcome node - Has at least one media node - All edge source/target IDs exist as nodes - No duplicate node IDs - No duplicate variable names - Edge types are valid for node types

Return type:: ValidationResult

Parameters¶

dagDAGSpec: The DAG to validate.

Returns¶

ValidationResult: Validation result with errors and warnings.

mmm_framework.dag_model_builder.validation.validate_dag_against_data(dag, panel)[source]¶

Validate DAG against available data.

Checks: - All variable names in DAG exist in the panel data - Dimension compatibility

Return type:: ValidationResult

Parameters¶

dagDAGSpec: The DAG to validate.
panelPanelDataset: The panel dataset to validate against.

Returns¶

ValidationResult: Validation result with errors and warnings.

mmm_framework.dag_model_builder.validation.validate_complete(dag, panel=None)[source]¶

Perform complete validation of DAG structure and data compatibility.

Return type:: ValidationResult

Parameters¶

dagDAGSpec: The DAG to validate.
panelPanelDataset | None: Optional panel dataset for data validation.

Returns¶

ValidationResult: Combined validation result.

Node Configurations¶

Node-Specific Configuration Classes

Provides typed configuration classes for each node type in the DAG. These configs are used to specify priors, transformations, and other node-specific settings.

class mmm_framework.dag_model_builder.node_configs.MediaNodeConfig(**data)[source]¶

Bases: BaseModel

Configuration for a media node.

Attributes¶

adstock_typeAdstockType: Type of adstock transformation.
adstock_lmaxint: Maximum lag for adstock.
adstock_normalizebool: Whether to normalize adstock weights.
adstock_alpha_prior_alphafloat: Alpha parameter for adstock decay prior (Beta distribution).
adstock_alpha_prior_betafloat: Beta parameter for adstock decay prior (Beta distribution).
saturation_typeSaturationType: Type of saturation transformation.
coefficient_prior_sigmafloat: Sigma for the coefficient prior (HalfNormal).
parent_channelstr | None: Parent channel for hierarchical media grouping.

adstock_type: AdstockType¶

adstock_lmax: int¶

adstock_normalize: bool¶

adstock_alpha_prior_alpha: float¶

adstock_alpha_prior_beta: float¶

saturation_type: SaturationType¶

saturation_kappa_prior_alpha: float¶

saturation_kappa_prior_beta: float¶

saturation_slope_prior_sigma: float¶

saturation_beta_prior_sigma: float¶

coefficient_prior_sigma: float¶

parent_channel: str | None¶

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

class mmm_framework.dag_model_builder.node_configs.ControlNodeConfig(**data)[source]¶

Bases: BaseModel

Configuration for a control node.

Attributes¶

allow_negativebool: Whether the coefficient can be negative.
coefficient_prior_mufloat: Mean of the coefficient prior (Normal distribution).
coefficient_prior_sigmafloat: Sigma of the coefficient prior (Normal distribution).
use_shrinkagebool: Whether to apply shrinkage (horseshoe-like) prior.

allow_negative: bool¶

coefficient_prior_mu: float¶

coefficient_prior_sigma: float¶

use_shrinkage: bool¶

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

class mmm_framework.dag_model_builder.node_configs.KPINodeConfig(**data)[source]¶

Bases: BaseModel

Configuration for a KPI (target) node.

Attributes¶

log_transformbool: Whether to log-transform the KPI (for multiplicative models).
floor_valuefloat: Minimum value for log safety.

log_transform: bool¶

floor_value: float¶

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

class mmm_framework.dag_model_builder.node_configs.MediatorNodeConfig(**data)[source]¶

Bases: BaseModel

Configuration for a mediator node.

Attributes¶

mediator_typestr: Type of mediator observation model. Options: “fully_observed”, “partially_observed”, “aggregated_survey”, “fully_latent”
observation_noise_sigmafloat: Observation noise sigma for observed mediators.
allow_direct_effectbool: Whether to allow direct media -> outcome effects (bypassing mediator).
direct_effect_sigmafloat: Prior sigma for direct effect.
media_effect_constraintstr: Constraint on media -> mediator effect. Options: “none”, “positive”, “negative”.
media_effect_sigmafloat: Prior sigma for media -> mediator effect.
outcome_effect_sigmafloat: Prior sigma for mediator -> outcome effect.
apply_adstockbool: Whether to apply adstock to media -> mediator pathway.
apply_saturationbool: Whether to apply saturation to media -> mediator pathway.

mediator_type: str¶

observation_noise_sigma: float¶

allow_direct_effect: bool¶

direct_effect_sigma: float¶

media_effect_constraint: str¶

media_effect_sigma: float¶

outcome_effect_sigma: float¶

apply_adstock: bool¶

apply_saturation: bool¶

dynamics: str | None¶

likelihood: str | None¶

trials_variable: str | None¶

category_variables: list[str] | None¶

design_effect: float¶

cutpoint_prior_sigma: float | None¶

rho_prior_alpha: float | None¶

rho_prior_beta: float | None¶

innovation_sigma: float | None¶

state_parameterization: str | None¶

affects_outcome: bool¶

parent_effect_sigma: float | None¶

control_effect_sigma: float | None¶

latent_factors: list[str] | None¶

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

class mmm_framework.dag_model_builder.node_configs.OutcomeNodeConfig(**data)[source]¶

Bases: BaseModel

Configuration for an outcome node (non-primary KPI).

Attributes¶

include_trendbool: Whether to include trend component.
include_seasonalitybool: Whether to include seasonality component.
intercept_prior_sigmafloat: Prior sigma for intercept.
media_effect_sigmafloat: Prior sigma for media effects.
log_transformbool: Whether to log-transform the outcome.

include_trend: bool¶

include_seasonality: bool¶

intercept_prior_sigma: float¶

media_effect_sigma: float¶

log_transform: bool¶

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

mmm_framework.dag_model_builder.node_configs.parse_node_config(node_type, config_dict)[source]¶

Parse a config dict into the appropriate NodeConfig type.

Return type:: MediaNodeConfig | ControlNodeConfig | KPINodeConfig | MediatorNodeConfig | OutcomeNodeConfig

Parameters¶

node_typestr: Type of node (“media”, “control”, “kpi”, “mediator”, “outcome”).
config_dictdict: Dictionary of configuration values.

Returns¶

NodeConfig: Parsed configuration object.

Raises¶

ValueError: If node_type is unknown.

Config Translator¶

Config Translator

Translates DAG specifications to framework configuration objects.

mmm_framework.dag_model_builder.config_translator.dag_to_mff_config(dag, date_format='%Y-%m-%d', frequency='W', enforce_identification=True)[source]¶

Translate DAG specification to MFFConfig.

Return type:: MFFConfig

Parameters¶

dagDAGSpec: The DAG specification.
date_formatstr: Date format string for parsing.
frequencystr: Data frequency (“W”, “D”, “M”).
enforce_identificationbool: When True (default), run backdoor identification and (a) tag each control with the causal role inferred from the adjustment set, so the model can refuse bad controls, and (b) warn when the effect is unidentified or when an identified confounder is missing from the controls. Set False to skip identification entirely (controls keep an unknown role).

Returns¶

MFFConfig: The generated MFF configuration.

Raises¶

ValueError: If no KPI node is found in the DAG.

Notes¶

INSTRUMENT nodes are intentionally NOT emitted into the MFFConfig: they are exogenous variation used only for graph-based IV identification checks (identification.iv_criterion()), not model regressors. IV estimation is a separate, not-yet-implemented feature.

mmm_framework.dag_model_builder.config_translator.dag_to_nested_config(dag)[source]¶

Extract nested model configuration from DAG.

Builds: - MediatorConfig for each mediator node - media_to_mediator_map from edges

Parameters¶

dagDAGSpec: The DAG specification.

Returns¶

NestedModelConfig: The nested model configuration.

mmm_framework.dag_model_builder.config_translator.dag_to_multivariate_config(dag)[source]¶

Extract multivariate model configuration from DAG.

Builds: - OutcomeConfig for each outcome node - CrossEffectConfig for cross-effect edges

Parameters¶

dagDAGSpec: The DAG specification.

Returns¶

MultivariateModelConfig: The multivariate model configuration.

mmm_framework.dag_model_builder.config_translator.dag_to_combined_config(dag)[source]¶

Build combined nested + multivariate config from DAG.

Parameters¶

dagDAGSpec: The DAG specification.

Returns¶

CombinedModelConfig: The combined model configuration.

mmm_framework.dag_model_builder.config_translator.dag_to_structural_config(dag)[source]¶

Extract a StructuralNestedConfig + per-mediator data requirements from a structural DAG.

Per mediator node: channels from MEDIA->mediator edges, parents from mediator->mediator edges, controls from CONTROL->mediator edges, latent factors + dynamics/measurement/priors from the node config – mapping ONLY keys present in the RAW config dict so MediatorSpec defaults (e.g. the tight direct-effect prior, dynamics-resolved adstock) hold when unset. Latent factors come from dag.metadata["latent_factors"] (a list of LatentFactorSpec-shaped dicts). outcome_controls is the set of CONTROL->KPI edges, so a control driving only a mediator (price -> consideration) stays out of the outcome equation.

Returns¶

(StructuralNestedConfig, list[dict]): The model config and, per non-latent mediator, the MFF data requirement: {"name", "likelihood", "variable_name", "trials_variable", "category_variables"}.

Model Type Resolver¶

Model Type Resolver

Determines which model class to use based on DAG structure.

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

Bases: str, Enum

Type of model to build based on DAG structure.

BAYESIAN_MMM = 'bayesian_mmm'¶

NESTED_MMM = 'nested_mmm'¶

STRUCTURAL_NESTED_MMM = 'structural_nested_mmm'¶

MULTIVARIATE_MMM = 'multivariate_mmm'¶

COMBINED_MMM = 'combined_mmm'¶

mmm_framework.dag_model_builder.model_type_resolver.resolve_model_type(dag)[source]¶

Determine the appropriate model class based on DAG structure.

Decision logic: 1. Has mediators + multiple outcomes → CombinedMMM 2. Has mediators only → NestedMMM 3. Multiple outcomes or cross-effects only → MultivariateMMM 4. Otherwise → BayesianMMM

Return type:: ModelType

Parameters¶

dagDAGSpec: The DAG specification.

Returns¶

ModelType: The resolved model type.

Examples¶

>>> dag = DAGSpec(
...     nodes=[
...         DAGNode(id="sales", variable_name="Sales", node_type=NodeType.KPI),
...         DAGNode(id="tv", variable_name="TV", node_type=NodeType.MEDIA),
...     ],
...     edges=[DAGEdge(source="tv", target="sales")]
... )
>>> resolve_model_type(dag)
<ModelType.BAYESIAN_MMM: 'bayesian_mmm'>

mmm_framework.dag_model_builder.model_type_resolver.get_model_class(model_type)[source]¶

Get the model class for a given model type.

Uses lazy imports to avoid loading PyMC unless needed.

Parameters¶

model_typeModelType: The model type.

Returns¶

type: The model class.

Raises¶

ValueError: If model_type is unknown.

mmm_framework.dag_model_builder.model_type_resolver.describe_model_type(dag)[source]¶

Get a human-readable description of the model type for a DAG.

Return type:: str

Parameters¶

dagDAGSpec: The DAG specification.

Returns¶

str: Description of the model type and why it was selected.

Frontend Adapter¶

Frontend Adapter

Converts between React Flow frontend JSON format and DAGSpec Python objects.

mmm_framework.dag_model_builder.frontend_adapter.react_flow_to_dag_spec(nodes, edges)[source]¶

Convert React Flow node/edge format to DAGSpec.

React Flow format (from frontend):

{
    "nodes": [
        {
            "id": "node_abc123",
            "type": "default",
            "position": {"x": 100, "y": 200},
            "data": {
                "label": "TV Spend",
                "type": "media",
                "variableName": "tv_spend",
                "config": {"adstockType": "geometric", ...}
            }
        }
    ],
    "edges": [
        {
            "id": "e1",
            "source": "node_abc",
            "target": "node_xyz",
            "data": {"edgeType": "direct"}
        }
    ]
}

Parameters:

nodes (list[dict]) – List of React Flow node objects.
edges (list[dict]) – List of React Flow edge objects.

Return type:

DAGSpec

Returns:

Converted DAG specification.

mmm_framework.dag_model_builder.frontend_adapter.dag_spec_to_react_flow(dag)[source]¶

Convert DAGSpec back to React Flow format for frontend.

Return type:: dict

Parameters¶

dagDAGSpec: The DAG specification.

Returns¶

dict: React Flow compatible JSON structure.

mmm_framework.dag_model_builder.frontend_adapter.create_simple_dag(kpi_name, media_names, control_names=None, dimensions=None)[source]¶

Create a simple DAG with all media and controls pointing to a single KPI.

This is a convenience function for creating basic model structures.

Return type:: DAGSpec

Parameters¶

kpi_namestr: Name of the KPI variable.
media_nameslist[str]: Names of media variables.
control_nameslist[str] | None: Names of control variables.
dimensionslist[str] | None: Dimensions for all variables.

Returns¶

DAGSpec: The created DAG specification.

Examples¶

>>> dag = create_simple_dag(
...     kpi_name="Sales",
...     media_names=["TV", "Digital", "Radio"],
...     control_names=["Price", "Distribution"],
... )

mmm_framework.dag_model_builder.frontend_adapter.create_mediation_dag(kpi_name, media_names, mediator_name, control_names=None, include_direct_effects=True, dimensions=None)[source]¶

Create a DAG with mediation structure.

All media → mediator → KPI, with optional direct media → KPI effects.

Return type:: DAGSpec

Parameters¶

kpi_namestr: Name of the KPI variable.
media_nameslist[str]: Names of media variables.
mediator_namestr: Name of the mediator variable.
control_nameslist[str] | None: Names of control variables.
include_direct_effectsbool: Whether to include direct media → KPI effects.
dimensionslist[str] | None: Dimensions for all variables.

Returns¶

DAGSpec: The created DAG specification.