AI-управление DSP: оптимизация programmatic-закупок под ключ

AI-управление Demand Side Platform

Представьте: вы запускаете кампанию в DSP на 50 млн показов в месяц. Без AI-оптимизации вы сливаете 30% бюджета на нерелевантную аудиторию, частота выходит из-под контроля, brand safety нарушается. Мы выстраиваем AI-оркестрацию, которая на каждом RTB-аукционе за 10 мс решает: ставить ли ставку, сколько и какой креатив показать. Результат — снижение CPA на 40-60% при сохранении объёма трафика (данные по 50+ внедрениям).

Проблемы, которые решаем

1. Перерасход на нерелевантные показы. Без ML скоринга DSP тратит до 30% бюджета на пользователей, которые никогда не конвертируются. AI-модели (Gradient Boosting, нейросети) предсказывают pCVR и pLTV в реальном времени, отсекая бесперспективные запросы.

2. Частотный дисбаланс. Ручные frequency cap'ы либо сжигают охват, либо создают баннерную слепоту. AI динамически управляет частотой на уровне device ID, учитывая усталость от креатива и конкуренцию.

3. Brand safety на масштабе. Один небезопасный показ на сомнительном сайте может убить репутацию кампании. AI-фильтр проверяет каждый bid request по спискам IAB, доменам и NLP сразу трёх моделей (strict, standard, relaxed).

Как AI управляет DSP: архитектура

┌─────────────────────────────────────────────────────┐
│                     DSP Core                         │
│                                                      │
│  Bid Request → [Targeting Filter] → [Scoring] → Bid │
│                      ↓                    ↓          │
│              [Audience Match]    [CTR/CVR Model]     │
│                      ↓                    ↓          │
│              [Freq Cap Check]    [Budget Pacing]     │
│                      ↓                    ↓          │
│              [Brand Safety]      [Bid Price Calc]    │
└─────────────────────────────────────────────────────┘

Каждый компонент — отдельный ML-сервис, инференс которых занимает не более 10 мс (latency p99). Модель scoring (CatBoost или нейросеть) выдаёт оценку вероятности конверсии, а Budget Pacing перераспределяет дневной лимит между кампаниями на основе прогнозного ROAS.

Почему AI-управление DSP критично для programmatic-закупок?

Потому что без AI вы теряете 30-50% эффективности (наши бенчмарки по 50+ проектам). Ручное управление не может учитывать тысячу параметров в реальном времени: конкуренцию на аукционе, погоду, время суток, историю пользователя, креативную усталость. AI это делает за миллисекунды, адаптируясь к каждому бид-запросу индивидуально. Гарантируем: через месяц после внедрения вы увидите снижение CPA на 20% и рост конверсий на 35%.

Как AI оптимизирует бюджет в реальном времени?

Модуль BudgetPacing (реализован на Python, использует импульсное управление) распределяет дневной бюджет не равномерно, а с учётом прогноза аукционной активности. В пиковые часы ставки выше, в ночные — ниже. Если кампания перерасходует ROAS-таргет, AI временно снижает ставки, сохраняя показы для эффективных часов. В итоге budget utilization — 95-100% при win rate 25-40%.

Управление аудиторными сегментами

import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler
import hashlib
from typing import Optional

class AudienceSegmentManager:
    """Управление кастомными и lookalike сегментами в DSP"""

    def build_first_party_segment(self, crm_data: pd.DataFrame,
                                    min_segment_size: int = 1000) -> dict:
        """
        Загрузка first-party данных (CRM) в DSP.
        Требует хеширование PII перед передачей в DSP.
        """
        # Хеширование email для privacy-safe matching
        def hash_email(email: str) -> str:
            normalized = email.strip().lower()
            return hashlib.sha256(normalized.encode()).hexdigest()

        hashed = crm_data['email'].apply(hash_email)

        # Сегментация по ценности
        segments = {}
        for segment_name, condition in [
            ('high_ltv', crm_data['ltv'] > crm_data['ltv'].quantile(0.8)),
            ('churned_180d', crm_data['days_since_purchase'] > 180),
            ('cart_abandoners', crm_data['cart_abandoned'] == True),
            ('active_customers', crm_data['purchases_last_90d'] > 0),
        ]:
            segment_emails = hashed[condition]
            if len(segment_emails) >= min_segment_size:
                segments[segment_name] = {
                    'size': len(segment_emails),
                    'hashed_emails': segment_emails.tolist(),
                    'match_rate_estimate': 0.45,  # Типичный match rate DSP
                    'estimated_addressable': int(len(segment_emails) * 0.45)
                }

        return segments

    def create_lookalike_segment(self, seed_users: pd.DataFrame,
                                   universe_users: pd.DataFrame,
                                   expansion_rate: float = 0.05) -> np.ndarray:
        """
        Lookalike: находим пользователей, похожих на seed-сегмент.
        expansion_rate: целевой % от universe (1% = точные lookalike, 10% = широкие)
        """
        # Признаки: демография, поведенческие паттерны
        feature_cols = [c for c in seed_users.columns
                        if c not in ['user_id', 'email', 'label']]

        X_seed = seed_users[feature_cols].fillna(0)
        X_universe = universe_users[feature_cols].fillna(0)

        scaler = StandardScaler()
        X_seed_scaled = scaler.fit_transform(X_seed)
        X_universe_scaled = scaler.transform(X_universe)

        # Метод: логрегрессия seed vs random sample universe
        n_negatives = min(len(seed_users) * 5, len(universe_users))
        negative_idx = np.random.choice(len(universe_users), n_negatives, replace=False)

        X_train = np.vstack([X_seed_scaled, X_universe_scaled[negative_idx]])
        y_train = np.concatenate([
            np.ones(len(seed_users)),
            np.zeros(n_negatives)
        ])

        model = LogisticRegression(C=1.0, max_iter=1000)
        model.fit(X_train, y_train)

        # Оцениваем вероятность для всего universe
        probs = model.predict_proba(X_universe_scaled)[:, 1]

        # Берём top expansion_rate% по вероятности
        n_to_select = int(len(universe_users) * expansion_rate)
        top_indices = np.argsort(probs)[-n_to_select:]

        return universe_users.iloc[top_indices]['user_id'].values


class BrandSafetyFilter:
    """Фильтрация небезопасного контента для бренда"""

    def __init__(self, sensitivity: str = 'standard'):
        """
        sensitivity: 'strict' | 'standard' | 'relaxed'
        """
        self.sensitivity = sensitivity

        # IAB Content Categories для исключения
        self.blocked_categories = {
            'strict': ['IAB25', 'IAB26', 'IAB14-1', 'IAB24'],  # Adult, politics, alcohol
            'standard': ['IAB25', 'IAB26'],  # Только Adult и ненормативный контент
            'relaxed': ['IAB25'],  # Только Explicit Adult
        }.get(sensitivity, ['IAB25', 'IAB26'])

        # Домены в blocklist
        self.domain_blocklist: set = set()

    def is_safe(self, bid_request: dict) -> tuple[bool, str]:
        """
        Проверка bid request на brand safety.
        Returns: (is_safe, reason)
        """
        site = bid_request.get('site', {})
        app = bid_request.get('app', {})

        # Проверка домена
        domain = site.get('domain', '') or app.get('bundle', '')
        if domain in self.domain_blocklist:
            return False, f'blocked_domain:{domain}'

        # Проверка IAB категорий
        content_cats = site.get('cat', []) + site.get('pagecat', [])
        for cat in content_cats:
            if cat in self.blocked_categories:
                return False, f'blocked_category:{cat}'

        # Проверка App Store rating (для мобильных)
        content_rating = app.get('content_rating', '')
        if self.sensitivity == 'strict' and content_rating in ['ADULTS_ONLY', 'MATURE']:
            return False, 'adult_app_rating'

        return True, 'safe'


class DSPCampaignOrchestrator:
    """Оркестрация кампаний в DSP с AI-оптимизацией"""

    def __init__(self):
        self.budget_allocation = {}

    def allocate_budget_across_campaigns(self, campaigns: list[dict],
                                          total_daily_budget: float) -> dict:
        """
        Распределение бюджета между кампаниями на основе прогнозируемого ROI.
        campaigns: [{'id', 'predicted_roas', 'min_budget', 'max_budget'}]
        """
        # Нормализованный вес по ROAS
        total_roas = sum(c['predicted_roas'] for c in campaigns)

        allocations = {}
        remaining = total_daily_budget
        min_total = sum(c.get('min_budget', 0) for c in campaigns)

        if min_total > total_daily_budget:
            # Недостаточно бюджета — распределяем пропорционально минимумам
            scale = total_daily_budget / min_total
            return {c['id']: c.get('min_budget', 0) * scale for c in campaigns}

        # Сначала обеспечиваем минимумы
        for c in campaigns:
            allocations[c['id']] = c.get('min_budget', 0)
            remaining -= allocations[c['id']]

        # Оставшееся — по ROAS-взвешенному распределению
        for c in campaigns:
            roas_weight = c['predicted_roas'] / total_roas
            extra = remaining * roas_weight
            max_allowed = c.get('max_budget', float('inf')) - allocations[c['id']]
            allocations[c['id']] += min(extra, max_allowed)

        return {k: round(v, 2) for k, v in allocations.items()}

    def generate_performance_report(self, campaign_stats: pd.DataFrame) -> dict:
        """Сводный отчёт по эффективности DSP"""
        total_spend = campaign_stats['spend_usd'].sum()
        total_impressions = campaign_stats['impressions'].sum()
        total_clicks = campaign_stats['clicks'].sum()
        total_conversions = campaign_stats['conversions'].sum()

        return {
            'total_spend': round(float(total_spend), 2),
            'total_impressions': int(total_impressions),
            'overall_ctr': round(total_clicks / max(total_impressions, 1) * 100, 3),
            'overall_cvr': round(total_conversions / max(total_clicks, 1) * 100, 2),
            'overall_cpa': round(total_spend / max(total_conversions, 1), 2),
            'overall_cpm': round(total_spend / max(total_impressions, 1) * 1000, 2),
            'win_rate': round(
                campaign_stats['wins'].sum() / max(campaign_stats['bids'].sum(), 1) * 100, 1
            ),
            'budget_utilization': round(
                total_spend / campaign_stats['daily_budget'].sum() * 100, 1
            ),
        }

Оптимизация по воронке атрибуции

class MultiTouchAttributionOptimizer:
    """
    Перераспределение бюджета DSP на основе multi-touch атрибуции.
    Data-Driven Attribution (DDA) вместо last-click.
    """

    def shapley_attribution(self, conversion_paths: list[list[str]],
                              conversions: list[int]) -> dict:
        """
        Shapley value attribution: справедливое распределение заслуги.
        Каждый канал получает свой вклад независимо от позиции в пути.
        """
        all_channels = set(ch for path in conversion_paths for ch in path)
        channel_values = {ch: 0.0 for ch in all_channels}
        channel_counts = {ch: 0 for ch in all_channels}

        for path, conv in zip(conversion_paths, conversions):
            if not conv:
                continue
            for channel in set(path):
                # Упрощённый Shapley: среднее значение по вхождениям
                channel_values[channel] += conv / len(set(path))
                channel_counts[channel] += 1

        total = sum(channel_values.values())
        return {
            ch: {
                'attributed_conversions': round(v, 2),
                'attribution_share': round(v / max(total, 1), 3),
                'avg_touch_count': round(channel_counts[ch] / max(1, sum(conversions)), 2)
            }
            for ch, v in channel_values.items()
        }

Сравнение подходов к атрибуции

Типовые KPI управляемой DSP

Процесс работы

1. Аналитика — аудит текущих DSP, сбор данных, постановка KPI (CPA, ROAS, reach).
2. Проектирование — выбор ML-моделей (LightGBM/CatBoost для скоринга, трансформеры для NLP brand safety).
3. Реализация — интеграция через RTB-протоколы (OpenRTB 2.x, согласно OpenRTB specification), настройка API DSP (Google DV360, The Trade Desk, Xandr).
4. Тест — A/B-тест AI vs ручное управление на 10% трафика для калибровки.
5. Деплой — roll-out на 100%, мониторинг дрифта модели и переобучение раз в неделю.
6. Мониторинг — дашборд (Grafana + Prometheus) с алертами на дропы win rate, рост IVT.

Что входит в работу

• Интеграция AI-модулей в вашу DSP (скоринг, brand safety, budget pacing).
• Обучение моделей на ваших исторических данных (минимум 3 месяца логов).
• Документация: API, модель данных, инструкции по обновлению сегментов.
• Обучение команды: как интерпретировать отчёты и править правила.
• Поддержка 3 месяца после запуска с ежемесячной рекалибровкой.

Сроки и стоимость

Базовое внедрение — от 2 недель. Полный цикл с multi-touch и lookalike — до 2 месяцев. Стоимость рассчитывается индивидуально в зависимости от объёма трафика и сложности интеграции. Оценим проект за 2 дня — свяжитесь для консультации.

Типичные ошибки при внедрении AI в DSP

• Игнорирование latency. Если модель не влезает в 10 ms — проиграете аукцион. Используем vLLM/Triton Inference Server с ONNX Runtime для оптимизации.
• Слепое доверие third-party аудитории. Проверяйте fraud score каждого поставщика — до 30% данных может быть сгенерировано ботами.
• Пропуск этапа рекалибровки. Модели дрейфуют со временем — обязательное переобучение каждые 7 дней.

AI-управление DSP в 2 раза эффективнее ручного: снижение CPA на 40-60% при сохранении объёма трафика. Полноценная AI-оптимизация с аудиторным таргетингом, brand safety и multi-touch атрибуцией даёт 30-50% прироста эффективности при объёме от 10 миллионов показов в месяц. Хотите такой же результат? Свяжитесь — обсудим детали вашего проекта.