AI DSP (Demand-Side Platform) Management System

AI-Management Demand Side Platform

DSPs are the operational center of programmatic buying. The goal of AI in DSPs is not simply to win auctions, but to win the right auctions at the right frequency with the right audience, without overspending or showing ads where they harm the brand. This is a multi-parameter optimization problem with strict latency constraints.

Architecture of AI DSP components

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

Managing Audience Segments

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
            ),
        }

Attribution Funnel Optimization

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()
        }

Typical KPIs for a managed DSP

Full-fledged AI-based DSP optimization with audience targeting, brand safety, and multi-touch attribution increases efficiency by 30-50% compared to manual management for volumes of 10 million impressions per month.