Как проводится A/B тест на нынешнем проекте?

Процесс проведения A/B тестирования в production

A/B тестирование — это ключевой инструмент для Product Analyst'а, чтобы принимать обоснованные решения о развитии продукта. Давайте разберемся в полном цикле — от планирования до анализа результатов и выводов.

1. Фаза 1: Планирование и гипотеза

Шаг 1: Определяем гипотезу

Все начинается с четкой гипотезы, которую мы хотим протестировать.

Пример гипотез:
1. ГИПОТЕЗА: Если мы добавим рекомендации товаров на главную,
   ТО конверсия увеличится минимум на 5%
   КОНТРОЛЬ: старый дизайн
   ВАРИАНТ: новый дизайн с рекомендациями

2. ГИПОТЕЗА: Если мы поменяем текст кнопки с "Buy" на "Get Now",
   ТО CTR кнопки увеличится на 3%

3. ГИПОТЕЗА: Если мы отправим push в 18:00 вместо 12:00,
   ТО rate открытия увеличится на 10%

Шаг 2: Определяем метрику успеха

Основная метрика (Primary Metric): конверсия, CTR, AOV
Дополнительные метрики (Secondary Metrics): retention, LTV, support tickets
Метрики риска (Guardrail Metrics): отвалы, bounce rate, negative feedback

2. Расчет размера выборки

from statsmodels.stats.power import proportions_ztest
import numpy as np

def calculate_sample_size(
    baseline_rate: float,  # Текущий уровень конверсии
    expected_lift: float,  # Ожидаемый прирост (5% = 0.05)
    alpha: float = 0.05,   # Уровень значимости (обычно 0.05)
    power: float = 0.80    # Мощность теста (обычно 0.80)
) -> int:
    """
    Расчет необходимого размера выборки для A/B теста
    """
    effect_size = expected_lift  # Относительный прирост
    
    # Для пропорций используем формулу
    from scipy.stats import norm
    
    p1 = baseline_rate
    p2 = baseline_rate * (1 + effect_size)
    
    # Объединенная доля
    p = (p1 + p2) / 2
    
    # Стандартные значения
    z_alpha = norm.ppf(1 - alpha/2)  # двусторонний тест
    z_beta = norm.ppf(power)
    
    # Формула размера выборки
    n = ((z_alpha + z_beta) ** 2 * p * (1 - p) * 2) / ((p2 - p1) ** 2)
    
    return int(np.ceil(n))

# Пример
baseline_cr = 0.025  # 2.5% конверсия
expected_lift = 0.05  # Ожидаем +5% (до 2.625%)

sample_size = calculate_sample_size(baseline_cr, expected_lift)
print(f"Необходимый размер на одну группу: {sample_size:,}")
print(f"Общий размер выборки (2 группы): {sample_size * 2:,}")
# Результат: ~77,000 на группу = 154,000 всего

3. Фаза 2: Техническая реализация

Как работает тестирование на уровне приложения:

# backend/features/ab_test.py
from enum import Enum
from dataclasses import dataclass
import hashlib

class TestVariant(Enum):
    CONTROL = 'control'
    TREATMENT = 'treatment'

@dataclass
class ABTestConfig:
    test_id: str
    name: str
    percentage_split: float = 0.5  # 50/50 распределение
    start_date: str
    end_date: str
    hypothesis: str
    primary_metric: str

class ABTestManager:
    def __init__(self, db_connection):
        self.conn = db_connection
        self.tests = {}  # Кеш активных тестов
    
    def get_variant_for_user(self, user_id: str, test_id: str) -> TestVariant:
        """
        Определяет, в какую группу попадает пользователь
        
        Важно: используем хеширование, чтобы один пользователь 
        всегда попадал в одну и ту же группу
        """
        # Комбинируем user_id и test_id
        combined = f"{user_id}_{test_id}"
        
        # Хешируем
        hash_value = int(hashlib.md5(combined.encode()).hexdigest(), 16)
        
        # Берем остаток по 100
        percentage = hash_value % 100
        
        # Если < 50%, то control, иначе treatment
        if percentage < 50:
            return TestVariant.CONTROL
        else:
            return TestVariant.TREATMENT
    
    def track_event(self, user_id: str, test_id: str, event_type: str, 
                   value: float = None, properties: dict = None):
        """
        Логирование события в контексте A/B теста
        """
        variant = self.get_variant_for_user(user_id, test_id)
        
        # Сохраняем в БД
        event_record = {
            'user_id': user_id,
            'test_id': test_id,
            'variant': variant.value,
            'event_type': event_type,
            'value': value,
            'properties': properties,
            'timestamp': datetime.now(UTC)
        }
        
        # Вставляем в таблицу
        insert_into_table('ab_test_events', event_record)
        
        return variant

# Использование в API
from fastapi import APIRouter, Depends

router = APIRouter()
ab_test_manager = ABTestManager(db_connection)

@router.get("/api/v1/homepage")
async def get_homepage(user_id: str = Header(...)):
    # Определяем, в какой вариант попадает пользователь
    variant = ab_test_manager.get_variant_for_user(user_id, test_id="homepage_design_v1")
    
    if variant == TestVariant.CONTROL:
        # Возвращаем старый дизайн
        content = get_old_homepage_design()
    else:
        # Возвращаем новый дизайн
        content = get_new_homepage_design()
    
    # Логируем просмотр
    ab_test_manager.track_event(
        user_id=user_id,
        test_id="homepage_design_v1",
        event_type="page_view",
        properties={'variant': variant.value}
    )
    
    return content

@router.post("/api/v1/purchase")
async def purchase(user_id: str = Header(...), amount: float = Body(...)):
    # Логируем покупку с информацией о тесте
    variant = ab_test_manager.get_variant_for_user(user_id, test_id="homepage_design_v1")
    
    ab_test_manager.track_event(
        user_id=user_id,
        test_id="homepage_design_v1",
        event_type="purchase",
        value=amount,
        properties={'variant': variant.value}
    )
    
    # Выполняем покупку
    return process_payment(user_id, amount)

4. Схема БД для отслеживания A/B тестов

-- Конфигурация тестов
CREATE TABLE ab_tests (
    test_id UUID PRIMARY KEY,
    test_name VARCHAR,
    hypothesis TEXT,
    control_name VARCHAR,
    treatment_name VARCHAR,
    primary_metric VARCHAR,
    secondary_metrics TEXT[],
    guardrail_metrics TEXT[],
    percentage_split DECIMAL(5,2),  -- 50.00 = 50/50
    started_at TIMESTAMPTZ,
    ended_at TIMESTAMPTZ,
    status VARCHAR,  -- 'running', 'completed', 'cancelled'
    expected_duration_days INT,
    created_at TIMESTAMPTZ
);

-- События в тестах
CREATE TABLE ab_test_events (
    event_id UUID PRIMARY KEY,
    user_id UUID,
    test_id UUID REFERENCES ab_tests(test_id),
    variant VARCHAR,  -- 'control' or 'treatment'
    event_type VARCHAR,  -- 'page_view', 'purchase', 'signup'
    event_value DECIMAL,
    properties JSONB,
    created_at TIMESTAMPTZ,
    created_at_date DATE
) PARTITION BY RANGE (created_at_date);  -- Партиции по датам

-- Индексы для быстрых запросов
CREATE INDEX idx_ab_test_events_test_user 
ON ab_test_events(test_id, user_id);
CREATE INDEX idx_ab_test_events_test_variant 
ON ab_test_events(test_id, variant, event_type);

5. Анализ результатов

-- Метрика: Конверсия по вариантам
WITH user_events AS (
    SELECT
        test_id,
        variant,
        user_id,
        COUNT(CASE WHEN event_type = 'page_view' THEN 1 END) as page_views,
        COUNT(CASE WHEN event_type = 'purchase' THEN 1 END) as purchases,
        SUM(CASE WHEN event_type = 'purchase' THEN event_value ELSE 0 END) as total_revenue
    FROM ab_test_events
    WHERE test_id = 'homepage_design_v1'
      AND created_at >= (SELECT started_at FROM ab_tests WHERE test_id = 'homepage_design_v1')
    GROUP BY 1, 2, 3
)
SELECT
    test_id,
    variant,
    COUNT(DISTINCT user_id) as users,
    SUM(page_views) as total_page_views,
    SUM(purchases) as total_purchases,
    ROUND(100.0 * SUM(purchases) / NULLIF(COUNT(DISTINCT user_id), 0), 2) as conversion_rate,
    ROUND(SUM(total_revenue), 2) as total_revenue,
    ROUND(SUM(total_revenue) / NULLIF(COUNT(DISTINCT user_id), 0), 2) as avg_revenue_per_user
FROM user_events
GROUP BY 1, 2
ORDER BY variant;

6. Статистический анализ в Python

import pandas as pd
from scipy.stats import chi2_contingency
import numpy as np

def analyze_ab_test(test_id: str, db_connection):
    """
    Полный анализ A/B теста
    """
    # Загружаем данные
    query = f"""
        SELECT
            user_id,
            variant,
            COUNT(CASE WHEN event_type = 'purchase' THEN 1 END) as purchased
        FROM ab_test_events
        WHERE test_id = '{test_id}'
        GROUP BY 1, 2
    """
    
    data = pd.read_sql(query, db_connection)
    
    # Построение contingency table
    contingency = pd.crosstab(data['variant'], data['purchased'])
    
    # Chi-square тест
    chi2, p_value, dof, expected = chi2_contingency(contingency)
    
    # Расчет конверсии
    control_users = len(data[data['variant'] == 'control'])
    treatment_users = len(data[data['variant'] == 'treatment'])
    
    control_purchases = len(data[(data['variant'] == 'control') & (data['purchased'] == 1)])
    treatment_purchases = len(data[(data['variant'] == 'treatment') & (data['purchased'] == 1)])
    
    control_cr = control_purchases / control_users
    treatment_cr = treatment_purchases / treatment_users
    
    lift = (treatment_cr - control_cr) / control_cr * 100
    
    # Результаты
    results = {
        'test_id': test_id,
        'chi2_statistic': chi2,
        'p_value': p_value,
        'is_significant': p_value < 0.05,
        'confidence_level': 95 if p_value < 0.05 else 0,
        'control_cr': round(control_cr, 4),
        'treatment_cr': round(treatment_cr, 4),
        'lift_percent': round(lift, 2),
        'control_users': control_users,
        'treatment_users': treatment_users,
        'sample_size_adequate': min(control_users, treatment_users) >= 77000,
        'recommendation': get_recommendation(p_value, lift)
    }
    
    return results

def get_recommendation(p_value: float, lift: float) -> str:
    if p_value < 0.05 and lift > 0:
        return "✅ DEPLOY: Результат статистически значим и положителен"
    elif p_value < 0.05 and lift < 0:
        return "❌ REJECT: Результат статистически значим, но негативен"
    elif p_value >= 0.05 and abs(lift) < 2:
        return "⚠️ INCONCLUSIVE: Нет статистической значимости, результат неопределен"
    else:
        return "❓ EXTEND: Нужно продолжить тест для большей уверенности"

7. Дашборд для мониторинга A/B тестов

-- SQL для real-time мониторинга текущего теста
WITH daily_metrics AS (
    SELECT
        DATE(created_at) as test_date,
        variant,
        COUNT(DISTINCT user_id) as daily_users,
        COUNT(CASE WHEN event_type = 'purchase' THEN 1 END) as daily_purchases,
        ROUND(
            100.0 * COUNT(CASE WHEN event_type = 'purchase' THEN 1 END) / 
            COUNT(DISTINCT user_id),
            2
        ) as daily_cr
    FROM ab_test_events
    WHERE test_id = 'homepage_design_v1'
      AND created_at >= CURRENT_DATE - INTERVAL '30 days'
    GROUP BY 1, 2
)
SELECT
    test_date,
    variant,
    daily_users,
    daily_purchases,
    daily_cr,
    -- Скользящее среднее за 7 дней
    ROUND(
        AVG(daily_cr) OVER (
            PARTITION BY variant 
            ORDER BY test_date 
            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
        ),
        2
    ) as ma7_cr
FROM daily_metrics
ORDER BY test_date DESC, variant;

8. Распространенные ошибки при проведении A/B тестов

class ABTestCommonMistakes:
    """
    Частые ошибки и как их избежать
    """
    
    @staticmethod
    def mistake_1_peeking():
        """
        Ошибка 1: Смотреть результаты до конца теста
        
        ПРОБЛЕМА: Если смотреть результаты каждый день, вероятность 
        получить ложный положительный результат растет.
        
        РЕШЕНИЕ: Не смотрите результаты до плановой даты окончания.
        Установите дату заранее.
        """
        pass
    
    @staticmethod
    def mistake_2_sample_size():
        """
        Ошибка 2: Недостаточный размер выборки
        
        ПРОБЛЕМА: Если выборка слишком мала, результаты неточные.
        
        РЕШЕНИЕ: Всегда рассчитывайте минимальный размер перед тестом.
        """
        pass
    
    @staticmethod
    def mistake_3_multiple_comparisons():
        """
        Ошибка 3: Множественные сравнения
        
        ПРОБЛЕМА: Если тестируем 10 метрик, вероятность ложного 
        положительного результата для одной из них = 40%!
        
        РЕШЕНИЕ: Определите PRIMARY METRIC заранее.
        Тестируйте одно изменение за раз.
        """
        pass
    
    @staticmethod
    def mistake_4_data_quality():
        """
        Ошибка 4: Плохое качество данных
        
        ПРОБЛЕМА: Боты, дублирующиеся события, некорректная фильтрация
        
        РЕШЕНИЕ: Проверьте данные перед анализом:
        - Нет ли ботов?
        - Сбалансированы ли группы?
        - Нет ли пропусков в данных?
        """
        pass

9. Чеклист для A/B теста

Перед запуском:
- ✅ Определена четкая гипотеза
- ✅ Выбрана primary метрика
- ✅ Рассчитан минимальный размер выборки
- ✅ Определена длительность теста (не менее 1-2 недель)
- ✅ Проверена техническая реализация (код не содержит ошибок)
- ✅ Проверена аллокация (50/50 распределение)

Во время теста:
- ✅ Не смотрим результаты до конца
- ✅ Мониторим quality metrics (аномалии, боты)
- ✅ Проверяем, что обе группы сбалансированы

После теста:
- ✅ Проверена статистическая значимость (p < 0.05)
- ✅ Проверен размер выборки
- ✅ Анализированы результаты по подгруппам
- ✅ Проверены guardrail metrics (нет негативных побочных эффектов)
- ✅ Написан отчет с рекомендациями

Имплементация:
- ✅ Если результат положительный, развернули в production
- ✅ Задокументировали результаты и выводы
- ✅ Удалили A/B тестирующий код для победившего варианта

Заключение

A/B тестирование — это строгий научный подход к улучшению продукта. Ключ к успеху: четкая гипотеза, правильный размер выборки, долгий достаточный период теста и честный анализ результатов. Не спешите с выводами и всегда проверяйте статистическую значимость перед тем, как развертывать изменения.