{ "cells": [ { "attachments": {}, "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "a02bb3a2-e5d7-4d14-b966-827457675b75", "showTitle": false, "title": "" } }, "source": [ "# Retail Demand Forecasting\n", "\n", "Retailers often deal with large datasets that exceed the memory capacity of a single machine. \n", "These datasets also exhibit diverse characteristics, particularly when considering the varying behaviors of different stores. \n", "Consequently, data scientists face several challenges, including:\n", "\n", "- Limitations of Pandas/Python in handling big datasets, necessitating the use of Pyspark.\n", "- The need to create separate models for different groups in the dataset (e.g., stores or items) to optimize performance.\n", "- Despite the dataset's overall size, each group (items/stores) is small enough to leverage models similar to those provided by ``scikit-learn``.\n", "\n", "This tutorial demonstrates how data scientists can overcome these challenges using ``ForecastFlowML``.\n", "\n", "## Situation\n", "\n", "Given the considerable size of our sample dataset, it exceeds the memory capacity for direct loading.\n", " Consequently, we must leverage PySpark, a distributed processing framework, to effectively manage the data processing operations.\n", "\n", "Fortunately, within this extensive dataset, we possess store-specific information.\n", "The individual store data, being relatively smaller in size, can comfortably fit within the memory of a single machine. \n", "Consequently, we can leverage this advantage to employ ``scikit-learn`` for model creation, without encountering any memory limitations.\n", "\n", "## Goal\n", "\n", "- Develop independent models for each store in the dataset.\n", "- Parallelize the feature engineering, training, and inference steps.\n", "- Employ ``LightGBM`` as the machine learning algorithm.\n", "- Utilize a direct multi-step forecasting approach.\n", "- Conduct backtesting to assess the model's performance." ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Import Packages" ] }, { "cell_type": "code", "execution_count": 38, "metadata": {}, "outputs": [], "source": [ "from forecastflowml import ForecastFlowML\n", "from forecastflowml import FeatureExtractor\n", "from forecastflowml.data.loader import load_walmart_m5\n", "from lightgbm import LGBMRegressor\n", "from pyspark.sql import SparkSession\n", "import pyspark.sql.functions as F\n", "import plotly.express as px\n", "import plotly.io as pio\n", "import pandas as pd\n", "import sys\n", "import os\n", "\n", "os.environ[\"PYSPARK_PYTHON\"] = sys.executable\n", "pd.set_option(\"display.max_columns\", 40)" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Initialize Spark" ] }, { "cell_type": "code", "execution_count": 39, "metadata": {}, "outputs": [], "source": [ "spark = (\n", " SparkSession.builder.master(\"local[4]\")\n", " .config(\"spark.driver.memory\", \"4g\")\n", " .config(\"spark.sql.shuffle.partitions\", \"4\")\n", " .config(\"spark.sql.execution.pyarrow.enabled\", \"true\")\n", " .getOrCreate()\n", ")" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Sample Dataset" ] }, { "cell_type": "code", "execution_count": 40, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------------------+-----------+-------+------+--------+--------+----------+-----+\n", "| id| item_id|dept_id|cat_id|store_id|state_id| date|sales|\n", "+--------------------+-----------+-------+------+--------+--------+----------+-----+\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-15| 3.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-16| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-17| 1.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-18| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-19| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-20| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-21| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-22| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-23| 0.0|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-24| 0.0|\n", "+--------------------+-----------+-------+------+--------+--------+----------+-----+\n", "only showing top 10 rows\n", "\n" ] } ], "source": [ "df = load_walmart_m5(spark).localCheckpoint()\n", "df.show(10)" ] }, { "cell_type": "code", "execution_count": 41, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------+\n", "|store_id|\n", "+--------+\n", "| TX_1|\n", "| WI_1|\n", "| CA_1|\n", "| TX_2|\n", "+--------+\n", "\n" ] } ], "source": [ "df.select(\"store_id\").dropDuplicates().show()" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Approach" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "There are 4 different stores in our dataset. Since each store might have different behaviours, \n", "it would be better to build an independent model for each of them. Moreover,\n", "there will be 4 models per store that is going to predict 7 days. \n", "In total, we are going to have 4 x 4 = 16 models.\n", "\n", "\n", "" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "As we are going to build 4 weekly independent direct forecasting models, each of them are going to use different lag features \n", "based upon their forecasting horizon.\n", "\n", "" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Feature Engineering\n", "\n", "Let's create the features that will be used for forecasting. \n", "Our package has a seperate module ``FeatureExtractor`` to extract features.\n", "Spesifically, we are going to extract:\n", "\n", "- Lags\n", "- Roling means\n", "- Date features\n", "- Stockout features (counting consecutive no sales periods)\n", "- History length (number of periods passed after the start of the time series)" ] }, { "cell_type": "code", "execution_count": 42, "metadata": {}, "outputs": [], "source": [ "feature_extractor = FeatureExtractor(\n", " id_col=\"id\",\n", " date_col=\"date\",\n", " target_col=\"sales\",\n", " lag_window_features={\n", " \"lag\": [7 * (i + 1) for i in range(7)],\n", " \"mean\": [[window, lag] for lag in [7, 14, 21, 28] for window in [7, 14, 30]],\n", " },\n", " date_features=[\n", " \"day_of_month\",\n", " \"day_of_week\",\n", " \"week_of_year\",\n", " \"quarter\",\n", " \"month\",\n", " \"year\",\n", " ],\n", " count_consecutive_values={\n", " \"value\": 0,\n", " \"lags\": [7, 14, 21, 28],\n", " },\n", " history_length=True,\n", ")" ] }, { "cell_type": "code", "execution_count": 43, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------------------+-----------+-------+------+--------+--------+----------+-----+-----+------+------+------+------+------+------+-------------------+--------------------+--------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+-----------------------------+------------------------------+------------------------------+------------------------------+--------------+------------+-----------+------------+-------+-----+----+\n", "| id| item_id|dept_id|cat_id|store_id|state_id| date|sales|lag_7|lag_14|lag_21|lag_28|lag_35|lag_42|lag_49|window_7_lag_7_mean|window_14_lag_7_mean|window_30_lag_7_mean|window_7_lag_14_mean|window_14_lag_14_mean|window_30_lag_14_mean|window_7_lag_21_mean|window_14_lag_21_mean|window_30_lag_21_mean|window_7_lag_28_mean|window_14_lag_28_mean|window_30_lag_28_mean|count_consecutive_value_lag_7|count_consecutive_value_lag_14|count_consecutive_value_lag_21|count_consecutive_value_lag_28|history_length|day_of_month|day_of_week|week_of_year|quarter|month|year|\n", "+--------------------+-----------+-------+------+--------+--------+----------+-----+-----+------+------+------+------+------+------+-------------------+--------------------+--------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+-----------------------------+------------------------------+------------------------------+------------------------------+--------------+------------+-----------+------------+-------+-----+----+\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-15| 3.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 1| 15| 4| 3| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-16| 0.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 2| 16| 5| 3| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-17| 1.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 3| 17| 6| 3| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-18| 0.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 4| 18| 7| 3| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-19| 0.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 5| 19| 1| 4| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-20| 0.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 6| 20| 2| 4| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-21| 0.0| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| null| 7| 21| 3| 4| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-22| 0.0| 3.0| null| null| null| null| null| null| 3.0| 3.0| 3.0| null| null| null| null| null| null| null| null| null| 0| null| null| null| 8| 22| 4| 4| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-23| 0.0| 0.0| null| null| null| null| null| null| 1.5| 1.5| 1.5| null| null| null| null| null| null| null| null| null| 1| null| null| null| 9| 23| 5| 4| 1| 1|2015|\n", "|FOODS_1_002_TX_1_...|FOODS_1_002|FOODS_1| FOODS| TX_1| TX|2015-01-24| 0.0| 1.0| null| null| null| null| null| null| 1.3333333333333333| 1.3333333333333333| 1.3333333333333333| null| null| null| null| null| null| null| null| null| 0| null| null| null| 10| 24| 6| 4| 1| 1|2015|\n", "+--------------------+-----------+-------+------+--------+--------+----------+-----+-----+------+------+------+------+------+------+-------------------+--------------------+--------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+--------------------+---------------------+---------------------+-----------------------------+------------------------------+------------------------------+------------------------------+--------------+------------+-----------+------------+-------+-----+----+\n", "only showing top 10 rows\n", "\n" ] } ], "source": [ "df_features = feature_extractor.transform(df).localCheckpoint()\n", "df_features.show(10)" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Training/Test Dataset" ] }, { "cell_type": "code", "execution_count": 44, "metadata": {}, "outputs": [], "source": [ "df_train = df_features.filter(F.col(\"date\") < \"2016-04-25\")\n", "df_test = df_features.filter(F.col(\"date\") >= \"2016-04-25\")" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Initialize ForecastFlowML" ] }, { "cell_type": "code", "execution_count": 45, "metadata": {}, "outputs": [], "source": [ "forecast_flow = ForecastFlowML(\n", " group_col=\"store_id\", # column to slice the dataframe\n", " id_col=\"id\", # time series identifier column\n", " date_col=\"date\", # date column\n", " target_col=\"sales\", # target column\n", " categorical_cols=[\n", " \"item_id\",\n", " \"dept_id\",\n", " \"cat_id\",\n", " ], # columns to be treated as categorical\n", " date_frequency=\"days\", # dataset date frequency\n", " model_horizon=7, # single model predicts 7 days\n", " max_forecast_horizon=28, # total forecast horizon will be 28 days\n", " model=LGBMRegressor(), # algorithm\n", " use_lag_range=21, # allow to use 21 days of extra lag features to use based upon the forecast horizon\n", ")" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "## Cross Validation\n", "\n", "First, let's perform some backtesting to understand the model performance. We are going to \n", "predict the last 3x28 days." ] }, { "cell_type": "code", "execution_count": 46, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+--------+--------------------+-------------------+---+-----+----------+\n", "|store_id| id| date| cv|sales|prediction|\n", "+--------+--------------------+-------------------+---+-----+----------+\n", "| CA_1|FOODS_1_064_CA_1_...|2016-03-28 00:00:00| 0| 2.0| 0.979212|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-03-29 00:00:00| 0| 0.0| 1.047827|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-03-30 00:00:00| 0| 0.0|0.92554504|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-03-31 00:00:00| 0| 0.0| 0.9589823|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-01 00:00:00| 0| 3.0| 1.305723|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-02 00:00:00| 0| 0.0| 1.5510178|\n", "| CA_1|FOODS_1_064_CA_1_...|2016-04-03 00:00:00| 0| 3.0| 1.4713657|\n", "| CA_1|FOODS_1_121_CA_1_...|2016-03-28 00:00:00| 0| 0.0| 0.560704|\n", "| CA_1|FOODS_1_121_CA_1_...|2016-03-29 00:00:00| 0| 0.0| 0.5866643|\n", "| CA_1|FOODS_1_121_CA_1_...|2016-03-30 00:00:00| 0| 0.0| 0.5995418|\n", "+--------+--------------------+-------------------+---+-----+----------+\n", "only showing top 10 rows\n", "\n" ] } ], "source": [ "cv_forecast = forecast_flow.cross_validate(df_train, n_cv_splits=3).localCheckpoint()\n", "cv_forecast.show(10)" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "### Calculate RMSE" ] }, { "cell_type": "code", "execution_count": 47, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+---+------------------+\n", "| cv| rmse|\n", "+---+------------------+\n", "| 0|1.9499041165007749|\n", "| 1|2.0947956593876733|\n", "| 2| 1.925051121492435|\n", "+---+------------------+\n", "\n" ] } ], "source": [ "(\n", " cv_forecast.withColumn(\n", " \"error_squared\", F.pow(F.abs(F.col(\"sales\") - F.col(\"prediction\")), 2)\n", " )\n", " .groupBy(\"cv\")\n", " .agg(F.pow(F.avg(\"error_squared\"), 0.5).alias(\"rmse\"))\n", " .orderBy(\"cv\")\n", " .show()\n", ")" ] }, { "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ "### Visualize Cross Validation Predictions" ] }, { "cell_type": "code", "execution_count": 48, "metadata": {}, "outputs": [], "source": [ "cv_past_future = (\n", " df_train.select(\"id\", \"store_id\", \"date\", \"sales\")\n", " .join(\n", " cv_forecast.select(\"id\", \"date\", \"cv\", \"prediction\"),\n", " on=[\"id\", \"date\"],\n", " how=\"left\",\n", " )\n", " .groupBy(\"id\", \"store_id\", \"date\", \"sales\")\n", " .pivot(\"cv\")\n", " .sum(\"prediction\")\n", " .groupBy(\"store_id\", \"date\")\n", " .agg(\n", " F.sum(\"sales\").alias(\"sales\"),\n", " *[F.sum(f\"{i}\").alias(f\"cv_{i}\") for i in range(3)],\n", " )\n", " .orderBy(\"store_id\", \"date\")\n", ").toPandas()" ] }, { "cell_type": "code", "execution_count": 49, "metadata": {}, "outputs": [ { "data": { "text/html": [ " \n", " " ] }, "metadata": {}, "output_type": "display_data" }, { "data": { "text/html": [ "
| \n", " | store_id | \n", "forecast_horizon | \n", "model | \n", "start_time | \n", "end_time | \n", "elapsed_seconds | \n", "
|---|---|---|---|---|---|---|
| 0 | \n", "CA_1 | \n", "[[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,... | \n", "[[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ... | \n", "19-May-2023 (23:41:17) | \n", "19-May-2023 (23:41:18) | \n", "0.6 | \n", "
| 1 | \n", "TX_1 | \n", "[[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,... | \n", "[[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ... | \n", "19-May-2023 (23:41:18) | \n", "19-May-2023 (23:41:19) | \n", "0.6 | \n", "
| 2 | \n", "WI_1 | \n", "[[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,... | \n", "[[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ... | \n", "19-May-2023 (23:41:19) | \n", "19-May-2023 (23:41:19) | \n", "0.5 | \n", "
| 3 | \n", "TX_2 | \n", "[[1, 2, 3, 4, 5, 6, 7], [8, 9, 10, 11, 12, 13,... | \n", "[[128, 4, 149, 236, 1, 0, 0, 0, 0, 0, 0, 140, ... | \n", "19-May-2023 (23:41:16) | \n", "19-May-2023 (23:41:17) | \n", "0.7 | \n", "