Реализация AI-предсказания Next Best Action для маркетинга

Реализация AI-системы Next Best Action для маркетинга

Next Best Action (NBA) — система, определяющая оптимальное следующее действие для каждого клиента в каждый момент времени. Не "что ему предложить" (product recommendation), а "что сделать" — позвонить, отправить email со скидкой, показать баннер, подключить менеджера или не делать ничего. Reinforcement Learning + бизнес-правила.

Контекстуальный бандит для NBA

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

class NextBestActionEngine:
    """Контекстуальный бандит для выбора следующего действия"""

    def __init__(self):
        self.actions = {
            'email_discount': {'cost': 10, 'type': 'outbound'},
            'push_notification': {'cost': 1, 'type': 'outbound'},
            'call_outbound': {'cost': 50, 'type': 'outbound'},
            'show_banner': {'cost': 0.5, 'type': 'inbound'},
            'offer_upgrade': {'cost': 0, 'type': 'inbound'},
            'no_action': {'cost': 0, 'type': 'none'}
        }
        self.action_models = {}  # Отдельная модель для каждого действия
        self.scaler = StandardScaler()

    def train(self, history: pd.DataFrame):
        """
        history: user_id, context_features..., action_taken, reward (LTV delta)
        """
        X = self.scaler.fit_transform(
            history.drop(columns=['user_id', 'action_taken', 'reward']).fillna(0)
        )

        for action in self.actions:
            mask = history['action_taken'] == action
            if mask.sum() < 50:
                continue

            # Модель: при каком контексте действие приносит высокий reward
            action_rewards = history.loc[mask, 'reward']
            y = (action_rewards > action_rewards.median()).astype(int)

            model = LogisticRegression(C=1.0, max_iter=200)
            model.fit(X[mask], y)
            self.action_models[action] = model

    def recommend_action(self, user: dict,
                          business_constraints: dict = None) -> dict:
        """Выбор оптимального действия"""
        context_features = self._extract_context(user)
        X = self.scaler.transform([context_features])

        action_scores = {}

        for action, meta in self.actions.items():
            # Применяем бизнес-ограничения
            if business_constraints:
                if meta['cost'] > business_constraints.get('max_action_cost', 1000):
                    continue
                if (action in business_constraints.get('blocked_actions', [])):
                    continue

            # Предсказание вознаграждения
            if action in self.action_models:
                reward_prob = self.action_models[action].predict_proba(X)[0][1]
            else:
                reward_prob = 0.3  # Prior для необученных действий

            # Ожидаемый ROI = ожидаемая выручка - стоимость
            expected_revenue = reward_prob * user.get('expected_clv', 100)
            expected_roi = expected_revenue - meta['cost']

            # Учёт усталости от коммуникаций
            if meta['type'] == 'outbound':
                communications_7d = user.get('communications_7d', 0)
                fatigue_penalty = max(0, 1 - 0.3 * communications_7d)
                expected_roi *= fatigue_penalty

            action_scores[action] = {
                'expected_roi': expected_roi,
                'reward_probability': reward_prob,
                'cost': meta['cost']
            }

        # Выбор действия с максимальным ROI
        best_action = max(action_scores, key=lambda x: action_scores[x]['expected_roi'])

        return {
            'recommended_action': best_action,
            'expected_roi': action_scores[best_action]['expected_roi'],
            'reward_probability': action_scores[best_action]['reward_probability'],
            'all_scores': action_scores
        }

    def _extract_context(self, user: dict) -> list:
        return [
            user.get('days_since_last_purchase', 30),
            user.get('total_orders', 0),
            user.get('ltv', 0),
            user.get('churn_probability', 0.5),
            user.get('email_open_rate', 0.2),
            user.get('age_months', 12),
            user.get('avg_order_value', 100),
            user.get('support_tickets_30d', 0),
            user.get('website_visits_7d', 0),
            user.get('communications_7d', 0)
        ]

Оркестрация NBA в реальном времени

class NBAOrchestrator:
    """Real-time оркестрация действий"""

    def __init__(self, nba_engine: NextBestActionEngine):
        self.engine = nba_engine
        self.action_executors = {}

    def process_customer_event(self, event_type: str,
                                user: dict) -> dict:
        """Триггерная обработка событий"""
        # Контекст события влияет на NBA
        event_modifiers = {
            'cart_abandoned': {'churn_probability': +0.2, 'urgency': 'high'},
            'product_viewed_3x': {'intent_score': 0.8},
            'price_page_viewed': {'price_sensitivity': +0.3},
            'support_ticket_opened': {'satisfaction': -0.5}
        }

        enriched_user = {**user}
        if event_type in event_modifiers:
            for key, delta in event_modifiers[event_type].items():
                if isinstance(delta, (int, float)):
                    enriched_user[key] = enriched_user.get(key, 0) + delta
                else:
                    enriched_user[key] = delta

        # Ограничения на основе события
        constraints = {}
        if event_type == 'cart_abandoned':
            constraints['allowed_actions'] = ['email_discount', 'push_notification', 'show_banner']

        decision = self.engine.recommend_action(enriched_user, constraints)

        # Логирование для обучения
        self._log_decision(user['user_id'], event_type, decision)

        return decision

    def _log_decision(self, user_id: str, event: str, decision: dict):
        """Запись решения для offline обучения"""
        import json
        import datetime
        log_entry = {
            'timestamp': datetime.datetime.now().isoformat(),
            'user_id': user_id,
            'trigger_event': event,
            'action_taken': decision['recommended_action'],
            'expected_roi': decision['expected_roi']
        }
        # В реальности: запись в Kafka/DB
        print(f"NBA: {json.dumps(log_entry)}")

NBA с RL показывает ROAS 4-8x на маркетинговые расходы против 2-3x у правил-based подхода. Главный insight: для ~30-40% клиентов оптимальное действие — "ничего не делать" (сэкономить деньги). Это не ошибка модели, а правильное решение — избыточные коммуникации повышают churn.