building_demand_forecasting_model.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyPnbkNaGEbocvWXS9nLVSUk",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/Parissai/ML-DS-Playground/blob/main/building_demand_forecasting_model.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Building a Demand Forecasting Model\n",
        "\n",
        "| Column     | Description              |\n",
        "|------------|--------------------------|\n",
        "| `'InvoiceNo'` | A 6-digit number uniquely assigned to each transaction |\n",
        "| `'StockCode'` | A 5-digit number uniquely assigned to each distinct product |\n",
        "| `'Description'` | The product name |\n",
        "| `'Quantity'` | The quantity of each product (item) per transaction |\n",
        "| `'UnitPrice'` | Product price per unit |\n",
        "| `'CustomerID'` | A 5-digit number uniquely assigned to each customer |\n",
        "| `'Country'` | The name of the country where each customer resides |\n",
        "| `'InvoiceDate'` | The day and time when each transaction was generated `\"MM/DD/YYYY\"` |\n",
        "| `'Year'` | The year when each transaction was generated |\n",
        "| `'Month'` | The month when each transaction was generated |\n",
        "| `'Week'` | The week when each transaction was generated (`1`-`52`) |\n",
        "| `'Day'` | The day of the month when each transaction was generated (`1`-`31`) |\n",
        "| `'DayOfWeek'` | The day of the weeke when each transaction was generated <br>(`0` = Monday, `6` = Sunday) |"
      ],
      "metadata": {
        "id": "Lz2p_D6kpqyI"
      }
    },
    {
      "cell_type": "code",
      "execution_count": 28,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "FlC4lxO5pl_X",
        "outputId": "4451116e-19ae-4c0f-d136-55982a1c8391"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Drive already mounted at /content/drive; to attempt to forcibly remount, call drive.mount(\"/content/drive\", force_remount=True).\n"
          ]
        }
      ],
      "source": [
        "from google.colab import drive\n",
        "drive.mount('/content/drive')"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Import required libraries\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.ml.feature import VectorAssembler\n",
        "from pyspark.ml import Pipeline\n",
        "from pyspark.ml.regression import RandomForestRegressor\n",
        "from pyspark.sql.functions import col, dayofmonth, month, year,  to_date, to_timestamp, weekofyear, dayofweek\n",
        "from pyspark.ml.feature import StringIndexer\n",
        "from pyspark.ml.evaluation import RegressionEvaluator\n",
        "\n",
        "# Initialize Spark session\n",
        "my_spark = SparkSession.builder.appName(\"SalesForecast\").getOrCreate()\n",
        "\n",
        "# Importing sales data\n",
        "sales_data = my_spark.read.csv(\n",
        "    \"/content/drive/MyDrive/data/online_retail.csv\", header=True, inferSchema=True, sep=\",\")\n",
        "\n",
        "# Convert InvoiceDate to datetime\n",
        "sales_data = sales_data.withColumn(\"InvoiceDate\", to_date(\n",
        "    to_timestamp(col(\"InvoiceDate\"), \"d/M/yyyy H:mm\")))"
      ],
      "metadata": {
        "id": "7UmXNmoJp2Ym"
      },
      "execution_count": 42,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Aggregate data into daily intervals\n",
        "daily_sales_data = sales_data.groupBy(\"Country\", \"StockCode\", \"InvoiceDate\", \"Year\", \"Month\", \"Day\", \"Week\", \"DayOfWeek\").agg({\"Quantity\": \"sum\",                                                                                                           \"UnitPrice\": \"avg\"})\n",
        "# Rename the target column\n",
        "daily_sales_data = daily_sales_data.withColumnRenamed(\n",
        "    \"sum(Quantity)\", \"Quantity\")"
      ],
      "metadata": {
        "id": "XO0osmWNp4zO"
      },
      "execution_count": 43,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Split the data into two sets based on the spliting date, \"2011-09-25\". All data up to and including this date should be in the training set, while data after this date should be in the testing set. Return a pandas Dataframe, pd_daily_train_data, containing, at least, the columns [\"Country\", \"StockCode\", \"InvoiceDate\", \"Quantity\"].\n",
        "split_date_train_test = \"2011-09-25\"\n",
        "\n",
        "# Creating the train and test datasets\n",
        "train_data = daily_sales_data.filter(\n",
        "    col(\"InvoiceDate\") <= split_date_train_test)\n",
        "test_data = daily_sales_data.filter(col(\"InvoiceDate\") > split_date_train_test)\n",
        "\n",
        "pd_daily_train_data = train_data.toPandas()"
      ],
      "metadata": {
        "id": "H1gw8_xEp5fO"
      },
      "execution_count": 44,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Creating indexer for categorical columns\n",
        "country_indexer = StringIndexer(\n",
        "    inputCol=\"Country\", outputCol=\"CountryIndex\").setHandleInvalid(\"keep\")\n",
        "stock_code_indexer = StringIndexer(\n",
        "    inputCol=\"StockCode\", outputCol=\"StockCodeIndex\").setHandleInvalid(\"keep\")"
      ],
      "metadata": {
        "id": "Vwl08NAQp8Fy"
      },
      "execution_count": 45,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Selectiong features columns\n",
        "feature_cols = [\"CountryIndex\", \"StockCodeIndex\", \"Month\", \"Year\",\n",
        "                \"DayOfWeek\", \"Day\", \"Week\"]\n",
        "\n",
        "# Using vector assembler to combine features\n",
        "assembler = VectorAssembler(inputCols=feature_cols, outputCol=\"features\")"
      ],
      "metadata": {
        "id": "1DwNle_Fp-ZS"
      },
      "execution_count": 33,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Initializing a Random Forest model\n",
        "rf = RandomForestRegressor(\n",
        "    featuresCol=\"features\",\n",
        "    labelCol=\"Quantity\",\n",
        "    maxBins=4000\n",
        ")\n",
        "\n",
        "# Create a pipeline for staging the processes\n",
        "pipeline = Pipeline(stages=[country_indexer, stock_code_indexer, assembler, rf])\n",
        "\n",
        "# Training the model\n",
        "model = pipeline.fit(train_data)\n",
        "\n",
        "# Getting test predictions\n",
        "test_predictions = model.transform(test_data)\n",
        "test_predictions = test_predictions.withColumn(\n",
        "    \"prediction\", col(\"prediction\").cast(\"double\"))"
      ],
      "metadata": {
        "id": "pYSlWKvUqApC"
      },
      "execution_count": 46,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Provide the Mean Absolute Error (MAE)\n",
        "\n",
        "# Initializing the evaluator\n",
        "mae_evaluator = RegressionEvaluator(\n",
        "    labelCol=\"Quantity\", predictionCol=\"prediction\", metricName=\"mae\")\n",
        "\n",
        "# Obtaining MAE\n",
        "mae = mae_evaluator.evaluate(test_predictions)\n",
        "print(mae)"
      ],
      "metadata": {
        "id": "6U956JUPqE_2",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "b4f646a8-3072-4349-f6b1-5be6678f00c2"
      },
      "execution_count": 35,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "9.731486041283297\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# How many units will be sold during the  week 39 of 2011?\n",
        "\n",
        "# Getting the weekly sales of all countries\n",
        "weekly_test_predictions = test_predictions.groupBy(\"Year\", \"Week\").agg({\"prediction\": \"sum\"})\n",
        "\n",
        "# Finding the quantity sold on the 39 week.\n",
        "promotion_week = weekly_test_predictions.filter(col('Week')==39)\n",
        "\n",
        "# Storing prediction as quantity_sold_w30\n",
        "quantity_sold_w39 = int(promotion_week.select(\"sum(prediction)\").collect()[0][0])\n",
        "print(quantity_sold_w39)\n",
        "\n",
        "# Stop the Spark session\n",
        "my_spark.stop()"
      ],
      "metadata": {
        "id": "kkyLLLUvqFy-",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "2966495d-4cb1-4f94-b944-658d3f95674a"
      },
      "execution_count": 47,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "89480\n"
          ]
        }
      ]
    }
  ]
}