utils/prediction_utils.py

# utils/prediction_utils.py
import warnings
warnings.filterwarnings("ignore")

import numpy as np

# Prophet 시도, 실패 시 Holt(지수 평활) 폴백
try:
    from prophet import Prophet
    _HAS_PROPHET = True
except Exception:
    _HAS_PROPHET = False

try:
    from statsmodels.tsa.holtwinters import ExponentialSmoothing
    _HAS_HOLT = True
except Exception:
    _HAS_HOLT = False

import pandas as pd

def prophet_emotion_forecast(scores, steps=3):
    """
    Prophet 기반 예측. 입력: 점수 리스트(0~100)
    반환: 예측 점수 리스트 길이=steps (float)
    참고: Taylor & Letham (2018) 'Forecasting at Scale'
    """
    if not _HAS_PROPHET or len(scores) < 5:
        return None

    df = pd.DataFrame({
        'ds': pd.date_range(start='2023-01-01', periods=len(scores), freq='D'),
        'y': scores
    })
    model = Prophet(
        daily_seasonality=False, weekly_seasonality=False, yearly_seasonality=False,
        changepoint_prior_scale=0.2  # 변화점 민감도
    )
    model.fit(df)
    future = model.make_future_dataframe(periods=steps)
    forecast = model.predict(future)
    tail = forecast[['yhat']].tail(steps)['yhat'].to_list()
    # 안전 클리핑
    return [float(np.clip(x, 0, 100)) for x in tail]

def holt_fallback_forecast(scores, steps=3):
    """
    Holt 지수평활(추세형) 폴백 예측.
    Prophet 사용 불가 환경에서 간단하고 안정적.
    """
    if not _HAS_HOLT or len(scores) < 4:
        return None
    try:
        model = ExponentialSmoothing(
            np.array(scores, dtype=float),
            trend='add', seasonal=None
        ).fit()
        fcast = model.forecast(steps)
        return [float(np.clip(x, 0, 100)) for x in fcast.tolist()]
    except Exception:
        return None

def forecast_scores(scores, steps=3):
    """
    통합 예측 엔진:
    1) Prophet 성공 시 Prophet 사용
    2) 아니면 Holt 폴백
    3) 그래도 안 되면, 마지막 기울기 선형 외삽(극단 폴백)
    """
    if len(scores) < 2:
        return [scores[-1]] * steps if scores else [0.0]*steps

    yhat = prophet_emotion_forecast(scores, steps=steps)
    if yhat is not None:
        return yhat

    yhat = holt_fallback_forecast(scores, steps=steps)
    if yhat is not None:
        return yhat

    # 마지막 두 점의 기울기로 외삽
    slope = scores[-1] - scores[-2]
    return [float(np.clip(scores[-1] + slope*(i+1), 0, 100)) for i in range(steps)]