3강 — 시계열 분석과 트렌드

# ① 환경 준비 (이 강의 첫 셀에서 한 번만)
!pip install openai -q

import json
import re
import random
from datetime import datetime   # 오늘 주제가 '시간'이라 새로 등장 — 파이썬 기본 시간 처리 라이브러리
from collections import defaultdict
from google.colab import userdata
from openai import OpenAI

# 시각화 (없으면 텍스트로 대체)
try:
    import matplotlib.pyplot as plt
    plt.rcParams['font.family'] = 'DejaVu Sans'
    plt.rcParams['axes.unicode_minus'] = False
    HAS_MPL = True
except ImportError:
    HAS_MPL = False

MODEL = "mistralai/devstral-2512:free"
api_key = userdata.get('OPENROUTER_API_KEY')
client = OpenAI(base_url="https://openrouter.ai/api/v1", api_key=api_key)

def clean_json_response(text):
    text = text.strip()
    if '```json' in text:
        m = re.search(r'```json\s*([\s\S]*?)\s*```', text)
        if m: return m.group(1).strip()
    if '```' in text:
        m = re.search(r'```\s*([\s\S]*?)\s*```', text)
        if m: return m.group(1).strip()
    m = re.search(r'(\{[\s\S]*\}|\[[\s\S]*\])', text)
    return m.group(1).strip() if m else text

def call_llm(system_prompt, user_message):
    res = client.chat.completions.create(
        model=MODEL,
        messages=[{"role": "system", "content": system_prompt},
                  {"role": "user", "content": user_message}],
    )
    return res.choices[0].message.content

def call_llm_json(system_prompt, user_message):
    return json.loads(clean_json_response(call_llm(system_prompt, user_message)))

print("준비 완료!")

def generate_monthly_data():
    """월별 감성 데이터를 시나리오에 맞춰 생성.
    1-2월 안정(긍정 65% 전후) → 3-4월 위기(45%) → 5-6월 회복(60%↑)."""
    base_rates = {1: 65, 2: 67, 3: 45, 4: 47, 5: 58, 6: 63}   # 월별 기본 긍정률(%)
    data = []
    for month in range(1, 7):
        total = random.randint(180, 220) if month in (3, 4) else random.randint(120, 160)
        rate = max(30, min(80, base_rates[month] + random.randint(-3, 3)))   # 약간의 노이즈
        positive = int(total * rate / 100)
        neutral = int((total - positive) * 0.3)
        negative = total - positive - neutral
        data.append({'month_name': f"{month}월", 'total': total,
                     'positive': positive, 'negative': negative, 'neutral': neutral,
                     'positive_rate': round(positive / total * 100, 1)})
    return data

monthly_data = generate_monthly_data()
print(f"{'월':^5}|{'총':^6}|{'긍정':^5}|{'부정':^5}|{'중립':^5}|{'긍정률':^7}")
for m in monthly_data:
    print(f"{m['month_name']:^5}|{m['total']:^6}|{m['positive']:^5}|{m['negative']:^5}|{m['neutral']:^5}|{m['positive_rate']:^6}%")

def moving_average(data, window=3):
    """최근 window개의 평균을 계산. 데이터가 부족한 앞부분은 None."""
    result = []
    for i in range(len(data)):
        if i < window - 1:
            result.append(None)                       # 아직 window개가 안 모임
        else:
            avg = sum(data[i - window + 1:i + 1]) / window
            result.append(round(avg, 1))
    return result

rates = [m['positive_rate'] for m in monthly_data]
months = [m['month_name'] for m in monthly_data]
ma_3 = moving_average(rates, window=3)               # 3개월 이동 평균

print(f"{'월':^5}|{'긍정률':^8}|{'3개월 이동평균':^14}")
for i, mth in enumerate(months):
    ma = f"{ma_3[i]}%" if ma_3[i] is not None else "-"
    print(f"{mth:^5}|{rates[i]:^7}%|{ma:^14}")

# 원본 vs 이동평균 시각화
if HAS_MPL:
    x = range(len(months))
    plt.figure(figsize=(10, 5))
    plt.plot(x, rates, 'o-', label='Original', linewidth=2)
    ma_x = [i for i, v in enumerate(ma_3) if v is not None]
    ma_y = [v for v in ma_3 if v is not None]
    plt.plot(ma_x, ma_y, 's--', label='3-Month MA', linewidth=2)
    plt.xticks(list(x), months); plt.ylim(30, 80)
    plt.ylabel('Positive Rate (%)'); plt.title('Monthly Positive Rate with Moving Average')
    plt.legend(); plt.grid(alpha=0.3); plt.tight_layout(); plt.show()

def detect_change_points(data, threshold=10):
    """이전 값 대비 변화량이 threshold(%p)를 넘으면 변화 지점으로 표시."""
    points = []
    for i in range(1, len(data)):
        change = data[i] - data[i - 1]
        if abs(change) >= threshold:
            points.append({'index': i, 'from': data[i - 1], 'to': data[i],
                           'change': round(change, 1),
                           'direction': '상승' if change > 0 else '하락'})
    return points

for cp in detect_change_points(rates, threshold=10):
    i = cp['index']
    print(f"{months[i-1]} → {months[i]}: {cp['from']}% → {cp['to']}% "
          f"({cp['change']:+.1f}%p, {cp['direction']})")
    if cp['direction'] == '하락':
        print("  → 부정적 변화. 그 시점에 무슨 일이 있었는지 원인 분석 필요")
    else:
        print("  → 긍정적 회복. 어떤 조치가 효과적이었는지 확인")

def analyze_time_series(monthly_data, ma_window=3, change_threshold=10):
    """이동평균·변화점·전체 추세 방향을 한 번에 종합한다."""
    values = [m['positive_rate'] for m in monthly_data]
    ma = moving_average(values, window=ma_window)
    changes = detect_change_points(values, threshold=change_threshold)
    first_ma = next(v for v in ma if v is not None)          # 첫 유효 이동평균
    diff = ma[-1] - first_ma                                 # 처음 대비 마지막
    trend = "상승 추세" if diff > 5 else "하락 추세" if diff < -5 else "안정적"
    return {'trend': trend, 'change_points': len(changes), 'current': values[-1]}

a = analyze_time_series(monthly_data)
print(f"전체 추세: {a['trend']} / 변화점 {a['change_points']}개 / 현재 긍정률 {a['current']}%")

# 이벤트 타임라인 — 데이터에 '맥락'을 입혀 주는 핵심
events = [
    {"month": 3, "event": "대규모 콘텐츠 업데이트 (버전 2.0)"},
    {"month": 4, "event": "버그 리포트 급증, 공식 사과문 발표"},
    {"month": 5, "event": "긴급 핫픽스 패치 배포"},
]

SYSTEM_PROMPT = """당신은 게임 리뷰 분석 전문가입니다.
월별 감성 추이와 이벤트 타임라인을 분석해 인사이트를 제공합니다.
다음 JSON 형식으로만 응답하세요:
{
  "summary": "한 줄 요약",
  "correlations": [{"period": "기간", "event": "관련 이벤트", "impact": "데이터에 미친 영향"}],
  "recommendations": ["권장사항 1", "권장사항 2"]
}"""

request = f"""## 월별 긍정률
{json.dumps([{'month': m['month_name'], 'positive_rate': m['positive_rate']} for m in monthly_data], ensure_ascii=False)}
## 이벤트 타임라인
{json.dumps(events, ensure_ascii=False)}
위 데이터를 분석해 인사이트를 제공해주세요."""

try:
    result = call_llm_json(SYSTEM_PROMPT, request)
    print("요약:", result.get('summary'))
    print("\n이벤트-데이터 연관성:")
    for c in result.get('correlations', []):
        print(f"  • {c.get('period')}: {c.get('event')} → {c.get('impact')}")
    print("\n권장사항:")
    for i, rec in enumerate(result.get('recommendations', []), 1):
        print(f"  {i}. {rec}")
except Exception as e:
    print(f"LLM 분석 실패: {e} (API 키나 네트워크를 확인하세요)")