Bí quyết phát hiện và xử lý Sample Ratio Mismatch (SRM) trong A/B Testing bằng Python

import scipy.stats as stats

# Hằng số định nghĩa tỷ lệ kỳ vọng của các biến thể (Control, Treatment)
EXPECTED_RATIO = [0.5, 0.5]

def check_sample_ratio_mismatch(observed_counts, expected_ratio, significance_level=0.05):
    """
    Hàm kiểm tra hiện tượng Sample Ratio Mismatch (SRM) sử dụng kiểm định Chi-square.
    
    Args:
        observed_counts (list): Số lượng mẫu thực tế thu được cho mỗi nhóm.
        expected_ratio (list): Tỷ lệ kỳ vọng thiết kế ban đầu.
        significance_level (float): Mức ý nghĩa thống kê (p-value threshold).
        
    Returns:
        dict: Kết quả kiểm định bao gồm chi2 stat, p-value và trạng thái SRM.
    """
    total_samples = sum(observed_counts)
    
    # Tính toán tần số kỳ vọng dựa trên tổng số mẫu thực tế thu về
    expected_counts = [total_samples * ratio for ratio in expected_ratio]
    
    # Thực hiện kiểm định Chi-square Goodness of Fit
    chi2_statistic, p_value = stats.chisquare(f_obs=observed_counts, f_exp=expected_counts)
    
    # Xác định có xuất hiện SRM hay không dựa vào p-value
    has_srm = p_value < significance_level
    
    return {
        "chi2_statistic": chi2_statistic,
        "p_value": p_value,
        "has_srm": has_srm
    }

# Ví dụ dữ liệu thực tế thu về: Nhóm Control = 49200, Nhóm Treatment = 48500
observed_data = [49200, 48500]
result = check_sample_ratio_mismatch(observed_data, EXPECTED_RATIO)

print(f"Chi2 Statistic: {result['chi2_statistic']:.4f}")
print(f"P-value: {result['p_value']:.4f}")

if result['has_srm']:
    print("CẢNH BÁO: Phát hiện hiện tượng SRM! Không nên tin tưởng kết quả thí nghiệm.")
else:
    print("Dữ liệu phân bổ đồng đều, không có dấu hiệu SRM.")

import scipy.stats as stats

# Hằng số định nghĩa tỷ lệ kỳ vọng cho 3 nhóm (Control, Treatment A, Treatment B)
EXPECTED_RATIO_MULTI = [0.5, 0.25, 0.25]

def check_multi_variant_srm(observed_counts, expected_ratio, significance_level=0.05):
    """
    Hàm kiểm tra hiện tượng SRM cho thử nghiệm A/B/n với Chi-square Goodness of Fit.
    
    Args:
        observed_counts (list): Số lượng mẫu thực tế cho từng nhóm.
        expected_ratio (list): Tỷ lệ kỳ vọng thiết kế.
        significance_level (float): Ngưỡng p-value để kết luận SRM.
        
    Returns:
        dict: Kết quả thống kê và trạng thái SRM.
    """
    total_samples = sum(observed_counts)
    expected_counts = [total_samples * ratio for ratio in expected_ratio]
    
    chi2_statistic, p_value = stats.chisquare(f_obs=observed_counts, f_exp=expected_counts)
    has_srm = p_value < significance_level
    
    return {
        "chi2_statistic": chi2_statistic,
        "p_value": p_value,
        "has_srm": has_srm
    }

# Dữ liệu thực tế thu về cho 3 nhóm: Control = 100500, Treatment A = 49200, Treatment B = 48800
observed_data_multi = [100500, 49200, 48800]
result_multi = check_multi_variant_srm(observed_data_multi, EXPECTED_RATIO_MULTI)

print(f"Chi2 Statistic: {result_multi['chi2_statistic']:.4f}")
print(f"P-value: {result_multi['p_value']:.4f}")
if result_multi['has_srm']:
    print("CẢNH BÁO: Phát hiện SRM ở thử nghiệm đa biến thể!")
else:
    print("Phân bổ lưu lượng hợp lệ.")

import pandas as pd
import scipy.stats as stats

# Dữ liệu giả lập ghi nhận theo từng ngày cho 2 nhóm (Control, Treatment)
data = {
    "date": ["2026-06-01", "2026-06-02", "2026-06-03"],
    "control_users": [5000, 4800, 5100],
    "treatment_users": [5000, 4500, 5050] # Ngày 2 có dấu hiệu lệch lớn
}

df_logs = pd.DataFrame(data)

def detect_daily_srm(df, expected_ratio=[0.5, 0.5], significance_level=0.05):
    """
    Hàm quét và phát hiện SRM theo từng ngày từ DataFrame chứa dữ liệu log.
    
    Args:
        df (pd.DataFrame): DataFrame chứa cột ngày và số lượng user từng nhóm.
        expected_ratio (list): Tỷ lệ phân bổ kỳ vọng.
        significance_level (float): Mức ý nghĩa thống kê.
        
    Returns:
        pd.DataFrame: DataFrame bổ sung kết quả p-value và cờ cảnh báo SRM.
    """
    p_values = []
    srm_flags = []
    
    for index, row in df.iterrows():
        observed = [row["control_users"], row["treatment_users"]]
        total = sum(observed)
        expected = [total * ratio for ratio in expected_ratio]
        
        chi2, p_val = stats.chisquare(f_obs=observed, f_exp=expected)
        p_values.append(p_val)
        srm_flags.append(p_val < significance_level)
        
    df["p_value"] = p_values
    df["has_srm"] = srm_flags
    return df

# Chạy kiểm tra SRM theo ngày
df_result = detect_daily_srm(df_logs)
print(df_result[["date", "p_value", "has_srm"]])