208 lines
8.3 KiB
Python
208 lines
8.3 KiB
Python
"""
|
|
forecast_helpers.py
|
|
====================
|
|
Modul untuk menghasilkan peramalan pada periode BEBAS (sesuai pilihan
|
|
user di date picker), karena `best_forecast` yang tersimpan di dalam
|
|
all_results.pkl hanya mencakup periode test asli (Jul-Des 2025).
|
|
|
|
Catatan penting soal transformasi:
|
|
- Kalau config['use_log'] == True, model dilatih pada log1p(y). Prophet
|
|
akan mengembalikan yhat dalam skala log lewat model.predict(), jadi
|
|
HARUS di-inverse-transform (np.expm1) supaya kembali ke satuan asli
|
|
(unit produk, misal Liter/Kg). Ini sudah dikonfirmasi dari data test:
|
|
yhat di best_forecast untuk skenario log_transform/log_regresor sudah
|
|
dalam skala asli (bukan skala log), jadi app harus meniru proses yang sama.
|
|
- Komponen dekomposisi (trend, weekly, yearly, holidays) TETAP dalam skala
|
|
model aslinya (skala log untuk skenario log_transform/log_regresor).
|
|
Ini karena komponen tidak bisa di-expm1 satu-satu lalu dijumlah (hasilnya
|
|
tidak akan sama dengan expm1 dari total). Ini perilaku standar Prophet,
|
|
bukan bug -- tapi HARUS diberi label jelas di UI ("skala log") supaya
|
|
tidak disalahartikan sebagai unit produk asli.
|
|
- Kalau config['use_regressors'] == True, model butuh kolom
|
|
'status_operasional' & 'ketersediaan_stok' (sudah di-scale MinMax(0,1))
|
|
untuk setiap tanggal yang mau diramal. Karena kita tidak tahu nilai
|
|
regresor di masa depan, dipakai ASUMSI KONSTAN "kondisi normal"
|
|
(nilai mentah 2 untuk kedua kolom, sesuai kesepakatan), lalu di-scale
|
|
pakai scaler yang tersimpan di entry['scaler'].
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
# Nilai mentah default untuk regresor pada periode masa depan
|
|
# (2 = kondisi paling umum/dominan di data historis, lihat analisis awal)
|
|
DEFAULT_STATUS_OPERASIONAL = 2
|
|
DEFAULT_KETERSEDIAAN_STOK = 2
|
|
|
|
# Kolom komponen dekomposisi yang ingin ditampilkan di UI
|
|
DECOMPOSITION_COLUMNS = ["trend", "weekly", "yearly", "holidays"]
|
|
|
|
|
|
class ForecastError(Exception):
|
|
"""Dilempar kalau forecast tidak bisa dibuat (mis. scaler hilang)."""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Membangun dataframe masa depan (input untuk model.predict())
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def build_future_dataframe(
|
|
product_entry: dict[str, Any],
|
|
scenario_name: str,
|
|
start_date,
|
|
end_date,
|
|
) -> pd.DataFrame:
|
|
"""
|
|
Bangun DataFrame `future` yang siap dipakai model.predict().
|
|
|
|
Kalau skenario butuh regresor, kolom status_operasional & ketersediaan_stok
|
|
diisi dengan asumsi "kondisi normal" (nilai mentah 2), lalu di-scale
|
|
pakai scaler produk tersebut.
|
|
"""
|
|
dates = pd.date_range(start=start_date, end=end_date, freq="D")
|
|
future = pd.DataFrame({"ds": dates})
|
|
|
|
scenario = product_entry["scenarios"][scenario_name]
|
|
needs_regressors = scenario["config"]["use_regressors"]
|
|
|
|
if needs_regressors:
|
|
scaler = product_entry.get("scaler")
|
|
if scaler is None:
|
|
raise ForecastError(
|
|
"Skenario ini butuh regresor eksternal, tapi produk tidak "
|
|
"punya scaler tersimpan. Tidak bisa membuat peramalan masa depan."
|
|
)
|
|
|
|
n = len(dates)
|
|
# dibungkus sebagai DataFrame bernama kolom (bukan array polos) karena
|
|
# scaler ini dilatih dari DataFrame -- kalau dikasih array biasa,
|
|
# scikit-learn akan memberi warning "X does not have valid feature names"
|
|
raw_status = pd.DataFrame({"status_operasional": [DEFAULT_STATUS_OPERASIONAL] * n})
|
|
raw_stok = pd.DataFrame({"ketersediaan_stok": [DEFAULT_KETERSEDIAAN_STOK] * n})
|
|
|
|
future["status_operasional"] = scaler["scaler_status"].transform(raw_status).flatten()
|
|
future["ketersediaan_stok"] = scaler["scaler_stok"].transform(raw_stok).flatten()
|
|
|
|
return future
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Menghasilkan forecast (dengan inverse-transform kalau perlu)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def generate_forecast(
|
|
product_entry: dict[str, Any],
|
|
scenario_name: str,
|
|
start_date,
|
|
end_date,
|
|
) -> tuple[pd.DataFrame, bool]:
|
|
"""
|
|
Hasilkan forecast untuk 1 produk, 1 skenario, pada rentang tanggal bebas.
|
|
|
|
Returns
|
|
-------
|
|
forecast : pd.DataFrame
|
|
Semua kolom asli hasil model.predict(), DITAMBAH:
|
|
- kolom yhat, yhat_lower, yhat_upper SUDAH dalam skala asli
|
|
(sudah di-expm1 kalau skenarionya pakai log).
|
|
- kolom komponen (trend, weekly, yearly, holidays, dst) MASIH
|
|
dalam skala model asli (lihat `is_log_scale_components`).
|
|
is_log_scale_components : bool
|
|
True kalau kolom komponen (selain yhat) masih dalam skala log,
|
|
sehingga UI wajib memberi label "skala log" saat menampilkannya.
|
|
"""
|
|
scenario = product_entry["scenarios"][scenario_name]
|
|
model = scenario["best_model"]
|
|
use_log = scenario["config"]["use_log"]
|
|
|
|
future = build_future_dataframe(product_entry, scenario_name, start_date, end_date)
|
|
forecast = model.predict(future)
|
|
|
|
if use_log:
|
|
for col in ("yhat", "yhat_lower", "yhat_upper"):
|
|
if col in forecast.columns:
|
|
forecast[col] = np.expm1(forecast[col])
|
|
|
|
return forecast, use_log
|
|
|
|
|
|
def generate_optimal_forecast(
|
|
product_entry: dict[str, Any],
|
|
start_date,
|
|
end_date,
|
|
) -> tuple[pd.DataFrame, str, bool]:
|
|
"""
|
|
Shortcut: hasilkan forecast pakai skenario OPTIMAL produk ini
|
|
(dipakai di Beranda & Detail Peramalan, di mana user tidak memilih
|
|
skenario secara manual -- sistem yang menentukan otomatis).
|
|
|
|
Returns
|
|
-------
|
|
forecast, optimal_scenario_name, is_log_scale_components
|
|
"""
|
|
optimal = product_entry["optimal_scenario"]
|
|
forecast, is_log = generate_forecast(product_entry, optimal, start_date, end_date)
|
|
return forecast, optimal, is_log
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Ekstraksi komponen dekomposisi (untuk Tab "Dekomposisi")
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_decomposition(forecast: pd.DataFrame) -> dict[str, pd.DataFrame]:
|
|
"""
|
|
Ambil komponen-komponen dekomposisi dari hasil forecast, dikelompokkan
|
|
supaya gampang dipakai untuk 4 mini-chart (Trend, Weekly, Yearly, Holidays).
|
|
|
|
Holiday effect dipecah lagi per nama hari libur (ramadhan, idul_fitri,
|
|
idul_adha, maulid) kalau kolom tersebut ada, karena itu lebih informatif
|
|
daripada cuma total 'holidays' gabungan.
|
|
"""
|
|
result: dict[str, pd.DataFrame] = {}
|
|
|
|
result["trend"] = forecast[["ds", "trend"]].copy()
|
|
|
|
if "weekly" in forecast.columns:
|
|
result["weekly"] = forecast[["ds", "weekly"]].copy()
|
|
|
|
if "yearly" in forecast.columns:
|
|
result["yearly"] = forecast[["ds", "yearly"]].copy()
|
|
|
|
holiday_cols = [c for c in ["ramadhan", "idul_fitri", "idul_adha", "maulid"] if c in forecast.columns]
|
|
if holiday_cols:
|
|
result["holidays"] = forecast[["ds", *holiday_cols]].copy()
|
|
elif "holidays" in forecast.columns:
|
|
result["holidays"] = forecast[["ds", "holidays"]].copy()
|
|
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Ringkasan angka (dipakai di header Detail Peramalan & agregasi Beranda)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def summarize_forecast(forecast: pd.DataFrame) -> dict[str, float]:
|
|
"""
|
|
Ringkasan sederhana dari 1 forecast: total, rata-rata harian, min, max.
|
|
Dipakai untuk kalimat semacam:
|
|
"Estimasi total penjualan 30 hari ke depan: 125 Liter (rata-rata 4.1 Liter/hari)"
|
|
"""
|
|
yhat = forecast["yhat"].clip(lower=0) # jaga-jaga, permintaan tidak boleh negatif
|
|
return {
|
|
"total": float(yhat.sum()),
|
|
"rata_rata_harian": float(yhat.mean()),
|
|
"minimum": float(yhat.min()),
|
|
"maksimum": float(yhat.max()),
|
|
"jumlah_hari": int(len(forecast)),
|
|
}
|
|
|
|
|
|
# build_future_dataframe → siapkan input tanggal (+ regresor asumsi) untuk Prophet
|
|
# generate_forecast → jalankan model.predict(), urus inverse-transform log kalau perlu
|
|
# generate_optimal_forecast → shortcut pakai skenario terbaik otomatis
|
|
# get_decomposition → pecah hasil jadi komponen-komponen buat chart
|
|
# summarize_forecast → ringkas jadi angka-angka kunci (total, rata-rata, min, max) |