Skip to content

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.

python
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.DataFrame

Parameters ​

  • airings (pd.DataFrame): Broadcast airings schedule conforming to AiringLogSchema (columns: spot_id, timestamp, channel, creative_id, cost, grp).
  • web_telemetry (pd.DataFrame): Minute-level web visits stream conforming to TelemetryStreamSchema (columns: timestamp, value, channel_source).
  • window_minutes (int, default 8): Half-open response window duration [t0,t0+W) in minutes.
  • baseline_fn (Callable, optional): Custom baseline estimator conforming to BaselineFn. Defaults to compute_diurnal_baseline. Pass simple_mean_baseline for short-history campaigns.
  • collision_attribution (str, default "weighted_share"): Attribution algorithm for overlapping spots ("weighted_share" or "naive_proportional").
  • apply_cr_model (bool, default True): If True and 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 in web_telemetry["channel_source"] (e.g., "key_event_subscription_purchase" or ["key_event_survey_submitted", "key_event_subscription_purchase"]). When a list is provided, appends cr_<event> and response_<event> columns for each event.
  • cr_window_hours (int, default 1): Half-width of the hourly neighborhood for conversion rate calculation.
  • clip_negative (bool, default False): If False, preserves signed lift to avoid upward rectification noise bias. If True, clamps negative lift at 0.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 (R(s)=αsβ) and marginal CPIV (mCPIV) equalization under boundary conditions.

python
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,
) -> BudgetOptimizationResult

Parameters ​

  • df (pd.DataFrame): DataFrame of attributed TV spots (output of analyze_spots).
  • total_budget (float, optional): Target total budget. If None, 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, default 0.35): Maximum relative budget reallocation allowed per channel (±35%).
  • elasticity (float, default 0.70): Diminishing returns power-law exponent β (0<β<1).
  • 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, default True): If True, computes Conservative (±15%), Balanced (±35%), and Aggressive (±60%) scenarios.
  • mmm_results (Any, optional): Optional macro MMM results (float total, DataFrame, CSV/Parquet/JSON path, dict/list, or MMMResultsInput). When provided, delegates to optimize_tv_with_mmm() to anchor macro scale, elasticity, and posterior uncertainty.
  • risk_aversion (float, default 0.0): Posterior uncertainty penalty weight used when mmm_results is provided.
  • n_posterior_draws (int, default 300): Number of posterior draws used when mmm_results is provided.

Returns ​

  • BudgetOptimizationResult: Container holding recommendations DataFrame, current and optimized totals, percentage gains, scenarios dict, MMM metadata (mmm_integrated, mmm_calibrated_beta, implied_mmm_multiplier, mmm_warnings), print_summary(), and summary_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 11−λ).

python
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,
) -> CalibrationEvidence

Returns ​

  • CalibrationEvidence: Structured evidence bundle whose .data DataFrame 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:

python
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),
) -> Any
  • mode="lower_bound": Adds a pm.Potential imposing an asymmetric soft HalfNormal penalty only when modeled MMM TV contribution falls below direct_lift_floor (modulo block-bootstrap sigma).
  • mode="latent_share": Introduces a learned immediate-share parameter ρ∼Beta(α,β) and links direct_lift_floor to ρ⋅CTV,t.

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.

python
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, or None for full-flight total).
  • adstock_retention (float, default 0.0): Optional geometric adstock retention λ∈[0,1). 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.

python
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-matched alphas and gammas ranges for Meta Robyn.
    • "google_meridian_priors": Support-matched half_saturation_prior and slope_prior dicts 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:

python
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,
) -> BudgetOptimizationResult

Returns ​

  • BudgetOptimizationResult: Includes recommendations with direct_lift_floor, projected_direct_lift, projected_lift, projected_lift_p10, projected_lift_p90, low_information_flag, and extrapolation_warning, plus mmm_calibrated_beta, implied_mmm_multiplier, and mmm_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 (∑gCg,t(m)=CTV,t(m)).

python
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.DataFrame

Returns ​

  • pd.DataFrame: Decomposed table with spend, direct_lift_floor, allocated_mmm_contribution, allocated_lower, allocated_upper, reconciliation_residual, share_of_spend, share_of_mmm, implied_multiplier, direct_cpiv, mmm_cpiv, and low_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.

python
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.

python
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.

python
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 grp with 0.0.

spotlift.load_telemetry() ​

Loads and validates minute-level digital telemetry.

python
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 with 0.0; otherwise raises TelemetryGapError.

Baseline Estimators ​

spotlift.compute_diurnal_baseline() ​

Calculates the decay-weighted cyclic rolling baseline.

python
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.

python
spotlift.simple_mean_baseline(
    telemetry_series: pd.Series,
    spot_timestamps: list[pd.Timestamp],
    window_minutes: int = 6,
    pre_window_minutes: int = 15,
) -> pd.Series

Collision Mathematics ​

spotlift.compute_continuous_weighted_share() ​

Analytical exponential-decay weighted share attribution for overlapping broadcast collision clusters.

python
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.

python
spotlift.estimate_conversion_rate_for_spot(
    spot_timestamp: pd.Timestamp,
    telemetry_df: pd.DataFrame,
    conversions_source: str = "ga4_conversions",
    window_hours: int = 1,
) -> float

Seamlessly 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).

python
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.DataFrame

Clean 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).

python
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,
) -> str

spotlift.generate_amc_telemetry_sql() ​

Generates an Amazon Marketing Cloud (AMC) clean room SQL query extracting hourly ordered units and sales.

python
spotlift.generate_amc_telemetry_sql(
    start_date: str,
    end_date: str,
) -> str

Synthetic 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.

python
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).

python
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).

python
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.

python
spotlift.install_pro(
    api_key: str,
    index_url: Optional[str] = None,
    local_path: Optional[str] = None,
) -> bool

spotlift.analyze_omnichannel_pro() ​

Pro entry point dispatches to compiled Cython binaries for Shifted-Gamma EM deconvolution, AMC retail halo, and Excel export.

python
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,
) -> dict

spotlift_pro.amc_ingestion.analyze_amc_spot_conversions() ​

Analyzes pre/post windowed Amazon Marketing Cloud (AMC) clean room tables (pageViewStarts, addToCart, purchases, totalProductSales, newToBrandPurchases).

python
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]

Released under the MIT License.