"""
Data loading and transformation utilities for MFF format.
Handles variable-dimension data, dimension alignment, and panel construction.
"""
from __future__ import annotations
import re
import warnings
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
import numpy as np
import pandas as pd
from .config import (
AllocationMethod,
DimensionType,
MediaChannelConfig,
MFFConfig,
VariableConfig,
VariableRole,
MFFColumnConfig,
)
if TYPE_CHECKING:
from numpy.typing import NDArray
# =============================================================================
# Data validation
# =============================================================================
[docs]
class MFFValidationError(Exception):
"""Raised when MFF data fails validation."""
pass
# Tokens that mean "missing", not "unparseable", in a value column.
_NULL_TOKENS = {"", "nan", "none", "null", "na", "n/a", "-", "—"}
# Currency symbols + whitespace to strip before parsing (NOT the , / . separators,
# which are handled locale-aware by _normalize_decimal).
_CURRENCY_WS_RE = re.compile(r"[\s$€£¥₹]")
def _normalize_decimal_separators(cleaned: pd.Series, decimal: str) -> pd.Series:
"""Map a value column's thousands/decimal separators to a plain ``.`` decimal.
``decimal`` is the source's decimal mark: ``"."`` (US: ``1,234.56``), ``","``
(EU: ``1.234,56``), or ``"auto"`` (best-effort per value: when both marks
appear, the LAST one is the decimal; a lone ``,`` before a 3-digit group is
treated as a thousands separator, otherwise as a decimal).
"""
if decimal == ".":
return cleaned.str.replace(",", "", regex=False) # ',' = thousands
if decimal == ",":
return cleaned.str.replace(".", "", regex=False).str.replace(
",", ".", regex=False
)
def _auto(v: str) -> str:
has_dot, has_comma = "." in v, "," in v
if has_dot and has_comma:
if v.rfind(",") > v.rfind("."): # comma is the decimal
return v.replace(".", "").replace(",", ".")
return v.replace(",", "") # dot is the decimal; comma = thousands
if has_comma:
tail = v.rsplit(",", 1)[-1]
if v.count(",") == 1 and len(tail) == 3:
return v.replace(",", "") # 1,234 -> thousands
return v.replace(",", ".") # 1,5 -> decimal
return v
return cleaned.map(_auto)
[docs]
def coerce_numeric_column(
series: pd.Series, *, column: str, decimal: str = "."
) -> pd.Series:
"""Coerce an MFF value column to float, tolerating real-world formatting.
Client exports routinely carry currency/thousands formatting (``"$1,234"``,
``"1.234,56"``) or accounting negatives (``"(123)"``). Previously these flowed
straight into ``groupby().sum()`` and raised an opaque ``TypeError`` deep in
the stack. This normalizes them up front:
* already-numeric columns pass through unchanged;
* ``$ € £ ¥ ₹`` and whitespace are stripped; ``(123)`` -> ``-123``;
* thousands/decimal separators are normalized per ``decimal`` (``"."`` US,
``","`` EU, or ``"auto"``) so non-US-formatted data ingests;
* recognized null tokens (and genuinely empty cells) become ``NaN``;
* any remaining unparseable value raises :class:`MFFValidationError` naming
the column and a few offending samples (an actionable message, not a
``TypeError``).
"""
if pd.api.types.is_numeric_dtype(series):
return series.astype(float)
text = series.astype(str).str.strip()
is_nullish = series.isna() | text.str.lower().isin(_NULL_TOKENS)
cleaned = text.str.replace(_CURRENCY_WS_RE, "", regex=True)
cleaned = cleaned.str.replace(r"^\((.*)\)$", r"-\1", regex=True) # (123) -> -123
cleaned = _normalize_decimal_separators(cleaned, decimal)
coerced = pd.to_numeric(cleaned, errors="coerce")
failed = coerced.isna() & ~is_nullish
if failed.any():
sample = list(pd.unique(series[failed]))[:5]
raise MFFValidationError(
f"Column '{column}' contains {int(failed.sum())} non-numeric value(s) "
f"that could not be parsed as numbers (e.g. {sample!r}). Fix the "
f"currency/text formatting or remove those rows before loading."
)
result = coerced.astype(float)
result[is_nullish.to_numpy()] = np.nan
return result
[docs]
def validate_mff_structure(df: pd.DataFrame, config: MFFConfig) -> list[str]:
"""
Validate MFF dataframe structure against config.
Returns list of warnings (empty if clean).
Raises MFFValidationError for critical issues.
"""
warnings_list = []
cols = config.columns
# Check required columns exist
missing_cols = set(cols.all_columns) - set(df.columns)
if missing_cols:
raise MFFValidationError(f"Missing required columns: {missing_cols}")
# Check for expected variables. A declared ``spend_column`` is referenced by
# a media channel (for impression-level ROI) but is not itself a modeled
# variable, so it is "expected" and must not warn as an ignored extra.
actual_vars = set(df[cols.variable_name].unique())
spend_cols = {
m.spend_column
for m in config.media_channels
if getattr(m, "spend_column", None) is not None
}
expected_vars = set(config.variable_names) | spend_cols
missing_vars = set(config.variable_names) - actual_vars
if missing_vars:
raise MFFValidationError(f"Missing expected variables: {missing_vars}")
missing_spend = spend_cols - actual_vars
if missing_spend:
warnings_list.append(
f"Declared spend_column variable(s) absent from data (ROI will fall "
f"back to per-volume efficiency): {missing_spend}"
)
extra_vars = actual_vars - expected_vars
if extra_vars:
warnings_list.append(f"Extra variables in data (will be ignored): {extra_vars}")
# Check for nulls in key columns
null_counts = df[cols.all_columns].isnull().sum()
if null_counts[cols.variable_value] > 0:
warnings_list.append(
f"Found {null_counts[cols.variable_value]} null values in {cols.variable_value}"
)
# Validate date parsing across the WHOLE period column (not just the first
# row): a file with one well-formed header date but corrupt later dates used
# to pass here and then crash mid-load with a raw pandas ValueError, breaking
# the framework's own MFFValidationError contract.
period_raw = df[cols.period]
parsed = pd.to_datetime(period_raw, format=config.date_format, errors="coerce")
bad_dates = parsed.isna() & period_raw.notna()
if bad_dates.any():
sample = list(pd.unique(period_raw[bad_dates]))[:5]
raise MFFValidationError(
f"Cannot parse {int(bad_dates.sum())} date(s) in column "
f"'{cols.period}' with format '{config.date_format}' (e.g. {sample!r})."
)
return warnings_list
[docs]
def validate_variable_dimensions(
df: pd.DataFrame,
var_config: VariableConfig,
config: MFFConfig,
) -> tuple[bool, str]:
"""
Validate that a variable's data matches its configured dimensions.
Returns (is_valid, message).
"""
cols = config.columns
var_data = df[df[cols.variable_name] == var_config.name]
if var_data.empty:
return False, f"No data found for variable '{var_config.name}'"
# Map dimension types to column names
dim_col_map = {
DimensionType.PERIOD: cols.period,
DimensionType.GEOGRAPHY: cols.geography,
DimensionType.PRODUCT: cols.product,
DimensionType.CAMPAIGN: cols.campaign,
DimensionType.OUTLET: cols.outlet,
DimensionType.CREATIVE: cols.creative,
}
# Check which dimensions have variation
active_dims = []
for dim_type in DimensionType:
col = dim_col_map[dim_type]
unique_vals = var_data[col].dropna().unique()
# Consider dimension active if more than one unique non-null value
# or if it's Period (always active)
if len(unique_vals) > 1 or dim_type == DimensionType.PERIOD:
active_dims.append(dim_type)
configured_dims = set(var_config.dimensions)
actual_dims = set(active_dims)
# Period should always be present
if DimensionType.PERIOD not in actual_dims:
return False, f"Variable '{var_config.name}' has no time variation"
# Check for unexpected dimensions
extra_dims = actual_dims - configured_dims - {DimensionType.PERIOD}
if extra_dims and DimensionType.PERIOD in configured_dims:
# Only warn if we expected Period-only but got more
if configured_dims == {DimensionType.PERIOD}:
return False, (
f"Variable '{var_config.name}' has unexpected dimensions: {extra_dims}. "
f"Configured for: {configured_dims}"
)
return True, "OK"
# =============================================================================
# Panel dataset container
# =============================================================================
[docs]
@dataclass
class PanelCoordinates:
"""Coordinate labels for panel dimensions."""
periods: pd.DatetimeIndex
geographies: list[str] | None = None
products: list[str] | None = None
channels: list[str] = field(default_factory=list)
controls: list[str] = field(default_factory=list)
@property
def has_geo(self) -> bool:
return self.geographies is not None and len(self.geographies) > 0
@property
def has_product(self) -> bool:
return self.products is not None and len(self.products) > 0
@property
def n_periods(self) -> int:
return len(self.periods)
@property
def n_geos(self) -> int:
return len(self.geographies) if self.geographies else 1
@property
def n_products(self) -> int:
return len(self.products) if self.products else 1
@property
def n_obs(self) -> int:
"""Total number of observations."""
return self.n_periods * self.n_geos * self.n_products
[docs]
def to_pymc_coords(self) -> dict[str, list]:
"""Convert to PyMC coordinate dictionary."""
coords = {
"date": self.periods.tolist(),
"channel": self.channels,
}
if self.has_geo:
coords["geo"] = self.geographies
if self.has_product:
coords["product"] = self.products
if self.controls:
coords["control"] = self.controls
return coords
[docs]
@dataclass
class PanelDataset:
"""
Structured panel data ready for modeling.
All arrays are aligned to the same index structure.
"""
# Target variable
y: pd.Series
# Feature matrices
X_media: pd.DataFrame
X_controls: pd.DataFrame | None
# Coordinate information
coords: PanelCoordinates
# Index for reconstruction
index: pd.MultiIndex | pd.DatetimeIndex
# Original config reference
config: MFFConfig
# Metadata
media_stats: dict[str, dict] = field(default_factory=dict)
# Per-variable boolean masks of values that were EXPLICITLY missing (NaN) in
# the source, as opposed to filled — populated by RaggedMFFLoader. ``None`` on
# the standard (fully-filled) loader path.
explicit_nan_mask: dict[str, pd.Series] | None = None
# Per-channel external dollar-spend series (impression-level ROI, option a):
# ``{channel_name: aligned per-obs spend array}`` for channels that declare a
# ``MediaChannelConfig.spend_column``. ``None`` (the default) when no channel
# uses a separate spend column — the modeled variable is the spend, so ROI is
# divided by it directly. Aligned to the same index as ``X_media``.
spend_raw: dict[str, NDArray] | None = None
@property
def n_obs(self) -> int:
return len(self.y)
@property
def n_channels(self) -> int:
return self.X_media.shape[1] if self.X_media is not None else 0
@property
def n_controls(self) -> int:
return self.X_controls.shape[1] if self.X_controls is not None else 0
@property
def is_panel(self) -> bool:
"""True if data has geo or product dimensions."""
return self.coords.has_geo or self.coords.has_product
[docs]
def to_numpy(self) -> tuple[NDArray, NDArray, NDArray | None]:
"""Convert to numpy arrays for modeling."""
y = self.y.values
X_media = self.X_media.values
X_controls = self.X_controls.values if self.X_controls is not None else None
return y, X_media, X_controls
[docs]
def as_dataset(self) -> "Dataset": # noqa: F821 - lazy import to avoid a cycle
"""View this panel through the flexible role-tagged :class:`Dataset`.
Additive adapter for the generalized data layer: the MMM roles are tagged
(kpi→target, media→predictor, control→control) and every reader keeps
working via the ``Dataset`` MMM views. See :mod:`mmm_framework.dataset`.
"""
from .dataset import Dataset
return Dataset.from_panel(self)
[docs]
def compute_spend_shares(self) -> pd.Series:
"""Compute share of total spend for each channel."""
totals = self.X_media.sum()
return totals / totals.sum()
[docs]
def summary(self) -> str:
"""Generate summary string."""
lines = [
"PanelDataset Summary",
"=" * 40,
f"Observations: {self.n_obs}",
f"Time periods: {self.coords.n_periods}",
f"Geographies: {self.coords.n_geos}",
f"Products: {self.coords.n_products}",
f"Media channels: {self.n_channels}",
f"Control variables: {self.n_controls}",
"",
"Target (y) stats:",
f" Mean: {self.y.mean():.2f}",
f" Std: {self.y.std():.2f}",
f" Min: {self.y.min():.2f}",
f" Max: {self.y.max():.2f}",
]
if self.n_channels > 0:
lines.append("")
lines.append("Media channel totals:")
for col in self.X_media.columns:
lines.append(f" {col}: {self.X_media[col].sum():,.0f}")
return "\n".join(lines)
# =============================================================================
# MFF Loader and Parser
# =============================================================================
[docs]
class MFFLoader:
"""
Loads and parses MFF format data into panel structure.
Handles:
- Variable extraction by name
- Dimension alignment (national to geo, etc.)
- Missing value handling
- Date parsing and frequency alignment
"""
[docs]
def __init__(self, config: MFFConfig):
self.config = config
self._raw_data: pd.DataFrame | None = None
self._allocation_weights: dict[str, pd.Series] = {}
[docs]
def load(self, data: pd.DataFrame | str) -> MFFLoader:
"""
Load MFF data from DataFrame or file path.
Parameters
----------
data : pd.DataFrame or str
Either a DataFrame in MFF format or path to CSV/parquet file.
Returns
-------
self : MFFLoader
For method chaining.
"""
if isinstance(data, str):
if data.endswith(".parquet"):
self._raw_data = pd.read_parquet(data)
else:
self._raw_data = pd.read_csv(data)
else:
self._raw_data = data.copy()
# Validate structure
warnings_list = validate_mff_structure(self._raw_data, self.config)
for w in warnings_list:
warnings.warn(w)
# Parse dates
cols = self.config.columns
self._raw_data[cols.period] = pd.to_datetime(
self._raw_data[cols.period], format=self.config.date_format
)
# Coerce the value column to numeric up front (tolerating currency /
# thousands formatting) so a "$1,234" cell raises a clear validation
# error instead of a TypeError deep inside a later groupby().sum().
self._raw_data[cols.variable_value] = coerce_numeric_column(
self._raw_data[cols.variable_value],
column=cols.variable_value,
decimal=getattr(self.config, "decimal_separator", "."),
)
return self
[docs]
def set_allocation_weights(
self,
dimension: DimensionType,
weights: pd.Series | dict[str, float],
) -> MFFLoader:
"""
Set custom allocation weights for dimension disaggregation.
Parameters
----------
dimension : DimensionType
The dimension to set weights for (GEOGRAPHY or PRODUCT).
weights : pd.Series or dict
Weights indexed by dimension level (e.g., geo names).
Will be normalized to sum to 1.
Returns
-------
self : MFFLoader
For method chaining.
"""
if isinstance(weights, dict):
weights = pd.Series(weights)
# Normalize
weights = weights / weights.sum()
self._allocation_weights[dimension.value] = weights
return self
def _extract_variable(
self,
var_config: VariableConfig,
) -> pd.DataFrame:
"""Extract a single variable from MFF data."""
cols = self.config.columns
# Filter to this variable
mask = self._raw_data[cols.variable_name] == var_config.name
var_data = self._raw_data[mask].copy()
if var_data.empty:
raise MFFValidationError(f"No data found for variable '{var_config.name}'")
# Build dimension columns based on config
dim_cols = [cols.period] # Always include period
if DimensionType.GEOGRAPHY in var_config.dimensions:
dim_cols.append(cols.geography)
if DimensionType.PRODUCT in var_config.dimensions:
dim_cols.append(cols.product)
# Handle split dimensions for media
if isinstance(var_config, MediaChannelConfig):
if DimensionType.OUTLET in var_config.split_dimensions:
dim_cols.append(cols.outlet)
if DimensionType.CAMPAIGN in var_config.split_dimensions:
dim_cols.append(cols.campaign)
# Guard against SILENT duplicate-row summing. A true duplicate is two raw
# rows sharing the FULL MFF key (period + every dimension column) for this
# variable -- the same cell measured twice. Summing those corrupts the
# value. (Rows that merely differ on a dimension the model does not split
# on are legitimate finer granularity and are aggregated up below.)
key_cols = [c for c in cols.dimension_columns if c in var_data.columns]
if key_cols and var_data.duplicated(subset=key_cols).any():
policy = getattr(self.config, "duplicate_policy", "error")
n_dups = int(var_data.duplicated(subset=key_cols).sum())
if policy == "error":
sample = (
var_data.loc[
var_data.duplicated(subset=key_cols, keep=False), key_cols
]
.drop_duplicates()
.head(5)
.to_dict("records")
)
raise MFFValidationError(
f"Variable '{var_config.name}' has {n_dups} duplicate row(s) "
f"sharing the same period/dimension key (e.g. {sample!r}). "
f"Summing them would silently corrupt the value. Deduplicate "
f"the data, or set MFFConfig.duplicate_policy to "
f"'sum'/'mean'/'first' to choose how to combine them."
)
if policy in ("mean", "first"):
# Collapse the duplicated cells first; finer granularity is then
# summed by the dim_cols aggregation below.
var_data = var_data.groupby(key_cols, as_index=False)[
cols.variable_value
].agg(policy)
# policy == "sum": the dim_cols groupby below sums them directly.
# Aggregate to configured dimensions
result = var_data.groupby(dim_cols, as_index=False)[cols.variable_value].sum()
return result
def _get_allocation_weights(
self,
dimension: DimensionType,
levels: list[str],
) -> pd.Series:
"""Get allocation weights for a dimension."""
dim_name = dimension.value
# Check for custom weights
if dim_name in self._allocation_weights:
weights = self._allocation_weights[dim_name]
# Ensure all levels are present
missing = set(levels) - set(weights.index)
if missing:
warnings.warn(
f"Missing allocation weights for {missing}. Using equal weights."
)
return pd.Series(1.0 / len(levels), index=levels)
return weights.loc[levels]
# Check config for method
if dimension == DimensionType.GEOGRAPHY:
method = self.config.alignment.geo_allocation
weight_var = self.config.alignment.geo_weight_variable
elif dimension == DimensionType.PRODUCT:
method = self.config.alignment.product_allocation
weight_var = self.config.alignment.product_weight_variable
else:
method = AllocationMethod.EQUAL
weight_var = None
if method == AllocationMethod.EQUAL:
return pd.Series(1.0 / len(levels), index=levels)
elif method == AllocationMethod.SALES and self._raw_data is not None:
# Use KPI totals as weights
cols = self.config.columns
kpi_data = self._raw_data[
self._raw_data[cols.variable_name] == self.config.kpi.name
]
dim_col = (
cols.geography if dimension == DimensionType.GEOGRAPHY else cols.product
)
totals = kpi_data.groupby(dim_col)[cols.variable_value].sum()
weights = totals / totals.sum()
return weights.reindex(levels).fillna(1.0 / len(levels))
elif method == AllocationMethod.CUSTOM and weight_var:
# Extract weight variable from data
cols = self.config.columns
weight_data = self._raw_data[
self._raw_data[cols.variable_name] == weight_var
]
dim_col = (
cols.geography if dimension == DimensionType.GEOGRAPHY else cols.product
)
weights = weight_data.groupby(dim_col)[cols.variable_value].sum()
weights = weights / weights.sum()
return weights.reindex(levels).fillna(1.0 / len(levels))
# Default to equal
return pd.Series(1.0 / len(levels), index=levels)
def _align_to_target_dimensions(
self,
var_data: pd.DataFrame,
var_config: VariableConfig,
target_index: pd.MultiIndex | pd.DatetimeIndex,
) -> pd.Series:
"""
Align variable data to target KPI dimensions.
Handles:
- Disaggregation (national → geo, national → geo+product)
- Aggregation (product-geo → geo)
- Direct mapping (same dimensions)
"""
cols = self.config.columns
target_dims = self.config.kpi.dimensions
var_dims = var_config.dimensions
# Get dimension names for indexing
target_has_geo = DimensionType.GEOGRAPHY in target_dims
target_has_product = DimensionType.PRODUCT in target_dims
var_has_geo = DimensionType.GEOGRAPHY in var_dims
var_has_product = DimensionType.PRODUCT in var_dims
# Build index columns
var_index_cols = [cols.period]
if var_has_geo:
var_index_cols.append(cols.geography)
if var_has_product:
var_index_cols.append(cols.product)
# Set index on variable data
var_series = var_data.set_index(var_index_cols)[cols.variable_value]
# Case 1: Same dimensions - direct reindex. Return WITH NaN for missing
# cells; the caller fills (per fill_missing_*) and surfaces how much was
# filled. (Previously this hardcoded fillna(0), which both hid the gaps
# AND made fill_missing_media/_controls ineffective.)
if set(var_dims) == set(target_dims):
return var_series.reindex(target_index)
# Case 2: Disaggregation needed - build full cross-product expansion
expand_geos = target_has_geo and not var_has_geo
expand_products = target_has_product and not var_has_product
if expand_geos or expand_products:
# Get the periods from original data
if isinstance(var_series.index, pd.MultiIndex):
periods = var_series.index.get_level_values(cols.period).unique()
else:
periods = var_series.index.unique()
# Build expansion dimensions and weights
expand_dims = []
expand_weights = []
if expand_geos:
geo_levels = (
target_index.get_level_values(cols.geography).unique().tolist()
)
geo_weights = self._get_allocation_weights(
DimensionType.GEOGRAPHY, geo_levels
)
expand_dims.append((cols.geography, geo_levels))
expand_weights.append(geo_weights)
if expand_products:
product_levels = (
target_index.get_level_values(cols.product).unique().tolist()
)
product_weights = self._get_allocation_weights(
DimensionType.PRODUCT, product_levels
)
expand_dims.append((cols.product, product_levels))
expand_weights.append(product_weights)
# Create expanded data
expanded_records = []
for period in periods:
# Get base value for this period
if isinstance(var_series.index, pd.MultiIndex):
base_val = var_series.xs(period, level=cols.period)
if isinstance(base_val, pd.Series):
base_val = base_val.iloc[0]
else:
base_val = var_series.loc[period]
# Generate all combinations
if len(expand_dims) == 1:
dim_name, levels = expand_dims[0]
weights = expand_weights[0]
for level in levels:
record = {cols.period: period, dim_name: level}
record["value"] = base_val * weights[level]
expanded_records.append(record)
elif len(expand_dims) == 2:
dim1_name, levels1 = expand_dims[0]
dim2_name, levels2 = expand_dims[1]
weights1, weights2 = expand_weights
for level1 in levels1:
for level2 in levels2:
record = {
cols.period: period,
dim1_name: level1,
dim2_name: level2,
}
record["value"] = (
base_val * weights1[level1] * weights2[level2]
)
expanded_records.append(record)
# Convert to series with proper index
expanded_df = pd.DataFrame(expanded_records)
# Build index in same order as target
index_cols = []
for name in target_index.names:
if name in expanded_df.columns:
index_cols.append(name)
var_series = expanded_df.set_index(index_cols)["value"]
# Case 3: Aggregation needed (less common)
if var_has_geo and not target_has_geo:
# Aggregate across geo
var_series = var_series.groupby(level=cols.period).sum()
if var_has_product and not target_has_product:
# Aggregate across product
group_levels = [n for n in var_series.index.names if n != cols.product]
var_series = var_series.groupby(level=group_levels).sum()
# Final reindex to target. Return WITH NaN for missing cells; the caller
# fills (per fill_missing_*) and surfaces how much was filled.
return var_series.reindex(target_index)
[docs]
def build_panel(self) -> PanelDataset:
"""
Build complete panel dataset from loaded MFF data.
Returns
-------
PanelDataset
Structured data ready for modeling.
"""
if self._raw_data is None:
raise RuntimeError("No data loaded. Call load() first.")
cols = self.config.columns
# 1. Extract and build KPI (target) index
kpi_data = self._extract_variable(self.config.kpi)
# Build index based on KPI dimensions
index_cols = [cols.period]
if self.config.kpi.has_geo:
index_cols.append(cols.geography)
if self.config.kpi.has_product:
index_cols.append(cols.product)
kpi_data = kpi_data.sort_values(index_cols)
if len(index_cols) > 1:
target_index = pd.MultiIndex.from_frame(kpi_data[index_cols])
else:
target_index = pd.DatetimeIndex(kpi_data[cols.period])
y = pd.Series(
kpi_data[cols.variable_value].values,
index=target_index,
name=self.config.kpi.name,
)
# Apply log transform if multiplicative
if self.config.kpi.log_transform:
y = np.log(y.clip(lower=self.config.kpi.floor_value))
# 2. Build coordinates
periods = kpi_data[cols.period].unique()
periods = pd.DatetimeIndex(sorted(periods))
# Surface a non-contiguous KPI series: gaps silently shorten the series
# and distort adstock/trend. The fit-path EDA gate blocks on this, but
# warn at load time too so direct load_mff users are not caught out.
if len(periods) >= 3:
try:
expected = generate_complete_date_range(
periods.min(), periods.max(), self.config.frequency
)
missing_periods = expected.difference(periods)
if len(missing_periods):
sample = [str(d.date()) for d in missing_periods[:3]]
warnings.warn(
f"KPI '{self.config.kpi.name}' has {len(missing_periods)} "
f"missing period(s) at frequency '{self.config.frequency}' "
f"(e.g. {sample}); the series is non-contiguous, which "
"distorts adstock carryover and trend estimation."
)
except Exception: # noqa: BLE001 - contiguity check is best-effort
pass
geos = None
if self.config.kpi.has_geo:
geos = sorted(kpi_data[cols.geography].unique().tolist())
products = None
if self.config.kpi.has_product:
products = sorted(kpi_data[cols.product].unique().tolist())
coords = PanelCoordinates(
periods=periods,
geographies=geos,
products=products,
channels=self.config.media_names,
controls=self.config.control_names,
)
# 3. Build media matrix
media_series = {}
media_stats = {}
for media_config in self.config.media_channels:
var_data = self._extract_variable(media_config)
aligned = self._align_to_target_dimensions(
var_data, media_config, target_index
)
# Surface (don't silently swallow) gap-filling: count the cells that
# were missing in the source and are about to be filled.
n_missing = int(aligned.isna().sum())
n_total = max(len(aligned), 1)
aligned = aligned.fillna(self.config.fill_missing_media)
if n_missing:
warnings.warn(
f"Media '{media_config.name}': {n_missing} of {n_total} cell(s) "
f"({n_missing / n_total * 100:.1f}%) were missing and filled with "
f"fill_missing_media={self.config.fill_missing_media}. Treating "
"absent spend rows as zero affects adstock carryover — confirm "
"they are truly zero-spend periods."
)
media_series[media_config.name] = aligned
media_stats[media_config.name] = {
"total": float(aligned.sum()),
"mean": float(aligned.mean()),
"std": float(aligned.std()),
"nonzero_pct": float((aligned > 0).mean()),
"filled_count": n_missing,
"filled_pct": float(n_missing / n_total),
}
X_media = pd.DataFrame(media_series)
X_media.index = target_index
# 3b. Build external spend series for impression/click channels that
# declare a separate ``spend_column`` (impression-level ROI, option a).
spend_raw = self._build_spend_raw(target_index)
# 4. Build control matrix
X_controls = None
if self.config.controls:
control_series = {}
for control_config in self.config.controls:
var_data = self._extract_variable(control_config)
aligned = self._align_to_target_dimensions(
var_data, control_config, target_index
)
# Fill missing controls (surfacing how much was filled)
n_missing = int(aligned.isna().sum())
if self.config.fill_missing_controls is not None:
aligned = aligned.fillna(self.config.fill_missing_controls)
how = f"fill value {self.config.fill_missing_controls}"
else:
aligned = aligned.ffill().bfill()
how = "forward/backward fill"
if n_missing:
warnings.warn(
f"Control '{control_config.name}': {n_missing} missing "
f"cell(s) filled via {how}."
)
control_series[control_config.name] = aligned
X_controls = pd.DataFrame(control_series)
X_controls.index = target_index
# 5. Construct panel dataset
panel = PanelDataset(
y=y,
X_media=X_media,
X_controls=X_controls,
coords=coords,
index=target_index,
config=self.config,
media_stats=media_stats,
spend_raw=spend_raw,
)
return panel
def _build_spend_raw(
self, target_index: pd.MultiIndex | pd.DatetimeIndex
) -> dict[str, NDArray] | None:
"""Aligned per-obs dollar-spend series for channels declaring a
``spend_column`` (impression-level ROI, option a).
Returns ``None`` when no channel uses a separate spend column. A missing
spend variable is surfaced as a warning (not a hard failure) — the ROI
resolver then degrades that channel to per-volume efficiency.
"""
spend_channels = [
m
for m in self.config.media_channels
if getattr(m, "spend_column", None) is not None
]
if not spend_channels:
return None
spend_raw: dict[str, NDArray] = {}
for media_config in spend_channels:
spend_cfg = VariableConfig(
name=media_config.spend_column,
role=VariableRole.MEDIA,
dimensions=list(media_config.dimensions),
)
try:
var_data = self._extract_variable(spend_cfg)
except MFFValidationError:
warnings.warn(
f"Channel '{media_config.name}' declares spend_column="
f"{media_config.spend_column!r} but that variable is absent from "
"the data; ROI will fall back to per-volume efficiency.",
stacklevel=2,
)
continue
aligned = self._align_to_target_dimensions(
var_data, spend_cfg, target_index
).fillna(0.0)
spend_raw[media_config.name] = np.asarray(aligned.values, dtype=np.float64)
return spend_raw or None
# =============================================================================
# Convenience functions
# =============================================================================
[docs]
def load_mff(
data: pd.DataFrame | str,
config: MFFConfig,
geo_weights: pd.Series | dict | None = None,
product_weights: pd.Series | dict | None = None,
) -> PanelDataset:
"""
Convenience function to load MFF data in one call.
Parameters
----------
data : pd.DataFrame or str
MFF data or path to file.
config : MFFConfig
Configuration specifying variables and dimensions.
geo_weights : pd.Series or dict, optional
Custom weights for geo allocation.
product_weights : pd.Series or dict, optional
Custom weights for product allocation.
Returns
-------
PanelDataset
Ready-to-use panel data.
Examples
--------
>>> config = create_simple_mff_config(
... kpi_name="Sales",
... media_names=["TV", "Digital", "Social"],
... control_names=["Price", "Distribution"],
... kpi_dimensions=[DimensionType.PERIOD, DimensionType.GEOGRAPHY],
... )
>>> panel = load_mff("data.csv", config)
>>> print(panel.summary())
"""
loader = MFFLoader(config)
loader.load(data)
if geo_weights is not None:
loader.set_allocation_weights(DimensionType.GEOGRAPHY, geo_weights)
if product_weights is not None:
loader.set_allocation_weights(DimensionType.PRODUCT, product_weights)
return loader.build_panel()
# =============================================================================
# Ragged data handling utilities
# =============================================================================
[docs]
def generate_complete_date_range(
start_date: pd.Timestamp,
end_date: pd.Timestamp,
frequency: str,
) -> pd.DatetimeIndex:
"""
Generate a complete date range for the given frequency.
Parameters
----------
start_date : pd.Timestamp
First date in range.
end_date : pd.Timestamp
Last date in range.
frequency : str
'W' for weekly, 'D' for daily, 'M' for monthly.
Returns
-------
pd.DatetimeIndex
Complete date range with no gaps.
"""
freq_map = {
"D": "D",
"M": "MS", # Month start
}
if frequency == "W":
# Anchor the weekly grid to the data's OWN weekday. Marketing panels are
# commonly Monday-anchored; a fixed "W-SUN" grid would land on Sundays and
# flag every Monday observation as "missing". Anchoring to start_date's
# weekday makes contiguous weekly data (on any weekday) validate cleanly,
# while genuine gaps in that grid are still detected.
anchor = ["MON", "TUE", "WED", "THU", "FRI", "SAT", "SUN"][start_date.weekday()]
pd_freq = f"W-{anchor}"
else:
pd_freq = freq_map.get(frequency, frequency)
return pd.date_range(start=start_date, end=end_date, freq=pd_freq)
[docs]
def build_complete_index(
periods: pd.DatetimeIndex,
geographies: list[str] | None = None,
products: list[str] | None = None,
period_col: str = "Period",
geo_col: str = "Geography",
product_col: str = "Product",
) -> pd.Index | pd.MultiIndex:
"""
Build a complete index covering all combinations of dimensions.
This ensures we have entries for every date x geo x product combination,
enabling proper detection of missing vs explicit NaN values.
"""
if geographies is None and products is None:
return pd.DatetimeIndex(periods, name=period_col)
index_parts = [(period_col, periods)]
if geographies is not None:
index_parts.append((geo_col, geographies))
if products is not None:
index_parts.append((product_col, products))
# Create MultiIndex from product of all dimensions
names = [name for name, _ in index_parts]
arrays = [values for _, values in index_parts]
return pd.MultiIndex.from_product(arrays, names=names)
# =============================================================================
# Enhanced MFF Loader
# =============================================================================
[docs]
class RaggedMFFLoader:
"""
Loads MFF data with proper handling of ragged/sparse data.
Key behaviors:
- Generates complete date range based on min/max dates in data
- Missing date/dimension combinations filled with configured value (default 0)
- Explicit NaN values in source data are preserved
- Tracks which values were explicitly NaN for downstream handling
"""
[docs]
def __init__(self, config: MFFConfig):
self.config = config
self._raw_data: pd.DataFrame | None = None
self._allocation_weights: dict[str, pd.Series] = {}
[docs]
def load(self, data: pd.DataFrame | str) -> RaggedMFFLoader:
"""Load MFF data from DataFrame or file path."""
if isinstance(data, str):
if data.endswith(".parquet"):
self._raw_data = pd.read_parquet(data)
else:
self._raw_data = pd.read_csv(data)
else:
self._raw_data = data.copy()
# Parse dates
cols = self.config.columns
self._raw_data[cols.period] = pd.to_datetime(
self._raw_data[cols.period], format=self.config.date_format
)
return self
def _build_complete_periods(self) -> pd.DatetimeIndex:
"""Build complete date range from min to max date in data."""
cols = self.config.columns
min_date = self._raw_data[cols.period].min()
max_date = self._raw_data[cols.period].max()
return generate_complete_date_range(min_date, max_date, self.config.frequency)
def _get_dimension_levels(self, var_config: VariableConfig) -> dict[str, list]:
"""Get all unique levels for each dimension from the KPI data."""
cols = self.config.columns
levels = {}
# Get KPI data to determine dimension levels
kpi_data = self._raw_data[
self._raw_data[cols.variable_name] == self.config.kpi.name
]
if var_config.has_geo:
levels[cols.geography] = sorted(
kpi_data[cols.geography].dropna().unique().tolist()
)
if var_config.has_product:
levels[cols.product] = sorted(
kpi_data[cols.product].dropna().unique().tolist()
)
return levels
[docs]
def build_panel(self) -> PanelDataset:
"""
Build complete panel dataset with ragged data handling.
Missing dates/combinations are filled with zeros (or configured value).
Explicit NaN values in source data are preserved.
"""
if self._raw_data is None:
raise RuntimeError("No data loaded. Call load() first.")
cols = self.config.columns
# 1. Build complete date range
complete_periods = self._build_complete_periods()
# 2. Get dimension levels from KPI data
dim_levels = self._get_dimension_levels(self.config.kpi)
geos = dim_levels.get(cols.geography)
products = dim_levels.get(cols.product)
# 3. Build complete target index
target_index = build_complete_index(
periods=complete_periods,
geographies=geos,
products=products,
period_col=cols.period,
geo_col=cols.geography,
product_col=cols.product,
)
# 4. Extract KPI with NaN tracking
y, y_nan_mask = extract_with_nan_tracking(
self._raw_data,
self.config.kpi.name,
cols,
target_index,
fill_value=0.0, # Usually don't want to fill KPI with 0
preserve_explicit_nan=True, # Always preserve NaN for KPI
)
# Warn if KPI has missing values
missing_count = y.isna().sum()
if missing_count > 0:
warnings.warn(
f"KPI '{self.config.kpi.name}' has {missing_count} missing values "
f"({missing_count / len(y) * 100:.1f}% of observations). "
f"Consider checking data completeness."
)
# 5. Build coordinates
coords = PanelCoordinates(
periods=complete_periods,
geographies=geos,
products=products,
channels=self.config.media_names,
controls=self.config.control_names,
)
# 6. Extract media variables
media_series = {}
media_stats = {}
explicit_nan_masks = {"kpi": y_nan_mask}
for media_config in self.config.media_channels:
values, nan_mask = extract_with_nan_tracking(
self._raw_data,
media_config.name,
cols,
target_index,
fill_value=self.config.fill_missing_media,
preserve_explicit_nan=self.config.preserve_explicit_nan,
)
media_series[media_config.name] = values
explicit_nan_masks[media_config.name] = nan_mask
media_stats[media_config.name] = {
"total": float(values.sum()),
"mean": float(values.mean()),
"std": float(values.std()),
"nonzero_pct": float((values > 0).mean()),
"explicit_nan_pct": float(nan_mask.mean()),
"filled_zero_pct": float(
((values == self.config.fill_missing_media) & ~nan_mask).mean()
),
}
X_media = pd.DataFrame(media_series)
X_media.index = target_index
# 7. Extract control variables
X_controls = None
if self.config.controls:
control_series = {}
for control_config in self.config.controls:
fill_val = self.config.fill_missing_controls
values, nan_mask = extract_with_nan_tracking(
self._raw_data,
control_config.name,
cols,
target_index,
fill_value=fill_val if fill_val is not None else 0.0,
preserve_explicit_nan=self.config.preserve_explicit_nan,
)
# If fill_missing_controls is None, use forward fill
if fill_val is None:
# Only forward fill non-explicit-NaN positions
if self.config.preserve_explicit_nan:
# Complex: need to ffill only missing, not explicit NaN
filled = values.ffill().bfill()
values = values.where(nan_mask, filled)
else:
values = values.ffill().bfill()
control_series[control_config.name] = values
explicit_nan_masks[control_config.name] = nan_mask
X_controls = pd.DataFrame(control_series)
X_controls.index = target_index
# 8. Build panel dataset
panel = PanelDataset(
y=y,
X_media=X_media,
X_controls=X_controls,
coords=coords,
index=target_index,
config=self.config,
media_stats=media_stats,
explicit_nan_mask=explicit_nan_masks,
)
return panel
# =============================================================================
# Convenience function
# =============================================================================
[docs]
def load_ragged_mff(
data: pd.DataFrame | str,
config: MFFConfig,
) -> PanelDataset:
"""
Load MFF data with ragged data handling.
Parameters
----------
data : pd.DataFrame or str
MFF data or path to file.
config : MFFConfig
Configuration specifying variables and dimensions.
Returns
-------
PanelDataset
Panel data with missing dates filled and explicit NaN preserved.
Examples
--------
>>> config = MFFConfig(
... kpi=VariableConfig(name="Sales", dimensions=[DimensionType.PERIOD]),
... media_channels=[
... VariableConfig(name="TV", dimensions=[DimensionType.PERIOD]),
... VariableConfig(name="Digital", dimensions=[DimensionType.PERIOD]),
... ],
... fill_missing_media=0.0,
... preserve_explicit_nan=True,
... )
>>> panel = load_ragged_mff("ragged_data.csv", config)
>>> print(f"Filled {(panel.X_media == 0).sum().sum()} missing values with 0")
"""
loader = RaggedMFFLoader(config)
loader.load(data)
return loader.build_panel()