TIF_E41253317/utils/demand_insights.py

48 lines
1.6 KiB
Python

from __future__ import annotations
from typing import Any
import pandas as pd
def detect_peak_days(forecast: pd.DataFrame, threshold_std: float = 1.0) -> pd.DataFrame:
"""
Filter tanggal-tanggal di mana yhat > mean(yhat) + threshold_std * std(yhat).
Hasil diurutkan dari yhat tertinggi ke terendah.
Kalau rentang periode cuma 1 hari, std tidak bisa dihitung (NaN) --
dalam kasus itu, tidak ada yang ditandai sebagai lonjakan (butuh
pembanding dari beberapa hari).
"""
yhat = forecast["yhat"].clip(lower=0)
if len(forecast) < 2:
return forecast.iloc[0:0][["ds", "yhat"]].copy()
threshold = yhat.mean() + threshold_std * yhat.std()
peaks = forecast.loc[yhat > threshold, ["ds", "yhat"]].copy()
return peaks.sort_values("yhat", ascending=False).reset_index(drop=True)
def nearest_holiday(model: Any, on_or_after) -> dict | None:
"""
Cari hari libur (dari atribut model.holidays milik Prophet) yang jatuh
paling dekat pada/setelah tanggal `on_or_after`.
Returns
-------
dict {"nama": str, "tanggal": pd.Timestamp} atau None kalau model ini
tidak dikonfigurasi dengan holidays, atau tidak ada hari libur yang
jatuh pada/setelah tanggal tersebut di dalam kalender yang didaftarkan.
"""
holidays = getattr(model, "holidays", None)
if holidays is None or holidays.empty:
return None
cutoff = pd.Timestamp(on_or_after)
upcoming = holidays.loc[holidays["ds"] >= cutoff].sort_values("ds")
if upcoming.empty:
return None
row = upcoming.iloc[0]
return {"nama": row["holiday"], "tanggal": row["ds"]}