매초마다:
    "미국 사용자가 배너 광고를 클릭했어요!"
    "한국 사용자가 리워드 비디오를 봤어요!"
    → Kafka ad-events 토픽으로 전송 (초당 80개)

# 1분마다 계산
- 미국 배너: impression 1,234회, click 45회
- CTR = 45 / 1,234 = 3.64%
- eCPM = $2.50

→ Redis 저장 (실시간 조회용)
→ ClickHouse 저장 (장기 분석용)

# 과거 1시간 평균과 비교
if 현재 CTR < 평균 CTR × 0.5:
    🚨 알림: "CTR 급락!"
    
if 현재 CTR > 평균 CTR × 2:
    🤖 알림: "CTR 급등 (봇 의심)!"
    
if 현재 트래픽 > 평균 트래픽 × 3:
    📈 알림: "트래픽 폭증!"
    
if 현재 eCPM < 평균 eCPM × 0.5:
    💸 알림: "수익 급락!"

# 이상 감지 시
alert = {
    "alert_type": "ctr_drop",
    "country": "US",
    "ad_format": "banner",
    "current_value": 2.0,
    "expected_value": 5.0,
    "deviation_percent": -60.0,
    "message": "🚨 CTR 급락! US banner: 5.00% → 2.00% (-60.0%)"
}

→ Kafka alerts 토픽으로 전송

consumer = KafkaConsumer('alerts', ...)

for message in consumer:
    alert = message.value
    # {"alert_type": "ctr_drop", "message": "🚨 CTR 급락!", ...}

# 같은 알림이 10분 내에 발생했나?
key = f"alert_sent:US:banner:ctr_drop"

if redis.exists(key):
    print("⏭️  중복 알림 (10분 내 동일 알림 발생)")
    continue

# 10분 동안 중복 방지
redis.setex(key, 600, "1")

slack_message = {
    "text": "🚨 📉 AdTech 이상 탐지",
    "blocks": [
        {"type": "header", "text": "🚨 CTR 급락! US banner: 5.00% → 2.00%"},
        {"type": "section", "fields": [
            {"type": "mrkdwn", "text": "*국가:*\nUS"},
            {"type": "mrkdwn", "text": "*광고 포맷:*\nbanner"},
            ...
        ]}
    ]
}

requests.post(SLACK_WEBHOOK_URL, json=slack_message)

# 1단계: 정상 데이터 전송 (30초)
for i in range(1000):
    send_event(type="impression")  # 1,000개
for i in range(50):
    send_event(type="click")        # 50개 (CTR 5%)

time.sleep(30)  # Spark가 평균 계산

# 2단계: 이상 데이터 전송
for i in range(1000):
    send_event(type="impression")  # 1,000개
for i in range(20):
    send_event(type="click")        # 20개 (CTR 2% ← 급락!)

# 메트릭 저장
metrics:US:banner:1m = {"ctr": 5.0, "impressions": 100, "clicks": 5}

# 히스토리 저장 (평균 계산용)
history:US:banner:1h = {"avg_ctr": 5.0, "avg_impressions": 100}

# 1분 윈도우 집계
current_ctr = 2.0%
avg_ctr = 5.0%  # Redis history에서 조회

# 이상 탐지 로직
if current_ctr < avg_ctr * 0.5:  # 2.0 < 2.5 → True!
    alert = {
        "message": "🚨 CTR 급락! US banner: 5.00% → 2.00% (-60.0%)",
        "severity": "high",
        ...
    }
    send_to_kafka(alert)

# Kafka에서 메시지 수신
message = consumer.poll()
alert = json.loads(message.value)

# 중복 체크
if not is_duplicate(alert):
    send_to_slack(alert)

avg = (5 + 5 + 5 + 20) / 4 = 8.75%

# 다음 시간
13시: CTR 10%
if 10% < 8.75% × 0.5:  # False (정상으로 판단)
    alert("CTR 급락!")

# 최근 데이터에 더 큰 가중치
new_avg = old_avg × 0.8 + current_value × 0.2

09시: avg = 5.0
10시: avg = 5.0 × 0.8 + 5.0 × 0.2 = 5.0
11시: avg = 5.0 × 0.8 + 5.0 × 0.2 = 5.0
12시: avg = 5.0 × 0.8 + 20.0 × 0.2 = 8.0  # 천천히 증가

13시: current = 10%
avg = 8.0 × 0.8 + 10.0 × 0.2 = 8.4
if 10% < 8.4% × 0.5:  # False (정상으로 판단)

# Redis에서 히스토리 조회
history = redis.get(f"history:{country}:{ad_format}:1h")
avg_ctr = history['avg_ctr']

# 이상 탐지
if current_ctr < avg_ctr * 0.5:
    send_alert("CTR 급락!")

# EMA로 히스토리 업데이트
new_avg_ctr = avg_ctr * 0.8 + current_ctr * 0.2
redis.setex(f"history:{country}:{ad_format}:1h", 3600, {
    "avg_ctr": new_avg_ctr
})

def is_duplicate_alert(alert):
    # 알림 키 생성 (국가 + 포맷 + 타입)
    key = f"alert_sent:{alert['country']}:{alert['ad_format']}:{alert['alert_type']}"
    # 예: "alert_sent:US:banner:ctr_drop"
    
    # 이미 보냈나?
    if redis.exists(key):
        return True  # 중복!
    
    # 10분 동안 이 알림 금지
    redis.setex(key, 600, "1")
    return False

def detect_anomalies(batch_df, batch_id):
    """
    배치 데이터에서 이상 패턴 탐지
    """
    import redis
    r = redis.Redis(host='redis', port=6379, db=0)
    alerts = []
    
    for row in batch_df.collect():
        # 현재 메트릭
        current_ctr = float(row.ctr) if row.ctr else 0.0
        
        # 과거 평균 조회
        history_key = f"history:{row.country}:{row.ad_format}:1h"
        history_data = r.get(history_key)
        
        if history_data:
            history = json.loads(history_data)
            avg_ctr = history.get('avg_ctr', current_ctr)
        else:
            avg_ctr = current_ctr  # 히스토리 없으면 현재값 사용
        
        # 🚨 이상 1: CTR 급락 (50% 이상 하락)
        if avg_ctr > 0 and current_ctr < avg_ctr * 0.5:
            deviation = ((current_ctr - avg_ctr) / avg_ctr) * 100
            alert = {
                "alert_id": str(uuid.uuid4()),
                "alert_time": row.window_end.isoformat(),
                "alert_type": "ctr_drop",
                "severity": "high" if deviation < -70 else "medium",
                "country": row.country,
                "ad_format": row.ad_format,
                "metric_name": "CTR",
                "current_value": current_ctr,
                "expected_value": avg_ctr,
                "deviation_percent": round(deviation, 2),
                "message": f"🚨 CTR 급락! {row.country} {row.ad_format}: {avg_ctr:.2f}% → {current_ctr:.2f}% ({deviation:.1f}%)"
            }
            alerts.append(alert)
            print(f"⚠️  {alert['message']}")
        
        # EMA로 히스토리 업데이트
        new_avg_ctr = avg_ctr * 0.8 + current_ctr * 0.2
        history_data = {
            "avg_ctr": round(new_avg_ctr, 4),
            "last_updated": row.last_updated.isoformat()
        }
        r.setex(history_key, 3600, json.dumps(history_data))
    
    # Kafka로 알림 전송
    if alerts:
        send_alerts_to_kafka(alerts)

from kafka import KafkaProducer
import json

def send_alerts_to_kafka(alerts):
    """
    탐지된 알림을 Kafka "alerts" 토픽으로 전송
    """
    try:
        producer = KafkaProducer(
            bootstrap_servers=['kafka:29092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        
        for alert in alerts:
            producer.send('alerts', value=alert)
        
        producer.flush()
        print(f"✅ {len(alerts)}개 알림을 Kafka로 전송 완료")
    except Exception as e:
        print(f"❌ Kafka 전송 실패: {str(e)}")

from kafka import KafkaConsumer
import json
import redis
import requests

def consume_alerts():
    """
    Kafka alerts 토픽을 구독하여 알림 처리
    """
    consumer = KafkaConsumer(
        'alerts',
        bootstrap_servers=['localhost:9092'],
        auto_offset_reset='latest',
        enable_auto_commit=True,
        group_id='slack-alert-consumer',
        value_deserializer=lambda x: json.loads(x.decode('utf-8'))
    )
    
    r = redis.Redis(host='localhost', port=6379, db=1)
    
    for message in consumer:
        alert = message.value
        
        print(f"\n🔔 알림 수신: {alert['message']}")
        
        # 중복 체크
        key = f"alert_sent:{alert['country']}:{alert['ad_format']}:{alert['alert_type']}"
        if r.exists(key):
            print(f"   ⏭️  중복 알림 (10분 내 동일 알림 발생)")
            continue
        
        # Slack 전송
        send_to_slack(alert)
        
        # 10분 cooldown
        r.setex(key, 600, "1")

def send_to_slack(alert):
    """
    Slack Webhook으로 알림 전송
    """
    severity_emoji = {
        "low": "ℹ️",
        "medium": "⚠️",
        "high": "🚨",
        "critical": "🔥"
    }
    
    emoji = severity_emoji.get(alert['severity'], "⚠️")
    
    message = {
        "text": f"{emoji} AdTech 이상 탐지",
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"{emoji} {alert['message']}"
                }
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*국가:*\n{alert['country']}"},
                    {"type": "mrkdwn", "text": f"*광고 포맷:*\n{alert['ad_format']}"},
                    {"type": "mrkdwn", "text": f"*메트릭:*\n{alert['metric_name']}"},
                    {"type": "mrkdwn", "text": f"*심각도:*\n{alert['severity'].upper()}"},
                    {"type": "mrkdwn", "text": f"*현재 값:*\n{alert['current_value']:.2f}"},
                    {"type": "mrkdwn", "text": f"*예상 값:*\n{alert['expected_value']:.2f}"},
                    {"type": "mrkdwn", "text": f"*변화율:*\n{alert['deviation_percent']:.1f}%"},
                    {"type": "mrkdwn", "text": f"*발생 시간:*\n{alert['alert_time']}"}
                ]
            }
        ]
    }
    
    response = requests.post(SLACK_WEBHOOK_URL, json=message)
    
    if response.status_code == 200:
        print(f"✅ Slack 전송 성공!")
    else:
        print(f"❌ Slack 전송 실패: {response.status_code}")

alert = {
    "alert_time": row.window_end,  # datetime 객체
    ...
}
json.dumps(alert)  # 에러!

alert = {
    "alert_time": row.window_end.isoformat(),  # 문자열 "2025-12-18T02:58:00"
    ...
}

# Spark는 이걸 Spark 함수로 해석
round((avg_ctr * 0.8 + current_ctr * 0.2), 4)

import builtins
builtins.round((avg_ctr * 0.8 + current_ctr * 0.2), 4)

if current_value < average * 0.5:
    alert("이상!")