# data-generator/ab_test_generator.py

AB_TESTS = {
    'banner_color_test_001': {
        'variants': {
            'A': {
                'name': 'blue_banner',
                'ctr': 0.04,       # 기존: CTR 4%
                'description': '파란색 배너 (기존)'
            },
            'B': {
                'name': 'red_banner',
                'ctr': 0.06,       # 신규: CTR 6% (50% 개선!)
                'description': '빨간색 배너 (신규)'
            }
        },
        'traffic_split': {'A': 0.5, 'B': 0.5},  # 50:50
        'countries': ['US', 'KR', 'JP'],
        'ad_format': 'banner'
    }
}

def generate_ab_test_event(test_id, test_config):
    # Traffic Split에 따라 variant 선택
    rand = random.random()
    cumulative = 0
    selected_variant = None
    
    for variant, split in test_config['traffic_split'].items():
        cumulative += split
        if rand < cumulative:
            selected_variant = variant
            break
    
    # 선택된 variant의 CTR로 클릭 여부 결정
    variant_config = test_config['variants'][selected_variant]
    ctr = variant_config['ctr']
    is_click = random.random() < ctr
    
    # 이벤트 생성
    return {
        'event_id': f"ABT{int(time.time()*1000)}{random.randint(100,999)}",
        'test_id': test_id,
        'variant': selected_variant,
        'variant_name': variant_config['name'],
        'event_type': 'click' if is_click else 'impression',
        # ... 기타 필드
    }

# 전체 CTR = 195 / 3,531 = 5.52%
expected_click_A = 1,772 × 0.0552 = 97.8
expected_click_B = 1,759 × 0.0552 = 97.1

chi_square = Σ [(관측값 - 기대값)² / 기대값]

= (83-97.8)²/97.8 + (1689-1674.2)²/1674.2 + ...
= 4.4766

from scipy.stats import chi2_contingency

observed = [
    [83, 1689],  # A: clicks, no-clicks
    [112, 1647]  # B: clicks, no-clicks
]

chi2, p_value, dof, expected = chi2_contingency(observed)
# p_value = 0.0344

if p_value < 0.05:  # 0.0344 < 0.05 ✅
    print("통계적으로 유의미! B가 승자!")
    confidence = 1 - p_value  # 96.56%
else:
    print("우연일 수 있음, 더 기다려야 함")

# src/spark_ab_test_analyzer.py

def perform_chi_square_test(variant_a_data, variant_b_data):
    """Chi-square 검정 수행"""
    from scipy.stats import chi2_contingency
    import builtins  # Python 내장 round() 사용 (Spark round 충돌 방지)
    
    # 데이터 추출
    a_clicks = variant_a_data['clicks']
    a_impressions = variant_a_data['impressions']
    a_no_clicks = a_impressions - a_clicks
    
    b_clicks = variant_b_data['clicks']
    b_impressions = variant_b_data['impressions']
    b_no_clicks = b_impressions - b_clicks
    
    # 최소 샘플 크기 체크
    if a_impressions < 100 or b_impressions < 100:
        return {
            "chi_square": 0.0,
            "p_value": 1.0,
            "is_significant": False,
            "message": "샘플 부족 (최소 100 impressions 필요)"
        }
    
    # Contingency Table 생성
    observed = [
        [a_clicks, a_no_clicks],
        [b_clicks, b_no_clicks]
    ]
    
    # Chi-square 검정
    chi2, p_value, dof, expected = chi2_contingency(observed)
    
    return {
        "chi_square": builtins.round(float(chi2), 4),
        "p_value": builtins.round(float(p_value), 4),
        "is_significant": bool(p_value < 0.05),
        "confidence_level": builtins.round(1 - float(p_value), 4),
        "message": "✅ 통계적으로 유의미!" if p_value < 0.05 else "⏳ 아직 판단 불가"
    }

def process_ab_test_batch(batch_df, batch_id):
    """각 배치에서 A/B Test 분석"""
    
    # A와 B 데이터 분리
    variant_a = batch_df.filter(col("variant") == "A").collect()[0]
    variant_b = batch_df.filter(col("variant") == "B").collect()[0]
    
    # Chi-square 검정
    test_result = perform_chi_square_test(
        {'impressions': int(variant_a.impressions), 'clicks': int(variant_a.clicks)},
        {'impressions': int(variant_b.impressions), 'clicks': int(variant_b.clicks)}
    )
    
    # 승자 판정
    winner = None
    if test_result['is_significant']:
        if variant_b.ctr > variant_a.ctr:
            winner = "B"
            print(f"🏆 Winner: B - CTR {variant_b.ctr:.2f}% > {variant_a.ctr:.2f}%")
        else:
            winner = "A"
    
    # Redis 저장
    redis_data = {
        "test_id": test_id,
        "variant_a": {...},
        "variant_b": {...},
        "test_result": test_result,
        "winner": winner
    }
    redis.setex(f"ab_test:{test_id}:latest", 3600, json.dumps(redis_data))

# src/fastapi_ab_test_api.py

@app.get("/ab-test/status/{test_id}")
def get_ab_test_status(test_id: str):
    """현재 A/B Test 상태 조회"""
    redis_key = f"ab_test:{test_id}:latest"
    data = redis_client.get(redis_key)
    
    if not data:
        raise HTTPException(status_code=404, detail=f"Test {test_id} not found")
    
    result = json.loads(data)
    
    return {
        "test_id": test_id,
        "window": {...},
        "variant_a": result["variant_a"],
        "variant_b": result["variant_b"],
        "test_result": result["test_result"],
        "winner": result.get("winner"),
        "recommendation": get_recommendation(result)
    }

@app.get("/ab-test/winner/{test_id}")
def get_ab_test_winner(test_id: str):
    """승자 확인"""
    data = get_test_data(test_id)
    
    if not data["test_result"]["is_significant"]:
        return {
            "winner": None,
            "is_significant": False,
            "message": "아직 통계적으로 유의미한 차이가 없습니다",
            "p_value": data["test_result"]["p_value"]
        }
    
    winner = data["winner"]
    winner_data = data[f"variant_{winner.lower()}"]
    loser_data = data[f"variant_{'b' if winner == 'A' else 'a'}"]
    
    improvement = ((winner_data["ctr"] - loser_data["ctr"]) / loser_data["ctr"] * 100)
    
    return {
        "winner": winner,
        "winner_name": winner_data["name"],
        "is_significant": True,
        "confidence_level": data["test_result"]["confidence_level"],
        "p_value": data["test_result"]["p_value"],
        "improvement_percent": round(improvement, 2),
        "message": f"🏆 {winner_data['name']}가 {improvement:.1f}% 더 우수합니다!"
    }

# Spark 컨텍스트에서 round() 호출
return {
    "chi_square": round(float(chi2), 4),  # Spark round()로 해석됨
    ...
}

import builtins

return {
    "chi_square": builtins.round(float(chi2), 4),  # Python 내장 함수 명시
    ...
}

# ClickHouse 저장 (선택사항)
# batch_df.write.format("jdbc")...

print(f"✅ Batch {batch_id}: 처리 완료 (Redis 저장)")

A: 10 impressions, 1 click (10%)
B: 10 impressions, 2 clicks (20%)

# p-value = 0.5 > 0.05 → 판단 불가!

A: 1,000 impressions, 100 clicks (10%)
B: 1,000 impressions, 200 clicks (20%)

# p-value = 0.0001 < 0.05 → B가 확실히 우수!