Какой дизайн эксперимента сделаешь для введения стикеров?

Дизайн эксперимента для введения стикеров

Этот вопрос проверяет понимание того, как правильно тестировать новую фичу. Стикеры — это интерактивный элемент, который может влиять на engagement, retention, monetization и другие ключевые метрики. Нужен тщательный дизайн эксперимента. Рассмотрим полный подход от гипотезы к анализу.

Шаг 1: Формулировка гипотезы

from dataclasses import dataclass
from datetime import datetime

@dataclass
class Hypothesis:
    statement: str
    hypothesis_type: str
    expected_impact: dict
    success_criteria: dict

# Формулируем нашу гипотезу
hypothesis = Hypothesis(
    statement="Введение стикеров увеличит engagement и retention пользователей",
    hypothesis_type="Behavioral / Engagement",
    expected_impact={
        'primary_metric': 'daily_active_messages_count',
        'expected_lift': 0.10,  # 10% увеличение
        'secondary_metrics': ['session_duration', 'retention_day7', 'daily_active_users']
    },
    success_criteria={
        'statistical_significance': 0.05,
        'minimum_effect_size': 0.08,  # 8% это OK
        'confidence_level': 0.80
    }
)

print("=== ГИПОТЕЗА ===")
print(f"H0 (Нулевая): Стикеры НЕ влияют на engagement")
print(f"H1 (Альтернативная): Стикеры УВЕЛИЧИВАЮТ engagement на {hypothesis.expected_impact['expected_lift']*100}%")
print(f"\nПервичная метрика: {hypothesis.expected_impact['primary_metric']}")
print(f"Вторичные метрики: {hypothesis.expected_impact['secondary_metrics']}")

Шаг 2: Расчёт размера выборки

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

def calculate_sample_size_for_experiment(
    baseline_mean: float,
    baseline_std: float,
    expected_lift_pct: float,
    alpha: float = 0.05,
    power: float = 0.80
) -> dict:
    """
    Рассчитать требуемый размер выборки
    
    Args:
        baseline_mean: базовое среднее значение метрики (напр., среднее кол-во сообщений)
        baseline_std: стандартное отклонение
        expected_lift_pct: ожидаемый лифт в %
        alpha: уровень значимости (обычно 0.05)
        power: статистическая мощность (обычно 0.80)
    """
    
    # Базовые метрики (предполагаемые для мессенджера со стикерами)
    # Среднее кол-во сообщений в день: 5
    # Стандартное отклонение: 3
    
    target_mean = baseline_mean * (1 + expected_lift_pct)
    effect_size = (target_mean - baseline_mean) / baseline_std
    
    # Рассчитываем требуемый размер выборки
    n_per_group = tt_ind_solve_power(
        effect_size=effect_size,
        alpha=alpha,
        power=power,
        ratio=1.0,  # Одинаковые группы
        alternative='two-sided'
    )
    
    return {
        'effect_size': effect_size,
        'n_per_group': int(n_per_group),
        'total_users': int(n_per_group * 2),
        'baseline_mean': baseline_mean,
        'target_mean': target_mean
    }

# Расчёты для стикеров
results = calculate_sample_size_for_experiment(
    baseline_mean=5,      # 5 сообщений в день
    baseline_std=3,       # Стандартное отклонение
    expected_lift_pct=0.10  # 10% лифт
)

print("\n=== РАЗМЕР ВЫБОРКИ ===")
print(f"Базовое среднее: {results['baseline_mean']:.1f} сообщений/день")
print(f"Ожидаемое среднее: {results['target_mean']:.2f} сообщений/день")
print(f"Effect size (Cohen's d): {results['effect_size']:.3f}")
print(f"Требуется пользователей на группу: {results['n_per_group']:,}")
print(f"Всего пользователей: {results['total_users']:,}")
print(f"\nПри 10000 DAU в день → {results['total_users'] / 10000:.1f} дней теста")

Шаг 3: Выбор дизайна эксперимента

Вариант A: Стандартный A/B тест (рекомендуется)

@dataclass
class ExperimentDesign:
    name: str
    control_group: dict
    treatment_group: dict
    split_ratio: tuple
    duration_days: int
    target_metrics: list
    exclusion_criteria: list

design_ab = ExperimentDesign(
    name="Standard A/B Test",
    control_group={
        'name': 'Control (без стикеров)',
        'treatment': None,
        'size_pct': 50
    },
    treatment_group={
        'name': 'Treatment (со стикерами)',
        'treatment': 'full_sticker_pack',  # Все стикеры включены
        'size_pct': 50
    },
    split_ratio=(0.5, 0.5),
    duration_days=14,  # 2 недели
    target_metrics=[
        'daily_messages_count',
        'session_duration',
        'daily_active_users',
        'retention_day7',
        'sticker_adoption_rate',  # Сколько % юзеров используют стикеры
        'sticker_usage_frequency'  # Как часто используют
    ],
    exclusion_criteria=[
        'new_users_less_than_7_days',  # Только активные юзеры
        'bot_accounts',
        'banned_users'
    ]
)

print(f"\n=== ДИЗАЙН ЭКСПЕРИМЕНТА: {design_ab.name} ===")
print(f"Контрольная группа: {design_ab.control_group['name']} ({design_ab.control_group['size_pct']}%)")
print(f"Экспериментальная группа: {design_ab.treatment_group['name']} ({design_ab.treatment_group['size_pct']}%)")
print(f"Длительность: {design_ab.duration_days} дней")
print(f"Исключение: {len(design_ab.exclusion_criteria)} критериев")

Вариант B: Градуальное раскатывание (Gradual Rollout)

Для снижения риска:

design_gradual = {
    'name': 'Gradual Rollout',
    'phase_1': {
        'duration_days': 7,
        'percentage': 0.05,  # 5% юзеров
        'goal': 'Обнаружить критические баги'
    },
    'phase_2': {
        'duration_days': 7,
        'percentage': 0.10,  # 10% юзеров
        'goal': 'Проверить стабильность на 10%'
    },
    'phase_3': {
        'duration_days': 7,
        'percentage': 0.50,  # 50% юзеров
        'goal': 'Полный эксперимент с достаточной выборкой'
    },
    'total_duration': 21,  # 3 недели
}

print("\n=== ГРАДУАЛЬНОЕ РАСКАТЫВАНИЕ ===")
for phase, details in design_gradual.items():
    if phase != 'total_duration':
        print(f"\n{phase.upper()}:")
        print(f"  Дней: {details['duration_days']}")
        print(f"  Процент юзеров: {details['percentage']*100:.0f}%")
        print(f"  Цель: {details['goal']}")

Вариант C: Многовариантный тест (Multivariate / A/B/C)

Если есть несколько вариантов стикеров:

design_multivariate = {
    'groups': [
        {'name': 'Control (без стикеров)', 'size_pct': 25},
        {'name': 'Variant A (стикеры, всегда видны)', 'size_pct': 25},
        {'name': 'Variant B (стикеры, иконка)', 'size_pct': 25},
        {'name': 'Variant C (стикеры, popup на первый чат)', 'size_pct': 25},
    ],
    'duration_days': 21,  # Нужно больше времени для 4 групп
    'analysis_type': 'One-way ANOVA',
}

print("\n=== МНОГОВАРИАНТНЫЙ ТЕСТ ===")
for group in design_multivariate['groups']:
    print(f"  {group['name']:45s}: {group['size_pct']:3.0f}%")

Шаг 4: Техническая реализация

from enum import Enum
import hashlib

class ExperimentVariant(Enum):
    CONTROL = 'control'
    TREATMENT = 'treatment_stickers'

def assign_user_to_variant(user_id: str, experiment_id: str) -> ExperimentVariant:
    """
    Детерминированное распределение юзера в группу
    (один юзер всегда в одной группе)
    """
    hash_input = f"{user_id}_{experiment_id}"
    hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
    
    # 50/50 split
    if hash_value % 2 == 0:
        return ExperimentVariant.CONTROL
    else:
        return ExperimentVariant.TREATMENT

# Пример
user_id = "user_12345"
experiment_id = "stickers_v1"
variant = assign_user_to_variant(user_id, experiment_id)
print(f"\nПользователь {user_id} → {variant.value}")

# Логирование события
def log_sticker_usage(user_id: str, event_type: str, variant: ExperimentVariant):
    """
    Логируем использование стикеров
    """
    event = {
        'timestamp': datetime.now().isoformat(),
        'user_id': user_id,
        'experiment_variant': variant.value,
        'event_type': event_type,  # 'sticker_viewed', 'sticker_sent', etc.
        'metadata': {}
    }
    # Отправляем в Analytics
    print(f"Event logged: {event}")

# Примеры логирования
log_sticker_usage(user_id, 'sticker_viewed', variant)
log_sticker_usage(user_id, 'sticker_sent', variant)

Шаг 5: Метрики для отслеживания

from typing import Dict, List

class MetricsDefinition:
    # PRIMARY METRICS (главные, для принятия решения)
    primary_metrics = [
        {
            'name': 'Messages per DAU',
            'definition': 'Среднее количество сообщений на активного пользователя в день',
            'unit': 'count',
            'expected_change': '+10%',
            'guardrail': False
        },
        {
            'name': 'Session Duration',
            'definition': 'Среднее время проведенное в приложении',
            'unit': 'seconds',
            'expected_change': '+15%',
            'guardrail': False
        },
    ]
    
    # SECONDARY METRICS (поддерживающие)
    secondary_metrics = [
        {
            'name': 'DAU',
            'definition': 'Количество дневных активных пользователей',
            'unit': 'count',
            'expected_change': 'No change (guardrail)',
            'guardrail': True  # Убедимся, что мы не теряем юзеров
        },
        {
            'name': 'Retention D7',
            'definition': '% пользователей, вернувшихся на день 7',
            'unit': '%',
            'expected_change': '+5%',
            'guardrail': False
        },
        {
            'name': 'Sticker Adoption',
            'definition': '% юзеров, которые хотя бы раз используют стикеры',
            'unit': '%',
            'expected_change': '+30%',
            'guardrail': False
        },
    ]
    
    # GUARDRAIL METRICS (чтобы не сломать ничего)
    guardrail_metrics = [
        {
            'name': 'App Crashes',
            'threshold_increase': '< 10%',  # Не может увеличиться на 10%+
            'critical': True
        },
        {
            'name': 'Error Rate',
            'threshold_increase': '< 5%',
            'critical': True
        },
        {
            'name': 'Churn Rate',
            'threshold_increase': '< 2%',
            'critical': False
        },
    ]

# Вывод
metrics = MetricsDefinition()
print("\n=== PRIMARY METRICS ===")
for m in metrics.primary_metrics:
    print(f"{m['name']:25s} | {m['definition']}")
    print(f"{'':25s}   Ожидаемо: {m['expected_change']}\n")

print("\n=== GUARDRAIL METRICS (не ломай!) ===")
for m in metrics.guardrail_metrics:
    print(f"{m['name']:25s} | Лимит: {m['threshold_increase']}")

Шаг 6: Анализ результатов

import pandas as pd
from scipy import stats

class ExperimentAnalysis:
    def __init__(self, control_data: pd.DataFrame, treatment_data: pd.DataFrame):
        self.control = control_data
        self.treatment = treatment_data
    
    def analyze(self) -> dict:
        results = {}
        
        # Анализируем каждую метрику
        for metric in ['messages', 'session_duration', 'retention']:
            control_values = self.control[metric]
            treatment_values = self.treatment[metric]
            
            # T-тест
            t_stat, p_value = stats.ttest_ind(control_values, treatment_values)
            
            control_mean = control_values.mean()
            treatment_mean = treatment_values.mean()
            lift = ((treatment_mean - control_mean) / control_mean) * 100
            
            # Cohen's d (effect size)
            pooled_std = np.sqrt((control_values.std()**2 + treatment_values.std()**2) / 2)
            cohens_d = (treatment_mean - control_mean) / pooled_std
            
            results[metric] = {
                'control_mean': control_mean,
                'treatment_mean': treatment_mean,
                'lift': lift,
                'p_value': p_value,
                'cohens_d': cohens_d,
                'significant': p_value < 0.05
            }
        
        return results
    
    def create_report(self, results: dict) -> str:
        report = "\n=== РЕЗУЛЬТАТЫ ЭКСПЕРИМЕНТА ==="
        
        for metric, stats_dict in results.items():
            report += f"\n\n{metric.upper()}:\n"
            report += f"  Control: {stats_dict['control_mean']:.2f}\n"
            report += f"  Treatment: {stats_dict['treatment_mean']:.2f}\n"
            report += f"  Лифт: {stats_dict['lift']:+.1f}%\n"
            report += f"  P-value: {stats_dict['p_value']:.4f}\n"
            report += f"  Effect size (Cohen's d): {stats_dict['cohens_d']:.3f}\n"
            
            if stats_dict['significant']:
                report += f"  ✓ СТАТИСТИЧЕСКИ ЗНАЧИМО (p < 0.05)\n"
            else:
                report += f"  ✗ НЕ значимо (p >= 0.05)\n"
        
        return report

# Пример вывода
print("\n=== ПРИМЕР ОТЧЁТА ===")
print("""
=== РЕЗУЛЬТАТЫ ЭКСПЕРИМЕНТА ===

MESSAGES PER DAU:
  Control: 5.23
  Treatment: 5.75
  Лифт: +10.0%
  P-value: 0.0234
  Effect size (Cohen's d): 0.312
  ✓ СТАТИСТИЧЕСКИ ЗНАЧИМО (p < 0.05)

SESSION DURATION:
  Control: 8.45 мин
  Treatment: 9.23 мин
  Лифт: +9.2%
  P-value: 0.0892
  Effect size (Cohen's d): 0.156
  ✗ НЕ значимо (p >= 0.05)

RETENTION D7:
  Control: 42.1%
  Treatment: 44.3%
  Лифт: +5.2%
  P-value: 0.1234
  Effect size (Cohen's d): 0.089
  ✗ НЕ значимо (p >= 0.05)
""")

Шаг 7: Решение и Рекомендация

class ExperimentDecision:
    def __init__(self, results: dict):
        self.results = results
    
    def make_decision(self) -> str:
        """
        Принять решение на основе результатов
        """
        significant_wins = 0
        significant_losses = 0
        
        for metric, stats_dict in self.results.items():
            if stats_dict['significant']:
                if stats_dict['lift'] > 0:
                    significant_wins += 1
                else:
                    significant_losses += 1
        
        if significant_losses > 0:
            return "ROLLBACK - Есть значимое ухудшение"
        elif significant_wins > 1:
            return "ROLLOUT - Есть значимое улучшение"
        elif significant_wins == 1:
            return "INCONCLUSIVE - Нужны уточнения или расширение теста"
        else:
            return "NO EFFECT - Стикеры не влияют на метрики"

# Пример
decision_maker = ExperimentDecision({
    'messages': {'significant': True, 'lift': 10.0},
    'session_duration': {'significant': False, 'lift': 9.2},
    'retention': {'significant': False, 'lift': 5.2}
})

print(f"\n=== РЕШЕНИЕ ===")
print(decision_maker.make_decision())
print("\nПримечание: Стикеры УВЕЛИЧИВАЮТ количество сообщений на 10% (значимо).")
print("Время в приложении и retention растут, но не значимо.")
print("РЕКОМЕНДАЦИЯ: Раскатать стикеры для всех пользователей.")

Практический План Эксперимента (Summary)

experiment_plan = {
    'Гипотеза': 'Стикеры увеличивают engagement (+10%)',
    'Дизайн': 'Стандартный A/B тест, 50/50 split',
    'Размер выборки': 50000 пользователей (25000 на группу)',
    'Длительность': 14 дней,
    'Первичная метрика': 'Messages per DAU',
    'Вторичные': ['Session duration', 'Retention D7'],
    'Guardrails': ['App crashes', 'Error rate'],
    'Уровень значимости': 0.05,
    'Мощность': 0.80,
    'Критерий успеха': 'Lift > 8% и p < 0.05',
    'Запуск': 'Gradual rollout: 5% → 10% → 50%',
    'Риски': ['User confusion with new feature', 'Technical issues'],
    'Rollback condition': 'Crashes > +10% OR Errors > +5%',
}

for key, value in experiment_plan.items():
    print(f"{key:25s}: {value}")