Create app.py

app.py

@@ -0,0 +1,168 @@
+import pandas as pd
+import numpy as np
+import datetime as dt
+import warnings
+
+from statsmodels.tsa.holtwinters import ExponentialSmoothing
+import plotly.graph_objects as go
+import gradio as gr
+
+warnings.filterwarnings("ignore")
+
+# -----------------------------
+# CONFIG
+# -----------------------------
+DATA_FILE = "202503-domae.parquet"  # 같은 경로에 놓여 있어야 함
+FORECAST_END_YEAR = 2030            # 예측 종료 연도(12월까지)
+SEASONAL_PERIODS = 12               # 월별 seasonality
+
+# -----------------------------
+# 1. 데이터 적재 & 전처리
+# -----------------------------
+
+def load_data(path: str) -> pd.DataFrame:
+    """Parquet → 월별 피벗 테이블(DateIndex, 열: 품목, 값: 가격)."""
+    df = pd.read_parquet(path)
+
+    # 날짜 컬럼 생성/정규화 (두 가지 케이스 지원)
+    if "date" in df.columns:
+        df["date"] = pd.to_datetime(df["date"])
+    elif "PRCE_REG_MM" in df.columns:
+        df["date"] = pd.to_datetime(df["PRCE_REG_MM"].astype(str), format="%Y%m")
+    else:
+        raise ValueError("지원되지 않는 날짜 컬럼 형식입니다.")
+
+    # 기본 컬럼명 통일
+    item_col = "PDLT_NM" if "PDLT_NM" in df.columns else "item"
+    price_col = "AVRG_PRCE" if "AVRG_PRCE" in df.columns else "price"
+
+    monthly = (
+        df.groupby(["date", item_col])[price_col]
+        .mean()
+        .reset_index()
+    )
+    pivot = (
+        monthly
+        .pivot(index="date", columns=item_col, values=price_col)
+        .sort_index()
+    )
+    # 월 시작일 MS 빈도로 정렬
+    pivot.index = pd.to_datetime(pivot.index).to_period("M").to_timestamp()
+    return pivot
+
+pivot = load_data(DATA_FILE)
+products = pivot.columns.tolist()
+
+# -----------------------------
+# 2. 고유 모델 정의 (Holt‑Winters + fallback)
+# -----------------------------
+
+def _fit_forecast(series: pd.Series) -> pd.Series:
+    """월별 시계열 → 2025‑04 이후 FORECAST_END_YEAR‑12까지 예측."""
+    # Ensure Monthly Start frequency
+    series = series.asfreq("MS")
+
+    # 예측 기간 계산
+    last_date = series.index[-1]
+    end_date = dt.datetime(FORECAST_END_YEAR, 12, 1)
+    horizon = (end_date.year - last_date.year) * 12 + (end_date.month - last_date.month)
+    if horizon <= 0:
+        return pd.Series(dtype=float)
+
+    try:
+        model = ExponentialSmoothing(
+            series,
+            trend="add",
+            seasonal="mul",
+            seasonal_periods=SEASONAL_PERIODS,
+            initialization_method="estimated",
+        )
+        res = model.fit(optimized=True)
+        fc = res.forecast(horizon)
+    except Exception:
+        # 홀트윈터스 학습 실패 시 단순 CAGR 기반 예측
+        growth = series.pct_change().fillna(0).mean()
+        fc = pd.Series(
+            [series.iloc[-1] * (1 + growth) ** i for i in range(1, horizon + 1)],
+            index=pd.date_range(
+                series.index[-1] + pd.DateOffset(months=1),
+                periods=horizon,
+                freq="MS",
+            ),
+        )
+    return fc
+
+# 품목별 전체 시리즈(과거+예측) 사전 구축 → 앱 반응 속도 개선
+FULL_SERIES = {}
+FORECASTS = {}
+for item in products:
+    hist = pivot[item].dropna()
+    fc = _fit_forecast(hist)
+    FULL_SERIES[item] = pd.concat([hist, fc])
+    FORECASTS[item] = fc
+
+# -----------------------------
+# 3. 내일 가격 예측 함수
+# -----------------------------
+
+today = dt.date.today()
+tomorrow = today + dt.timedelta(days=1)
+
+def build_tomorrow_df() -> pd.DataFrame:
+    """내일(일 단위) 예상 가격 DataFrame 반환."""
+    preds = {}
+    for item, series in FULL_SERIES.items():
+        # 일 단위 선형 보간
+        daily = series.resample("D").interpolate("linear")
+        preds[item] = round(daily.loc[tomorrow], 2) if tomorrow in daily.index else np.nan
+    return (
+        pd.DataFrame.from_dict(preds, orient="index", columns=[f"내일({tomorrow}) 예상가(KRW)"])
+        .sort_index()
+    )
+
+tomorrow_df = build_tomorrow_df()
+
+# -----------------------------
+# 4. 시각화 함수
+# -----------------------------
+
+def plot_item(item: str):
+    hist = pivot[item].dropna().asfreq("MS")
+    fc = FORECASTS[item]
+
+    fig = go.Figure()
+    fig.add_trace(go.Scatter(x=hist.index, y=hist.values, mode="lines", name="Historical"))
+    fig.add_trace(go.Scatter(x=fc.index, y=fc.values, mode="lines", name="Forecast"))
+    fig.update_layout(
+        title=f"{item} – Monthly Avg Price (1996‑2025) & Forecast(2025‑04→2030‑12)",
+        xaxis_title="Date",
+        yaxis_title="Price (KRW)",
+        legend=dict(orientation="h", y=1.02, x=0.01),
+        margin=dict(l=40, r=20, t=60, b=40),
+    )
+    return fig
+
+# -----------------------------
+# 5. Gradio UI
+# -----------------------------
+with gr.Blocks(title="도매 가격 예측 App") as demo:
+    gr.Markdown("## 📈 도매 가격 예측 대시보드 (1996‑2030)")
+
+    # 품목 선택 → 그래프 업데이트
+    item_dd = gr.Dropdown(products, value=products[0], label="품목 선택")
+    chart_out = gr.Plot(label="가격 추세")
+
+    # 내일 가격 표 (초기 고정)
+    gr.Markdown(f"### 내일({tomorrow}) 각 품목 예상가 (KRW)")
+    tomorrow_table = gr.Dataframe(tomorrow_df, interactive=False, height=400)
+
+    def update_chart(product):
+        return plot_item(product)
+
+    item_dd.change(update_chart, inputs=item_dd, outputs=chart_out, queue=False)
+
+# -----------------------------
+# 6. 실행 스크립트 엔트리포인트
+# -----------------------------
+if __name__ == "__main__":
+    demo.launch()