TIF_E41253317/views/detail_peramalan.py

1045 lines
46 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
views/detail_peramalan.py
==========================
Halaman detail per-produk (semua role bisa akses). Alurnya:
1. FORM-GATE: kalau user belum pernah pilih produk & periode di sesi ini
(baik lewat form di halaman ini, ATAU lewat tombol "Lihat detail" di
Beranda), tampilkan form pemilihan dulu -- jangan langsung hitung
apa pun. Ini menghindari beban render tiap kali halaman dibuka.
2. Begitu ada pilihan, tampilkan kontrol ringkas (bisa diganti kapan saja)
+ 2 tab:
- Tab "Peramalan": grafik historis+prediksi, highlight box insight,
tabel harian + tombol unduh CSV.
- Tab "Dekomposisi": 4 komponen model (trend, weekly, yearly, holidays)
+ Rekomendasi Aksi Bisnis Dinamis.
"""
from __future__ import annotations
import datetime
import pandas as pd
import plotly.graph_objects as go
import streamlit as st
from utils import pdf_report, report_insights as ri, ui_components
from utils.data_loader import get_product_list, get_forecast_date_bounds
from utils.forecast_helpers import generate_optimal_forecast, get_decomposition, summarize_forecast
PRODUCT_UNITS = {
"minyak": "Kg",
"gula": "Kg",
"tepung": "Kg",
"telur": "Kg",
"sedap goreng": "Pcs",
}
HARI_URUTAN = ["Senin", "Selasa", "Rabu", "Kamis", "Jumat", "Sabtu", "Minggu"]
# ---------------------------------------------------------------------------
# Cache forecast
# ---------------------------------------------------------------------------
# @st.cache_data(show_spinner="Menghitung peramalan...")
@st.cache_data(show_spinner=False)
def _cached_forecast(_all_results: dict, product: str, start: str, end: str):
entry = _all_results[product]
return generate_optimal_forecast(entry, start, end)
# ---------------------------------------------------------------------------
# FORM-GATE
# ---------------------------------------------------------------------------
def _render_form_gate(all_results: dict, products: list[str]) -> None:
_, col, _ = st.columns([1, 1.4, 1])
with col:
st.markdown(
"""
<style>
/* >>> FIX: rule "gap" via CSS dihapus -- ganti pakai parameter
resmi `gap=` di st.container() di bawah (lebih pasti bekerja,
tidak bergantung struktur DOM internal Streamlit yang bisa
beda-beda antar versi/rilis -- ada known-issue di GitHub
Streamlit soal class "st-key-..." tidak selalu nempel di
elemen yang diharapkan). Cuma sisa aturan spacing header saja
di sini. */
.gate-header {
text-align: center;
margin-top: 1.5rem;
margin-bottom: 1rem;
}
.gate-header p {
font-weight: 500;
margin-top: 6px !important;
margin-bottom: 0px !important;
}
</style>
""",
unsafe_allow_html=True,
)
st.markdown(
"<div class='gate-header'>"
"<span style='font-size:28px;'>🔍</span>"
"<p>Pilih produk & periode untuk mulai peramalan</p>"
"</div>",
unsafe_allow_html=True,
)
with st.container(key="form_gate_box", gap="small", border=True):
product = st.selectbox(
"Produk FMCG", products, format_func=str.title, key="form_gate_product",
)
today = datetime.date.today()
min_d, max_d = get_forecast_date_bounds(all_results, product)
start, end = ui_components.date_range_picker(
st, "Periode", today, today, key="form_gate_period",
min_date=min_d, max_date=max_d,
)
submitted = st.button(
"Tampilkan hasil peramalan", width="stretch", type="primary", key="form_gate_submit",
)
if submitted:
if end < start:
st.error("Tanggal akhir harus setelah atau sama dengan tanggal awal.")
else:
st.session_state["selected_product"] = product
st.session_state["selected_period"] = (start, end)
st.rerun()
# ---------------------------------------------------------------------------
# Kontrol ringkas
# ---------------------------------------------------------------------------
def _render_inline_controls(all_results: dict,products: list[str], product: str, start, end):
c1, c2, c3 = st.columns([1, 1, 0.5], vertical_alignment="bottom")
new_product = c1.selectbox(
"Produk", products, index=products.index(product) if product in products else 0,
format_func=str.title, key="detail_product_select",
)
# new_start, new_end = ui_components.date_range_picker(
# c2, "Periode", start, end, key="detail_period_select",
# )
min_d, max_d = get_forecast_date_bounds(all_results, new_product)
new_start, new_end = ui_components.date_range_picker(
c2, "Periode", start, end, key="detail_period_select",
min_date=min_d, max_date=max_d,
)
if new_end < new_start:
st.error("Tanggal akhir harus setelah atau sama dengan tanggal awal.")
st.stop()
if new_product != product or (new_start, new_end) != (start, end):
st.session_state["selected_product"] = new_product
st.session_state["selected_period"] = (new_start, new_end)
st.rerun()
with c3:
st.markdown(
"""
<style>
div[data-testid="stDownloadButton"] button {
height: 42px !important;
margin-bottom: 0px !important;
}
</style>
""",
unsafe_allow_html=True,
)
try:
forecast_dl, scenario_dl, is_log_dl = _cached_forecast(all_results, new_product, str(new_start), str(new_end))
unit_dl = PRODUCT_UNITS.get(new_product, "unit")
summary_dl = summarize_forecast(forecast_dl)
_render_pdf_export(
all_results[new_product], new_product, unit_dl, forecast_dl, summary_dl,
new_start, new_end, scenario_dl, is_log_dl,
)
except Exception:
# Kalau forecast gagal, biarkan render() yang menampilkan pesan
# errornya (di bawah) -- di sini cukup diam saja, tidak usah
# menampilkan error dobel.
pass
return new_product, new_start, new_end
# ---------------------------------------------------------------------------
# Tab 1: Peramalan
# ---------------------------------------------------------------------------
def _main_chart(product_entry: dict, forecast: pd.DataFrame, start, end, unit: str = "unit") -> go.Figure:
fig = go.Figure()
yhat_formatted = forecast["yhat"].clip(lower=0).apply(
lambda x: f"{x:,.3f}".replace(",", "X").replace(".", ",").replace("X", ".")
)
scenario_name = product_entry["optimal_scenario"]
model = product_entry["scenarios"][scenario_name]["best_model"]
event_tags = []
if hasattr(model, "holidays") and model.holidays is not None and not model.holidays.empty:
holidays_df = model.holidays
for ds_val in forecast["ds"]:
curr_date = pd.to_datetime(ds_val).date()
matched_events = []
for _, row in holidays_df.iterrows():
ev_date = pd.to_datetime(row["ds"]).date()
ev_name = str(row["holiday"]).replace("_", " ").title()
lower_win = int(row["lower_window"]) if "lower_window" in row and not pd.isna(row["lower_window"]) else 0
upper_win = int(row["upper_window"]) if "upper_window" in row and not pd.isna(row["upper_window"]) else 0
start_win = ev_date + datetime.timedelta(days=lower_win)
end_win = ev_date + datetime.timedelta(days=upper_win)
if start_win <= curr_date <= end_win:
selisih = (curr_date - ev_date).days
if selisih < 0:
status = f"H{selisih}"
elif selisih == 0:
status = "Hari H"
else:
status = f"H+{selisih}"
matched_events.append(f"{status} {ev_name}")
# tidak di-break -- lanjut cek baris lain, siapa tahu ada event lain
# yang windownya juga menyentuh tanggal yang sama
tag = f" ({', '.join(matched_events)})" if matched_events else ""
event_tags.append(tag)
else:
event_tags = [""] * len(forecast)
customdata_combined = list(zip(yhat_formatted, event_tags))
fig.add_trace(go.Scatter(
x=pd.concat([forecast["ds"], forecast["ds"][::-1]]),
y=pd.concat([forecast["yhat_upper"], forecast["yhat_lower"][::-1]]),
fill="toself", fillcolor="rgba(37,99,235,0.12)",
line=dict(color="rgba(0,0,0,0)"), hoverinfo="skip",
name="Rentang estimasi", showlegend=True,
))
fig.add_trace(go.Scatter(
x=forecast["ds"], y=forecast["yhat"].clip(lower=0),
mode="lines+markers", name="Peramalan",
line=dict(color="#2563eb", width=2),
marker=dict(size=4),
customdata=customdata_combined,
hovertemplate="%{x|%d-%m-%Y}%{customdata[1]}<br>Peramalan: %{customdata[0]} " + unit + "<extra></extra>"
))
fig.update_layout(
height=320, margin=dict(l=10, r=10, t=10, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left", x=0),
hovermode="closest",
plot_bgcolor="rgba(0,0,0,0)",
xaxis=dict(
hoverformat="%d-%m-%Y",
tickformat="%d-%m-%Y"
)
)
return fig
def _render_highlight_box(
product: str, unit: str, forecast: pd.DataFrame, summary: dict,
product_entry: dict, start: datetime.date, end: datetime.date,
) -> None:
n_hari = summary["jumlah_hari"]
total_fmt = f"{summary['total']:,.0f}".replace(",", ".")
rata_fmt = f"{summary['rata_rata_harian']:.3f}".replace(".", ",")
scenario_name = product_entry["optimal_scenario"]
model = product_entry["scenarios"][scenario_name]["best_model"]
bullets: list[str] = []
is_event_period = False # menentukan warna box: oranye kalau ada event yang menyentuh periode ini
# ---------- Pola mingguan & event -- DIHITUNG untuk SEMUA n_hari (termasuk 1 hari) ----------
# >>> FIX: sebelumnya blok ini ada di dalam "if n_hari > 1:", jadi untuk
# periode 1 hari, is_event_period SELALU False walau tanggal itu memang
# H-/Hari H/H+ dari suatu event -- box tidak akan pernah oranye. Sekarang
# dipindah ke sini, dihitung sekali untuk semua kasus.
decomposition = get_decomposition(forecast)
weekly_top_day, weekly_bottom_day = None, None
pola_mingguan = pd.Series(dtype=float) # Inisialisasi default agar safe
if "weekly" in decomposition:
pola_mingguan = _weekly_pattern(decomposition["weekly"]).dropna()
# FIX: Jika pola_mingguan kurang dari 7 hari (misal user cuma pilih 1 hari),
# buat sampel forecast 1 minggu penuh (Senin-Minggu) agar tahu pola 7 hari sebenarnya.
if len(pola_mingguan) < 7:
try:
week_start = start - datetime.timedelta(days=start.weekday()) # Senin
week_end = week_start + datetime.timedelta(days=6) # Minggu
full_week_fc, _, _ = generate_optimal_forecast(product_entry, str(week_start), str(week_end))
full_decomp = get_decomposition(full_week_fc)
if "weekly" in full_decomp:
pola_mingguan = _weekly_pattern(full_decomp["weekly"]).dropna()
except Exception:
pass
if not pola_mingguan.empty and (pola_mingguan.max() - pola_mingguan.min()) > 1e-9:
weekly_top_day = pola_mingguan.idxmax()
weekly_bottom_day = pola_mingguan.idxmin()
events_in_period = []
holidays_df = getattr(model, "holidays", None)
if holidays_df is not None and not holidays_df.empty:
seen_events = set()
for _, row in holidays_df.iterrows():
ev_date = pd.to_datetime(row["ds"]).date()
ev_name_raw = str(row["holiday"])
lower_win = int(row["lower_window"]) if "lower_window" in row and not pd.isna(row["lower_window"]) else 0
upper_win = int(row["upper_window"]) if "upper_window" in row and not pd.isna(row["upper_window"]) else 0
win_start = ev_date + datetime.timedelta(days=lower_win)
win_end = ev_date + datetime.timedelta(days=upper_win)
if win_end < start or win_start > end:
continue # window event tidak menyentuh periode yang dipilih sama sekali
key = (ev_name_raw, ev_date)
if key in seen_events:
continue
seen_events.add(key)
events_in_period.append({
"name": ev_name_raw.replace("_", " ").title(),
"ev_date": ev_date,
"win_start": win_start,
"win_end": win_end,
"hari_h_masuk_periode": start <= ev_date <= end,
})
events_in_period.sort(key=lambda e: e["ev_date"])
is_event_period = len(events_in_period) > 0
# ================= 1. KISARAN ESTIMASI KESELURUHAN PERIODE =================
overall_lower = forecast["yhat_lower"].clip(lower=0).min()
overall_upper = forecast["yhat_upper"].max()
overall_lower_fmt = f"{overall_lower:,.0f}".replace(",", ".")
overall_upper_fmt = f"{overall_upper:,.0f}".replace(",", ".")
overall_yhatmin = forecast["yhat"].clip(lower=0).min()
overall_yhatmax = forecast["yhat"].max()
overall_yhatmin_fmt = f"{overall_yhatmin:,.0f}".replace(",", ".")
overall_yhatmax_fmt = f"{overall_yhatmax:,.0f}".replace(",", ".")
start_str = start.strftime("%d-%m-%Y")
end_str = end.strftime("%d-%m-%Y")
if n_hari == 1:
# >>> FIX: insight 1 hari dirombak total supaya SELENGKAP periode
# multi-hari -- mencakup (a) nama hari + konteks pola mingguan,
# (b) keterangan event kalau tanggal itu H-/Hari H/H+ suatu event,
# (c) saran restock yang eksplisit sebut batas atas (yhat_upper).
the_date = start
the_dow = HARI_URUTAN[the_date.weekday()]
if weekly_top_day and the_dow == weekly_top_day:
weekly_txt = f" Hari <b>{the_dow}</b> merupakan hari dengan <b>permintaan tertinggi</b> dibandingkan hari lain berdasarkan pola mingguan."
hubungkata = " Ditambah lagi,"
elif weekly_bottom_day and the_dow == weekly_bottom_day:
weekly_txt = f" Hari <b>{the_dow}</b> merupakan hari dengan <b>permintaan terendah</b> dibandingkan hari lain berdasarkan pola mingguan."
hubungkata = " Meski begitu,"
else:
weekly_txt = f" Hari <b>{the_dow}</b> merupakan hari dengan <b>permintaan normal</b> berdasarkan pola mingguan."
hubungkata = " Selain itu,"
active_events_here = []
for ev in events_in_period:
if ev["win_start"] <= the_date <= ev["win_end"]:
selisih = (the_date - ev["ev_date"]).days
status_waktu = "Hari H" if selisih == 0 else (f"H{selisih}" if selisih < 0 else f"H+{selisih}")
active_events_here.append(f"{status_waktu} {ev['name']}")
event_txt = ""
if active_events_here:
event_txt = f" {hubungkata} tanggal ini bertepatan dengan <b>{', '.join(active_events_here)}</b>, yang berpotensi memengaruhi permintaan."
bullets.append(
f"Peramalan permintaan pada <b>{the_dow}, {start_str}</b> diperkirakan "
f"<b>{total_fmt} {unit}</b> (kisaran estimasi: <b>{overall_lower_fmt}-{overall_upper_fmt} {unit}</b>)."
f"{weekly_txt}{event_txt} Siapkan stok utama sebesar <b>{total_fmt} {unit}</b> dan pertimbangkan "
f"stok cadangan hingga <b>{overall_upper_fmt} {unit}</b> untuk mengantisipasi lonjakan mendadak."
)
else:
bullets.append(
f"Peramalan permintaan periode {start_str} - {end_str} berkisar antara <b>{overall_yhatmin_fmt} {unit} (terendah)</b> hingga <b>{overall_yhatmax_fmt} {unit} (tertinggi)</b>."
)
# ================= 2. TOTAL & RATA-RATA =================
bullets.append(
f"Total peramalan permintaan <b>{n_hari} hari</b> tersebut: "
f"<b>{total_fmt} {unit}</b> (rata-rata {rata_fmt} {unit}/hari)."
)
if n_hari > 1:
rata_harian = summary["rata_rata_harian"]
# ================= 3. TITIK TERENDAH =================
min_idx = forecast["yhat"].idxmin()
min_row = forecast.loc[min_idx]
min_ds = min_row["ds"].date()
min_yhat = min_row["yhat"]
min_dow = HARI_URUTAN[min_ds.weekday()]
min_ds_str = min_ds.strftime("%d-%m-%Y")
min_yhat_fmt = f"{min_yhat:,.0f}".replace(",", ".")
pct_low_diff = ((min_yhat / rata_harian) - 1) * 100 if rata_harian > 0 else 0
pct_low_fmt = f"{abs(pct_low_diff):.0f}".replace(".", ",")
low_sebab = ""
if weekly_bottom_day and min_dow == weekly_bottom_day:
low_sebab = f" Ini sejalan dengan pola mingguan di hari <b>{min_dow}</b>, yang biasanya merupakan hari dengan permintaan rendah dibandingkan hari lain (lihat di tab Dekomposisi)."
bullets.append(
f"Permintaan <b>TERENDAH</b> diperkirakan pada <b>{min_dow}, {min_ds_str}</b>: "
f"<b>{min_yhat_fmt} {unit}</b> sekitar <b>{pct_low_fmt}%</b> di bawah rata-rata periode ini."
f"{low_sebab}"
f" Bisa jadi waktu yang tepat untuk <i>restock</i>."
)
# ================= 4. TITIK TERTINGGI =================
max_idx = forecast["yhat"].idxmax()
max_row = forecast.loc[max_idx]
max_ds = max_row["ds"].date()
max_yhat = max_row["yhat"]
max_lower = max(max_row["yhat_lower"], 0)
max_upper = max_row["yhat_upper"]
max_dow = HARI_URUTAN[max_ds.weekday()]
max_ds_str = max_ds.strftime("%d-%m-%Y")
max_yhat_fmt = f"{max_yhat:,.0f}".replace(",", ".")
max_lower_fmt = f"{max_lower:,.0f}".replace(",", ".")
max_upper_fmt = f"{max_upper:,.0f}".replace(",", ".")
pct_diff = ((max_yhat / rata_harian) - 1) * 100 if rata_harian > 0 else 0
pct_fmt = f"{pct_diff:.0f}".replace(".", ",")
# Cari SEMUA event yang aktif TEPAT di tanggal puncak (bisa lebih dari 1), untuk badge & penjelasan sebab
peak_events_here = []
for ev in events_in_period:
if ev["win_start"] <= max_ds <= ev["win_end"]:
selisih = (max_ds - ev["ev_date"]).days
status_waktu = "Hari H" if selisih == 0 else (f"H{selisih}" if selisih < 0 else f"H+{selisih}")
peak_events_here.append(f"{status_waktu} {ev['name']}")
badge_info = f"{', '.join(peak_events_here)}" if peak_events_here else ""
peak_event_names = {
ev["name"] for ev in events_in_period if ev["win_start"] <= max_ds <= ev["win_end"]
}
sebab_list = []
if peak_event_names:
sebab_list.append(f"efek event <b>{', '.join(sorted(peak_event_names))}</b>")
if weekly_top_day and max_dow == weekly_top_day:
sebab_list.append(f"pola mingguan di hari <b>{max_dow}</b>, yang biasanya merupakan hari dengan permintaan tinggi dibandingkan hari lain")
sebab_str = " & ".join(sebab_list)
peak_txt = (
f"Permintaan <b>TERTINGGI</b> diperkirakan pada <b>{max_dow}, {max_ds_str}</b>: "
f"<b>{max_yhat_fmt} {unit}</b> sekitar <b>{pct_fmt}%</b> di atas rata-rata periode ini."
)
if sebab_str:
peak_txt += f" Kenaikan ini dipengaruhi oleh {sebab_str} (lihat di tab Dekomposisi)."
peak_txt += (
f" Pastikan stok mendekati batas atas tersedia."
)
bullets.append(peak_txt)
# ================= 5. EVENT LAIN DALAM PERIODE (selain yang sudah disebut di titik puncak) =================
other_events = [ev for ev in events_in_period if ev["name"] not in peak_event_names]
if other_events:
nama_event_lain = ", ".join(sorted({ev["name"] for ev in other_events}))
bullets.append(
f"Periode ini terdeteksi efek event <b>{nama_event_lain}</b>. "
f"Arahkan kursor ke titik pada grafik di atas untuk melihat tanggal & detail pengaruhnya."
)
# Rangkai semua poin jadi bullet list kecuali 1 hari
if len(bullets) == 1:
msg = bullets[0]
else:
list_html = "".join(f"<li style='margin-bottom: 6px;'>{b}</li>" for b in bullets)
msg = f"<ul style='margin: 0; padding-left: 18px;'>{list_html}</ul>"
# ================= CATATAN ASUMSI MODEL (di luar bullet, sebagai footnote) =================
msg += (
"<div style='margin-top: 10px; font-size: 0.8rem; opacity: 0.85;'>"
"ℹ️ Catatan: peramalan ini mengasumsikan stok tersedia & operasional toko berjalan normal "
"(tidak ada gangguan pasokan atau tutup mendadak). Kalau ada kendala operasional pada tanggal "
"tertentu, angka aktual bisa berbeda dari peramalan ini."
"</div>"
)
bg_color = "#fff2e0" if is_event_period else "#e7f7ee"
border_color = "#f8d35a" if is_event_period else "#a7f3d0"
text_color = "#b45309" if is_event_period else "#1a7f4e"
st.markdown(
f"""
<div style="
background-color: {bg_color};
border: 1px solid {border_color};
color: {text_color};
padding: 10px 14px;
border-radius: 5px;
font-size: 0.875rem;
line-height: 1.5;
margin-top: 8px;
margin-bottom: 8px;
">
{msg}
</div>
""",
unsafe_allow_html=True,
)
def _render_table_and_download(product: str, unit: str, forecast: pd.DataFrame, start, end) -> None:
tabel = forecast[["ds", "yhat", "yhat_lower", "yhat_upper"]].copy()
tabel.columns = ["Tanggal", f"Peramalan ({unit})", "Estimasi Bawah Peramalan", "Estimasi Atas Peramalan"]
tabel["Tanggal"] = tabel["Tanggal"].dt.strftime("%d-%m-%Y")
for col in tabel.columns[1:]:
tabel[col] = tabel[col].clip(lower=0).apply(lambda x: f"{x:.1f}".replace(".", ","))
st.markdown(
"""
<style>
.table-title {
font-size: 1rem;
font-weight: 600;
margin-bottom: 12px !important;
color: #1e293b;
}
.table-caption {
font-size: 0.85rem;
color: #64748b;
margin-bottom: 17px !important;
line-height: 1.4;
}
[data-testid="stElementToolbar"] button svg {
transform: scale(1.3) !important;
}
[data-testid="stElementToolbar"] button {
padding: 6px 10px !important;
}
[data-testid="stElementToolbar"] {
padding: 4px 6px !important;
gap: 4px !important;
}
</style>
<div class="table-title">Tabel hasil peramalan</div>
<div class="table-caption">
💡 Gunakan angka <b>peramalan</b> sebagai acuan utama dalam perencanaan jumlah stok.
Siapkan stok cadangan untuk mengantisipasi kemungkinan permintaan naik hingga <b>estimasi atas peramalan.</b>
</div>
""",
unsafe_allow_html=True,
)
st.dataframe(tabel, width="stretch", hide_index=True)
def _render_tab_peramalan(product_entry, product, unit, forecast, summary, start, end):
st.plotly_chart(
_main_chart(product_entry, forecast, start, end, unit),
width="stretch", config={"displayModeBar": False},
)
_render_highlight_box(product, unit, forecast, summary, product_entry, start, end)
st.divider()
_render_table_and_download(product, unit, forecast, start, end)
# ---------------------------------------------------------------------------
# Tab 2: Dekomposisi & Rekomendasi Bisnis
# ---------------------------------------------------------------------------
def _weekly_pattern(weekly_df: pd.DataFrame) -> pd.DataFrame:
df = weekly_df.copy()
df["hari"] = df["ds"].dt.day_name().map({
"Monday": "Senin", "Tuesday": "Selasa", "Wednesday": "Rabu",
"Thursday": "Kamis", "Friday": "Jumat", "Saturday": "Sabtu", "Sunday": "Minggu",
})
ringkas = df.groupby("hari")["weekly"].mean().reindex(HARI_URUTAN)
return ringkas
def _mini_bar(x, y, color="#2563eb", is_weekly: bool = False) -> go.Figure:
y_formatted = [f"{val:,.5f}".replace(",", "X").replace(".", ",").replace("X", ".") for val in y]
fig = go.Figure(go.Bar(
x=x, y=y, marker_color=color,
customdata=y_formatted,
hovertemplate="%{x}: %{customdata}<extra></extra>" if is_weekly else "%{x|%d-%m-%Y}: %{customdata}<extra></extra>"
))
fig.update_layout(
height=160, margin=dict(l=10, r=10, t=10, b=10),
plot_bgcolor="rgba(0,0,0,0)",
)
return fig
def _mini_line(x, y, color="#2563eb", hover_labels: list[str] | None = None) -> go.Figure:
y_formatted = [f"{val:,.5f}".replace(",", "X").replace(".", ",").replace("X", ".") for val in y]
fig = go.Figure()
if hover_labels:
customdata_combined = list(zip(hover_labels, y_formatted))
fig.add_trace(go.Scatter(
x=x, y=y,
mode="lines",
line=dict(color=color, width=2),
marker=dict(size=4),
customdata=customdata_combined,
hovertemplate="%{customdata[0]}<br>%{x|%d-%m-%Y}: %{customdata[1]}<extra></extra>"
))
else:
fig.add_trace(go.Scatter(
x=x, y=y,
mode="lines",
line=dict(color=color, width=2),
customdata=y_formatted,
hovertemplate="%{x|%d-%m-%Y}: %{customdata}<extra></extra>"
))
fig.update_layout(
height=160, margin=dict(l=10, r=10, t=10, b=10),
plot_bgcolor="rgba(0,0,0,0)",
xaxis=dict(
hoverformat="%d-%m-%Y",
tickformat="%d-%m-%Y"
)
)
return fig
def _mini_multiline(x, series_dict: dict[str, list], colors: list[str], label_dict: dict[str, list] | None = None) -> go.Figure:
"""Chart mini dengan 1 garis TERPISAH per event (bukan dijumlah jadi 1
garis total) -- supaya visual chart konsisten dengan insight teks yang
juga menganalisis tiap event secara terpisah (lihat holiday_insight_bullets).
`label_dict`: opsional, {nama_event: [teks baris pertama hover per
titik]} -- format hover disamakan dengan chart lain di tab Peramalan:
tidak ada teks bold, baris pertama "Nama (H-X)" atau "Tidak ada event"
polos (tanpa nama event) kalau titik itu di luar window efek."""
fig = go.Figure()
for i, (nama, y) in enumerate(series_dict.items()):
y_formatted = [f"{val:,.5f}".replace(",", "X").replace(".", ",").replace("X", ".") for val in y]
color = colors[i % len(colors)]
labels = label_dict.get(nama) if label_dict else [nama] * len(y)
customdata = list(zip(labels, y_formatted))
fig.add_trace(go.Scatter(
x=x, y=y,
mode="lines",
name=nama,
line=dict(color=color, width=2),
customdata=customdata,
hovertemplate="%{customdata[0]}<br>%{x|%d-%m-%Y}: %{customdata[1]}<extra></extra>",
))
fig.update_layout(
height=200, margin=dict(l=10, r=10, t=10, b=10),
plot_bgcolor="rgba(0,0,0,0)",
xaxis=dict(hoverformat="%d-%m-%Y", tickformat="%d-%m-%Y"),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left", x=0),
)
return fig
def _render_tab_dekomposisi(
forecast: pd.DataFrame, is_log: bool, scenario: str, product_entry: dict, product: str,
start: datetime.date, end: datetime.date,
) -> None:
scenario_name = product_entry["optimal_scenario"]
model = product_entry["scenarios"][scenario_name]["best_model"]
seasonality_mode = getattr(model, "seasonality_mode", "additive")
is_relative = is_log or (seasonality_mode == "multiplicative") or ("multiplicative" in scenario.lower())
if is_relative:
penjelasan_pola = (
"merupakan <b>bobot penyesuaian relatif (persentase)</b> terhadap tren. "
"Contoh: angka <b>-0.08</b> berarti penjualan pada hari tersebut cenderung <b>turun ~8%</b> dari tren utama."
)
else:
penjelasan_pola = (
"merupakan <b>penyesuaian jumlah unit barang</b> terhadap tren. "
"Contoh: angka <b>-0.08</b> berarti penjualan pada hari tersebut berkurang sekitar <b>0.08 unit</b> dari tren utama."
)
st.markdown(
f"""
<div style='font-size: 0.85rem; color: #64748b; margin-bottom: 0.8rem; line-height: 1.5;'>
<div style='font-weight: 600; color: #475569; margin-bottom: 5px;'>💡 Cara Membaca Grafik Dekomposisi:</div>
<div style='margin-bottom: 4px;'>• <b>Tren dasar penjualan</b> merupakan angka perkiraan utama penjualan barang Anda.</div>
<div>• <b>Pola mingguan, tahunan, dan efek event</b> {penjelasan_pola}</div>
</div>
""",
unsafe_allow_html=True,
)
if is_log:
st.caption(
"\n ⚠️ Skenario optimal produk ini memakai **transformasi logaritma**. Grafik di Tab Peramalan "
"sudah dalam satuan asli, tapi grafik di bawah ini masih dalam "
"**skala log** -- dimana polanya (naik/turun/musiman) tetap valid dibaca, hanya "
"**angkanya tidak langsung sepadan** dengan satuan produk aslinya"
)
decomposition = get_decomposition(forecast)
# Styling CSS untuk teks insight di bawah grafik
style_insight = (
"font-size: 0.8rem; color: #475569; background-color: #f8fafc; "
"padding: 8px 12px; border-radius: 6px; border-left: 3px solid #3b82f6; "
"margin-top: -5px; margin-bottom: 1.2rem; line-height: 1.4;"
)
c1, c2 = st.columns(2)
c3, c4 = st.columns(2)
# --- 1. TREN ---
with c1:
st.markdown("###### Tren Dasar Penjualan")
if "trend" in decomposition:
st.plotly_chart(
_mini_line(decomposition["trend"]["ds"], decomposition["trend"]["trend"]),
width="stretch", config={"displayModeBar": False}, key="dekomp_trend",
)
# Insight dipusatkan lewat utils/report_insights.py (BUKAN dihitung
# ulang di sini) supaya UI & PDF selalu konsisten -- lihat catatan
# yang sama di bagian "Efek Event Lokal / Hari Libur" di bawah.
txt_trend = ri.trend_insight_text(decomposition)
st.markdown(f"<div style='{style_insight}'>💡 {txt_trend}</div>", unsafe_allow_html=True)
else:
st.caption("Tidak tersedia untuk skenario ini.")
# --- 2. MINGGUAN ---
with c2:
st.markdown("###### Pola Musiman Mingguan")
if "weekly" in decomposition:
pola = _weekly_pattern(decomposition["weekly"])
st.plotly_chart(
_mini_bar(pola.index, pola.values, is_weekly=True),
width="stretch", config={"displayModeBar": False}, key="dekomp_weekly",
)
# Insight dipusatkan lewat utils/report_insights.py -- lihat
# catatan yang sama di bagian "Efek Event Lokal / Hari Libur".
txt_weekly = ri.weekly_insight_text(pola)
st.markdown(f"<div style='{style_insight}'>💡 {txt_weekly}</div>", unsafe_allow_html=True)
else:
st.caption("Tidak tersedia untuk skenario ini.")
# --- 3. TAHUNAN ---
with c3:
st.markdown("###### Pola Musiman Tahunan")
if "yearly" in decomposition:
st.plotly_chart(
_mini_line(decomposition["yearly"]["ds"], decomposition["yearly"]["yearly"], color="#7c3aed"),
width="stretch", config={"displayModeBar": False}, key="dekomp_yearly",
)
# Insight dipusatkan lewat utils/report_insights.py -- lihat
# catatan yang sama di bagian "Efek Event Lokal / Hari Libur".
# PENTING: sebelumnya bagian ini punya logika duplikat (dihitung
# ulang inline di sini, terpisah dari yearly_insight_text() di
# report_insights.py) -- akibatnya perbaikan/fix yang diterapkan
# ke yearly_insight_text() (mis. deteksi arah pakai regresi
# linear, bukan cuma titik awal-akhir) TIDAK ikut kepakai di
# sini. Sekarang dipanggil langsung dari fungsi terpusatnya,
# supaya fix apa pun ke depannya otomatis konsisten di UI & PDF.
txt_yearly = ri.yearly_insight_text(decomposition["yearly"]["yearly"].values)
st.markdown(f"<div style='{style_insight}'>💡 {txt_yearly}</div>", unsafe_allow_html=True)
else:
st.caption("Tidak tersedia untuk skenario ini.")
# --- 4. EVENT LOKAL ---
# with c4:
# st.markdown("###### Efek Event Lokal / Hari Libur")
# if "holidays" in decomposition:
# hol = decomposition["holidays"]
# hol_cols = [c for c in hol.columns if c != "ds"]
# total_holiday = hol[hol_cols].sum(axis=1)
# scenario_name = product_entry["optimal_scenario"]
# model = product_entry["scenarios"][scenario_name]["best_model"]
# holidays_df = getattr(model, "holidays", None)
# hover_labels = []
# for _, row in hol.iterrows():
# curr_date = pd.to_datetime(row["ds"]).date()
# matched_events = []
# for col in hol_cols:
# if abs(row[col]) > 1e-6:
# event_name_title = col.replace("_", " ").title()
# status_str = ""
# if holidays_df is not None and not holidays_df.empty:
# ev_rows = holidays_df[holidays_df["holiday"] == col]
# for _, h_row in ev_rows.iterrows():
# ev_date = pd.to_datetime(h_row["ds"]).date()
# lower_win = int(h_row["lower_window"]) if "lower_window" in h_row and not pd.isna(h_row["lower_window"]) else 0
# upper_win = int(h_row["upper_window"]) if "upper_window" in h_row and not pd.isna(h_row["upper_window"]) else 0
# if ev_date + datetime.timedelta(days=lower_win) <= curr_date <= ev_date + datetime.timedelta(days=upper_win):
# selisih = (curr_date - ev_date).days
# if selisih < 0:
# status_str = f"H{selisih} "
# elif selisih == 0:
# status_str = "Hari H "
# else:
# status_str = f"H+{selisih} "
# break
# matched_events.append(f"{status_str}{event_name_title}".strip())
# if matched_events:
# hover_labels.append(", ".join(matched_events))
# else:
# hover_labels.append("Tidak ada event")
# st.plotly_chart(
# _mini_line(hol["ds"], total_holiday, color="#f59e0b", hover_labels=hover_labels),
# width="stretch", config={"displayModeBar": False}, key="dekomp_holidays",
# )
# # Insight Event -- lead time restock dihitung dari lower_window ASLI event
# # yang aktif di hari dengan efek terkuat (bukan angka karangan).
# if (total_holiday.abs() > 1e-6).any():
# max_idx = total_holiday.abs().idxmax()
# max_row_date = hol.loc[max_idx, "ds"].strftime("%d-%m-%Y")
# active_cols = [col for col in hol_cols if abs(hol.loc[max_idx, col]) > 1e-6]
# active_events = [c.replace("_", " ").title() for c in active_cols]
# ev_str = ", ".join(active_events) if active_events else "event lokal"
# # Cari lower_window terpanjang (paling negatif) dari event-event yang aktif
# # di hari puncak ini, supaya saran lead-time restock sesuai data model.
# lead_time_days = None
# if holidays_df is not None and not holidays_df.empty:
# for col in active_cols:
# ev_rows = holidays_df[holidays_df["holiday"] == col]
# for _, h_row in ev_rows.iterrows():
# lw = int(h_row["lower_window"]) if "lower_window" in h_row and not pd.isna(h_row["lower_window"]) else 0
# if lead_time_days is None or lw < lead_time_days:
# lead_time_days = lw
# if lead_time_days is not None and lead_time_days < 0:
# saran_lead_time = f" Siapkan stok tambahan mulai sekitar <b>{abs(lead_time_days)} hari sebelum</b> tanggal tersebut."
# else:
# saran_lead_time = " Siapkan stok tambahan menjelang tanggal tersebut."
# txt_event = (
# f"Terdeteksi puncak pengaruh event <b>{ev_str}</b> terjadi sekitar tanggal <b>{max_row_date}</b>."
# f"{saran_lead_time}"
# )
# # Kalau ada event LAIN (berbeda nama) di hari lain dalam periode yang sama,
# # sebutkan supaya tidak ada event penting yang terlewat.
# semua_event_periode = set()
# for _, row in hol.iterrows():
# for col in hol_cols:
# if abs(row[col]) > 1e-6:
# semua_event_periode.add(col)
# event_lain = semua_event_periode - set(active_cols)
# if event_lain:
# nama_lain = ", ".join(sorted(c.replace("_", " ").title() for c in event_lain))
# txt_event += f" <br>Ada juga efek event lain di periode ini: <b>{nama_lain}</b>."
# else:
# txt_event = "Operasional berjalan normal tanpa lonjakan akibat hari libur/event pada periode tanggal ini."
# st.markdown(f"<div style='{style_insight}'>💡 {txt_event}</div>", unsafe_allow_html=True)
# else:
# st.caption("Tidak tersedia untuk skenario ini.")
with c4:
st.markdown("###### Efek Event Lokal / Hari Libur")
if "holidays" in decomposition:
hol = decomposition["holidays"]
hol_cols = [c for c in hol.columns if c != "ds"]
scenario_name = product_entry["optimal_scenario"]
model = product_entry["scenarios"][scenario_name]["best_model"]
holidays_df = getattr(model, "holidays", None)
# Chart digambar SEBAGAI GARIS TERPISAH per event (bukan
# dijumlah jadi 1 garis total) -- supaya visual chart SELALU
# konsisten dengan insight teks di bawahnya, yang juga
# menganalisis tiap event secara terpisah/terisolasi lewat
# holiday_insight_bullets(). Sebelumnya chart menjumlahkan
# semua kolom event jadi 1 garis ("total_holiday"), yang bisa
# menyesatkan kalau 2 event bertumpang-tindih jadwalnya (mis.
# window Ramadhan berakhir persis di tanggal window Idul Fitri
# mulai) -- garis gabungan jadi tinggi di titik itu walau
# kolom event yang sedang dibahas di teks nilainya rendah.
EVENT_COLORS = ["#f59e0b", "#2563eb", "#16a34a", "#dc2626", "#7c3aed", "#0891b2", "#db2777"]
active_cols = [c for c in hol_cols if hol[c].abs().max() > 1e-6]
series_dict = {
col.replace("_", " ").title(): hol[col].tolist()
for col in active_cols
}
# Label baris pertama hover per titik, per event -- dihitung
# terhadap Hari H event ITU SENDIRI (bukan dicampur dengan
# event lain seperti versi lama), format disamakan dengan
# chart lain di tab Peramalan: "Nama (H-X)" kalau di dalam
# window efek, atau "Tidak ada event" POLOS (tanpa nama event)
# kalau titik itu di luar window efeknya.
label_dict: dict[str, list[str]] = {}
for col in active_cols:
nama = col.replace("_", " ").title()
ev_rows = holidays_df[holidays_df["holiday"] == col] if (holidays_df is not None and not holidays_df.empty) else None
labels = []
for curr_date in hol["ds"]:
curr_date = pd.to_datetime(curr_date).date()
status = None
if ev_rows is not None:
for _, h_row in ev_rows.iterrows():
ev_date = pd.to_datetime(h_row["ds"]).date()
lower_win = int(h_row["lower_window"]) if "lower_window" in h_row and not pd.isna(h_row["lower_window"]) else 0
upper_win = int(h_row["upper_window"]) if "upper_window" in h_row and not pd.isna(h_row["upper_window"]) else 0
if ev_date + datetime.timedelta(days=lower_win) <= curr_date <= ev_date + datetime.timedelta(days=upper_win):
selisih = (curr_date - ev_date).days
status = "Hari H" if selisih == 0 else (f"H{selisih}" if selisih < 0 else f"H+{selisih}")
break
labels.append(f"{nama} ({status})" if status else "Tidak ada event")
label_dict[nama] = labels
if series_dict:
st.plotly_chart(
_mini_multiline(hol["ds"], series_dict, EVENT_COLORS, label_dict),
width="stretch", config={"displayModeBar": False}, key="dekomp_holidays",
)
else:
# Seluruh periode yang dipilih user TIDAK ADA event sama
# sekali -- tetap tampilkan grafiknya (bukan cuma teks
# caption), sebagai garis datar di angka 0, warna hijau
# (menggambarkan kondisi "tenang"/tidak ada efek event).
flat_zero = [0.0] * len(hol)
st.plotly_chart(
_mini_multiline(
hol["ds"],
{"Tidak ada event": flat_zero},
["#16a34a"],
{"Tidak ada event": ["Tidak ada event"] * len(hol)},
),
width="stretch", config={"displayModeBar": False}, key="dekomp_holidays",
)
# Insight Event -- dipusatkan lewat utils/report_insights.py supaya
# UI & PDF SELALU konsisten. Dianalisis PER EVENT (bukan sinyal
# gabungan semua event dijumlah), jadi event yang tumpang tindih
# tanggalnya (mis. Ramadhan & Idulfitri) tetap dinarasikan
# terpisah -- dan tiap event dipisah fase naik vs turunnya.
event_bullets = ri.holiday_insight_bullets(hol, hol_cols, holidays_df)
if len(event_bullets) == 1:
st.markdown(f"<div style='{style_insight}'>💡 {event_bullets[0]}</div>", unsafe_allow_html=True)
else:
items_html = "".join(f"<li style='margin-bottom:6px;'>{b}</li>" for b in event_bullets)
st.markdown(
f"<div style='{style_insight}'>💡 "
f"<ul style='margin:4px 0 0 18px; padding:0;'>{items_html}</ul></div>",
unsafe_allow_html=True,
)
else:
st.caption("Tidak tersedia untuk skenario ini.")
# ---------------------------------------------------------------------------
# Unduh Laporan PDF
# ---------------------------------------------------------------------------
def _render_pdf_export(product_entry, product, unit, forecast, summary, start, end, scenario, is_log) -> None:
"""
Tombol unduh langsung untuk laporan PDF (ringkasan + grafik + tabel +
dekomposisi) untuk produk & periode yang sedang dipilih.
PDF-nya di-cache di session_state per kombinasi (produk, periode) --
baru disusun ulang kalau produk/periodenya beda dari yang terakhir
di-cache, supaya interaksi lain di halaman (ganti tab, dst.) tidak
memicu pembuatan PDF berulang-ulang.
"""
cache_key = f"pdf_report::{product}::{start}::{end}"
slot = st.empty()
if cache_key not in st.session_state:
# Tampilkan tombol status "Loading..." menggantikan posisi tombol unduh
slot.button(
"⏳ Menyusun PDF...",
disabled=True,
width="stretch",
key="btn_unduh_pdf_loading",
)
try:
pdf_bytes = pdf_report.build_report_pdf(
product_entry, product, unit, forecast, summary, start, end, scenario, is_log,
)
st.session_state[cache_key] = pdf_bytes
except Exception as e: # noqa: BLE001
slot.error(f"Gagal membuat laporan PDF: {e}")
return
# Setelah proses penyusunan selesai / dari cache, ganti slot dengan tombol unduh
file_name = f"laporan_peramalan_{product}_{start}_{end}.pdf"
slot.download_button(
"📄 Unduh Laporan",
data=st.session_state[cache_key],
file_name=file_name,
mime="application/pdf",
width="stretch",
key="btn_unduh_pdf_report",
)
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def render() -> None:
ui_components.inject_calendar_id_script()
all_results = st.session_state.get("all_results", {})
if not all_results:
st.error("Data model belum dimuat.")
return
products = get_product_list(all_results)
product = st.session_state.get("selected_product")
if product is None:
st.title("Detail Peramalan")
_render_form_gate(all_results, products)
return
start, end = st.session_state.get("selected_period", (datetime.date.today(), datetime.date.today()))
st.title("Detail Peramalan")
product, start, end = _render_inline_controls(all_results, products, product, start, end)
try:
forecast, scenario, is_log = _cached_forecast(all_results, product, str(start), str(end))
except Exception as e: # noqa: BLE001
st.error(f"Gagal membuat peramalan untuk **{product.title()}**: {e}")
return
unit = PRODUCT_UNITS.get(product, "unit")
summary = summarize_forecast(forecast)
st.markdown(f"### {product.title()}")
product_entry = all_results[product]
tab1, tab2 = st.tabs(["📈 Peramalan", "🧩 Dekomposisi"])
product_entry = all_results[product]
with tab1:
_render_tab_peramalan(product_entry, product, unit, forecast, summary, start, end)
with tab2:
_render_tab_dekomposisi(forecast, is_log, scenario, product_entry, product, start, end)