AI Ad Budget Cross-Channel Optimization System

AI-оптимизация рекламного бюджета по каналам

Распределение бюджета между Google Ads, Meta, programmatic display, YouTube и другими каналами — задача с нелинейными эффектами насыщения, кросс-канальными взаимодействиями и delayed attribution. Ручная оптимизация даже с хорошей аналитикой теряет 20-40% эффективности по сравнению с математическими моделями.

Media Mix Modeling (MMM)

import pandas as pd
import numpy as np
from scipy.optimize import minimize
from scipy.special import expit

class MediaMixModel:
    """
    Байесовский Media Mix Model для анализа вклада каналов.
    Учитывает saturation (убывающая отдача) и adstock (остаточный эффект).
    """

    def adstock_transform(self, spend: np.ndarray,
                           decay_rate: float = 0.7) -> np.ndarray:
        """
        Adstock: реклама имеет остаточный эффект на следующие периоды.
        decay_rate: 0.5-0.9 в зависимости от канала (TV > 0.8, Search < 0.4)
        """
        adstocked = np.zeros_like(spend, dtype=float)
        adstocked[0] = spend[0]
        for t in range(1, len(spend)):
            adstocked[t] = spend[t] + decay_rate * adstocked[t - 1]
        return adstocked

    def saturation_transform(self, adstocked: np.ndarray,
                               alpha: float = 2.0,
                               gamma: float = 0.5) -> np.ndarray:
        """
        Hill-функция насыщения: убывающая отдача при увеличении инвестиций.
        alpha: крутизна кривой насыщения
        gamma: точка полунасыщения (при каком spend ROI = 50% от максимального)
        """
        return adstocked ** alpha / (adstocked ** alpha + gamma ** alpha)

    def fit_channel_contributions(self, weekly_data: pd.DataFrame,
                                    channel_cols: list[str],
                                    revenue_col: str = 'revenue') -> dict:
        """
        Регрессия для оценки вклада каждого канала.
        weekly_data: недельные данные по spend и revenue
        """
        from sklearn.linear_model import Ridge

        X_transformed = {}
        for ch in channel_cols:
            adstocked = self.adstock_transform(weekly_data[ch].values)
            saturated = self.saturation_transform(adstocked)
            X_transformed[ch] = saturated

        X = pd.DataFrame(X_transformed)
        y = weekly_data[revenue_col].values

        model = Ridge(alpha=1.0, fit_intercept=True)
        model.fit(X, y)

        # ROAS по каналу = coefficient * mean_saturation / mean_spend
        contributions = {}
        for i, ch in enumerate(channel_cols):
            coef = model.coef_[i]
            mean_saturation = X[ch].mean()
            mean_spend = weekly_data[ch].mean()
            marginal_roas = coef * mean_saturation / max(mean_spend, 1)

            contributions[ch] = {
                'coefficient': round(float(coef), 4),
                'revenue_contribution_pct': round(
                    float(coef * X[ch].sum() / max(y.sum(), 1) * 100), 1
                ),
                'marginal_roas': round(float(marginal_roas), 2),
            }

        return contributions


class BudgetAllocator:
    """Оптимальное распределение бюджета между каналами"""

    def __init__(self, channel_params: dict):
        """
        channel_params: {channel_name: {'alpha': float, 'gamma': float, 'max_spend': float}}
        """
        self.channels = channel_params

    def marginal_roi(self, channel: str, spend: float) -> float:
        """Предельная отдача от дополнительного рубля в канале"""
        p = self.channels[channel]
        alpha = p.get('alpha', 2.0)
        gamma = p.get('gamma', 1000.0)
        base_roas = p.get('base_roas', 3.0)

        # Производная Hill-функции
        numerator = alpha * gamma ** alpha * spend ** (alpha - 1)
        denominator = (spend ** alpha + gamma ** alpha) ** 2
        saturation_derivative = numerator / max(denominator, 1e-10)

        return base_roas * saturation_derivative

    def optimize_allocation(self, total_budget: float,
                              min_channel_budget: float = 100.0) -> dict:
        """
        Решение задачи оптимизации: максимизировать суммарный revenue.
        Используем метод Lagrange multipliers (выравнивание предельных ROI).
        """
        channels = list(self.channels.keys())
        n = len(channels)

        def total_negative_revenue(budgets):
            """Минус суммарный revenue для минимизации"""
            total = 0
            for i, ch in enumerate(channels):
                p = self.channels[ch]
                adstocked = budgets[i]
                sat = adstocked ** p.get('alpha', 2) / (
                    adstocked ** p.get('alpha', 2) + p.get('gamma', 1000) ** p.get('alpha', 2)
                )
                total += sat * p.get('base_roas', 3.0) * budgets[i]
            return -total

        constraints = [
            {'type': 'eq', 'fun': lambda b: sum(b) - total_budget}
        ]

        bounds = [
            (min_channel_budget, self.channels[ch].get('max_spend', total_budget))
            for ch in channels
        ]

        x0 = [total_budget / n] * n

        result = minimize(
            total_negative_revenue,
            x0,
            method='SLSQP',
            bounds=bounds,
            constraints=constraints,
            options={'maxiter': 1000}
        )

        optimal = {ch: round(float(b), 2) for ch, b in zip(channels, result.x)}

        # Прогнозируемый Revenue
        rev_estimate = -result.fun
        current_rev = -total_negative_revenue(x0)

        return {
            'allocation': optimal,
            'projected_revenue': round(float(rev_estimate), 2),
            'vs_equal_split': round((rev_estimate - current_rev) / max(current_rev, 1) * 100, 1),
            'optimization_converged': result.success,
        }

Кросс-канальная атрибуция и incrementality

class IncrementalityMeasurement:
    """Измерение инкрементальности через geo experiments"""

    def design_geo_holdout(self, geos: list[str],
                             treatment_fraction: float = 0.5) -> dict:
        """
        Geo holdout experiment: часть регионов не получает рекламу.
        Чистая мера инкрементальности без selection bias.
        """
        np.random.shuffle(geos)
        split = int(len(geos) * treatment_fraction)

        return {
            'treatment_geos': geos[:split],
            'control_geos': geos[split:],
            'n_treatment': split,
            'n_control': len(geos) - split,
            'recommendation': 'Run for minimum 4 weeks, measure revenue lift'
        }

    def calculate_incremental_roas(self, treatment_revenue: float,
                                    control_revenue: float,
                                    treatment_spend: float,
                                    treatment_population: int,
                                    control_population: int) -> dict:
        """True incrementality ROAS (iROAS)"""
        # Нормализуем на численность населения
        treatment_rev_per_capita = treatment_revenue / max(treatment_population, 1)
        control_rev_per_capita = control_revenue / max(control_population, 1)

        incremental_rev_per_capita = treatment_rev_per_capita - control_rev_per_capita
        incremental_total_rev = incremental_rev_per_capita * treatment_population

        iroas = incremental_total_rev / max(treatment_spend, 1)

        return {
            'incremental_revenue': round(incremental_total_rev, 2),
            'iroas': round(iroas, 2),
            'reported_roas': round(treatment_revenue / max(treatment_spend, 1), 2),
            'incrementality_ratio': round(iroas / max(treatment_revenue / max(treatment_spend, 1), 0.01), 2),
            'interpretation': 'iROAS < reported ROAS means significant organic/direct traffic'
        }

Типовые результаты оптимизации бюджета

Горизонт реализации: базовый MMM — 6-8 недель (сбор данных, моделирование, валидация). Полный цикл с incrementality experiments — 3-4 месяца. Окупается при monthly ad spend от $50 тысяч.