087 종합 프로젝트 - A/B 테스트 분석 자동화

키워드: A/B 테스트, 실험 분석, 통계 분석

개요

A/B 테스트는 두 가지 버전을 비교하여 더 나은 것을 선택하는 실험 방법입니다. 이 글에서는 FLAML AutoML을 활용하여 A/B 테스트 결과 분석을 자동화하고, 전환율 예측 모델을 구축합니다.

실습 환경

• Python 버전: 3.11 권장
• 필요 패키지: flaml[automl], scipy

pip install flaml[automl] pandas numpy scipy scikit-learn

A/B 테스트 데이터 생성

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

np.random.seed(42)

# 087 실험 데이터 생성
n_users = 10000

# 087 사용자 특성
users = pd.DataFrame({
    'user_id': range(n_users),
    'age': np.random.randint(18, 65, n_users),
    'gender': np.random.choice(['M', 'F'], n_users),
    'country': np.random.choice(['US', 'UK', 'DE', 'FR', 'JP'], n_users),
    'device': np.random.choice(['mobile', 'desktop', 'tablet'], n_users, p=[0.6, 0.3, 0.1]),
    'is_new_user': np.random.choice([0, 1], n_users, p=[0.7, 0.3]),
    'visit_count': np.random.poisson(5, n_users),
})

# 087 A/B 그룹 할당
users['group'] = np.random.choice(['control', 'treatment'], n_users)

# 087 전환율 시뮬레이션 (그룹별 다른 기본 확률)
base_conversion = 0.10  # 기본 전환율 10%
treatment_lift = 0.02   # 실험군 리프트 2%

# 087 전환 확률 계산 (특성 기반)
conversion_prob = base_conversion + \
    0.02 * (users['device'] == 'desktop').astype(int) + \
    0.01 * (users['is_new_user'] == 1).astype(int) + \
    0.005 * (users['visit_count'] > 5).astype(int) + \
    treatment_lift * (users['group'] == 'treatment').astype(int)

users['converted'] = np.random.binomial(1, np.clip(conversion_prob, 0, 1))

print("=== A/B 테스트 데이터 ===")
print(f"총 사용자: {len(users)}")
print(f"\n그룹 분포:")
print(users['group'].value_counts())
print(f"\n그룹별 전환율:")
print(users.groupby('group')['converted'].mean())

전통적 A/B 테스트 분석

def ab_test_analysis(data, group_col='group', conversion_col='converted'):
    """전통적 A/B 테스트 분석"""

    # 그룹별 통계
    control = data[data[group_col] == 'control'][conversion_col]
    treatment = data[data[group_col] == 'treatment'][conversion_col]

    n_control = len(control)
    n_treatment = len(treatment)

    conv_control = control.mean()
    conv_treatment = treatment.mean()

    # 절대 리프트
    absolute_lift = conv_treatment - conv_control

    # 상대 리프트
    relative_lift = (conv_treatment - conv_control) / conv_control * 100

    # 카이제곱 검정
    contingency = pd.crosstab(data[group_col], data[conversion_col])
    chi2, p_value, dof, expected = stats.chi2_contingency(contingency)

    # z-검정
    p_pooled = (control.sum() + treatment.sum()) / (n_control + n_treatment)
    se = np.sqrt(p_pooled * (1 - p_pooled) * (1/n_control + 1/n_treatment))
    z_score = absolute_lift / se
    p_value_z = 2 * (1 - stats.norm.cdf(abs(z_score)))

    results = {
        '지표': ['Control 전환율', 'Treatment 전환율', '절대 리프트',
                 '상대 리프트', 'Chi-square', 'p-value', '통계적 유의성'],
        '값': [
            f"{conv_control:.4f}",
            f"{conv_treatment:.4f}",
            f"{absolute_lift:.4f}",
            f"{relative_lift:.2f}%",
            f"{chi2:.2f}",
            f"{p_value:.4f}",
            "유의함" if p_value < 0.05 else "유의하지 않음"
        ]
    }

    return pd.DataFrame(results)

# 087 A/B 테스트 분석
print("\n=== 전통적 A/B 테스트 분석 ===")
ab_results = ab_test_analysis(users)
print(ab_results.to_string(index=False))

세그먼트별 분석

def segment_analysis(data, segment_col, group_col='group', conversion_col='converted'):
    """세그먼트별 A/B 테스트 분석"""
    results = []

    for segment in data[segment_col].unique():
        segment_data = data[data[segment_col] == segment]

        control = segment_data[segment_data[group_col] == 'control'][conversion_col]
        treatment = segment_data[segment_data[group_col] == 'treatment'][conversion_col]

        lift = treatment.mean() - control.mean()

        # 간단한 z-검정
        n_c, n_t = len(control), len(treatment)
        if n_c > 0 and n_t > 0:
            p_pooled = (control.sum() + treatment.sum()) / (n_c + n_t)
            if p_pooled > 0 and p_pooled < 1:
                se = np.sqrt(p_pooled * (1 - p_pooled) * (1/n_c + 1/n_t))
                z = lift / se if se > 0 else 0
                p_val = 2 * (1 - stats.norm.cdf(abs(z)))
            else:
                p_val = 1.0
        else:
            p_val = 1.0

        results.append({
            'segment': segment,
            'n_users': len(segment_data),
            'control_conv': control.mean(),
            'treatment_conv': treatment.mean(),
            'lift': lift,
            'p_value': p_val,
            'significant': p_val < 0.05
        })

    return pd.DataFrame(results)

# 087 세그먼트별 분석
print("\n=== 디바이스별 분석 ===")
device_analysis = segment_analysis(users, 'device')
print(device_analysis.to_string(index=False))

print("\n=== 국가별 분석 ===")
country_analysis = segment_analysis(users, 'country')
print(country_analysis.to_string(index=False))

FLAML로 전환 예측 모델

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
from flaml import AutoML

# 087 특성 준비
df = users.copy()
categorical_cols = ['gender', 'country', 'device', 'group']

for col in categorical_cols:
    le = LabelEncoder()
    df[col] = le.fit_transform(df[col])

feature_cols = ['age', 'gender', 'country', 'device', 'is_new_user', 'visit_count', 'group']
X = df[feature_cols]
y = df['converted']

# 087 데이터 분할
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# 087 FLAML 학습
automl = AutoML()
automl.fit(
    X_train, y_train,
    task="classification",
    metric="roc_auc",
    time_budget=60,
    n_jobs=-1,
    seed=42,
    verbose=1
)

print(f"\n=== 전환 예측 모델 ===")
print(f"최적 모델: {automl.best_estimator}")

모델 평가

from sklearn.metrics import accuracy_score, roc_auc_score, classification_report

y_pred = automl.predict(X_test)
y_proba = automl.predict_proba(X_test)[:, 1]

print("\n=== 모델 성능 ===")
print(f"정확도: {accuracy_score(y_test, y_pred):.4f}")
print(f"ROC AUC: {roc_auc_score(y_test, y_proba):.4f}")
print("\n분류 리포트:")
print(classification_report(y_test, y_pred, target_names=['Not Converted', 'Converted']))

특성 중요도 분석

import matplotlib.pyplot as plt

# 087 특성 중요도
if hasattr(automl.model.estimator, 'feature_importances_'):
    importance = pd.DataFrame({
        'feature': feature_cols,
        'importance': automl.model.estimator.feature_importances_
    }).sort_values('importance', ascending=False)

    print("\n=== 특성 중요도 ===")
    print(importance.to_string(index=False))

    # 시각화
    plt.figure(figsize=(10, 6))
    plt.barh(importance['feature'], importance['importance'])
    plt.xlabel('중요도')
    plt.title('전환 예측 특성 중요도')
    plt.gca().invert_yaxis()
    plt.tight_layout()
    plt.show()

처리 효과 이질성 분석 (CATE)

def estimate_cate(model, user_data, feature_cols):
    """조건부 평균 처리 효과 추정"""
    # Control 상태 예측
    user_control = user_data.copy()
    user_control['group'] = 0  # control
    control_proba = model.predict_proba(user_control[feature_cols])[:, 1]

    # Treatment 상태 예측
    user_treatment = user_data.copy()
    user_treatment['group'] = 1  # treatment
    treatment_proba = model.predict_proba(user_treatment[feature_cols])[:, 1]

    # CATE = Treatment - Control
    cate = treatment_proba - control_proba

    return cate

# 087 CATE 추정
df_test = X_test.copy()
df_test['cate'] = estimate_cate(automl, df_test, feature_cols)

print("\n=== 처리 효과 이질성 (CATE) ===")
print(f"평균 CATE: {df_test['cate'].mean():.4f}")
print(f"CATE 범위: [{df_test['cate'].min():.4f}, {df_test['cate'].max():.4f}]")

# 087 세그먼트별 CATE
print("\n디바이스별 평균 CATE:")
print(df_test.groupby('device')['cate'].mean())

최적 타겟팅 전략

def targeting_strategy(cate_data, threshold=0.02):
    """CATE 기반 타겟팅 전략"""

    # 높은 CATE 사용자 식별
    high_effect = cate_data['cate'] >= threshold
    low_effect = cate_data['cate'] < threshold

    strategy = {
        '전략': ['전체 적용', 'CATE 기반 타겟팅'],
        '타겟 비율': ['100%', f'{high_effect.mean()*100:.1f}%'],
        '예상 평균 리프트': [
            f"{cate_data['cate'].mean()*100:.2f}%p",
            f"{cate_data[high_effect]['cate'].mean()*100:.2f}%p"
        ],
        '효율성': ['기준', f'{cate_data[high_effect]["cate"].mean() / cate_data["cate"].mean():.2f}x']
    }

    return pd.DataFrame(strategy)

# 087 타겟팅 전략
print("\n=== 최적 타겟팅 전략 ===")
targeting = targeting_strategy(df_test, threshold=0.02)
print(targeting.to_string(index=False))

A/B 테스트 자동화 클래스

class ABTestAnalyzer:
    """A/B 테스트 자동화 분석기"""

    def __init__(self, time_budget=60):
        self.time_budget = time_budget
        self.model = None
        self.feature_cols = None

    def analyze(self, data, feature_cols, group_col='group', conversion_col='converted'):
        """전체 분석 수행"""
        self.feature_cols = feature_cols

        # 1. 기본 통계 분석
        print("=== 기본 통계 분석 ===")
        basic_results = ab_test_analysis(data, group_col, conversion_col)
        print(basic_results.to_string(index=False))

        # 2. 예측 모델 학습
        print("\n=== 예측 모델 학습 ===")
        X = data[feature_cols]
        y = data[conversion_col]

        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42, stratify=y
        )

        self.model = AutoML()
        self.model.fit(X_train, y_train, task="classification",
                       metric="roc_auc", time_budget=self.time_budget, verbose=0)

        y_proba = self.model.predict_proba(X_test)[:, 1]
        print(f"모델 ROC AUC: {roc_auc_score(y_test, y_proba):.4f}")

        # 3. CATE 분석
        print("\n=== CATE 분석 ===")
        cate = estimate_cate(self.model, X_test, feature_cols)
        print(f"평균 CATE: {cate.mean():.4f}")

        # 4. 타겟팅 전략
        cate_data = X_test.copy()
        cate_data['cate'] = cate
        print("\n=== 타겟팅 전략 ===")
        strategy = targeting_strategy(cate_data)
        print(strategy.to_string(index=False))

        return {
            'basic_analysis': basic_results,
            'model_auc': roc_auc_score(y_test, y_proba),
            'avg_cate': cate.mean(),
            'targeting_strategy': strategy
        }

# 087 자동화 분석기 사용
analyzer = ABTestAnalyzer(time_budget=30)

# 087 인코딩된 데이터로 분석
results = analyzer.analyze(df, feature_cols, 'group', 'converted')

정리

• 전통적 분석: 카이제곱, z-검정으로 통계적 유의성 확인
• 세그먼트 분석: 디바이스, 국가별 효과 차이 파악
• ML 모델: FLAML로 전환 예측 모델 구축
• CATE: 개인별 처리 효과 추정
• 타겟팅: CATE 기반 효율적 타겟팅 전략

다음 글 예고

다음 글에서는 MLflow로 실험 추적하기를 알아봅니다. FLAML 실험을 체계적으로 관리하고 추적하는 방법을 다룹니다.

---

FLAML AutoML 마스터 시리즈 #087