Реализация AI-атрибуции маркетинговых каналов

Реализация AI-атрибуции маркетинговых каналов

Атрибуция last-click даёт Google последнего перехода 100% кредита за конверсию, хотя клиент до этого видел баннер, читал статью в блоге и смотрел видео. Multi-touch атрибуция через ML распределяет ценность конверсии между всеми точками контакта пропорционально их реальному вкладу — что позволяет правильно перераспределять бюджет.

Сбор данных о touchpoints

import pandas as pd
import numpy as np
from anthropic import Anthropic
from itertools import combinations
import json

class MarketingAttribution:
    def __init__(self, touchpoints_df: pd.DataFrame, conversions_df: pd.DataFrame):
        """
        touchpoints_df: user_id, channel, timestamp, campaign, cost
        conversions_df: user_id, conversion_time, value
        """
        self.touchpoints = touchpoints_df
        self.conversions = conversions_df
        self.llm = Anthropic()

    def build_user_journeys(self, lookback_days: int = 30) -> pd.DataFrame:
        """Строит путь каждого пользователя до конверсии"""
        journeys = []

        for _, conv in self.conversions.iterrows():
            user_id = conv['user_id']
            conv_time = pd.to_datetime(conv['conversion_time'])
            lookback_start = conv_time - pd.Timedelta(days=lookback_days)

            # Touchpoints до конверсии
            user_touches = self.touchpoints[
                (self.touchpoints['user_id'] == user_id) &
                (pd.to_datetime(self.touchpoints['timestamp']) >= lookback_start) &
                (pd.to_datetime(self.touchpoints['timestamp']) <= conv_time)
            ].sort_values('timestamp')

            if len(user_touches) == 0:
                continue

            journeys.append({
                'user_id': user_id,
                'conversion_value': conv['value'],
                'conversion_time': conv_time,
                'journey': user_touches['channel'].tolist(),
                'timestamps': user_touches['timestamp'].tolist(),
                'total_touchpoints': len(user_touches),
                'journey_days': (conv_time - pd.to_datetime(user_touches['timestamp'].iloc[0])).days
            })

        return pd.DataFrame(journeys)

Data-Driven атрибуция (Shapley Values)

    def shapley_attribution(self, journeys_df: pd.DataFrame) -> pd.DataFrame:
        """
        Game-theoretic атрибуция через Shapley values.
        Каждый канал получает свой справедливый вклад.
        """
        # Уникальные каналы
        all_channels = set()
        for journey in journeys_df['journey']:
            all_channels.update(journey)

        # Конверсионная ценность каждой коалиции каналов
        coalition_values = {}

        for _, row in journeys_df.iterrows():
            journey_set = frozenset(row['journey'])
            if journey_set not in coalition_values:
                coalition_values[journey_set] = {'conversions': 0, 'value': 0}
            coalition_values[journey_set]['conversions'] += 1
            coalition_values[journey_set]['value'] += row['conversion_value']

        # Shapley value для каждого канала
        shapley_values = {ch: 0.0 for ch in all_channels}

        for channel in all_channels:
            # Маргинальный вклад при добавлении канала к каждому подмножеству
            other_channels = all_channels - {channel}

            for r in range(len(other_channels) + 1):
                for coalition in combinations(other_channels, r):
                    coalition_set = frozenset(coalition)
                    coalition_with = frozenset(coalition) | {channel}

                    v_with = coalition_values.get(coalition_with, {}).get('value', 0)
                    v_without = coalition_values.get(coalition_set, {}).get('value', 0)

                    marginal = v_with - v_without
                    n = len(all_channels)
                    weight = (
                        np.math.factorial(r) * np.math.factorial(n - r - 1) /
                        np.math.factorial(n)
                    )
                    shapley_values[channel] += weight * marginal

        # Нормализация
        total = sum(shapley_values.values())
        attribution = pd.DataFrame([
            {
                'channel': ch,
                'attributed_value': val,
                'attribution_pct': val / total * 100 if total > 0 else 0
            }
            for ch, val in shapley_values.items()
        ]).sort_values('attributed_value', ascending=False)

        return attribution

Markov Chain атрибуция

    def markov_chain_attribution(self, journeys_df: pd.DataFrame) -> pd.DataFrame:
        """
        Removal effect: насколько упадёт конверсия без каждого канала.
        Быстрее Shapley, хорошо работает для длинных цепочек.
        """
        # Строим матрицу переходов
        transitions = {}

        for _, row in journeys_df.iterrows():
            journey = ['START'] + row['journey'] + ['CONVERSION']

            for i in range(len(journey) - 1):
                state_from = journey[i]
                state_to = journey[i + 1]

                if state_from not in transitions:
                    transitions[state_from] = {}
                transitions[state_from][state_to] = transitions[state_from].get(state_to, 0) + 1

        # Добавляем null-переходы для non-converted
        non_converted = self.touchpoints[
            ~self.touchpoints['user_id'].isin(self.conversions['user_id'])
        ]
        for _, row in non_converted.groupby('user_id').last().iterrows():
            channel = self.touchpoints[self.touchpoints['user_id'] == row.name]['channel'].iloc[-1]
            if channel not in transitions:
                transitions[channel] = {}
            transitions[channel]['NULL'] = transitions[channel].get('NULL', 0) + 1

        # Вычисление conversion rate для каждой цепочки
        def compute_conversion_rate(transition_matrix):
            """Вероятность достижения CONVERSION из START"""
            # Упрощённый расчёт через статистику переходов
            total_start = sum(transition_matrix.get('START', {}).values())
            conv_from_start = transition_matrix.get('START', {}).get('CONVERSION', 0)
            return conv_from_start / total_start if total_start > 0 else 0

        base_cr = compute_conversion_rate(transitions)

        all_channels = set()
        for journey in journeys_df['journey']:
            all_channels.update(journey)

        removal_effects = {}
        for channel in all_channels:
            # Убираем канал из матрицы переходов
            modified_transitions = {
                k: {v: c for v, c in vals.items() if v != channel}
                for k, vals in transitions.items()
                if k != channel
            }
            modified_cr = compute_conversion_rate(modified_transitions)
            removal_effects[channel] = max(0, base_cr - modified_cr)

        total_removal = sum(removal_effects.values())
        total_conversion_value = journeys_df['conversion_value'].sum()

        attribution = pd.DataFrame([
            {
                'channel': ch,
                'removal_effect': effect,
                'attributed_value': effect / total_removal * total_conversion_value if total_removal > 0 else 0,
                'attribution_pct': effect / total_removal * 100 if total_removal > 0 else 0
            }
            for ch, effect in removal_effects.items()
        ]).sort_values('attributed_value', ascending=False)

        return attribution

LLM-анализ результатов атрибуции

    def generate_attribution_report(self, shapley_df: pd.DataFrame,
                                     channel_costs: dict) -> str:
        """Интерпретация результатов атрибуции через LLM"""
        # ROI по каналам
        roi_data = []
        for _, row in shapley_df.iterrows():
            ch = row['channel']
            cost = channel_costs.get(ch, 0)
            attributed = row['attributed_value']
            roi = (attributed - cost) / cost * 100 if cost > 0 else float('inf')
            roi_data.append({
                'channel': ch,
                'cost': cost,
                'attributed_revenue': attributed,
                'roi': roi,
                'attribution_pct': row['attribution_pct']
            })

        roi_data.sort(key=lambda x: x['roi'], reverse=True)

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=600,
            messages=[{
                "role": "user",
                "content": f"""Ты маркетинговый аналитик. Проанализируй результаты multi-touch атрибуции.

Данные по каналам:
{json.dumps(roi_data, ensure_ascii=False, indent=2)}

Дай анализ:
1. Какие каналы недооценены (высокий вклад, низкие расходы)?
2. Какие переоценены (низкий вклад, высокие расходы)?
3. Конкретные рекомендации по перераспределению бюджета (с числами)
4. Каналы для экспериментов

Будь конкретным, называй каналы по имени."""
            }]
        )

        return response.content[0].text

Сравнение моделей атрибуции

Внедрение data-driven атрибуции вместо last-click обычно приводит к перераспределению 20-35% бюджета. Типичный рост ROAS после оптимизации: 15-30% за первый квартал.