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

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

Атрибуция last-click даёт Google последнего перехода 100% кредита за конверсию, хотя клиент до этого видел баннер, читал статью в блоге и смотрел видео. Это приводит к перекосам в бюджете: вы льёте деньги в каналы, которые лишь закрывают сделку, а не привлекают. Мы решаем эту проблему с помощью data-driven атрибуции на базе ML — строим модель, которая справедливо распределяет ценность между всеми touchpoints.

Наш подход использует Shapley values и Markov chains, дополненные LLM-аналитикой. Результат — прозрачное распределение конверсий, выявление недооценённых каналов и рост ROAS до 30% за квартал. Мы внедрили такие системы для 15+ проектов, включая e-commerce с оборотами от $10M.

Пример: ритейлер с оборотом $50M использовал last-click, считая контекстную рекламу главным каналом. После внедрения Shapley-атрибуции выяснилось, что 35% конверсий начинались с email-рассылок, которые считались неэффективными. Перераспределение 20% бюджета в email дало +28% ROAS за два месяца.

Почему last-click атрибуция губит ваш бюджет?

Last-click — это baseline, который показывают все системы аналитики. Но он слеп: если клиент пришёл с баннера, потом через email и наконец конвертировался с контекстной рекламы, последней точке достаётся всё. Вы сокращаете баннерный бюджет, думая, что он не работает, хотя именно он запустил воронку. На практике перераспределение на основе multi-touch атрибуции даёт +20-35% к эффективности расходов.

Как мы строим multi-touch атрибуцию на ML

Мы используем комбинацию методов: Shapley value (теоретически справедливое распределение) для 5-8 каналов и Markov chain для масштабирования. Для интерпретации результатов подключаем LLM — Claude или GPT-4 пишут отчёт с конкретными рекомендациями по бюджету.

Вот фрагмент нашего пайплайна:

Сбор данных о 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

        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

        def compute_conversion_rate(transition_matrix):
            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_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

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

Как LLM улучшает интерпретацию атрибуции?

После расчёта Shapley или Markov chain мы передаём таблицу с attributions и costs в LLM (Claude или GPT-4). Модель генерирует отчёт на естественном языке: указывает, какие каналы недооценены (высокий вклад, низкие расходы), какие переоценены, и даёт конкретные рекомендации по перераспределению бюджета с процентами. Это экономит время аналитиков и снижает риск человеческих ошибок.

Этапы внедрения

Сроки: от 2 недель для базового решения, до 6 недель с LLM и дашбордами. Стоимость рассчитывается индивидуально под объём данных и сложность интеграции. Получите консультацию по вашему проекту — свяжитесь с нами, и мы покажем, сколько бюджета уходит впустую.

Почему клиенты выбирают нас?

Мы занимаемся AI-решениями для маркетинга 5+ лет. За это время реализовали более 15 проектов по атрибуции для e-commerce, SaaS и fintech. Наша система гарантирует прозрачность: вы видите вклад каждого канала и можете экспериментально проверять гипотезы. Средний рост ROAS после внедрения — 20-30% за первый квартал. Закажите предварительный аудит ваших данных — это бесплатно.

Data Engineering для ML: пайплайны, разметка и качество данных

«У нас много данных» — фраза, которая на деле часто означает «у нас много сырых логов в S3, которые никто не трогал два года». Перед тем как обучить модель, нужно понять, что вообще есть: какова структура, есть ли дубли, как часто меняется схема, насколько репрезентативна выборка.

Data Engineering для ML — не просто ETL. Это построение воспроизводимой инфраструктуры данных, которая делает обучение моделей надёжным, а переобучение — предсказуемым. По опыту нашей команды (8 лет в дата-инжиниринге, более 30 проектов в ML) каждая вторая проблема в продакшене связана не с архитектурой модели, а с качеством данных.

ETЛ-пайплайны для ML: чем отличаются от BI

ETL для аналитики и ETL для ML — разные задачи. В аналитике важна агрегация, в ML — индивидуальные записи с историей. В аналитике train/val/test split не нужен, в ML — критичен. В аналитике skew данных мешает интерпретации, в ML — напрямую влияет на качество модели.

Инструменты. Apache Spark (Wikipedia) для больших объёмов (10GB+): PySpark с DataFrames, оптимизации через partitioning и caching. dbt для трансформаций поверх DWH (Snowflake, BigQuery, Redshift) — декларативно, версионируется, тестируется. Pandas + Polars для объёмов до нескольких GB — Polars в 5-10x быстрее Pandas на типичных трансформациях.

Temporal splits. Для ML важно, что split по времени, а не случайный. Если данные временные (транзакции, события пользователей), случайный split даёт data leakage: модель видит «будущие» данные при обучении. Правило: train на периоде T1-T2, validation на T2-T3 (с gap для предотвращения leakage), test на T3-T4. Неправильный split может стоить 10–15% качества модели на валидации.

Инкрементальные пайплайны. Модель переобучается еженедельно на новых данных. Нужен пайплайн, который инкрементально добавляет новые записи к обучающей выборке, не перегружая всё с нуля. Delta Lake или Apache Iceberg — форматы с ACID-транзакциями, Change Data Capture, time travel.

Как избежать training-serving skew с помощью Feature Store

Feature Store решает проблему рассинхронизации между обучением и инференсом. Самая коварная ошибка в ML-инфраструктуре — training-serving skew: признак считается по-разному в обучении и в продакшене. Модель учится на «правильных» данных, а инференс получает другие.

Feast (open source) — офлайн store на Parquet/Delta в S3 для обучения, онлайн store на Redis для low-latency инференса (<10ms). Feature definitions как Python-код:

from feast import FeatureView, Field
from feast.types import Float32, Int64

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_session_duration", dtype=Float32),
    ],
    ttl=timedelta(days=7),
    source=user_features_source,
)

Один definition, используется везде. Нет расхождений.

Потоковые признаки. Когда признак должен обновляться в реальном времени (количество транзакций за последние 10 минут), нужна потоковая обработка. Apache Kafka + Apache Flink или Kafka Streams для вычисления признаков в реальном времени → запись в онлайн store. Сложнее, дороже, нужно только когда staleness признаков критична для качества.

Разметка данных: как не потратить бюджет впустую

Разметка — самая трудоёмкая и недооцениваемая часть ML-проекта. Плохо размеченные данные не исправит никакая архитектура.

Label Studio — open source, поддерживает разметку изображений (bounding box, polygon, segmentation), текста (NER, классификация), аудио, видео. Поднимается за 10 минут через Docker. Для небольших команд — первый выбор.

Оценка качества разметки. Inter-annotator agreement — насколько согласны разметчики между собой. Cohen's Kappa > 0.8 — хорошо, 0.6-0.8 — приемлемо, < 0.6 — задача неоднозначна или инструкция плохая. Пересечение разметок (10-20% примеров размечают два независимых аннотатора) — обязательная практика.

Active learning. Не размечать случайные примеры, а выбирать те, на которых модель наиболее неуверена (low confidence, high uncertainty). Позволяет добиться того же качества при 50-70% объёма разметки. Modals, Prodigy, Label Studio поддерживают active learning workflows. На одном из проектов для NLP мы сократили бюджет на разметку в 2,5 раза за счёт active learning.

Синтетические данные. Когда реальных данных мало или получить их дорого. Для CV: рендеринг в Blender/Unity с реалистичными текстурами (domain randomization). Для NLP: parafrase через LLM, backtranslation. Риск: модель обучается на distribution синтетических данных, а не реальных — нужна осторожность и проверка на реальном holdout.

Качество данных: валидация и мониторинг

Great Expectations — de facto стандарт для data validation в ML-пайплайнах. Expectations — это декларативные утверждения о данных: «колонка age содержит значения от 0 до 120», «колонка user_id не содержит null», «распределение amount не отклоняется более чем на 20% от baseline». Запускается в пайплайне, при провале — блокирует прохождение.

Pandera — Pythonic alternative для pandas/polars DataFrames. Schema-based validation с type hints:

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, nullable=False),
    "score": pa.Column(float, pa.Check.between(0, 1)),
    "label": pa.Column(str, pa.Check.isin(["positive", "negative", "neutral"])),
})

Data freshness. Модель ожидает данные за последние N дней. ETL упал, данные не обновились — модель использует устаревшие признаки. Мониторинг свежести данных: timestamp последней записи в каждой таблице, алерт при задержке > порога.

Дедупликация. Дубликаты в обучающей выборке завышают метрики (одни и те же примеры в train и val) и искажают веса модели. MinHash LSH для приближённой дедупликации больших датасетов. Для точной — хэш по нормализованному контенту.

Инструменты валидации: сравнение

Хранилища и форматы

Для большинства ML-проектов на старте: Parquet в S3 + DVC для версионирования. Delta Lake или Iceberg — когда появляется потребность в инкрементальных обновлениях или time travel.

Что входит в проект по дата-инжинирингу для ML

Мы предоставляем полный цикл:

• Аудит существующих данных и пайплайнов (1 неделя).
• Проектирование архитектуры: выбор инструментов, форматов, способов разметки.
• Реализация ETL/ELT пайплайна с валидацией и мониторингом.
• Документация кода и процессов (model card, data card).
• Обучение вашей команды работе с пайплайном.
• SLA на сопровождение и поддержку.

Как мы строим пайплайн: пошагово

1. Аудит существующих данных. Профилирование: ydata-profiling (бывший pandas-profiling) генерирует HTML-репорт со статистиками, дистрибуциями, корреляциями, missing values за минуты.
2. Проектирование пайплайна. Определяем источники данных, частоту обновления, требования к latency признаков, объёмы.
3. Реализация и тестирование. Unit-тесты на трансформации, integration-тесты на пайплайн, data validation через Great Expectations.
4. Деплой и мониторинг. Алерты на freshness, quality checks, аномалии в объёмах данных.

Почему стоит доверить это нам

Мы занимаемся дата-инжинирингом и ML с 2016 года. За это время реализовали более 40 проектов — от построения пайплайнов для NLP-моделей до разметки датасетов для компьютерного зрения. Гарантируем воспроизводимость пайплайнов и полную прозрачность процессов. В каждом проекте используем инструменты с открытым исходным кодом, чтобы вы не были привязаны к вендору.

Свяжитесь с нами для бесплатного аудита ваших данных — оценим текущий пайплайн и предложим roadmap. Закажите построение ML-пайплайна под ключ.