Python API Reference
Complete reference for all functions, classes, and contracts in the spotlift open-core library.
Attribution Core
spotlift.analyze_spots()
The primary high-frequency linear TV attribution function.
spotlift.analyze_spots(
airings: pd.DataFrame,
web_telemetry: pd.DataFrame,
window_minutes: int = 8,
pre_window_minutes: int = 15,
baseline_days: int = 21,
clip_negative: bool = False,
apply_cr_model: bool = True,
conversions_source: str | Sequence[str] = "ga4_conversions",
cr_window_hours: int = 1,
baseline_fn: Optional[Callable] = None,
collision_attribution: str = "weighted_share",
) -> pd.DataFrameParameters
airings(pd.DataFrame): Broadcast airings schedule conforming toAiringLogSchema(columns:spot_id,timestamp,channel,creative_id,cost,grp).web_telemetry(pd.DataFrame): Minute-level web visits stream conforming toTelemetryStreamSchema(columns:timestamp,value,channel_source).window_minutes(int, default8): Half-open response window durationin minutes. baseline_fn(Callable, optional): Custom baseline estimator conforming toBaselineFn. Defaults tocompute_diurnal_baseline. Passsimple_mean_baselinefor short-history campaigns.collision_attribution(str, default"weighted_share"): Attribution algorithm for overlapping spots ("weighted_share"or"naive_proportional").apply_cr_model(bool, defaultTrue): IfTrueand conversion events are present in telemetry, bridges session lift to key events via local hourly conversion rates.conversions_source(str | Sequence[str], default"ga4_conversions"): Label or list of custom GA4 sales event names inweb_telemetry["channel_source"](e.g.,"key_event_subscription_purchase"or["key_event_survey_submitted", "key_event_subscription_purchase"]). When a list is provided, appendscr_<event>andresponse_<event>columns for each event.cr_window_hours(int, default1): Half-width of the hourly neighborhood for conversion rate calculation.clip_negative(bool, defaultFalse): IfFalse, preserves signed lift to avoid upward rectification noise bias. IfTrue, clamps negative lift at0.0.
Returns
pd.DataFrame: Enriched DataFrame with spot-level attribution metrics (baseline_volume,observed_volume,incremental_lift,cpiv,conversion_rate,key_event_response,collision_cluster_id,collision_method).
spotlift.optimize_budget()
Solves the constrained portfolio budget optimization problem using diminishing returns response curves (
spotlift.optimize_budget(
df: pd.DataFrame,
*,
total_budget: Optional[float] = None,
channel_col: str = "channel",
cost_col: str = "cost",
lift_col: str = "incremental_lift",
max_shift: float = 0.35,
elasticity: float = 0.70,
elasticity_map: Optional[Mapping[str, float]] = None,
bounds: Optional[Mapping[str, tuple[float, float]]] = None,
group_constraints: Optional[Mapping[str, Any]] = None,
compute_scenarios: bool = True,
mmm_results: Any = None,
risk_aversion: float = 0.0,
n_posterior_draws: int = 300,
) -> BudgetOptimizationResultParameters
df(pd.DataFrame): DataFrame of attributed TV spots (output ofanalyze_spots).total_budget(float, optional): Target total budget. IfNone, defaults to current flight spend (budget-neutral).channel_col(str, default"channel"): Column identifying media channel or station.cost_col(str, default"cost"): Column containing spot media cost.lift_col(str, default"incremental_lift"): Column containing attributed incremental visitor lift.max_shift(float, default0.35): Maximum relative budget reallocation allowed per channel (). elasticity(float, default0.70): Diminishing returns power-law exponent( ). elasticity_map(Mapping[str, float], optional): Channel-specific elasticity overrides.bounds(Mapping[str, tuple[float, float]], optional): Explicit(min_spend, max_spend)bounds per channel.group_constraints(Mapping[str, Any], optional): Broadcaster group share commitments (e.g.{"Ad Alliance": {"channels": ["RTL", "VOX"], "min_share": 0.45}}).compute_scenarios(bool, defaultTrue): IfTrue, computes Conservative (), Balanced ( ), and Aggressive ( ) scenarios. mmm_results(Any, optional): Optional macro MMM results (float total, DataFrame, CSV/Parquet/JSON path, dict/list, orMMMResultsInput). When provided, delegates tooptimize_tv_with_mmm()to anchor macro scale, elasticity, and posterior uncertainty.risk_aversion(float, default0.0): Posterior uncertainty penalty weight used whenmmm_resultsis provided.n_posterior_draws(int, default300): Number of posterior draws used whenmmm_resultsis provided.
Returns
BudgetOptimizationResult: Container holdingrecommendationsDataFrame, current and optimized totals, percentage gains,scenariosdict, MMM metadata (mmm_integrated,mmm_calibrated_beta,implied_mmm_multiplier,mmm_warnings),print_summary(), andsummary_dict().
spotlift.export_causal_lower_bound()
Exports SpotLift 0–8 minute attribution as period-aligned causal lower-bound calibration evidence (CalibrationEvidence) with cluster-block bootstrap + Empirical-Bayes pooled standard errors and explicit provenance (carryover_mode="none" by default; does not scale immediate lift by
spotlift.export_causal_lower_bound(
df: pd.DataFrame,
*,
mmm_channel_name: Optional[str] = None,
channel_col: str = "channel",
timestamp_col: str = "timestamp",
cost_col: str = "cost",
lift_col: str = "incremental_lift",
freq: Optional[str] = "W-MON",
elasticity: float = 0.70,
elasticity_map: Optional[Mapping[str, float]] = None,
outcome_units: str = "sessions",
window_minutes: int = 8,
overlap_with_mmm_outcome: str = "unknown",
discrepancy_rel: float = 0.10,
n_boot: int = 300,
random_state: int = 42,
) -> CalibrationEvidenceReturns
CalibrationEvidence: Structured evidence bundle whose.dataDataFrame contains['channel', 'period', 'period_start', 'period_end', 'spend', 'direct_lift_floor', 'sigma', 'lower_95_floor', 'n_spots', 'n_blocks', 'evidence_class', 'estimand', 'adstock_scaled']alongside.to_metadata_dict()provenance.
spotlift.attach_pymc_lower_bound() & spotlift.generate_pymc_lower_bound_snippet()
Attaches SpotLift CalibrationEvidence to a PyMC / PyMC-Marketing model without assuming equality between immediate 8-minute lift and total long-term MMM contribution:
spotlift.attach_pymc_lower_bound(
model: Any,
evidence: CalibrationEvidence,
*,
tv_contribution_tensor: Any,
mode: Literal["lower_bound", "latent_share"] = "lower_bound",
rho_prior: tuple[float, float] = (2.0, 3.0),
) -> Anymode="lower_bound": Adds apm.Potentialimposing an asymmetric soft HalfNormal penalty only when modeled MMM TV contribution falls belowdirect_lift_floor(modulo block-bootstrapsigma).mode="latent_share": Introduces a learned immediate-share parameterand links direct_lift_floorto.
spotlift.export_pymc_lift_test()
Exports an exploratory lift-test table ([channel, x, delta_x, delta_y, sigma], or CalibrationEvidence when return_evidence=True) aligned to MMM time periods using cluster-block bootstrap + EB standard errors and explicit discrepancy_rel.
spotlift.export_pymc_lift_test(
df: pd.DataFrame,
*,
mmm_channel_name: Optional[str] = None,
channel_col: str = "channel",
timestamp_col: str = "timestamp",
cost_col: str = "cost",
lift_col: str = "incremental_lift",
freq: Optional[str] = "W-MON",
elasticity: float = 0.70,
elasticity_map: Optional[Mapping[str, float]] = None,
adstock_retention: float = 0.0,
carryover_mode: Optional[str] = None,
horizon_periods: Optional[int] = None,
min_sigma_rel: float = 0.05,
discrepancy_rel: float = 0.10,
overlap_with_mmm_outcome: str = "unknown",
return_evidence: bool = False,
) -> Union[pd.DataFrame, CalibrationEvidence]Parameters
freq(str, optional, default"W-MON"): Pandas period offset matching the target MMM time grain ("W-MON"for weekly,"D"for daily, orNonefor full-flight total).adstock_retention(float, default0.0): Optional geometric adstock retention. Requires explicit carryover_mode("finite_horizon"or"infinite_horizon") to enable carryover scaling; by default (carryover_mode=None) no adstock inflation is applied.overlap_with_mmm_outcome(str, default"unknown"): Outcome provenance relative to the MMM target ("unknown","independent_stream", or"shared_same_series").mmm_channel_name(str, optional): Optional unified channel name (e.g."tv_broadcast") to aggregate stations into a single macro MMM media column.
spotlift.export_mmm_priors()
Exports causal lower-bound evidence, exploratory lift-test records, and support-matched Hill/Robyn priors for Media Mix Modeling.
spotlift.export_mmm_priors(
df: pd.DataFrame,
*,
channel_col: str = "channel",
timestamp_col: str = "timestamp",
cost_col: str = "cost",
lift_col: str = "incremental_lift",
freq: str = "W-MON",
default_elasticity: float = 0.70,
elasticity_map: Optional[Mapping[str, float]] = None,
adstock_retention: float = 0.0,
carryover_mode: Optional[str] = None,
overlap_with_mmm_outcome: str = "unknown",
) -> dict[str, Any]Returns
dict[str, Any]: Dictionary containing:"channels": Channel-level OLS power-law specs and support-matched Hill parameters."causal_lower_bound": Period-aligned causal lower-bound records."pymc_marketing_lift_test": Exploratory period-aligned lift-test records."meta_robyn_hyperparameters": Support-matchedalphasandgammasranges for Meta Robyn."google_meridian_priors": Support-matchedhalf_saturation_priorandslope_priordicts for Google Meridian."provenance": Metadata clarifying observational direct-response evidence class and carryover mode.
spotlift.load_mmm_results() & spotlift.optimize_tv_with_mmm()
Loads macro MMM TV contribution outputs and runs Direction 2 (MMM → SpotLift) portfolio optimization anchored to MMM macro total lift, calibrated elasticity, and posterior draws:
spotlift.load_mmm_results(
source: Union[float, Mapping[str, Any], pd.DataFrame, MMMResultsInput, str, Path],
*,
date_col: str = "date",
contribution_col: str = "contribution",
lower_col: Optional[str] = "lower",
upper_col: Optional[str] = "upper",
spend_col: Optional[str] = "spend",
adstock_retention: float = 0.0,
elasticity: Optional[float] = None,
freq: str = "W-MON",
overlap_with_spots_outcome: str = "unknown",
) -> MMMResultsInput
spotlift.optimize_tv_with_mmm(
spots_df: pd.DataFrame,
mmm_results: Union[float, Mapping[str, Any], pd.DataFrame, MMMResultsInput, str, Path],
*,
by: str = "channel",
total_budget: Optional[float] = None,
timestamp_col: str = "timestamp",
cost_col: str = "cost",
lift_col: str = "incremental_lift",
max_shift: float = 0.35,
default_elasticity: float = 0.70,
bounds: Optional[Mapping[str, tuple[float, float]]] = None,
group_constraints: Optional[Mapping[str, Any]] = None,
risk_aversion: float = 0.0,
n_posterior_draws: int = 300,
compute_scenarios: bool = True,
random_state: int = 42,
) -> BudgetOptimizationResultReturns
BudgetOptimizationResult: Includesrecommendationswithdirect_lift_floor,projected_direct_lift,projected_lift,projected_lift_p10,projected_lift_p90,low_information_flag, andextrapolation_warning, plusmmm_calibrated_beta,implied_mmm_multiplier, andmmm_warnings.
spotlift.disaggregate_mmm_contributions()
Disaggregates macro MMM TV contributions (and posterior draws / intervals) down to individual TV stations ("channel"), creatives ("creative_id"), or multi-dimensional slices (["channel", "creative_id"]) using Empirical-Bayes shrunk efficiency weights and joint posterior-draw conservation (
spotlift.disaggregate_mmm_contributions(
spots_df: pd.DataFrame,
mmm_contributions: Union[float, pd.DataFrame, MMMResultsInput],
*,
by: Union[str, Sequence[str]] = "channel",
timestamp_col: str = "timestamp",
cost_col: str = "cost",
lift_col: str = "incremental_lift",
date_col: str = "date",
contribution_col: str = "contribution",
lower_col: Optional[str] = "lower",
upper_col: Optional[str] = "upper",
freq: str = "W-MON",
adstock_retention: Optional[float] = None,
return_period_grain: bool = False,
n_draws: int = 400,
random_state: int = 42,
) -> pd.DataFrameReturns
pd.DataFrame: Decomposed table withspend,direct_lift_floor,allocated_mmm_contribution,allocated_lower,allocated_upper,reconciliation_residual,share_of_spend,share_of_mmm,implied_multiplier,direct_cpiv,mmm_cpiv, andlow_information_flag.
spotlift.export_meridian_package() (Pro)
Exports Google Meridian interchange calibration data (calibration_data.csv) and custom interchange config (meridian_model_spec.yaml under root key interchange_spec) with support-matched Hill priors and explicit observational_direct_response provenance.
spotlift.export_meridian_package(
spots_df: pd.DataFrame,
*,
api_key: Optional[str] = None,
channel_col: str = "channel",
timestamp_col: str = "timestamp",
cost_col: str = "cost",
lift_col: str = "incremental_lift",
freq: str = "W-MON",
elasticity: float = 0.70,
adstock_retention: float = 0.0,
carryover_mode: Optional[str] = None,
mmm_channel_name: Optional[str] = None,
min_sigma_rel: float = 0.05,
discrepancy_rel: float = 0.10,
overlap_with_mmm_outcome: str = "unknown",
output_dir: Optional[Union[str, Path]] = None,
) -> dict[str, Any]spotlift.export_robyn_package() (Pro)
Exports Meta Robyn calibration table (calibration_input.csv) and exploratory R execution script (run_robyn_calibration.R) with support-matched hyperparameter ranges (alphas, gammas, thetas). confidence is left as missing (NA) unless an independently measured experimental_confidence in (0, 1] is provided.
spotlift.export_robyn_package(
spots_df: pd.DataFrame,
*,
api_key: Optional[str] = None,
channel_col: str = "channel",
timestamp_col: str = "timestamp",
cost_col: str = "cost",
lift_col: str = "incremental_lift",
metric_name: str = "conversions",
freq: str = "W-MON",
elasticity: float = 0.70,
adstock_retention: float = 0.0,
carryover_mode: Optional[str] = None,
experimental_confidence: Optional[float] = None,
mmm_channel_name: Optional[str] = None,
overlap_with_mmm_outcome: str = "unknown",
output_dir: Optional[Union[str, Path]] = None,
) -> dict[str, Any]Data Ingestion & Loaders
spotlift.load_airings()
Loads and validates a broadcast airings log from a CSV file path, buffer, or DataFrame.
spotlift.load_airings(filepath_or_df: Union[str, Path, pd.DataFrame]) -> pd.DataFrame- Validates required columns (
spot_id,timestamp,channel,creative_id,cost). - Parses and normalizes timestamps with explicit timezone handling.
- Fills missing
grpwith0.0.
spotlift.load_telemetry()
Loads and validates minute-level digital telemetry.
spotlift.load_telemetry(
filepath_or_df: Union[str, Path, pd.DataFrame],
zero_fill_gaps: bool = False,
) -> pd.DataFrame- Verifies strict monotonic minute increments.
- Checks for missing minutes. If
zero_fill_gaps=True, fills missing timestamps with0.0; otherwise raisesTelemetryGapError.
Baseline Estimators
spotlift.compute_diurnal_baseline()
Calculates the decay-weighted cyclic rolling baseline.
spotlift.compute_diurnal_baseline(
telemetry_series: pd.Series,
spot_timestamps: list[pd.Timestamp],
window_minutes: int = 6,
) -> tuple[pd.Series, dict[pd.Timestamp, float]]Returns a tuple of (calibrated_baseline_series, scaling_factors_dict).
spotlift.simple_mean_baseline()
Fallback baseline computing the pre-spot moving average for short-history flights without multi-week diurnal history.
spotlift.simple_mean_baseline(
telemetry_series: pd.Series,
spot_timestamps: list[pd.Timestamp],
window_minutes: int = 6,
pre_window_minutes: int = 15,
) -> pd.SeriesCollision Mathematics
spotlift.compute_continuous_weighted_share()
Analytical exponential-decay weighted share attribution for overlapping broadcast collision clusters.
spotlift.compute_continuous_weighted_share(
cluster_df: pd.DataFrame,
telemetry_series: pd.Series,
baseline_series: pd.Series,
window_minutes: int = 6,
half_life_minutes: float = 1.5,
) -> dict[Any, float]Returns a dictionary mapping each spot_id to its attributed incremental lift.
Conversion Rate Bridge
spotlift.estimate_conversion_rate_for_spot()
Calculates the volume-weighted conversion rate for a spot based on the surrounding hourly window.
spotlift.estimate_conversion_rate_for_spot(
spot_timestamp: pd.Timestamp,
telemetry_df: pd.DataFrame,
conversions_source: str = "ga4_conversions",
window_hours: int = 1,
) -> floatSeamlessly crosses midnight boundaries using continuous timestamp slices.
spotlift.apply_conversion_model()
Enriches an attributed spots DataFrame with conversion_rate and key_event_response (plus cr_<event> and response_<event> columns when conversions_source is a sequence of multiple custom GA4 events).
spotlift.apply_conversion_model(
spots_df: pd.DataFrame,
web_telemetry: pd.DataFrame,
sessions_source: str = "ga4_direct_organic_sessions",
conversions_source: str | Sequence[str] = "ga4_conversions",
cr_window_hours: int = 1,
) -> pd.DataFrameClean Room SQL Generators
spotlift.generate_ga4_bigquery_sql()
Generates a production Google BigQuery SQL query extracting minute-level direct/organic sessions and conversion events (supports single or multiple custom GA4 key events).
spotlift.generate_ga4_bigquery_sql(
project_id: str,
dataset_id: str,
start_date: str,
end_date: str,
conversion_event_name: str | Sequence[str] = "purchase",
include_paid_search: bool = False,
) -> strspotlift.generate_amc_telemetry_sql()
Generates an Amazon Marketing Cloud (AMC) clean room SQL query extracting hourly ordered units and sales.
spotlift.generate_amc_telemetry_sql(
start_date: str,
end_date: str,
) -> strSynthetic Benchmark Dataset Generators
spotlift.generate_dach_benchmark_dataset()
Generates a 14-day DACH TV campaign with 250 spots, GA4 minute telemetry, and hourly Amazon telemetry.
spotlift.generate_dach_benchmark_dataset(
n_days: int = 14,
n_spots: int = 250,
seed: int = 42,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]spotlift.generate_omnichannel_amc_dataset()
Generates an omnichannel campaign with correlated GA4 web telemetry and dual-format Amazon AMC telemetry (amc_spot_conversions with pageViewStarts, addToCart, purchases, totalProductSales across pre/post windows + amazon_hourly_telemetry).
spotlift.generate_omnichannel_amc_dataset(
n_days: int = 14,
n_spots: int = 180,
seed: int = 101,
amc_window_minutes: int = 10,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame]spotlift.generate_us_state_dataset()
Generates a US regional state-targeted TV campaign (TX, FL, PA active states vs. CA, NY, IL zero-spot holdout control states + US_NATIONAL aggregate).
spotlift.generate_us_state_dataset(
n_days: int = 14,
n_spots: int = 130,
seed: int = 202,
active_states: tuple[str, ...] = ("TX", "FL", "PA"),
control_states: tuple[str, ...] = ("CA", "NY", "IL"),
) -> tuple[pd.DataFrame, pd.DataFrame]SpotLift Pro Dispatch
spotlift.install_pro()
Validates a commercial license key and unlocks Pro capabilities.
spotlift.install_pro(
api_key: str,
index_url: Optional[str] = None,
local_path: Optional[str] = None,
) -> boolspotlift.analyze_omnichannel_pro()
Pro entry point dispatches to compiled Cython binaries for Shifted-Gamma EM deconvolution, AMC retail halo, and Excel export.
spotlift.analyze_omnichannel_pro(
airings: pd.DataFrame,
web_telemetry: pd.DataFrame,
amazon_telemetry: Optional[pd.DataFrame] = None,
window_minutes: int = 8,
export_html: Optional[str] = None,
export_excel: Optional[str] = None,
white_label_config: Optional[dict] = None,
clip_negative: bool = False,
apply_cr_model: bool = True,
conversions_source: str | Sequence[str] = "ga4_conversions",
cr_window_hours: int = 1,
) -> dictspotlift_pro.amc_ingestion.analyze_amc_spot_conversions()
Analyzes pre/post windowed Amazon Marketing Cloud (AMC) clean room tables (pageViewStarts, addToCart, purchases, totalProductSales, newToBrandPurchases).
spotlift_pro.amc_ingestion.analyze_amc_spot_conversions(
amc_df: pd.DataFrame,
airings_df: Optional[pd.DataFrame] = None,
clip_negative: bool = False,
) -> dict[str, Any]