319 lines
12 KiB
Python
319 lines
12 KiB
Python
"""
|
|
data_loader.py (sebagai "Gudang Data" atau "Penyedia Bahan Baku")
|
|
==============
|
|
Modul untuk membaca dan menyimpan cache file master `all_results.pkl`,
|
|
serta menyediakan fungsi-fungsi bantu untuk mengambil informasi ringkas
|
|
dari struktur data (daftar produk, skenario optimal, tabel metrik, dsb).
|
|
|
|
Struktur `all_results.pkl` (per produk):
|
|
{
|
|
"<nama_produk>": {
|
|
"data": {
|
|
"train": <DataFrame kolom: ds, y, status_operasional, ketersediaan_stok>,
|
|
"test": <DataFrame kolom sama>
|
|
},
|
|
"scenarios": {
|
|
"standar": {config, best_params, best_metrics, best_model, best_forecast, all_results},
|
|
"log_transform":{...},
|
|
"regresor": {...},
|
|
"log_regresor": {...},
|
|
},
|
|
"optimal_scenario": "regresor", # salah satu dari 4 nama skenario di atas
|
|
"scaler": { # HANYA ADA jika produk punya skenario berbasis regresor
|
|
"scaler_status": MinMaxScaler(...),
|
|
"scaler_stok": MinMaxScaler(...),
|
|
}
|
|
},
|
|
...
|
|
}
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import pickle
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
import pandas as pd
|
|
import streamlit as st
|
|
|
|
# Skenario yang WAJIB ada di setiap produk
|
|
EXPECTED_SCENARIOS = {"standar", "log_transform", "regresor", "log_regresor"}
|
|
|
|
# Skenario yang membutuhkan regresor eksternal (dan karenanya, scaler)
|
|
REGRESSOR_SCENARIOS = {"regresor", "log_regresor"}
|
|
|
|
DEFAULT_DATA_PATH = os.path.join("data", "all_results.pkl")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pembacaan file master (dengan cache, supaya tidak baca ulang tiap interaksi)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@st.cache_resource(show_spinner="Memuat data model...")
|
|
def load_all_results(path: str = DEFAULT_DATA_PATH) -> dict[str, Any]:
|
|
"""
|
|
Baca file all_results.pkl dan simpan di cache resource Streamlit.
|
|
|
|
Menggunakan st.cache_resource (bukan st.cache_data) karena isinya
|
|
mengandung objek Prophet (model) yang tidak bisa/tidak perlu di-hash
|
|
seperti data biasa.
|
|
|
|
Raises
|
|
------
|
|
FileNotFoundError
|
|
Jika file tidak ditemukan di path yang diberikan.
|
|
ValueError
|
|
Jika file berhasil dibaca tapi bukan dictionary yang valid.
|
|
"""
|
|
if not os.path.exists(path):
|
|
raise FileNotFoundError(
|
|
f"File model tidak ditemukan di '{path}'. "
|
|
"Pastikan all_results.pkl sudah ditempatkan di folder data/."
|
|
)
|
|
|
|
with open(path, "rb") as f:
|
|
data = pickle.load(f)
|
|
|
|
if not isinstance(data, dict) or len(data) == 0:
|
|
raise ValueError("File all_results.pkl tidak berisi dictionary produk yang valid.")
|
|
|
|
return data
|
|
|
|
|
|
def reload_all_results(path: str = DEFAULT_DATA_PATH) -> dict[str, Any]:
|
|
"""
|
|
Paksa baca ulang file dari disk (melewati cache).
|
|
Dipanggil setelah admin berhasil upload & merge model baru,
|
|
supaya perubahan langsung terlihat tanpa perlu restart aplikasi.
|
|
"""
|
|
load_all_results.clear() # kosongkan cache st.cache_resource
|
|
return load_all_results(path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fungsi bantu: info dasar
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_product_list(all_results: dict[str, Any]) -> list[str]:
|
|
"""Daftar nama produk, urutan sesuai urutan di file pkl."""
|
|
return list(all_results.keys())
|
|
|
|
|
|
def get_optimal_scenario(all_results: dict[str, Any], product: str) -> str:
|
|
"""Nama skenario optimal untuk satu produk (mis. 'log_regresor')."""
|
|
return all_results[product]["optimal_scenario"]
|
|
|
|
|
|
def get_scenario_names(all_results: dict[str, Any], product: str) -> list[str]:
|
|
"""Daftar nama skenario yang tersedia untuk satu produk."""
|
|
return list(all_results[product]["scenarios"].keys())
|
|
|
|
|
|
def uses_regressors(all_results: dict[str, Any], product: str, scenario: str) -> bool:
|
|
"""True kalau skenario ini butuh regresor eksternal (status_operasional & ketersediaan_stok)."""
|
|
return all_results[product]["scenarios"][scenario]["config"]["use_regressors"]
|
|
|
|
|
|
def uses_log_transform(all_results: dict[str, Any], product: str, scenario: str) -> bool:
|
|
"""True kalau target y dilatih dalam skala log (butuh inverse-transform expm1 saat prediksi)."""
|
|
return all_results[product]["scenarios"][scenario]["config"]["use_log"]
|
|
|
|
|
|
def has_scaler(all_results: dict[str, Any], product: str) -> bool:
|
|
"""Cek apakah produk punya scaler tersimpan (dibutuhkan kalau skenario optimalnya pakai regresor)."""
|
|
return "scaler" in all_results[product]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Batas tanggal peramalan OTOMATIS per produk (bukan hardcode global)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_forecast_date_bounds(all_results: dict[str, Any], product: str) -> tuple[datetime, datetime]:
|
|
"""
|
|
Hitung batas tanggal peramalan yang VALID untuk 1 produk, otomatis dari
|
|
isi .pkl-nya sendiri -- bukan angka hardcode yang sama untuk semua produk.
|
|
|
|
Batas Awal (min_date): 1 hari setelah tanggal terakhir di data.test
|
|
produk ini. Karena data test adalah data riil terakhir yang dipakai
|
|
untuk evaluasi model -- peramalan baru masuk akal dimulai SETELAH itu.
|
|
|
|
Batas Akhir (max_date): 31 Desember dari tahun TERAKHIR yang tercatat
|
|
di dataset hari libur (`model.holidays`) yang dipakai skenario
|
|
optimal produk ini. Karena Prophet butuh definisi holiday untuk
|
|
tanggal yang diramal -- di luar tahun yang terdaftar, efek
|
|
holiday-nya jadi tidak valid/tidak representatif.
|
|
|
|
Kalau produk tidak punya data holiday sama sekali (kasus langka),
|
|
fallback: max_date = min_date + 1 tahun, supaya tetap ada rentang
|
|
yang bisa dipilih (bukan error/rentang kosong).
|
|
|
|
Returns
|
|
-------
|
|
(min_date, max_date) : tuple[datetime.date, datetime.date]
|
|
"""
|
|
entry = all_results[product]
|
|
|
|
# --- Batas awal: dari data test produk ini ---
|
|
test_df = entry["data"]["test"]
|
|
last_test_date = pd.to_datetime(test_df["ds"]).max()
|
|
min_date = (last_test_date + pd.Timedelta(days=1)).date()
|
|
|
|
# --- Batas akhir: dari holidays yang dipakai skenario optimal produk ini ---
|
|
optimal_scenario = entry["optimal_scenario"]
|
|
model = entry["scenarios"][optimal_scenario]["best_model"]
|
|
holidays_df = getattr(model, "holidays", None)
|
|
|
|
if holidays_df is not None and not holidays_df.empty:
|
|
last_holiday_year = int(pd.to_datetime(holidays_df["ds"]).dt.year.max())
|
|
max_date = datetime(last_holiday_year, 12, 31).date()
|
|
else:
|
|
# Fallback kalau model ini kebetulan tidak punya data holiday sama sekali.
|
|
max_date = (last_test_date + pd.DateOffset(years=1)).date()
|
|
|
|
# Jaga-jaga: kalau karena suatu hal max_date < min_date (data holiday
|
|
# sudah kadaluarsa dibanding data test), paksa max_date ikut min_date
|
|
# supaya rentang kalender tidak pernah kosong/terbalik.
|
|
if max_date < min_date:
|
|
max_date = min_date
|
|
|
|
return min_date, max_date
|
|
|
|
|
|
|
|
def get_aggregate_forecast_date_bounds(
|
|
all_results: dict[str, Any], products: list[str] | None = None,
|
|
) -> tuple[datetime, datetime]:
|
|
"""
|
|
>>> FIX: Versi AGREGAT dari get_forecast_date_bounds() -- dipakai di
|
|
halaman yang menampilkan BANYAK produk sekaligus dengan 1 periode yang
|
|
sama (mis. Beranda), bukan 1 produk seperti Detail Peramalan.
|
|
|
|
Karena semua produk berbagi 1 periode yang sama, batasnya harus berupa
|
|
IRISAN (intersection) supaya periode yang dipilih valid untuk SEMUA
|
|
produk sekaligus -- bukan gabungan (union) yang malah bisa membuat
|
|
sebagian produk gagal diramal.
|
|
|
|
(Prinsip IRISAN ini dipakai supaya user tidak bisa memilih rentang tanggal
|
|
di kalender yang bakal bikin salah satu atau sebagian produk error/gagal
|
|
diramal di halaman Beranda.)
|
|
|
|
- min_date = tanggal paling AKHIR di antara semua produk (produk dengan data test paling baru yang menentukan)
|
|
- max_date = tanggal paling AWAL di antara semua produk (produk dengan cakupan data hari libur paling pendek yang menentukan)
|
|
|
|
Kalau `products` tidak diisi, otomatis pakai SEMUA produk yang ada
|
|
di `all_results`.
|
|
|
|
Returns
|
|
-------
|
|
(min_date, max_date) : tuple[datetime.date, datetime.date]
|
|
Kalau `products` kosong (tidak ada produk sama sekali), fallback
|
|
ke hari ini sebagai min_date dan +1 tahun sebagai max_date --
|
|
supaya tetap ada rentang yang bisa dipilih, bukan error.
|
|
"""
|
|
if products is None:
|
|
products = list(all_results.keys())
|
|
|
|
if not products:
|
|
today = datetime.now().date()
|
|
return today, (pd.Timestamp(today) + pd.DateOffset(years=1)).date()
|
|
|
|
bounds = [get_forecast_date_bounds(all_results, p) for p in products]
|
|
min_date = max(b[0] for b in bounds) # paling akhir di antara semua produk
|
|
max_date = min(b[1] for b in bounds) # paling awal di antara semua produk
|
|
|
|
# Jaga-jaga: kalau irisan tidak ketemu (satu produk sangat baru datanya,
|
|
# produk lain cakupan holiday-nya pendek) sehingga min > max, jangan
|
|
# sampai rentang kalender jadi kosong/terbalik.
|
|
if max_date < min_date:
|
|
max_date = min_date
|
|
|
|
return min_date, max_date
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fungsi bantu: tabel metrik & parameter (untuk halaman Perbandingan Konfigurasi)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_metric_comparison_table(all_results: dict[str, Any], metric: str = "MAPE") -> pd.DataFrame:
|
|
"""
|
|
Bangun tabel perbandingan 1 metrik (default MAPE) untuk semua produk x semua skenario.
|
|
|
|
Hasil: DataFrame dengan baris = produk, kolom = 4 skenario, nilai = metrik (%),
|
|
plus kolom 'optimal_scenario' untuk penanda skenario mana yang terbaik per produk.
|
|
|
|
Parameters
|
|
----------
|
|
metric : str
|
|
Salah satu dari 'MAE', 'RMSE', 'MAPE'.
|
|
"""
|
|
rows = []
|
|
for product, entry in all_results.items():
|
|
row = {"produk": product}
|
|
for scenario_name, scenario in entry["scenarios"].items():
|
|
row[scenario_name] = scenario["best_metrics"].get(metric)
|
|
row["optimal_scenario"] = entry["optimal_scenario"]
|
|
rows.append(row)
|
|
|
|
df = pd.DataFrame(rows).set_index("produk")
|
|
return df
|
|
|
|
|
|
def get_product_metrics_detail(all_results: dict[str, Any], product: str) -> pd.DataFrame:
|
|
"""
|
|
Tabel MAE/RMSE/MAPE untuk 4 skenario, KHUSUS 1 produk.
|
|
Dipakai di tab "Kinerja" pada halaman Detail Peramalan.
|
|
"""
|
|
entry = all_results[product]
|
|
rows = []
|
|
for scenario_name, scenario in entry["scenarios"].items():
|
|
m = scenario["best_metrics"]
|
|
rows.append({
|
|
"skenario": scenario_name,
|
|
"MAE": m.get("MAE"),
|
|
"RMSE": m.get("RMSE"),
|
|
"MAPE": m.get("MAPE"),
|
|
"optimal": scenario_name == entry["optimal_scenario"],
|
|
})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def get_hyperparameter_table(all_results: dict[str, Any]) -> pd.DataFrame:
|
|
"""
|
|
Tabel parameter terbaik (hasil grid search) dari SKENARIO OPTIMAL tiap produk.
|
|
Dipakai di halaman Perbandingan Konfigurasi (khusus admin).
|
|
"""
|
|
rows = []
|
|
for product, entry in all_results.items():
|
|
optimal = entry["optimal_scenario"]
|
|
params = entry["scenarios"][optimal]["best_params"]
|
|
row = {"produk": product, "skenario_optimal": optimal, **params}
|
|
rows.append(row)
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fungsi bantu: metadata pembaruan (untuk panel "Perbarui Model")
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_last_updated_info(all_results: dict[str, Any]) -> pd.DataFrame:
|
|
"""
|
|
Tabel ringkas: produk, skenario optimal, status scaler.
|
|
NOTE: timestamp "terakhir diperbarui" per produk belum tersedia dari
|
|
struktur pkl saat ini (tidak ada field waktu di dalam entry produk).
|
|
Kalau mau ditampilkan seperti di mockup, perlu ditambah field baru,
|
|
misalnya entry[produk]['updated_at'], diisi otomatis oleh model_manager.py
|
|
setiap kali produk itu di-update lewat panel upload.
|
|
"""
|
|
rows = []
|
|
for product, entry in all_results.items():
|
|
rows.append({
|
|
"produk": product,
|
|
"skenario_optimal": entry["optimal_scenario"],
|
|
"scaler_tersedia": has_scaler(all_results, product),
|
|
"diperbarui_terakhir": entry.get("updated_at", "-"),
|
|
})
|
|
return pd.DataFrame(rows)
|