""" 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): { "": { "data": { "train": , "test": }, "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)