AI-система управления DSP (Demand-Side Platform) с AI

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

DSP — операционный центр programmatic-закупки. Задача AI в DSP: не просто выигрывать аукционы, а выигрывать нужные аукционы с нужной частотой у нужной аудитории, при этом не перерасходуя бюджет и не показывая рекламу там, где она вредит бренду. Это многопараметрическая задача оптимизации с жёсткими latency-ограничениями.

Архитектура 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]    │
└─────────────────────────────────────────────────────┘

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

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

Полноценная AI-оптимизация DSP с аудиторным таргетингом, brand safety и multi-touch атрибуцией добавляет 30-50% эффективности по сравнению с ручным управлением при объёме от 10 миллионов показов в месяц.