examples/rossmann/rossmann-store-sales-preproc.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:22.650773Z",
     "iopub.status.busy": "2020-12-09T23:19:22.650209Z",
     "iopub.status.idle": "2020-12-09T23:19:22.652025Z",
     "shell.execute_reply": "2020-12-09T23:19:22.652531Z"
    }
   },
   "outputs": [],
   "source": [
    "# Copyright 2020 NVIDIA Corporation. All Rights Reserved.\n",
    "#\n",
    "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "# you may not use this file except in compliance with the License.\n",
    "# You may obtain a copy of the License at\n",
    "#\n",
    "#     http://www.apache.org/licenses/LICENSE-2.0\n",
    "#\n",
    "# Unless required by applicable law or agreed to in writing, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License.\n",
    "# =============================================================================="
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"http://developer.download.nvidia.com/compute/machine-learning/frameworks/nvidia_logo.png\" style=\"width: 90px; float: right;\">\n",
    "\n",
    "# Preprocessing the Rossmann Store Sales Dataset\n",
    "Here we implement some feature engineering outlined by FastAI in [their example solution](https://github.com/fastai/course-v3/blob/master/nbs/dl1/lesson6-rossmann.ipynb) to the [Kaggle Rossmann Store Sales competition](https://www.kaggle.com/c/rossmann-store-sales) using Rapids cuDF library (GPU DataFrame library). We've simplified some sections and left out some of the documentation to keep things neat, so feel free to consult the original preprocessing [notebook](https://github.com/fastai/course-v3/blob/master/nbs/dl1/rossman_data_clean.ipynb) for explanations of the feature engineering going on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:22.655774Z",
     "iopub.status.busy": "2020-12-09T23:19:22.655247Z",
     "iopub.status.idle": "2020-12-09T23:19:24.293212Z",
     "shell.execute_reply": "2020-12-09T23:19:24.292751Z"
    }
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import pandas as pd\n",
    "import cudf\n",
    "import dask_cudf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:24.297086Z",
     "iopub.status.busy": "2020-12-09T23:19:24.296438Z",
     "iopub.status.idle": "2020-12-09T23:19:24.298245Z",
     "shell.execute_reply": "2020-12-09T23:19:24.298635Z"
    }
   },
   "outputs": [],
   "source": [
    "INPUT_DATA_DIR = os.environ.get('INPUT_DATA_DIR', './rossmann')\n",
    "OUTPUT_DATA_DIR = os.environ.get('OUTPUT_DATA_DIR', './data')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The cell below will download the data into the `INPUT_DATA_DIR` if you do not have it already."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:24.302556Z",
     "iopub.status.busy": "2020-12-09T23:19:24.302069Z",
     "iopub.status.idle": "2020-12-09T23:19:24.303932Z",
     "shell.execute_reply": "2020-12-09T23:19:24.304337Z"
    }
   },
   "outputs": [],
   "source": [
    "if not os.path.isfile(os.path.join(INPUT_DATA_DIR, \"rossmann.tgz\")):\n",
    "    os.makedirs(INPUT_DATA_DIR, exist_ok=True)\n",
    "    os.system(f'wget -O {INPUT_DATA_DIR}/rossmann.tgz http://files.fast.ai/part2/lesson14/rossmann.tgz')\n",
    "    os.system(f'cd {INPUT_DATA_DIR} && tar -xzvf rossmann.tgz')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's read each csv file as a cudf dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:24.308996Z",
     "iopub.status.busy": "2020-12-09T23:19:24.308506Z",
     "iopub.status.idle": "2020-12-09T23:19:26.162756Z",
     "shell.execute_reply": "2020-12-09T23:19:26.162293Z"
    }
   },
   "outputs": [],
   "source": [
    "def read_table(table_name):\n",
    "    return cudf.read_csv(os.path.join(INPUT_DATA_DIR, f'{table_name}.csv'))\n",
    "\n",
    "train = read_table('train')\n",
    "store = read_table('store')\n",
    "store_states = read_table('store_states')\n",
    "state_names = read_table('state_names')\n",
    "googletrend = read_table('googletrend')\n",
    "weather = read_table('weather')\n",
    "test = read_table('test')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "StateHoliday column indicates a state holiday. Normally all stores, with few exceptions, are closed on state holidays. Note that all schools are closed on public holidays and weekends. Below, we turn state Holidays to booleans to make them more convenient for modeling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.167748Z",
     "iopub.status.busy": "2020-12-09T23:19:26.167233Z",
     "iopub.status.idle": "2020-12-09T23:19:26.175705Z",
     "shell.execute_reply": "2020-12-09T23:19:26.175176Z"
    }
   },
   "outputs": [],
   "source": [
    "train.StateHoliday = train.StateHoliday!='0'\n",
    "test.StateHoliday = test.StateHoliday!='0'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here, new features are added by extracting dates and state names from the given data. Also all instances of state name 'NI' are replaced to match the usage in the rest of the data: 'HB,NI'. \n",
    "\n",
    "We create a new data frame `trend_de` from the Google trends data that has a special category for the whole of the Germany."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.180338Z",
     "iopub.status.busy": "2020-12-09T23:19:26.179844Z",
     "iopub.status.idle": "2020-12-09T23:19:26.196813Z",
     "shell.execute_reply": "2020-12-09T23:19:26.197197Z"
    }
   },
   "outputs": [],
   "source": [
    "googletrend['Date'] = googletrend.week.str.split(' - ', expand=True)[0]\n",
    "googletrend['State'] = googletrend.file.str.split('_', expand=True)[2]\n",
    "googletrend['State'] = googletrend.State.where(googletrend['State']!='NI', 'HB,NI')\n",
    "trend_de = googletrend.loc[googletrend.file == 'Rossmann_DE'].copy()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following extracts particular date fields from a complete datetime for the purpose of constructing categoricals. When working with date-time featues, you should always consider this feature extraction step. You can capture the trend/cyclical behavior as a function of time at any of these granularities by expanding your date-time into these additional fields. Here, we create `Day, Week, Month and Year` features, and add these to every table with a date field. Note that cudf currently does not have `dt.week` property, thus, we applied a custom method to create `Week` column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.202872Z",
     "iopub.status.busy": "2020-12-09T23:19:26.202401Z",
     "iopub.status.idle": "2020-12-09T23:19:26.284688Z",
     "shell.execute_reply": "2020-12-09T23:19:26.285071Z"
    }
   },
   "outputs": [],
   "source": [
    "for df in (weather, googletrend, train, test, trend_de):\n",
    "    df['Date'] = df['Date'].astype('datetime64[ns]')\n",
    "    df['Month'] = df.Date.dt.month\n",
    "    df['Day'] = df.Date.dt.day\n",
    "    df['Year'] = df.Date.dt.year.astype(str)+ '-01-01'\n",
    "    df['Week']= (((df['Date'] - df['Year'].astype('datetime64[ns]')).dt.days)/7).astype('int16') +1\n",
    "    df['Week']=  df.Week.where(df['Week']!=53, 52)\n",
    "    df['Year'] = df.Date.dt.year"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will use the function below to perform left outer join operation. The suffixes argument describes the naming convention for duplicate fields."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.288962Z",
     "iopub.status.busy": "2020-12-09T23:19:26.288466Z",
     "iopub.status.idle": "2020-12-09T23:19:26.290095Z",
     "shell.execute_reply": "2020-12-09T23:19:26.290475Z"
    }
   },
   "outputs": [],
   "source": [
    "def merge(df, right, left_on, right_on=None, suffix=None):\n",
    "    df = df.merge(right, how='left', left_on=left_on, right_on=right_on or left_on, suffixes=('', suffix or '_y'))\n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This function will be used to drop redundant or unwanted columns generated via join operations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.294299Z",
     "iopub.status.busy": "2020-12-09T23:19:26.293823Z",
     "iopub.status.idle": "2020-12-09T23:19:26.295703Z",
     "shell.execute_reply": "2020-12-09T23:19:26.296097Z"
    }
   },
   "outputs": [],
   "source": [
    "def drop_cols(gdf):\n",
    "    for c in gdf.columns:\n",
    "        if c.endswith('_y'):\n",
    "            if c in gdf.columns: gdf.drop(c, inplace=True, axis=1)\n",
    "    return gdf"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can outer join all of our data into a single dataframe that we will use to train our DL model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.300251Z",
     "iopub.status.busy": "2020-12-09T23:19:26.299733Z",
     "iopub.status.idle": "2020-12-09T23:19:26.334575Z",
     "shell.execute_reply": "2020-12-09T23:19:26.334956Z"
    }
   },
   "outputs": [],
   "source": [
    "weather = merge(weather, state_names, 'file', right_on='StateName')\n",
    "store = merge(store, store_states, 'Store')\n",
    "train_df = merge(train, store, 'Store')\n",
    "test_df = merge(test, store, 'Store')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.338767Z",
     "iopub.status.busy": "2020-12-09T23:19:26.338294Z",
     "iopub.status.idle": "2020-12-09T23:19:26.384665Z",
     "shell.execute_reply": "2020-12-09T23:19:26.385049Z"
    }
   },
   "outputs": [],
   "source": [
    "train_df = merge(train_df, googletrend, ['State', 'Year', 'Week'])\n",
    "test_df = merge(test_df, googletrend, ['State', 'Year', 'Week'])\n",
    "train = test = googletrend = None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.389153Z",
     "iopub.status.busy": "2020-12-09T23:19:26.388648Z",
     "iopub.status.idle": "2020-12-09T23:19:26.453225Z",
     "shell.execute_reply": "2020-12-09T23:19:26.453903Z"
    }
   },
   "outputs": [],
   "source": [
    "train_df = merge(train_df, trend_de, ['Year', 'Week'], right_on=['Year', 'Week'], suffix='_DE')\n",
    "test_df = merge(test_df, trend_de, ['Year', 'Week'], right_on=['Year', 'Week'], suffix='_DE')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.460776Z",
     "iopub.status.busy": "2020-12-09T23:19:26.460089Z",
     "iopub.status.idle": "2020-12-09T23:19:26.566319Z",
     "shell.execute_reply": "2020-12-09T23:19:26.566873Z"
    }
   },
   "outputs": [],
   "source": [
    "train_df = drop_cols(train_df)\n",
    "train_df = merge(train_df, weather, ['State', 'Date'], right_on=['State', 'Date'], suffix='_y')\n",
    "test_df = drop_cols(test_df)\n",
    "test_df = merge(test_df, weather, ['State', 'Date'], right_on=['State', 'Date'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.576468Z",
     "iopub.status.busy": "2020-12-09T23:19:26.575772Z",
     "iopub.status.idle": "2020-12-09T23:19:26.577874Z",
     "shell.execute_reply": "2020-12-09T23:19:26.577233Z"
    }
   },
   "outputs": [],
   "source": [
    "# drop the redundant columns with `_y` suffix.\n",
    "train_df = drop_cols(train_df)\n",
    "test_df = drop_cols(test_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next we'll fill in missing values to avoid complications with NA's. Many models have problems when missing values are present, so it's important to think about how to deal with them. Here, we are picking arbitrary signal values and filling the missing values with them."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.582704Z",
     "iopub.status.busy": "2020-12-09T23:19:26.582080Z",
     "iopub.status.idle": "2020-12-09T23:19:26.607354Z",
     "shell.execute_reply": "2020-12-09T23:19:26.608219Z"
    }
   },
   "outputs": [],
   "source": [
    "for df in [train_df, test_df]:\n",
    "    df['CompetitionOpenSinceYear'] = df.CompetitionOpenSinceYear.fillna(1900).astype('int32')\n",
    "    df['CompetitionOpenSinceMonth'] = df.CompetitionOpenSinceMonth.fillna(1).astype('int32')\n",
    "    df['Promo2SinceYear'] = df.Promo2SinceYear.fillna(1900).astype('int32')\n",
    "    df['Promo2SinceWeek'] = df.Promo2SinceWeek.fillna(1).astype('int32')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next we'll extract features \"CompetitionOpenSince\" and \"CompetitionDaysOpen\". "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.613757Z",
     "iopub.status.busy": "2020-12-09T23:19:26.612743Z",
     "iopub.status.idle": "2020-12-09T23:19:26.936809Z",
     "shell.execute_reply": "2020-12-09T23:19:26.937196Z"
    }
   },
   "outputs": [],
   "source": [
    "for df in [train_df, test_df]:\n",
    "    df['year']= df['CompetitionOpenSinceYear']\n",
    "    df['month']= df['CompetitionOpenSinceMonth']\n",
    "    df['day']= '15'\n",
    "    df['CompetitionOpenSince'] = cudf.to_datetime(df[[\"year\", \"month\", \"day\"]])\n",
    "    df[\"CompetitionDaysOpen\"] = (df['Date'] - df['CompetitionOpenSince']).dt.days"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's replace some erroneous / outlying data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.941005Z",
     "iopub.status.busy": "2020-12-09T23:19:26.940527Z",
     "iopub.status.idle": "2020-12-09T23:19:26.964789Z",
     "shell.execute_reply": "2020-12-09T23:19:26.964384Z"
    }
   },
   "outputs": [],
   "source": [
    "for df in [train_df, test_df]:\n",
    "    df.loc[df.CompetitionDaysOpen<0, \"CompetitionDaysOpen\"] = 0\n",
    "    df.loc[df.CompetitionOpenSinceYear<1990, \"CompetitionDaysOpen\"] = 0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll now add \"CompetitionMonthsOpen\" field, limiting the maximum to two years to limit number of unique categories."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.968510Z",
     "iopub.status.busy": "2020-12-09T23:19:26.968026Z",
     "iopub.status.idle": "2020-12-09T23:19:26.976894Z",
     "shell.execute_reply": "2020-12-09T23:19:26.977277Z"
    }
   },
   "outputs": [],
   "source": [
    "for df in [train_df, test_df]:\n",
    "    df[\"CompetitionMonthsOpen\"] = df[\"CompetitionDaysOpen\"]//30\n",
    "    df.loc[df.CompetitionMonthsOpen>24, \"CompetitionMonthsOpen\"] = 24"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Same process is applied for Promo dates. We'll create \"Promo2Days\"  and \"Promo2Weeks\" fields, limiting the maximum to 25 weeks to limit number of unique categories."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:26.981935Z",
     "iopub.status.busy": "2020-12-09T23:19:26.981470Z",
     "iopub.status.idle": "2020-12-09T23:19:27.004175Z",
     "shell.execute_reply": "2020-12-09T23:19:27.004557Z"
    }
   },
   "outputs": [],
   "source": [
    "for df in [train_df, test_df]:\n",
    "    df['Promo2SinceYear_tmp']= df.Promo2SinceYear.astype(str)+ '-01-01'\n",
    "    dt = cudf.to_datetime(df.Promo2SinceYear_tmp, format='%Y').astype('int64') // 10**9\n",
    "    dt += 7*24*3600*df.Promo2SinceWeek\n",
    "    df[\"Promo2Since\"] = cudf.to_datetime(dt*10**9)\n",
    "    df[\"Promo2Days\"] = (df['Date']- df[\"Promo2Since\"]).dt.days"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:27.009307Z",
     "iopub.status.busy": "2020-12-09T23:19:27.008589Z",
     "iopub.status.idle": "2020-12-09T23:19:27.033146Z",
     "shell.execute_reply": "2020-12-09T23:19:27.033549Z"
    }
   },
   "outputs": [],
   "source": [
    "for df in [train_df, test_df]:\n",
    "    df.loc[df.Promo2Days<0, \"Promo2Days\"] = 0\n",
    "    df.loc[df.Promo2SinceYear<1990, \"Promo2Days\"] = 0\n",
    "    df[\"Promo2Weeks\"] = df[\"Promo2Days\"]//7\n",
    "    df.loc[df.Promo2Weeks<0, \"Promo2Weeks\"] = 0\n",
    "    df.loc[df.Promo2Weeks>25, \"Promo2Weeks\"] = 25"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:27.040061Z",
     "iopub.status.busy": "2020-12-09T23:19:27.039563Z",
     "iopub.status.idle": "2020-12-09T23:19:27.086562Z",
     "shell.execute_reply": "2020-12-09T23:19:27.086946Z"
    }
   },
   "outputs": [],
   "source": [
    "df = train_df.append(test_df, ignore_index=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:27.093626Z",
     "iopub.status.busy": "2020-12-09T23:19:27.093162Z",
     "iopub.status.idle": "2020-12-09T23:19:27.094842Z",
     "shell.execute_reply": "2020-12-09T23:19:27.095223Z"
    }
   },
   "outputs": [],
   "source": [
    "#let's drop these dummy columns.\n",
    "df.drop(['year', 'month', 'day', 'Promo2SinceYear_tmp'], inplace=True, axis=1)\n",
    "train_df.drop(['year', 'month', 'day', 'Promo2SinceYear_tmp'], inplace=True, axis=1)\n",
    "test_df.drop(['year', 'month', 'day', 'Promo2SinceYear_tmp'], inplace=True, axis=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:27.098488Z",
     "iopub.status.busy": "2020-12-09T23:19:27.098022Z",
     "iopub.status.idle": "2020-12-09T23:19:27.105066Z",
     "shell.execute_reply": "2020-12-09T23:19:27.104652Z"
    }
   },
   "outputs": [],
   "source": [
    "# # cast SchoolHoliday to int32.\n",
    "df['SchoolHoliday'] = df['SchoolHoliday'].astype('int32')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is modififed version of the original `get_elapsed` function from [rossmann_data_clean nb](https://github.com/fastai/course-v3/blob/master/nbs/dl1/rossman_data_clean.ipynb). The `get_elapsed` function is defined for cumulative counting across a sorted dataframe. Given a particular field to monitor, this function starts tracking time since the last occurrence of that field. When the field is seen again, the counter is set to zero.\n",
    "\n",
    "This section is still not working as intended with cuDF, therefore we are converting the cudf df to pandas df to perform the `bfill` and `ffill` operations below. We can explain the code below by looking at School Holiday as an exp. <i>SchoolHoliday</i> column indicates if the (Store, Date) was affected by the closure of public schools. First every row of the dataframe is sorted in order of store and date. Then, we will add to the dataframe the days since seeing a School Holiday, and the days until another holiday."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:27.108710Z",
     "iopub.status.busy": "2020-12-09T23:19:27.108246Z",
     "iopub.status.idle": "2020-12-09T23:19:28.250446Z",
     "shell.execute_reply": "2020-12-09T23:19:28.250832Z"
    }
   },
   "outputs": [],
   "source": [
    "# ops: masking, ffill, bfill, timedelta\n",
    "df = df.sort_values(by=['Store', 'Date'])\n",
    "# We convert cudf dataframe to the pandas df be able to do `bfill` and `ffill` operations below.\n",
    "pdf = df.to_pandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:28.257972Z",
     "iopub.status.busy": "2020-12-09T23:19:28.257491Z",
     "iopub.status.idle": "2020-12-09T23:19:30.022507Z",
     "shell.execute_reply": "2020-12-09T23:19:30.022891Z"
    }
   },
   "outputs": [],
   "source": [
    "# first build a mask indicating where stores start and end\n",
    "first_indices = pdf.Store.diff() != 0\n",
    "last_indices = pdf.Store.diff().iloc[1:].append(pd.Series([1]))\n",
    "last_indices.index = first_indices.index\n",
    "idx_mask = ~(first_indices | last_indices)\n",
    "\n",
    "event_fields = ['SchoolHoliday', 'StateHoliday', 'Promo']\n",
    "for field in event_fields:\n",
    "    # use the mask from above to mask save dates from the start and end\n",
    "    # of a given store's range, as well as all dates that have an event\n",
    "    pdf['tmp'] = pdf.Date\n",
    "    pdf.loc[(pdf[field] == 0) & idx_mask, 'tmp'] = float('nan')\n",
    "    # then use ffill and bbfill to give the input to the time delta\n",
    "    pdf['After'+field] = pdf.tmp.ffill()\n",
    "    pdf['Before'+field] = pdf.tmp.bfill()\n",
    "\n",
    "    # compute deltas between bfilled and ffilled dates and the current date\n",
    "    pdf['After'+field] = (pdf['Date'] - pdf['After'+field]).astype('timedelta64[D]')\n",
    "    pdf['Before'+field] = (pdf['Before'+field] - pdf['Date']).astype('timedelta64[D]')\n",
    "\n",
    "# get rid of our dummy column\n",
    "pdf = pdf.drop(columns=['tmp'])\n",
    "\n",
    "#let's convert pandas back to cudf df\n",
    "df = cudf.from_pandas(pdf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:30.028523Z",
     "iopub.status.busy": "2020-12-09T23:19:30.028033Z",
     "iopub.status.idle": "2020-12-09T23:19:30.123857Z",
     "shell.execute_reply": "2020-12-09T23:19:30.124267Z"
    }
   },
   "outputs": [],
   "source": [
    "pd.DataFrame(pdf.dtypes).to_csv('dtypes_none.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:30.127854Z",
     "iopub.status.busy": "2020-12-09T23:19:30.127337Z",
     "iopub.status.idle": "2020-12-09T23:19:30.190597Z",
     "shell.execute_reply": "2020-12-09T23:19:30.190978Z"
    }
   },
   "outputs": [],
   "source": [
    "# Set the active index to Date.\n",
    "df = df.set_index(\"Date\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next we'll apply window calculations in cudf to calculate rolling quantities. Here, we're sorting by date (sort_index()), grouping by Store column, and counting the number of events of interest (sum()) defined in columns in the following week (rolling()). We do the same in the opposite direction."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:30.195428Z",
     "iopub.status.busy": "2020-12-09T23:19:30.194962Z",
     "iopub.status.idle": "2020-12-09T23:19:30.494960Z",
     "shell.execute_reply": "2020-12-09T23:19:30.495384Z"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/conda/envs/rapids/lib/python3.7/site-packages/numba/cuda/envvars.py:17: NumbaWarning: \u001b[1m\n",
      "Environment variables with the 'NUMBAPRO' prefix are deprecated and consequently ignored, found use of NUMBAPRO_NVVM=/usr/local/cuda/nvvm/lib64/libnvvm.so.\n",
      "\n",
      "For more information about alternatives visit: ('https://numba.pydata.org/numba-doc/latest/cuda/overview.html', '#cudatoolkit-lookup')\u001b[0m\n",
      "  warnings.warn(errors.NumbaWarning(msg))\n",
      "/opt/conda/envs/rapids/lib/python3.7/site-packages/numba/cuda/envvars.py:17: NumbaWarning: \u001b[1m\n",
      "Environment variables with the 'NUMBAPRO' prefix are deprecated and consequently ignored, found use of NUMBAPRO_LIBDEVICE=/usr/local/cuda/nvvm/libdevice/.\n",
      "\n",
      "For more information about alternatives visit: ('https://numba.pydata.org/numba-doc/latest/cuda/overview.html', '#cudatoolkit-lookup')\u001b[0m\n",
      "  warnings.warn(errors.NumbaWarning(msg))\n"
     ]
    }
   ],
   "source": [
    "event_fields = ['SchoolHoliday', 'StateHoliday', 'Promo']\n",
    "bwd = df[['Store']+event_fields].sort_index().groupby(\"Store\").rolling(7, min_periods=1).sum()\n",
    "fwd = df[['Store']+event_fields].sort_index(ascending=False).groupby(\"Store\").rolling(7, min_periods=1).sum()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next we drop the Store indices grouped together in the window function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:30.500549Z",
     "iopub.status.busy": "2020-12-09T23:19:30.500056Z",
     "iopub.status.idle": "2020-12-09T23:19:30.501770Z",
     "shell.execute_reply": "2020-12-09T23:19:30.502160Z"
    }
   },
   "outputs": [],
   "source": [
    "for d in (bwd, fwd):\n",
    "    d.drop('Store', 1, inplace=True)\n",
    "    d.reset_index(inplace=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:30.505540Z",
     "iopub.status.busy": "2020-12-09T23:19:30.505070Z",
     "iopub.status.idle": "2020-12-09T23:19:30.506762Z",
     "shell.execute_reply": "2020-12-09T23:19:30.507177Z"
    }
   },
   "outputs": [],
   "source": [
    "df.reset_index(inplace=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "Let's merge these values onto the df, and drop the unwanted columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:30.511708Z",
     "iopub.status.busy": "2020-12-09T23:19:30.511173Z",
     "iopub.status.idle": "2020-12-09T23:19:30.967998Z",
     "shell.execute_reply": "2020-12-09T23:19:30.968403Z"
    }
   },
   "outputs": [],
   "source": [
    "df = df.merge(bwd, left_on=['Date', 'Store'], right_on=['Date', 'Store'], how='left', suffixes=['', '_bw'])\n",
    "df = df.merge(fwd, left_on=['Date', 'Store'], right_on=['Date', 'Store'], how= 'left', suffixes=['', '_fw'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:30.972806Z",
     "iopub.status.busy": "2020-12-09T23:19:30.972329Z",
     "iopub.status.idle": "2020-12-09T23:19:31.412217Z",
     "shell.execute_reply": "2020-12-09T23:19:31.412774Z"
    }
   },
   "outputs": [],
   "source": [
    "train_df = merge(train_df, df, ['Store', 'Date'], right_on=['Store', 'Date'])\n",
    "test_df = merge(test_df, df, ['Store', 'Date'], right_on=['Store', 'Date'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:31.463545Z",
     "iopub.status.busy": "2020-12-09T23:19:31.461853Z",
     "iopub.status.idle": "2020-12-09T23:19:31.465138Z",
     "shell.execute_reply": "2020-12-09T23:19:31.465521Z"
    }
   },
   "outputs": [],
   "source": [
    "train_df = drop_cols(train_df)\n",
    "test_df = drop_cols(test_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we are removing all instances where the store had zero sale / was closed, and we sort our training set by Date."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:31.469608Z",
     "iopub.status.busy": "2020-12-09T23:19:31.469125Z",
     "iopub.status.idle": "2020-12-09T23:19:32.100953Z",
     "shell.execute_reply": "2020-12-09T23:19:32.101335Z"
    }
   },
   "outputs": [],
   "source": [
    "train_df = train_df[train_df.Sales != 0]\n",
    "train_df = train_df.sort_values(by='Date', ascending=True)\n",
    "train_df=train_df.reset_index(drop=True)\n",
    "test_df=test_df.reset_index(drop=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:32.104937Z",
     "iopub.status.busy": "2020-12-09T23:19:32.104486Z",
     "iopub.status.idle": "2020-12-09T23:19:32.299683Z",
     "shell.execute_reply": "2020-12-09T23:19:32.300093Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "42171"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# determine a cut-off point to create a validation dataset.\n",
    "cut = train_df['Date'][(train_df['Date'] == train_df['Date'][len(test_df)])].index.max()\n",
    "cut"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:32.304369Z",
     "iopub.status.busy": "2020-12-09T23:19:32.303916Z",
     "iopub.status.idle": "2020-12-09T23:19:32.312213Z",
     "shell.execute_reply": "2020-12-09T23:19:32.311795Z"
    }
   },
   "outputs": [],
   "source": [
    "# split the train_df as training and validation data sets.\n",
    "num_valid = cut\n",
    "valid_df = train_df[-num_valid:]\n",
    "train_df = train_df[:-num_valid]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's save our preprocessed files as csv files to be used in [Rossmann examples](https://github.com/NVIDIA/NVTabular/blob/main/examples/rossmann/) for further preprocessing with NVTabular, then for model training and testing with PyT, TF and FastAI frameworks."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:32.317043Z",
     "iopub.status.busy": "2020-12-09T23:19:32.316579Z",
     "iopub.status.idle": "2020-12-09T23:19:34.105631Z",
     "shell.execute_reply": "2020-12-09T23:19:34.106039Z"
    }
   },
   "outputs": [],
   "source": [
    "!mkdir -p $OUTPUT_DATA_DIR\n",
    "\n",
    "train_df.to_csv(os.path.join(OUTPUT_DATA_DIR, 'train.csv'), index=False)\n",
    "valid_df.to_csv(os.path.join(OUTPUT_DATA_DIR, 'valid.csv'), index=False)\n",
    "test_df.to_csv(os.path.join(OUTPUT_DATA_DIR, 'test.csv'), index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2020-12-09T23:19:34.151329Z",
     "iopub.status.busy": "2020-12-09T23:19:34.110182Z",
     "iopub.status.idle": "2020-12-09T23:19:34.725760Z",
     "shell.execute_reply": "2020-12-09T23:19:34.726147Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "test.csv  train.csv  valid.csv\r\n"
     ]
    }
   ],
   "source": [
    "!ls $OUTPUT_DATA_DIR"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}