Programmatic Advertising AI: RTB, прогнози та оптимізація бюджету

import numpy as np
import pandas as pd
import torch
import torch.nn as nn
from sklearn.ensemble import GradientBoostingClassifier, GradientBoostingRegressor
from sklearn.calibration import CalibratedClassifierCV
import lightgbm as lgb
import json

class BidRequestFeaturizer:
    """Извлечение признаков из bid request за < 5ms"""

    def featurize(self, bid_request: dict) -> np.ndarray:
        """
        bid_request: стандартный OpenRTB 2.5 объект
        Возвращает признаковый вектор для модели за < 1ms
        """
        return np.array([
            self._hash_encode(bid_request.get('user', {}).get('id', ''), 100),
            bid_request.get('user', {}).get('yob', 1990),
            int(bid_request.get('user', {}).get('gender') == 'M'),
            len(bid_request.get('user', {}).get('segments', [])),
            self._device_type_encode(bid_request.get('device', {}).get('devicetype')),
            int(bid_request.get('device', {}).get('os', '') in ['iOS', 'Android']),
            self._hash_encode(bid_request.get('device', {}).get('model', ''), 50),
            bid_request.get('imp', [{}])[0].get('banner', {}).get('w', 300),
            bid_request.get('imp', [{}])[0].get('banner', {}).get('h', 250),
            int(bid_request.get('imp', [{}])[0].get('instl') == 1),
            self._hash_encode(bid_request.get('site', {}).get('domain', ''), 200),
            self._hash_encode(bid_request.get('site', {}).get('cat', ['IAB1'])[0], 20),
            pd.Timestamp.now().hour,
            pd.Timestamp.now().weekday(),
            int(pd.Timestamp.now().weekday() >= 5),
            bid_request.get('imp', [{}])[0].get('bidfloor', 0),
        ], dtype=np.float32)

    def _hash_encode(self, value: str, n_buckets: int) -> int:
        return hash(value) % n_buckets

    def _device_type_encode(self, device_type) -> int:
        mapping = {1: 1, 2: 2, 3: 3, 4: 4, 5: 5}
        return mapping.get(device_type, 0)


class CTRPredictor:
    """Предсказание CTR (Click-Through Rate) для bid. LightGBM обычно лучше нейросетей для tabular bid data."""

    def __init__(self):
        self.model = lgb.LGBMClassifier(
            n_estimators=500,
            learning_rate=0.05,
            num_leaves=127,
            min_child_samples=50,
            subsample=0.8,
            colsample_bytree=0.8,
            random_state=42,
            n_jobs=-1
        )

    def train(self, X, y, X_val, y_val):
        """Обучение с ранней остановкой"""
        self.model.fit(X, y, eval_set=[(X_val, y_val)], eval_metric='auc',
                        callbacks=[lgb.early_stopping(50), lgb.log_evaluation(100)])

    def predict_ctr(self, X):
        return self.model.predict_proba(X)[:, 1]


class ConversionRatePredictor:
    """CVR: вероятность конверсии при клике"""
    def __init__(self):
        self.model = lgb.LGBMClassifier(
            n_estimators=200, learning_rate=0.05, num_leaves=63,
            min_child_samples=100, random_state=42
        )

    def predict_cvr(self, X):
        return self.model.predict_proba(X)[:, 1]


class BiddingEngine:
    """Движок принятия решений о ставках"""
    def __init__(self, ctr_model, cvr_model, featurizer):
        self.ctr_model = ctr_model
        self.cvr_model = cvr_model
        self.featurizer = featurizer

    def compute_bid(self, bid_request, campaign_config):
        """Вычисление оптимальной ставки за <10ms"""
        features = self.featurizer.featurize(bid_request)
        ctr = float(self.ctr_model.predict_ctr(features.reshape(1, -1))[0])
        cvr = float(self.cvr_model.predict_cvr(features.reshape(1, -1))[0])
        pctcvr = ctr * cvr
        target_cpa = campaign_config.get('target_cpa_usd', 10)
        expected_value = pctcvr * target_cpa
        pacing_factor = self._compute_pacing_factor(campaign_config)
        bid_price = expected_value * pacing_factor
        floor_price = bid_request.get('imp', [{}])[0].get('bidfloor', 0)
        max_bid = campaign_config.get('max_bid_cpm', 10)
        if bid_price < floor_price:
            return {'bid': 0, 'reason': 'below_floor', 'predicted_ctr': ctr}
        final_bid = min(bid_price, max_bid)
        return {
            'bid': round(final_bid, 4),
            'predicted_ctr': round(ctr, 5),
            'predicted_cvr': round(cvr, 5),
            'predicted_pctcvr': round(pctcvr, 6),
            'pacing_factor': round(pacing_factor, 3),
            'auction_win_probability': self._estimate_win_prob(final_bid, floor_price)
        }

    def _compute_pacing_factor(self, campaign):
        budget_total = campaign.get('daily_budget_usd', 1000)
        spent_today = campaign.get('spent_today_usd', 0)
        hours_elapsed = campaign.get('hours_elapsed_today', 12)
        total_hours = 24
        expected_spent_ratio = hours_elapsed / total_hours
        actual_spent_ratio = spent_today / max(budget_total, 1)
        if actual_spent_ratio > expected_spent_ratio * 1.1:
            return 0.8
        elif actual_spent_ratio < expected_spent_ratio * 0.9:
            return 1.2
        return 1.0

    def _estimate_win_prob(self, bid, floor):
        if bid < floor:
            return 0.0
        margin = (bid - floor) / max(floor, 0.01)
        return min(0.95, 0.3 + margin * 0.5)


class BudgetPacingController:
    """Управление равномерностью расходования бюджета"""
    def throttle_bid_rate(self, campaign_stats, current_qps):
        budget = campaign_stats.get('daily_budget', 1000)
        spent = campaign_stats.get('spent', 0)
        hours = campaign_stats.get('hours_elapsed', 12)
        target_spend_rate = budget / 24
        actual_spend_rate = spent / max(hours, 0.1)
        if actual_spend_rate > target_spend_rate * 1.2:
            throttle = target_spend_rate / actual_spend_rate
            return float(np.clip(throttle, 0.1, 1.0))
        return 1.0

    def compute_optimal_frequency_cap(self, user_stats, campaign_config):
        base_cap = campaign_config.get('frequency_cap', {'hour': 2, 'day': 5, 'week': 15})
        if user_stats.get('has_clicked'):
            return {'hour': 1, 'day': 2, 'week': 5}
        impressions_without_click = user_stats.get('impressions_no_click', 0)
        if impressions_without_click > 20:
            return {'hour': 0, 'day': 1, 'week': 3}
        return base_cap


class AuctionOptimizer:
    """Оптимизация стратегии в аукционе первой и второй цены"""
    def optimal_bid_second_price(self, valuation, bid_landscape):
        return valuation

    def bid_shading_first_price(self, valuation, historical_clearing_prices):
        if len(historical_clearing_prices) == 0:
            return valuation * 0.8
        best_bid = valuation * 0.5
        best_profit = -float('inf')
        for bid_pct in np.arange(0.5, 1.0, 0.05):
            bid = valuation * bid_pct
            win_prob = (historical_clearing_prices < bid).mean()
            expected_profit = win_prob * (valuation - bid)
            if expected_profit > best_profit:
                best_profit = expected_profit
                best_bid = bid
        return round(best_bid, 4)

    def evaluate_campaign_performance(self, impressions):
        return {
            'impressions': len(impressions),
            'clicks': impressions['clicked'].sum(),
            'conversions': impressions['converted'].sum(),
            'spend_usd': impressions['bid_price'].sum(),
            'ctr': impressions['clicked'].mean(),
            'cvr': impressions['converted'].sum() / max(impressions['clicked'].sum(), 1),
            'cpa_usd': impressions['bid_price'].sum() / max(impressions['converted'].sum(), 1),
            'roas': impressions.get('revenue', pd.Series([0])).sum() / max(impressions['bid_price'].sum(), 1),
            'effective_cpm': impressions['bid_price'].mean() * 1000,
        }