Um die oben genannten Konzepte zu veranschaulichen, wird in diesem Abschnitt das Beispielproblem einer 7-Tage-Umsatzprognose für bestimmte Artikel/SKUs auf der Grundlage ihrer bisherigen Verkaufshistorie vorgestellt.
Daten
Die im Rahmen des Projekts verwendeten Daten stammen aus einem Hackathon, der von einem großen nordamerikanischen Einzelhändler für Georgia Tech MS Analytics-Studenten organisiert wurde. Die besagten Daten können mit Hilfe von DVC erfasst werden Hier. Die Daten umfassen den täglichen Verkaufsverlauf für eine Reihe von SKUs. Das Ziel hier besteht darin, diese Historie zu nutzen, um sie an ein HMM-Modell anzupassen und damit Umsätze für eine Woche im Voraus zu prognostizieren.
*Einen schnellen Einstieg in DVC finden Sie hier toller TDS-Artikel!
HMM passend
Um das HMM-Modell anzupassen, benötigen wir eine Folge von Y-Beobachtungen. In diesem Fall erstellen wir die Y-Beobachtungen am Beispiel verzögerter Tagesverkäufe und der entsprechenden Differenz zwischen verzögerten Verkäufen und heutigen Verkäufen.
Indem wir die verzögerten Variablen zum Trainieren des HMM verwenden, stellen wir sicher, dass es zwischen dem Trainingszeitraum und dem Zeitraum außerhalb des Zeitraums kein Datenleck gibt.
"""
# Created by ashish1610dhiman at 23/12/22
Contact at ashish1610dhiman@gmail.com
"""import pandas as pd
from hmmlearn.hmm import GaussianHMM
import itertools
import numpy as np
class sku_predict():
#initialise
def __init__(self,train_test, sku_id):
self.sku_id = sku_id
train_test_sku = train_test[train_test.Encoded_SKU_ID == sku_id]
train_test_sku.index = train_test_sku["SALES_DATE"]
train_test_sku = train_test_sku.sort_index()
self.train_test_sku = train_test_sku
self.sales_data = None
self.n_lags = None
self.X = None
self.mannequin = None
def get_features(self, n_lags = 3):
self.n_lags = n_lags
sales_data = self.train_test_sku[["DAILY_UNITS"]]
for lag in vary(1,n_lags+1):
du_lag = f"DAILY_UNITS_lag{lag}"
sales_data[du_lag] = sales_data["DAILY_UNITS"].shift(lag)
sales_data[f"change_lag{lag}"] = (sales_data["DAILY_UNITS"] - sales_data[du_lag])
print (f"Created {n_lags} lag options")
self.sales_data = sales_data #featurised knowledge
return sales_data
def split_train_test(self,start_dt):
train1 = self.sales_data[:pd.to_datetime(start_dt)+pd.DateOffset(-1)]
valid1 = self.sales_data[pd.to_datetime(start_dt):]
return (train1,valid1)
def fit_hmm(self,prepare, start_dt, n_components1 = 2):
hmm = GaussianHMM(n_components=n_components1)
# match hmm to pct_change and DAILY_UNITS_lag1
lag_price_cols = [f"DAILY_UNITS_lag{lag}" for lag in range(1,self.n_lags+1)]
lag_change_cols = [f"change_lag{lag}" for lag in range(1,self.n_lags+1)]
print ("Coaching on :",lag_price_cols,lag_change_cols)
X = prepare[lag_price_cols + lag_change_cols]
X = X[start_dt:]
self.X = X
hmm.match(X.dropna())
self.mannequin = hmm
Out-of-Time-Prognose
Schritt 1: Kandidatensatz generieren
Wie im obigen Abschnitt beschrieben, besteht der nächste Schritt für die Zeitvorhersage mit einem HMM-Modell darin, eine Suche im Raum möglicher beobachteter Variablen durchzuführen. Dies erfolgt durch die Erstellung eines Rasters möglicher Werte für die verzögerten Verkäufe und die Umsatzdifferenz. Anschließend nehmen wir ein Kreuzprodukt dieser Gitter, um zum Kandidatensatz zu gelangen.
def _compute_all_possible_outcomes(self,n_steps_pct,n_steps_price_lag):
# pct_change_range = np.linspace(-0.61, 0.61, n_steps_pct)
# change_range = np.linspace(self.X["change_lag1"].min(),self.X["change_lag1"].max(), n_steps_pct)
#TODO
DAILY_UNITS_range = np.distinctive(self.X.dropna()[[col for col in self.X.columns if "UNITS" in col]])
change_range = np.distinctive(self.X.dropna()[[col for col in self.X.columns if "change" in col]])
# DAILY_UNITS_range = np.linspace(self.X["DAILY_UNITS_lag1"].min(),
# self.X["DAILY_UNITS_lag1"].quantile(0.75), n_steps_price_lag)
change_list = [DAILY_UNITS_range]*self.n_lags + [change_range]*self.n_lags
all_outcomes = np.array(record(itertools.product(*change_list)))
return (all_outcomes)
Beachten Sie, dass wir die einzelnen Suchraster mit einem beliebigen Granularitätsgrad erstellen können. Für das obige Beispiel verwenden wir das Suchraster einfach als eindeutige Werte der Variablen dieser Variablen, vorausgesetzt, die Variable weist eine sehr geringe Variabilität auf.
Schritt 2: Maximierung der A-Posteriori-Wahrscheinlichkeit
Jedes dieser Ergebnisse wird dann mit dem angepassten HMM bewertet, um die A-Posteriori-Wahrscheinlichkeit zu ermitteln, dass das Ergebnis das nächste in der Sequenz ist. Das Ergebnis, das diese Wahrscheinlichkeit maximiert, wird dann als prognostiziertes Ergebnis ausgewählt. Dieses prognostizierte Ergebnis wird dann zur Sequenz hinzugefügt und der Vorgang wird für das folgende Ergebnis wiederholt (dynamische Prognose).
def _get_most_probable_outcome(self,prev_data,n_steps_pct,n_steps_price_lag):
hmm = self.mannequin
all_outcomes = self._compute_all_possible_outcomes(n_steps_pct,n_steps_price_lag)
outcome_score = record(map(lambda x: hmm.rating(np.row_stack((prev_data, x))),
all_outcomes))
most_probable_outcome = all_outcomes[np.argmax(outcome_score)]
return (most_probable_outcome)
Schritt 3: Vorhersage
Nachdem das wahrscheinlichste Ergebnis ermittelt wurde, kann die Prognose wie folgt berechnet werden:
Nachdem wir die Prognose für den Zeitraum t+1 erstellt haben, fügen wir sie der Sequenz hinzu und wiederholen die obigen Schritte.
def predict(self,valid1, pred_latency, n_steps_pct = 100,n_steps_price_lag = 100, w1=0.55, w2 = 0.45):
prev_data_init = self.X[-pred_latency:]
predict_df = pd.DataFrame()
print ("Beginning Prediction")
for i, dt in enumerate(valid1.index):
if i == 0:
prev_data = prev_data_init
else:
# print(prev_data.form)
self.mannequin.match(prev_data[-pred_latency:])
print (f"-----> Predicting for day:{i}")
most_probable_outcome = self._get_most_probable_outcome(prev_data, 100, 100)
temp = pd.DataFrame(most_probable_outcome).T
temp.index = [dt]
temp.columns = prev_data_init.columns
prev_data = pd.concat([prev_data, temp])
predict_df = pd.concat([predict_df, temp])
# print(most_probable_outcome)
predict_df["predicted1"] = predict_df["DAILY_UNITS_lag1"] + (predict_df["change_lag1"])
predict_df["predicted2"] = predict_df["DAILY_UNITS_lag2"] + (predict_df["change_lag2"])
predict_df["predicted"] = (w1*predict_df["predicted1"] + w2*predict_df["predicted2"])
return predict_df
Ergebnisse
Im Beispielszenario lieferte das HMM-Modell einen erheblichen Mehrwert für SKUs, die in den letzten Monaten eine deutliche Umsatzverschiebung oder einen Regimewechsel aufweisen.
Wie wir in der obigen Grafik deutlich sehen können, konnte HMM in der letzten Woche einen Umsatzanstieg prognostizieren, auch wenn es im Vormonat keine Verkäufe gab. Es wurde festgestellt, dass andere konventionelle Modelle wie Holt Winters und ARIMA (SARIMA) sowie die STL-Zerlegung solche Szenarien schlecht vorhersagen.
Das HMM-Modell galt auch für SKUs, die eine hohe Volatilität in den Daten aufweisen.
