app/forecaster.py

"""
Future Forecaster — N+1 only
Prediksi 1 bulan ke depan langsung dari data terakhir.
"""
import numpy as np
import pandas as pd
from dateutil.relativedelta import relativedelta


FITUR_LGBM = [
    "Price", "Is_Holiday", "Bulan", "Kuartal", "Tahun",
    "Category_enc", "Types_enc",
    "Qty_lag1", "Qty_lag2", "Qty_lag3", "Qty_lag12",
    "Qty_roll3", "Qty_roll6", "Qty_roll12"
]


def _get_holiday(target_date: pd.Timestamp, country_code: str) -> int:
    try:
        import holidays
        hl = getattr(holidays, country_code)(years=[target_date.year])
        days_in_month = pd.date_range(
            start=target_date.replace(day=1), end=target_date, freq="D"
        )
        return int(any(d.date() in hl for d in days_in_month))
    except Exception:
        return 0


def forecast_one_product(
    df_cont     : pd.DataFrame,
    product_id  : str,
    model,
    le_cat,
    le_types,
    country_code: str = "ID",
) -> dict:
    """
    Prediksi stok 1 bulan ke depan (N+1) dari data terakhir.
    N   = bulan data terakhir
    N+1 = bulan yang diprediksi
    """
    df_prod = df_cont[df_cont["Product_ID"] == product_id].copy()
    df_prod = df_prod.sort_values("Date").reset_index(drop=True)

    if df_prod.empty:
        return {"error": f"Produk {product_id} tidak ditemukan."}

    last_date   = df_prod["Date"].max()
    target_date = last_date + relativedelta(months=1)

    qty_history = df_prod["Quantity"].values

    def safe_lag(n):
        return float(qty_history[-n]) if len(qty_history) >= n else 0.0

    qty_lag1   = safe_lag(1)
    qty_lag2   = safe_lag(2)
    qty_lag3   = safe_lag(3)
    qty_lag12  = safe_lag(12)
    qty_roll3  = float(np.mean(qty_history[-3:]))  if len(qty_history) >= 3  else float(np.mean(qty_history))
    qty_roll6  = float(np.mean(qty_history[-6:]))  if len(qty_history) >= 6  else float(np.mean(qty_history))
    qty_roll12 = float(np.mean(qty_history[-12:])) if len(qty_history) >= 12 else float(np.mean(qty_history))

    price    = float(df_prod["Price"].iloc[-1])
    category = str(df_prod["Category"].iloc[-1])
    types    = str(df_prod["Types"].iloc[-1])
    name     = str(df_prod["Product_Name"].iloc[-1])

    try:    cat_enc   = int(le_cat.transform([category])[0])
    except: cat_enc   = 0
    try:    types_enc = int(le_types.transform([types])[0])
    except: types_enc = 0

    is_holiday = _get_holiday(target_date, country_code)

    feat_row = pd.DataFrame([{
        "Price": price, "Is_Holiday": is_holiday,
        "Bulan": target_date.month, "Kuartal": (target_date.month - 1) // 3 + 1,
        "Tahun": target_date.year, "Category_enc": cat_enc, "Types_enc": types_enc,
        "Qty_lag1": qty_lag1, "Qty_lag2": qty_lag2, "Qty_lag3": qty_lag3,
        "Qty_lag12": qty_lag12, "Qty_roll3": qty_roll3,
        "Qty_roll6": qty_roll6, "Qty_roll12": qty_roll12,
    }])[FITUR_LGBM]

    prediksi = max(0, round(model.predict(feat_row)[0]))

    df_hist     = df_prod.tail(12).copy()
    hist_labels = pd.to_datetime(df_hist["Date"]).dt.strftime("%b %Y").tolist()
    hist_qty    = df_hist["Quantity"].tolist()

    return {
        "product_id"  : product_id,
        "product_name": name,
        "category"    : category,
        "types"       : types,
        "last_data"   : last_date.strftime("%b %Y"),
        "target_month": target_date.strftime("%b %Y"),
        "prediksi"    : int(prediksi),
        "fitur_input" : {
            "qty_lag1" : qty_lag1, "qty_lag2": qty_lag2,
            "qty_lag3" : qty_lag3, "qty_lag12": qty_lag12,
            "qty_roll3": round(qty_roll3, 2), "is_holiday": is_holiday,
        },
        "hist_labels" : hist_labels,
        "hist_qty"    : hist_qty,
    }


def forecast_all_products(
    df_cont     : pd.DataFrame,
    model,
    le_cat,
    le_types,
    country_code: str = "ID",
    top_n       : int = None,
) -> list:
    if top_n:
        products = (
            df_cont.groupby("Product_ID")["Quantity"]
            .sum().sort_values(ascending=False)
            .head(top_n).index.tolist()
        )
    else:
        products = df_cont["Product_ID"].unique().tolist()

    results = []
    for pid in products:
        r = forecast_one_product(df_cont, pid, model, le_cat, le_types, country_code)
        if "error" not in r:
            results.append(r)
    return results