Как оптимизировать рекламный бюджет на основе данных?

Оптимизация рекламного бюджета на основе данных

Данный вопрос касается применения данных и аналитики для оптимизации маркетинговых затрат. Data Engineer в этом процессе играет ключевую роль, обеспечивая инфраструктуру для сбора, обработки и анализа рекламных метрик.

Роль Data Engineer

Data Engineer отвечает за:

• Сбор данных из различных рекламных платформ (Google Ads, Facebook Ads, Yandex)
• ETL процессы для консолидации данных
• Создание витрин для анализа и ROI
• Интеграция с CRM и финансовыми системами
• Обеспечение real-time или near real-time доступности метрик

Архитектура сбора данных

# Пример: интеграция с Google Ads API
import pandas as pd
from google.ads.googleads.client import GoogleAdsClient
from sqlalchemy import create_engine

client = GoogleAdsClient.load_from_storage('google-ads.yaml')

def extract_google_ads_metrics():
    """Загружает метрики из Google Ads"""
    service = client.get_service('GoogleAdsService')
    customer_ids = ['1234567890']
    
    for customer_id in customer_ids:
        query = """
        SELECT
            campaign.id,
            campaign.name,
            metrics.impressions,
            metrics.clicks,
            metrics.cost_micros,
            metrics.conversions,
            metrics.conversion_value,
            segments.date
        FROM campaign
        WHERE segments.date >= '2024-03-01'
        """
        
        response = service.search_stream(customer_id=customer_id, query=query)
        
        rows = []
        for batch in response:
            for row in batch.results:
                rows.append({
                    'campaign_id': row.campaign.id,
                    'campaign_name': row.campaign.name,
                    'impressions': row.metrics.impressions,
                    'clicks': row.metrics.clicks,
                    'cost': row.metrics.cost_micros / 1_000_000,
                    'conversions': row.metrics.conversions,
                    'conversion_value': row.metrics.conversion_value,
                    'date': row.segments.date,
                    'source': 'google_ads'
                })
        
        return pd.DataFrame(rows)

# Сохраняем в БД
engine = create_engine('postgresql://...')
df = extract_google_ads_metrics()
df.to_sql('stg_google_ads_daily', engine, if_exists='append', index=False)

ETL процесс для консолидации

-- Создаём unified витрину для всех рекламных каналов
CREATE TABLE dim_ad_channel (
    channel_id INT PRIMARY KEY,
    channel_name VARCHAR(50),  -- 'google_ads', 'facebook', 'yandex'
    platform_code VARCHAR(20)
);

CREATE TABLE fact_ad_spend (
    spend_id INT PRIMARY KEY,
    date DATE,
    channel_id INT,
    campaign_id VARCHAR(50),
    campaign_name VARCHAR(200),
    impressions BIGINT,
    clicks BIGINT,
    cost DECIMAL(12, 2),
    conversions INT,
    conversion_value DECIMAL(14, 2),
    ctr DECIMAL(6, 4),  -- click-through rate
    cpc DECIMAL(8, 2),  -- cost per click
    cpa DECIMAL(8, 2),  -- cost per acquisition
    roas DECIMAL(6, 2), -- return on ad spend
    FOREIGN KEY (channel_id) REFERENCES dim_ad_channel(channel_id)
);

-- ETL скрипт для расчёта метрик
INSERT INTO fact_ad_spend (
    date, channel_id, campaign_id, campaign_name, 
    impressions, clicks, cost, conversions, conversion_value
)
SELECT 
    date,
    channel_id,
    campaign_id,
    campaign_name,
    SUM(impressions),
    SUM(clicks),
    SUM(cost),
    SUM(conversions),
    SUM(conversion_value)
FROM stg_google_ads_daily
GROUP BY 1, 2, 3, 4;

-- Вычисляем производные метрики
UPDATE fact_ad_spend SET
    ctr = CASE WHEN impressions > 0 THEN clicks::DECIMAL / impressions ELSE 0 END,
    cpc = CASE WHEN clicks > 0 THEN cost / clicks ELSE 0 END,
    cpa = CASE WHEN conversions > 0 THEN cost / conversions ELSE 0 END,
    roas = CASE WHEN cost > 0 THEN conversion_value / cost ELSE 0 END;

Аналитические витрины для оптимизации

-- Витрина для анализа ROI по каналам
CREATE MATERIALIZED VIEW v_channel_roi AS
SELECT 
    c.channel_name,
    f.date,
    SUM(f.cost) as total_spend,
    SUM(f.conversions) as total_conversions,
    SUM(f.conversion_value) as total_value,
    SUM(f.conversion_value) / NULLIF(SUM(f.cost), 0) as roi,
    SUM(f.cost) / NULLIF(SUM(f.conversions), 0) as cpa
FROM fact_ad_spend f
JOIN dim_ad_channel c ON f.channel_id = c.channel_id
GROUP BY 1, 2
ORDER BY 2 DESC, 3 DESC;

-- Витрина для выявления неэффективных кампаний
CREATE MATERIALIZED VIEW v_campaign_efficiency AS
SELECT 
    campaign_name,
    SUM(cost) as total_spend,
    SUM(conversions) as conversions,
    SUM(conversion_value) / NULLIF(SUM(cost), 0) as roi,
    AVG(ctr) as avg_ctr,
    AVG(cpc) as avg_cpc,
    COUNT(DISTINCT date) as days_active,
    CASE 
        WHEN SUM(conversion_value) / NULLIF(SUM(cost), 0) < 1.5 THEN 'LOW_ROI'
        WHEN AVG(cpc) > (SELECT AVG(cpc) FROM fact_ad_spend) THEN 'HIGH_CPC'
        WHEN AVG(ctr) < (SELECT AVG(ctr) FROM fact_ad_spend) THEN 'LOW_CTR'
        ELSE 'EFFICIENT'
    END as status
FROM fact_ad_spend
WHERE date >= CURRENT_DATE - INTERVAL 30 DAYS
GROUP BY 1;

Алгоритм оптимизации бюджета

import numpy as np
from sklearn.linear_model import LinearRegression

def optimize_budget_allocation(campaigns_data, total_budget):
    """
    Оптимизирует распределение рекламного бюджета
    на основе исторических ROI данных
    """
    
    # campaigns_data: list of dicts с полями
    # {'campaign_name', 'current_spend', 'roi', 'conversions', 'cost'}
    
    # Шаг 1: Вычисляем исторический ROI
    campaign_rois = {c['name']: c['roi'] for c in campaigns_data}
    
    # Шаг 2: Вычисляем elasticity (чувствительность к бюджету)
    # Если увеличим бюджет на 10%, насколько вырастут конверсии?
    elasticity = {}
    for campaign in campaigns_data:
        # Используем историческую динамику
        elasticity[campaign['name']] = min(
            campaign['roi'] * 0.8,  # маржа деградации
            0.9  # максимальная elasticity
        )
    
    # Шаг 3: Распределяем бюджет пропорционально ROI
    total_roi = sum(campaign_rois.values())
    new_allocation = {}
    
    for campaign_name, roi in campaign_rois.items():
        # Выделяем долю бюджета пропорционально ROI
        allocation = (roi / total_roi) * total_budget
        new_allocation[campaign_name] = allocation
    
    return new_allocation

# Пример использования
campaigns = [
    {'name': 'Brand_Search', 'current_spend': 5000, 'roi': 4.2, 'conversions': 420},
    {'name': 'Performance_Search', 'current_spend': 3000, 'roi': 2.8, 'conversions': 210},
    {'name': 'Display', 'current_spend': 2000, 'roi': 1.5, 'conversions': 60}
]

optimal_budget = optimize_budget_allocation(campaigns, total_budget=10000)
# Результат: {'Brand_Search': 5500, 'Performance_Search': 3200, 'Display': 1300}

Многоканальная атрибуция

-- Для корректной оценки ROI нужна атрибуция
CREATE TABLE fact_customer_journey (
    journey_id INT PRIMARY KEY,
    customer_id INT,
    first_click_channel VARCHAR(50),
    last_click_channel VARCHAR(50),
    all_channels VARCHAR(500),  -- 'google_ads -> facebook -> direct'
    conversion_value DECIMAL(12, 2),
    conversion_date DATE
);

-- Last-click attribution (простейший вариант)
CREATE VIEW v_last_click_attribution AS
SELECT 
    last_click_channel,
    SUM(conversion_value) as attributed_value,
    COUNT(*) as conversions
FROM fact_customer_journey
WHERE conversion_date >= CURRENT_DATE - INTERVAL 30 DAYS
GROUP BY 1;

-- Multi-touch attribution (распределяем value между всеми каналами)
CREATE VIEW v_multitouch_attribution AS
SELECT 
    channel,
    SUM(share * conversion_value) as attributed_value
FROM (
    SELECT 
        UNNEST(STRING_TO_ARRAY(all_channels, ' -> ')) as channel,
        conversion_value,
        1.0 / ARRAY_LENGTH(STRING_TO_ARRAY(all_channels, ' -> '), 1) as share
    FROM fact_customer_journey
) t
GROUP BY 1;

Real-time мониторинг и алерты

# Пример: настройка алерты на падение ROI
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

def check_campaign_roi():
    """Проверяет, не упал ли ROI ниже целевого уровня"""
    
    query = """
    SELECT 
        campaign_name,
        SUM(conversion_value) / NULLIF(SUM(cost), 0) as current_roi
    FROM fact_ad_spend
    WHERE date >= CURRENT_DATE - INTERVAL 7 DAYS
    GROUP BY 1
    HAVING SUM(conversion_value) / NULLIF(SUM(cost), 0) < 2.0
    """
    
    low_roi_campaigns = db.execute(query).fetchall()
    
    if low_roi_campaigns:
        # Отправляем алерт
        for campaign in low_roi_campaigns:
            send_slack_alert(
                f"Campaign {campaign.name} has low ROI: {campaign.current_roi}"
            )
        
        # Рекомендуем снижение бюджета или паузу
        return 'RECOMMENDATION_REDUCE'
    
    return 'OK'

dag = DAG('campaign_roi_check', start_date=datetime(2024, 1, 1))
task = PythonOperator(python_callable=check_campaign_roi, dag=dag)

Best Practices

• Real-time данные: используй API отчётов для близкой к real-time актуальности
• Мультиканальность: не забывай про cross-channel эффекты
• Историческая глубина: держи минимум 1-2 года истории для сезонных паттернов
• Фреквенция обновления: скорее всего достаточно дневной актуальности
• Quality Control: проверяй целостность данных (NaN, outliers)
• Версионирование: сохраняй версии алгоритмов оптимизации для A/B тестирования