import numpy as np
from scipy import stats
[docs]
def normalize_experiments_list(experiments, default_channel="Generic"):
"""Normalizes various single-channel experiment input formats into a list of dicts."""
if experiments is None:
return []
if isinstance(experiments, dict):
if "spend" in experiments or "raw_spend" in experiments or "adstocked_spend" in experiments or "lift" in experiments:
return [experiments]
if default_channel in experiments:
val = experiments[default_channel]
return [val] if isinstance(val, dict) else list(val)
# Check if single key dictionary with experiment
first_val = next(iter(experiments.values()))
if isinstance(first_val, dict) and ("spend" in first_val or "lift" in first_val):
return [first_val]
elif isinstance(first_val, (list, tuple)):
return list(first_val)
return [experiments]
elif isinstance(experiments, (list, tuple)):
return list(experiments)
else:
raise TypeError(f"experiments must be a dict or list of dicts, got {type(experiments)}")
[docs]
def normalize_multichannel_experiments(multichannel_model, experiments=None):
"""Normalizes multi-channel experiments into a structured dict of {channel_name: list[exp]}."""
channel_exp_map = {}
if experiments is None:
if hasattr(multichannel_model, "calibration_experiments") and multichannel_model.calibration_experiments:
return normalize_multichannel_experiments(multichannel_model, multichannel_model.calibration_experiments)
for ch_name, ch_curve in multichannel_model.channels.items():
if hasattr(ch_curve, "calibration_experiments") and ch_curve.calibration_experiments:
channel_exp_map[ch_name] = list(ch_curve.calibration_experiments)
return channel_exp_map
if isinstance(experiments, dict):
# Check if keyed by channel name (e.g. {"YouTube": [exp1, exp2], "Search": exp3})
matched_keys = [k for k in experiments.keys() if k in multichannel_model.channels]
if matched_keys:
for ch_name in matched_keys:
exp_val = experiments[ch_name]
if isinstance(exp_val, dict):
channel_exp_map.setdefault(ch_name, []).append(dict(exp_val, channel=ch_name))
elif isinstance(exp_val, (list, tuple)):
for single_e in exp_val:
channel_exp_map.setdefault(ch_name, []).append(dict(single_e, channel=ch_name))
return channel_exp_map
else:
exp_list = [experiments]
elif isinstance(experiments, (list, tuple)):
exp_list = list(experiments)
else:
raise TypeError(f"experiments must be a dict or list of dicts, got {type(experiments)}")
for exp in exp_list:
ch = exp.get("channel")
if not ch:
raise KeyError("Each experiment in multi-channel validation must specify a 'channel' key or be keyed by channel name.")
if ch not in multichannel_model.channels:
raise ValueError(f"Channel '{ch}' not found in model channels: {list(multichannel_model.channels.keys())}")
channel_exp_map.setdefault(ch, []).append(exp)
return channel_exp_map
[docs]
def validate_curve_experiments(model, experiments=None, spend_is_raw=True, verbose=False):
"""Evaluates a fitted MarketingReturnCurve against one or more incrementality experiments.
Args:
model: An instance of MarketingReturnCurve.
experiments: A dictionary or list of dictionaries containing experimental results.
If None, uses experiments attached to the model (model.calibration_experiments).
Supported keys per experiment:
- 'spend' (or 'raw_spend', 'adstocked_spend'): Media spend level tested.
- 'lift' (or 'incremental_return', 'conversions'): Incremental response measured.
- 'se' (or 'std_error'): Standard error of the measured lift (optional).
- 'ci' (or 'confidence_interval'): (lower, upper) 95% confidence interval (optional).
- 'name' (or 'channel', 'test_name'): Descriptive label for the test (optional).
spend_is_raw: If True and the curve has theta > 0, converts raw daily test spend to
steady-state effective adstocked spend (S_eff = S_raw / (1 - theta)).
verbose: If True, prints a formatted validation report to stdout.
Returns:
dict: Detailed evaluation metrics including per-experiment errors, Z-scores,
confidence interval coverage, and aggregate goodness-of-fit statistics.
"""
if experiments is None:
experiments = getattr(model, "calibration_experiments", None)
if experiments is None:
raise ValueError(f"No experiments provided or attached to channel model '{model.channel_name}' to validate.")
exp_list = normalize_experiments_list(experiments, default_channel=model.channel_name)
if not exp_list:
raise ValueError(f"No experiments provided or attached to channel model '{model.channel_name}' to validate.")
per_exp_results = []
for idx, exp in enumerate(exp_list):
name = exp.get("name") or exp.get("test_name") or exp.get("channel") or f"Experiment_{idx+1}"
# Extract spend
if "adstocked_spend" in exp:
eval_spend = float(exp["adstocked_spend"])
raw_spend_val = eval_spend * (1.0 - model.theta) if (0.0 < model.theta < 1.0) else eval_spend
elif "raw_spend" in exp:
raw_spend_val = float(exp["raw_spend"])
eval_spend = raw_spend_val / (1.0 - model.theta) if (spend_is_raw and 0.0 < model.theta < 1.0) else raw_spend_val
elif "spend" in exp:
raw_spend_val = float(exp["spend"])
eval_spend = raw_spend_val / (1.0 - model.theta) if (spend_is_raw and 0.0 < model.theta < 1.0) else raw_spend_val
else:
raise KeyError(f"Experiment {name} missing 'spend', 'raw_spend', or 'adstocked_spend'.")
# Extract measured lift
if "lift" in exp:
y_true = float(exp["lift"])
elif "incremental_return" in exp:
y_true = float(exp["incremental_return"])
elif "conversions" in exp:
y_true = float(exp["conversions"])
else:
raise KeyError(f"Experiment {name} missing 'lift', 'incremental_return', or 'conversions'.")
# Model predicted incremental return (excluding organic baseline)
y_pred = float(model.predict_incremental_return(eval_spend, include_baseline=False))
error = y_pred - y_true
abs_error = abs(error)
pct_error = (error / y_true * 100.0) if y_true != 0.0 else 0.0
# Extract standard error & confidence interval
se = exp.get("se") or exp.get("std_error")
ci = exp.get("ci") or exp.get("confidence_interval")
ci_lower, ci_upper = None, None
if ci is not None:
ci_lower = float(ci[0])
ci_upper = float(ci[1])
if se is None:
se = (ci_upper - ci_lower) / (2.0 * 1.96)
elif se is not None:
se = float(se)
ci_lower = y_true - 1.96 * se
ci_upper = y_true + 1.96 * se
if se is not None and se > 0:
z_score = float(error / se)
p_value = float(2.0 * (1.0 - stats.norm.cdf(abs(z_score))))
in_ci = bool(ci_lower <= y_pred <= ci_upper)
else:
z_score = None
p_value = None
in_ci = None
per_exp_results.append({
"name": name,
"spend": raw_spend_val,
"effective_spend": eval_spend,
"measured_lift": y_true,
"predicted_lift": y_pred,
"error": error,
"abs_error": abs_error,
"pct_error": pct_error,
"std_error": se,
"ci": (ci_lower, ci_upper) if ci_lower is not None else None,
"z_score": z_score,
"p_value": p_value,
"in_95_ci": in_ci
})
n_total = len(per_exp_results)
mae = float(np.mean([r["abs_error"] for r in per_exp_results]))
mape = float(np.mean([abs(r["pct_error"]) for r in per_exp_results]))
valid_z = [r["z_score"] for r in per_exp_results if r["z_score"] is not None]
if valid_z:
chi2_stat = float(np.sum([z**2 for z in valid_z]))
dof = len(valid_z)
chi2_reduced = float(chi2_stat / dof)
omnibus_p_value = float(1.0 - stats.chi2.cdf(chi2_stat, df=dof))
n_in_ci = sum(1 for r in per_exp_results if r["in_95_ci"] is True)
coverage_pct = float((n_in_ci / dof) * 100.0)
else:
chi2_stat = None
dof = 0
chi2_reduced = None
omnibus_p_value = None
coverage_pct = None
# Determine validation verdict
if chi2_reduced is not None:
if coverage_pct >= 90.0 and chi2_reduced <= 1.5:
verdict = "EXCELLENT"
elif coverage_pct >= 66.7 and chi2_reduced <= 3.84:
verdict = "ALIGNED"
else:
verdict = "MISALIGNED"
else:
if mape <= 10.0:
verdict = "EXCELLENT"
elif mape <= 25.0:
verdict = "ALIGNED"
else:
verdict = "MISALIGNED"
result = {
"channel": model.channel_name,
"num_experiments": n_total,
"verdict": verdict,
"mae": mae,
"mape": mape,
"chi2": chi2_stat,
"chi2_reduced": chi2_reduced,
"omnibus_p_value": omnibus_p_value,
"ci_coverage_pct": coverage_pct,
"experiments": per_exp_results
}
if verbose:
print(format_validation_report(result))
return result
[docs]
def validate_multichannel_experiments(multichannel_model, experiments=None, spend_is_raw=True, verbose=False):
"""Evaluates a multi-channel model across experiments across different channels in parallel.
Args:
multichannel_model: MultiChannelModel instance or PortfolioAllocator instance.
experiments: List of experiment dictionaries or dict keyed by channel name.
If None, evaluates experiments attached across channels.
spend_is_raw: Whether spends are raw daily spend.
verbose: If True, prints a summary report.
Returns:
dict: Multi-channel validation summary with per-channel breakdown and global metrics.
"""
channel_exp_map = normalize_multichannel_experiments(multichannel_model, experiments)
if not channel_exp_map:
raise ValueError("No experiments provided or attached across channels to validate.")
channel_reports = {}
all_exp_results = []
for ch, exps in channel_exp_map.items():
if ch in multichannel_model.channels:
ch_curve = multichannel_model.channels[ch]
rep = validate_curve_experiments(ch_curve, exps, spend_is_raw=spend_is_raw, verbose=False)
channel_reports[ch] = rep
all_exp_results.extend(rep["experiments"])
if not all_exp_results:
raise ValueError(f"No matching channel experiments found for channels: {list(multichannel_model.channels.keys())}")
n_total = len(all_exp_results)
mae = float(np.mean([r["abs_error"] for r in all_exp_results]))
mape = float(np.mean([abs(r["pct_error"]) for r in all_exp_results]))
valid_z = [r["z_score"] for r in all_exp_results if r["z_score"] is not None]
if valid_z:
chi2_stat = float(np.sum([z**2 for z in valid_z]))
dof = len(valid_z)
chi2_reduced = float(chi2_stat / dof)
omnibus_p_value = float(1.0 - stats.chi2.cdf(chi2_stat, df=dof))
coverage_pct = float((sum(1 for r in all_exp_results if r["in_95_ci"] is True) / dof) * 100.0)
else:
chi2_stat = None
chi2_reduced = None
omnibus_p_value = None
coverage_pct = None
if chi2_reduced is not None:
if coverage_pct >= 90.0 and chi2_reduced <= 1.5:
verdict = "EXCELLENT"
elif coverage_pct >= 66.7 and chi2_reduced <= 3.84:
verdict = "ALIGNED"
else:
verdict = "MISALIGNED"
else:
if mape <= 10.0:
verdict = "EXCELLENT"
elif mape <= 25.0:
verdict = "ALIGNED"
else:
verdict = "MISALIGNED"
result = {
"num_experiments": n_total,
"verdict": verdict,
"mae": mae,
"mape": mape,
"chi2": chi2_stat,
"chi2_reduced": chi2_reduced,
"omnibus_p_value": omnibus_p_value,
"ci_coverage_pct": coverage_pct,
"channels": channel_reports,
"all_experiments": all_exp_results
}
if verbose:
print(format_multichannel_validation_report(result))
return result