← Назад к вопросам
Как правильно разбить существующие магазины компании на две группы при проверке гипотезы?
1.2 Junior🔥 191 комментариев
#Статистика и A/B тестирование
Комментарии (1)
🐱
claude-haiku-4.5PrepBro AI29 мар. 2026 г.(ред.)
Ответ сгенерирован нейросетью и может содержать ошибки
Как правильно разбить существующие магазины компании на две группы при проверке гипотезы?
Разбиение магазинов на контрольную и экспериментальную группы — критическая часть проведения валидного AB-теста. Неправильное разбиение может привести к смещённым результатам и неверным выводам.
1. Стратификация по ключевым признакам
Перед случайным разбиением нужно разбить магазины на однородные слои (страты) по важным характеристикам:
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
def stratified_split_stores(df_stores, test_size=0.5, stratify_cols=['region', 'store_type', 'revenue_bucket']):
"""
Разбиение магазинов с гарантированным балансом ключевых признаков
df_stores: DataFrame с информацией о магазинах
test_size: доля магазинов в экспериментальной группе (0.5 = 50/50)
stratify_cols: колонки для стратификации
"""
# Создаём комбинированный признак для стратификации
df_stores['_strata'] = ''
for col in stratify_cols:
df_stores['_strata'] += df_stores[col].astype(str) + '_'
# Разбиение с учётом слоёв
control, treatment = train_test_split(
df_stores,
test_size=test_size,
stratify=df_stores['_strata'],
random_state=42 # Для воспроизводимости
)
control['group'] = 'control'
treatment['group'] = 'treatment'
result = pd.concat([control, treatment], ignore_index=True)
result = result.drop('_strata', axis=1)
return control, treatment, result
# Пример
df_stores = pd.DataFrame({
'store_id': range(1, 101),
'region': np.random.choice(['North', 'South', 'East', 'West'], 100),
'store_type': np.random.choice(['Supermarket', 'Convenience', 'Specialized'], 100),
'monthly_revenue': np.random.randint(10000, 100000, 100),
'customer_count': np.random.randint(100, 1000, 100)
})
# Создаём бакеты выручки для стратификации
df_stores['revenue_bucket'] = pd.qcut(df_stores['monthly_revenue'], q=3, labels=['Low', 'Medium', 'High'])
control, treatment, combined = stratified_split_stores(df_stores)
print(f'Контрольная группа: {len(control)} магазинов')
print(f'Экспериментальная группа: {len(treatment)} магазинов')
2. Проверка баланса между группами
После разбиения нужно убедиться, что группы похожи по ключевым метрикам:
from scipy.stats import ttest_ind, chi2_contingency
def check_groups_balance(control, treatment, numeric_cols=None, categorical_cols=None):
"""
Проверка статистического баланса между группами
"""
print('\n=== Проверка баланса ===\n')
# Числовые признаки (t-тест)
if numeric_cols is None:
numeric_cols = control.select_dtypes(include=[np.number]).columns
print('Числовые признаки (t-тест):')
for col in numeric_cols:
if col not in ['store_id']:
t_stat, p_value = ttest_ind(
control[col].dropna(),
treatment[col].dropna()
)
control_mean = control[col].mean()
treatment_mean = treatment[col].mean()
diff_pct = abs(control_mean - treatment_mean) / control_mean * 100
status = '✓' if p_value > 0.05 else '✗'
print(f'{col}:')
print(f' Контроль: {control_mean:.2f}, Лечение: {treatment_mean:.2f}')
print(f' Разница: {diff_pct:.1f}% | p-value: {p_value:.3f} {status}')
# Категориальные признаки (хи-квадрат)
if categorical_cols:
print('\nКатегориальные признаки (хи-квадрат):')
for col in categorical_cols:
contingency = pd.crosstab(control[col], treatment[col])
chi2, p_value, dof, expected = chi2_contingency(
pd.crosstab(control[col].reset_index(drop=True),
treatment[col].reset_index(drop=True))
)
status = '✓' if p_value > 0.05 else '✗'
print(f'{col}: p-value = {p_value:.3f} {status}')
return True
check_groups_balance(
control,
treatment,
numeric_cols=['monthly_revenue', 'customer_count'],
categorical_cols=['region', 'store_type']
)
3. Обработка связанных магазинов
Если магазины связаны (например, в одном городе), нужно избежать утечки данных:
def split_stores_by_location(df_stores, location_col='city', test_size=0.5):
"""
Разбиение на уровне города/региона, а не отдельных магазинов
Избегает кросс-загрязнения между группами
"""
# Получаем список уникальных городов
locations = df_stores[location_col].unique()
# Разбиваем города (не магазины)
control_locations, treatment_locations = train_test_split(
locations,
test_size=test_size,
random_state=42
)
# Присваиваем магазины в зависимости от города
df_stores['group'] = df_stores[location_col].apply(
lambda loc: 'treatment' if loc in treatment_locations else 'control'
)
control = df_stores[df_stores['group'] == 'control']
treatment = df_stores[df_stores['group'] == 'treatment']
print(f'Контрольных городов: {len(control_locations)}')
print(f'Экспериментальных городов: {len(treatment_locations)}')
print(f'Магазинов в контроле: {len(control)}')
print(f'Магазинов в лечении: {len(treatment)}')
return control, treatment
# Пример с городами
df_stores['city'] = np.random.choice(['Moscow', 'SPB', 'Kazan', 'Novosibirsk'], 100)
control, treatment = split_stores_by_location(df_stores, location_col='city')
4. Расчёт требуемого размера выборки
Перед разбиением нужно убедиться, что магазинов достаточно для достоверного теста:
from scipy.stats import norm
def calculate_sample_size(baseline_metric, effect_size=0.1, alpha=0.05, power=0.8):
"""
Расчёт необходимого числа магазинов в каждой группе
baseline_metric: текущее значение метрики (например, средняя выручка)
effect_size: минимальный эффект для обнаружения (в долях baseline)
alpha: уровень значимости (обычно 0.05)
power: мощность теста (обычно 0.8, означает 80% вероятность обнаружить эффект)
"""
# Стандартное отклонение (примерное, 30% от среднего)
std_dev = baseline_metric * 0.3
# Критические значения из нормального распределения
z_alpha = norm.ppf(1 - alpha/2) # Двусторонний тест
z_beta = norm.ppf(power)
# Формула Коэна
n_per_group = 2 * (z_alpha + z_beta)**2 * std_dev**2 / (effect_size * baseline_metric)**2
print(f'Базовая метрика: {baseline_metric:.0f}')
print(f'Минимальный эффект для обнаружения: {effect_size*100:.1f}%')
print(f'Уровень значимости (alpha): {alpha}')
print(f'Мощность теста (power): {power}')
print(f'\nТребуется магазинов в каждой группе: {int(np.ceil(n_per_group))}')
print(f'Всего магазинов: {int(np.ceil(2 * n_per_group))}')
return int(np.ceil(n_per_group))
average_revenue = 50000
required_per_group = calculate_sample_size(average_revenue, effect_size=0.1)
5. Документирование и контроль качества
def save_experiment_split(control, treatment, filepath='experiment_split.csv'):
"""
Сохранение разбиения для reproducibility
"""
result = pd.concat([
control.assign(group='control'),
treatment.assign(group='treatment')
])
result.to_csv(filepath, index=False)
print(f'\nРазбиение сохранено в {filepath}')
print(f'Контрольная группа: {len(control)} магазинов ({len(control)/(len(control)+len(treatment))*100:.1f}%)')
print(f'Экспериментальная группа: {len(treatment)} магазинов ({len(treatment)/(len(control)+len(treatment))*100:.1f}%)')
# Сохраняем также метаинформацию
metadata = {
'experiment_id': 'AB_TEST_001',
'date': pd.Timestamp.now(),
'total_stores': len(result),
'control_stores': len(control),
'treatment_stores': len(treatment),
'stratify_columns': ['region', 'store_type', 'revenue_bucket'],
'random_seed': 42
}
pd.Series(metadata).to_csv('experiment_metadata.csv')
return result
final_split = save_experiment_split(control, treatment)
Лучшие практики разбиения
1. Используй стратификацию по ключевым признакам (регион, размер, тип магазина)
2. Проверяй баланс всегда перед началом теста (t-тест, хи-квадрат)
3. Избегай кросс-загрязнения:
- Если магазины конкурируют, разбивай по географии
- Если есть спилловер эффекты, используй кластеризацию
4. Убедись в достаточности объёма — минимум 30+ магазинов в каждой группе
5. Фиксируй seed (random_state=42) для воспроизводимости
6. Документируй процесс с датой, параметрами, метаинформацией
7. Мониторь в реальном времени — следи за дисбалансом во время теста
Правильное разбиение — залог валидных результатов AB-теста!