notebooks/model_training_2.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import mlflow\n",
    "import mlflow.xgboost\n",
    "from xgboost import XGBRegressor\n",
    "from sklearn.model_selection import GridSearchCV\n",
    "import subprocess\n",
    "from mlflow.tracking import MlflowClient\n",
    "from sklearn.metrics import mean_absolute_percentage_error, mean_absolute_error"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Preprocessing:\n",
    "    \"\"\"\n",
    "        Class functions to retrieve and preprocess time series data\n",
    "    \"\"\"\n",
    "    \n",
    "    def get_data(path: str) -> pd.DataFrame:\n",
    "        data = pd.read_csv(path, index_col=0)\n",
    "        return data\n",
    "\n",
    "    def preprocessing(data: pd.DataFrame) -> pd.DataFrame:\n",
    "        \"\"\"\n",
    "        Returns data with index and frequency of index set\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        data: pd.DataFrame\n",
    "\n",
    "        col: str\n",
    "            name of the column that will be kept\n",
    "        \"\"\"\n",
    "        data.index = pd.to_datetime(data.index)\n",
    "        #data = data[col]\n",
    "        data = data.div(1000)\n",
    "        data.index.freq = pd.infer_freq(data.index)\n",
    "        return data\n",
    "\n",
    "    def train_test_split_series(data: pd.DataFrame, n_test: int) -> pd.DataFrame:\n",
    "        return data.iloc[:-n_test], data.iloc[-n_test:]\n",
    "\n",
    "    def train_test_split_df(data: pd.DataFrame, n_test: int) -> pd.DataFrame:\n",
    "        return data.iloc[:-n_test], data.iloc[-n_test:]\n",
    "\n",
    "    def series_to_supervised(\n",
    "        data: pd.Series, n_in: int = 1, dropnan: bool = True\n",
    "    ) -> np.array:\n",
    "        \"\"\"\n",
    "        Converts a sequence of numbers, i.e. a univariate time series, into a matrix\n",
    "        with one array (series at time t) plus one more array for each n_in\n",
    "        (lags at times t-1, t-2, .., t-n_in).\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        data: pd.Series\n",
    "\n",
    "        n_in: int\n",
    "            number of lags to create from the original series.\n",
    "            For each lag required, one more column will be added,\n",
    "            at the cost of one row of observations.\n",
    "\n",
    "        dropnan: bool\n",
    "\n",
    "        \"\"\"\n",
    "        df = pd.DataFrame(data)\n",
    "        cols = list()\n",
    "        # input sequence (t-n, ... t-1)\n",
    "        for i in range(n_in, 0, -1):\n",
    "            cols.append(df.shift(i))\n",
    "        cols.append(df)\n",
    "        # put it all together\n",
    "        agg = pd.concat(cols, axis=1)\n",
    "        # drop rows with NaN values (in particular the first and the last rows)\n",
    "        if dropnan:\n",
    "            agg.dropna(inplace=True)\n",
    "\n",
    "        return agg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "class XGBForecaster:\n",
    "    \"\"\"\n",
    "        XGBoost model used for univariate or multivariate forecasting.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self):\n",
    "        self.model = XGBRegressor()\n",
    "\n",
    "    def fit(self, train_ensamble: pd.DataFrame) -> XGBRegressor:\n",
    "        data = np.asarray(train_ensamble)\n",
    "        X, y = data[:, :-1], data[:, -1]\n",
    "        self.fitted_model = self.model.fit(X, y)\n",
    "        return self.fitted_model\n",
    "\n",
    "    def forecast(self, row_just_before: int, steps_ahead: int) -> list:\n",
    "        \"\"\"\n",
    "            Rolling prediction with the model_fitted for predicting n=steps_ahead new instances.\n",
    "            This instances will immediately follow row_just_before, which is the last row of the dataframe available\n",
    "        \"\"\"\n",
    "        row_just_before = np.asarray(row_just_before)[1:]\n",
    "        current_row = row_just_before.reshape(1, -1)\n",
    "        forecast = []\n",
    "        for _ in range(steps_ahead):\n",
    "            pred = self.fitted_model.predict(current_row)\n",
    "            forecast.append(pred[0])\n",
    "            current_row = np.concatenate((current_row[0][1:], pred)).reshape(1, -1)\n",
    "        return forecast\n",
    "\n",
    "    def grid_search(self, parameters: dict, n_folds: int, train_df: pd.DataFrame, test_size, n_jobs=1, verbose=0):\n",
    "        \"\"\"\n",
    "            Grid Search for time series forecasting with XGBoost\n",
    "        \"\"\"\n",
    "        grid = GridSearchCV(\n",
    "            xgb, parameters, cv=n_folds, n_jobs=n_jobs, verbose=verbose\n",
    "        )\n",
    "        grid = XGBForecaster.fit(train_df, grid)\n",
    "        predictions = XGBForecaster.forecast(train_df.iloc[-1, :], grid, test_size)\n",
    "        return grid, predictions\n",
    "\n",
    "    def preprocess(self, data: pd.DataFrame, experiment_name: str, frac: float):\n",
    "        \"\"\"\n",
    "            Creates an experiment run for the model to be trained and preprocess the data\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        data: pd.DataFrame\n",
    "            data to use for training.\n",
    "        experiment_name: str\n",
    "            name of the experiment for training the model; might refer to the commodity to forecast.\n",
    "        frac: float\n",
    "            percentage of data to hold out for testing the model.\n",
    "\n",
    "        \"\"\"\n",
    "        # Create the experiment if it does not exist\n",
    "        experiment = mlflow.get_experiment_by_name(experiment_name)\n",
    "        if experiment is None:\n",
    "            mlflow.create_experiment(experiment_name)\n",
    "            experiment = mlflow.get_experiment_by_name(experiment_name)\n",
    "\n",
    "        with mlflow.start_run(experiment_id=experiment.experiment_id):\n",
    "            # logging information on input data\n",
    "            data = Preprocessing.preprocessing(data)\n",
    "\n",
    "            train, test = Preprocessing.train_test_split_df(\n",
    "                data=data, n_test=round(len(data) * frac)\n",
    "            )\n",
    "\n",
    "            proc_training_data = Preprocessing.series_to_supervised(\n",
    "                data=train, n_in=1, dropnan=True\n",
    "            )\n",
    "            proc_testing_data = Preprocessing.series_to_supervised(\n",
    "                data=test, n_in=1, dropnan=False\n",
    "            )\n",
    "\n",
    "            mlflow.log_param(key=\"pct_data_for_training\", value=(1 - frac))\n",
    "            mlflow.log_param(key=\"pct_data_for_testing\", value=(frac))\n",
    "\n",
    "            return proc_training_data, proc_testing_data\n",
    "        \n",
    "    def train_model(\n",
    "        self, experiment_name: str, train_data: pd.DataFrame, test_data: pd.DataFrame\n",
    "    ) -> XGBRegressor:\n",
    "\n",
    "        test_data.fillna(train_data.iloc[-1, -1])\n",
    "        X_train = train_data.iloc[:, :-1].values\n",
    "        X_test = test_data.iloc[:, :-1].values\n",
    "        y_train = train_data.iloc[:, -1].values\n",
    "        y_test = test_data.iloc[:, -1].values\n",
    "\n",
    "        # n-folds\n",
    "        effective_df_length = len(train_data) - len(test_data)\n",
    "        max_folds = effective_df_length // len(test_data)\n",
    "        n_folds = min(max_folds, 10)\n",
    "\n",
    "        # Create the experiment if it does not exist\n",
    "        client = MlflowClient()\n",
    "        experiment = mlflow.get_experiment_by_name(experiment_name)\n",
    "        experiment_id = experiment.experiment_id\n",
    "        latest_run = client.search_runs(\n",
    "            experiment_id, order_by=[\"start_time desc\"], max_results=1\n",
    "        )[0]\n",
    "            \n",
    "        # enable auto logging\n",
    "        mlflow.xgboost.autolog()\n",
    "\n",
    "        with mlflow.start_run(run_id=latest_run.info.run_id):\n",
    "            # log the script\n",
    "            #mlflow.log_artifact(__file__)\n",
    "\n",
    "            # Get current commit hash\n",
    "            commit_hash = (\n",
    "                subprocess.check_output([\"git\", \"rev-parse\", \"HEAD\"])\n",
    "                .strip()\n",
    "                .decode(\"utf-8\")\n",
    "            )\n",
    "            # Log Git commit hash as a parameter\n",
    "            mlflow.log_param(\"commit_hash\", commit_hash)\n",
    "\n",
    "            #xgb = XGBRegressor()\n",
    "            parameters_xgb = {\n",
    "                \"gamma\": [0, 30, 100, 200],\n",
    "                \"eta\": [0.3, 0.03, 0.003],\n",
    "                \"max_depth\": [6, 12, 30],\n",
    "            }\n",
    "            xgb_grid, predictions_xgb = XGBForecaster.grid_search(\n",
    "                parameters=parameters_xgb,\n",
    "                n_folds=n_folds,\n",
    "                train_df=train_data,\n",
    "                test_size=len(test_data),\n",
    "                n_jobs=-1,\n",
    "                verbose=1,\n",
    "            )\n",
    "            mae = mean_absolute_error(y_test, predictions_xgb)\n",
    "            mape = mean_absolute_percentage_error(y_test, predictions_xgb)\n",
    "\n",
    "            # log metrics\n",
    "            mlflow.log_metrics({\"MAE\": mae, \"MAPE\": mape})\n",
    "\n",
    "            return xgb_grid"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>BENZINA</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>DATA_RILEVAZIONE</th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>2005-01-03</th>\n",
       "      <td>1115.75</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2005-01-10</th>\n",
       "      <td>1088.00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2005-01-17</th>\n",
       "      <td>1088.14</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2005-01-24</th>\n",
       "      <td>1090.01</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2005-01-31</th>\n",
       "      <td>1132.11</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                  BENZINA\n",
       "DATA_RILEVAZIONE         \n",
       "2005-01-03        1115.75\n",
       "2005-01-10        1088.00\n",
       "2005-01-17        1088.14\n",
       "2005-01-24        1090.01\n",
       "2005-01-31        1132.11"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "gasoline = pd.read_csv(\"../data/fuel_prices.csv\", index_col=0)\n",
    "gasoline = gasoline[[\"BENZINA\"]]\n",
    "gasoline.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "ename": "TypeError",
     "evalue": "grid_search() missing 1 required positional argument: 'self'",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
      "Cell \u001b[0;32mIn[40], line 11\u001b[0m\n\u001b[1;32m      3\u001b[0m xgb_forecaster \u001b[39m=\u001b[39m XGBForecaster()\n\u001b[1;32m      5\u001b[0m training_data, testing_data\u001b[39m=\u001b[39m xgb_forecaster\u001b[39m.\u001b[39mpreprocess(\n\u001b[1;32m      6\u001b[0m     experiment_name\u001b[39m=\u001b[39mexperiment_name, \n\u001b[1;32m      7\u001b[0m     data\u001b[39m=\u001b[39mgasoline,\n\u001b[1;32m      8\u001b[0m     frac\u001b[39m=\u001b[39mfrac\n\u001b[1;32m      9\u001b[0m )\n\u001b[0;32m---> 11\u001b[0m xgb_forecaster\u001b[39m.\u001b[39;49mtrain_model(\n\u001b[1;32m     12\u001b[0m     experiment_name\u001b[39m=\u001b[39;49mexperiment_name,\n\u001b[1;32m     13\u001b[0m     train_data\u001b[39m=\u001b[39;49mtraining_data, \n\u001b[1;32m     14\u001b[0m     test_data\u001b[39m=\u001b[39;49mtesting_data\n\u001b[1;32m     15\u001b[0m )\n",
      "Cell \u001b[0;32mIn[36], line 126\u001b[0m, in \u001b[0;36mXGBForecaster.train_model\u001b[0;34m(self, experiment_name, train_data, test_data)\u001b[0m\n\u001b[1;32m    120\u001b[0m \u001b[39m#xgb = XGBRegressor()\u001b[39;00m\n\u001b[1;32m    121\u001b[0m parameters_xgb \u001b[39m=\u001b[39m {\n\u001b[1;32m    122\u001b[0m     \u001b[39m\"\u001b[39m\u001b[39mgamma\u001b[39m\u001b[39m\"\u001b[39m: [\u001b[39m0\u001b[39m, \u001b[39m30\u001b[39m, \u001b[39m100\u001b[39m, \u001b[39m200\u001b[39m],\n\u001b[1;32m    123\u001b[0m     \u001b[39m\"\u001b[39m\u001b[39meta\u001b[39m\u001b[39m\"\u001b[39m: [\u001b[39m0.3\u001b[39m, \u001b[39m0.03\u001b[39m, \u001b[39m0.003\u001b[39m],\n\u001b[1;32m    124\u001b[0m     \u001b[39m\"\u001b[39m\u001b[39mmax_depth\u001b[39m\u001b[39m\"\u001b[39m: [\u001b[39m6\u001b[39m, \u001b[39m12\u001b[39m, \u001b[39m30\u001b[39m],\n\u001b[1;32m    125\u001b[0m }\n\u001b[0;32m--> 126\u001b[0m xgb_grid, predictions_xgb \u001b[39m=\u001b[39m XGBForecaster\u001b[39m.\u001b[39;49mgrid_search(\n\u001b[1;32m    127\u001b[0m     parameters\u001b[39m=\u001b[39;49mparameters_xgb,\n\u001b[1;32m    128\u001b[0m     n_folds\u001b[39m=\u001b[39;49mn_folds,\n\u001b[1;32m    129\u001b[0m     train_df\u001b[39m=\u001b[39;49mtrain_data,\n\u001b[1;32m    130\u001b[0m     test_size\u001b[39m=\u001b[39;49m\u001b[39mlen\u001b[39;49m(test_data),\n\u001b[1;32m    131\u001b[0m     n_jobs\u001b[39m=\u001b[39;49m\u001b[39m-\u001b[39;49m\u001b[39m1\u001b[39;49m,\n\u001b[1;32m    132\u001b[0m     verbose\u001b[39m=\u001b[39;49m\u001b[39m1\u001b[39;49m,\n\u001b[1;32m    133\u001b[0m )\n\u001b[1;32m    134\u001b[0m mae \u001b[39m=\u001b[39m mean_absolute_error(y_test, predictions_xgb)\n\u001b[1;32m    135\u001b[0m mape \u001b[39m=\u001b[39m mean_absolute_percentage_error(y_test, predictions_xgb)\n",
      "\u001b[0;31mTypeError\u001b[0m: grid_search() missing 1 required positional argument: 'self'"
     ]
    }
   ],
   "source": [
    "experiment_name = \"xgboost_predictor_gasoline\"\n",
    "frac = 0.2\n",
    "xgb_forecaster = XGBForecaster()\n",
    "\n",
    "training_data, testing_data= xgb_forecaster.preprocess(\n",
    "    experiment_name=experiment_name, \n",
    "    data=gasoline,\n",
    "    frac=frac\n",
    ")\n",
    "\n",
    "xgb_forecaster.train_model(\n",
    "    experiment_name=experiment_name,\n",
    "    train_data=training_data, \n",
    "    test_data=testing_data\n",
    ")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "epm",
   "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.9.16"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}