local_autoReg.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "lined-finland",
   "metadata": {},
   "source": [
    "# Bitcoin Price Prediction - Local Mode ARIMA/VectorARIMA"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "efficient-concord",
   "metadata": {},
   "source": [
    "* **Description**: COMP4103(Big Data)--Group Project\n",
    "* **Author**: Aaron\n",
    "* **Version**: 0.2"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "broken-transfer",
   "metadata": {},
   "source": [
    "**Updates:**\n",
    "1. Update the way to calcute Training time accurately.\n",
    "2. Add Cross Validation part"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "hired-cyprus",
   "metadata": {},
   "source": [
    "**Issues:**  \n",
    "1. N/A"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "invisible-knowing",
   "metadata": {},
   "source": [
    "**To be done:**  \n",
    "1. Visualize the influence of the different number of partitions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "particular-comfort",
   "metadata": {},
   "source": [
    "## 1. Load related packages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "id": "selective-institute",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Apache Spark\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions as F\n",
    "\n",
    "# Python\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from itertools import product\n",
    "import time\n",
    "import io\n",
    "from contextlib import redirect_stdout\n",
    "\n",
    "# Graph packages\n",
    "# https://plotly.com/python/getting-started/#jupyterlab-support\n",
    "# https://plotly.com/python/time-series/\n",
    "import plotly.express as px\n",
    "\n",
    "# ARIMA\n",
    "# https://alkaline-ml.com/pmdarima/index.html\n",
    "import pmdarima as pm\n",
    "from pmdarima.arima import ndiffs\n",
    "\n",
    "#Vector Autoregressions\n",
    "# https://www.statsmodels.org/dev/vector_ar.html\n",
    "from statsmodels.tsa.api import VAR\n",
    "\n",
    "# Scikit-learn\n",
    "from sklearn.metrics import explained_variance_score, mean_absolute_error, mean_squared_error, mean_squared_log_error, mean_absolute_percentage_error, r2_score\n",
    "\n",
    "# Load the customized Time Series Cross Validation\n",
    "from tsCrossValidation import mulTsCrossValidation, blockedTsCrossValidation, wfTsCrossValidation, modelComparison"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "english-yield",
   "metadata": {},
   "source": [
    "## 2. Create a Spark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "id": "particular-announcement",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start a SparkSession\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .master(\"local[*]\") \\\n",
    "    .appName(\"Bitcoin Prediction - Local Auto Regressions\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "id": "colored-heather",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://172.16.92.172:4041\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.1.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Bitcoin Prediction - Local Auto Regressions</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7fdb5a3e4a00>"
      ]
     },
     "execution_count": 74,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "stuffed-denmark",
   "metadata": {},
   "source": [
    "## 3. load Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "id": "frequent-rouge",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read csv file\n",
    "filename = \"bitcoin_1m_1min.csv\"\n",
    "dataset = spark.read.format(\"csv\") \\\n",
    "          .option(\"inferSchema\",'True') \\\n",
    "          .option(\"header\",True) \\\n",
    "          .load(filename)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "continuing-magazine",
   "metadata": {},
   "source": [
    "## 4. Train/Test Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "id": "important-receiver",
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Description: Split and keep the original time-series order\n",
    "Args:\n",
    "    dataSet: The dataSet which needs to be splited\n",
    "    proportion: A number represents the split proportion\n",
    "\n",
    "Return: \n",
    "    train_data: The train dataSet\n",
    "    test_data: The test dataSet\n",
    "'''\n",
    "def trainSplit(dataSet, proportion):\n",
    "    records_num = dataset.count()\n",
    "    split_point = round(records_num * proportion)\n",
    "    \n",
    "    train_data = dataset.filter(F.col(\"id\") < split_point)\n",
    "    test_data = dataset.filter(F.col(\"id\") >= split_point)\n",
    "    \n",
    "    return (train_data,test_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "id": "demonstrated-summary",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------------------+--------+\n",
      "| id|          Timestamp|   Close|\n",
      "+---+-------------------+--------+\n",
      "|  0|2021-03-02 00:00:00|49657.53|\n",
      "|  1|2021-03-02 00:01:00|49706.37|\n",
      "|  2|2021-03-02 00:02:00|49714.75|\n",
      "|  3|2021-03-02 00:03:00|49814.72|\n",
      "|  4|2021-03-02 00:04:00|49847.43|\n",
      "+---+-------------------+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Have a look on the data\n",
    "dataset.select(\"id\",\"Timestamp\",\"Close\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "id": "cooperative-reader",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Training data: 41718\n",
      "Test data: 42\n"
     ]
    }
   ],
   "source": [
    "# Split the dataSet\n",
    "proportion = 0.9990\n",
    "#proportion = 0.7\n",
    "train_data,test_data = trainSplit(dataset, proportion)\n",
    "\n",
    "# Cache it\n",
    "train_data.cache()\n",
    "test_data.cache()\n",
    "\n",
    "# Number of train and test dataSets\n",
    "print(f\"Training data: {train_data.count()}\\nTest data: {test_data.count()}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "id": "demanding-hobby",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[None, None, None]"
      ]
     },
     "execution_count": 79,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Save column name \n",
    "column_names = dataset.columns\n",
    "# labels and features\n",
    "feature_cols = dataset.columns\n",
    "# Gain the column list of features\n",
    "non_feature_cols  = ['id',\"NEXT_BTC_CLOSE\",'Timestamp']\n",
    "[feature_cols.remove(non_feature) for non_feature in non_feature_cols]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "floating-insight",
   "metadata": {},
   "source": [
    "## 5. Local Mode building"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "id": "considered-soldier",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a function to plot line-like graph\n",
    "# https://plotly.com/python/time-series/#time-series-with-range-selector-buttons\n",
    "'''\n",
    "Description: Plot the line graph by plotly(custom design)\n",
    "Args:\n",
    "    data: The data(pandas dataframe) which you want to ploy by line\n",
    "    graph_title: The title of the graph\n",
    "    \n",
    "Return: None\n",
    "'''\n",
    "def line_plot(data,graph_title):\n",
    "    plot = px.line(data,title=graph_title)\n",
    "    plot.update_xaxes(\n",
    "        rangeslider_visible=True,\n",
    "        rangeselector=dict(\n",
    "            buttons=list([\n",
    "                dict(count=7, label=\"1w\", step=\"day\", stepmode=\"backward\"),\n",
    "                dict(count=1, label=\"1m\", step=\"month\", stepmode=\"backward\"),\n",
    "                dict(count=6, label=\"6m\", step=\"month\", stepmode=\"backward\"),\n",
    "                dict(count=1, label=\"1y\", step=\"year\", stepmode=\"backward\"),\n",
    "                dict(step=\"all\")\n",
    "            ])\n",
    "        )\n",
    "    )\n",
    "    plot.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "id": "cross-actress",
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Description: Transform each partition of Spark to pandas dataframe\n",
    "Args:\n",
    "    partition_rdd: RDD of each partition\n",
    "    \n",
    "Return: \n",
    "    pandas_df: Data in pandas dataframe\n",
    "'''\n",
    "def partitionToPandas(partition_rdd):\n",
    "    pandas_df = pd.DataFrame(columns = column_names)\n",
    "    \n",
    "    # each_row is Row() type in Spark\n",
    "    for each_row in partition_rdd:\n",
    "        pandas_df = pandas_df.append(each_row.asDict(),ignore_index=True)\n",
    "    return [pandas_df]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "id": "balanced-treaty",
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Description: Build ARIMA model on each partition\n",
    "Args:\n",
    "    partition_rdd: RDD of each partition\n",
    "    \n",
    "Return: \n",
    "    arima_model: ARIMA model\n",
    "'''\n",
    "def buildARIMA(pandas_df):\n",
    "    \n",
    "    # Only choose Close as prediction\n",
    "    pandas_df = pandas_df[['Timestamp','Close']].set_index(\"Timestamp\")\n",
    "    \n",
    "    # Choose the best degree of differencing\n",
    "    kpss_diffs = ndiffs(pandas_df, alpha=0.05, test='kpss', max_d=6)\n",
    "    adf_diffs = ndiffs(pandas_df, alpha=0.05, test='adf', max_d=6)\n",
    "    n_diffs = max(adf_diffs, kpss_diffs)\n",
    "    \n",
    "    # Auto training\n",
    "    # p: AR (i.e., the number of lag observations)\n",
    "    # d: The degree of differencing.\n",
    "    # q: MA (the size of the “window”)\n",
    "    output = io.StringIO()\n",
    "    # Capture the trace time output to get the model training time\n",
    "    with redirect_stdout(output):\n",
    "        arima_model = pm.auto_arima(pandas_df, start_p=1, seasonal=False,\n",
    "                             d=n_diffs, trace=True,\n",
    "                             suppress_warnings=True,\n",
    "                             error_action=\"ignore\",\n",
    "                             max_order=None,\n",
    "                             stepwise=True)\n",
    "    # Get the model training time from trace time output\n",
    "    model_results = output.getvalue().split('\\n')\n",
    "    model_results = [ line for line in model_results if \"AIC\" in line ]\n",
    "    model_results = [line.split(':')[1].split(',') for line in model_results]\n",
    "    AIC_lst = [ line[0].split('=')[1] for line in model_results ]\n",
    "    time_lst = [ line[1].split('=')[1].split(' ')[0] for line in model_results ]\n",
    "    model_results_dict = {\"AIC\": AIC_lst, \"time\": time_lst}\n",
    "    model_results_df = pd.DataFrame(model_results_dict)\n",
    "    train_time = model_results_df.sort_values(\"AIC\").iloc[0,1]\n",
    "    \n",
    "    # Save the (p,d,q)\n",
    "    order_info = arima_model.order\n",
    "    return (arima_model,float(train_time))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "id": "played-premises",
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Description: Make prediction on each partition\n",
    "Args:\n",
    "    pandas_df: Data in pandas dataframe\n",
    "    broadcast_models: Trained Models\n",
    "    model_name: specify which model to make prediction\n",
    "    \n",
    "Return: \n",
    "    partition_pred: Predictions on the partition in a list\n",
    "'''\n",
    "def makePrediction(pandas_df, broadcast_models, model_name):\n",
    "    prediction_lst = []\n",
    "    num_pred = pandas_df.shape[0]\n",
    "    num_models = len(broadcast_models.value)\n",
    "    \n",
    "    if model_name == \"VectorARIMA\":\n",
    "        pandas_df.drop(['id'], axis=1, inplace=True)\n",
    "        pandas_df.set_index(\"Timestamp\", inplace=True)\n",
    "        \n",
    "        # Get the prediction from each model, then save to a list\n",
    "        for model in broadcast_models.value:\n",
    "            results = model.fit(maxlags=6, ic='aic')\n",
    "            lag_order = results.k_ar\n",
    "            prediction = results.forecast(pandas_df.values[-lag_order:],num_pred)\n",
    "            close_prediction = [lst[3] for lst in prediction]\n",
    "            prediction_lst.append(close_prediction)\n",
    "            \n",
    "    elif model_name == \"ARIMA\":\n",
    "        # Get the prediction from each model, then save to a list\n",
    "        for model in broadcast_models.value:\n",
    "            prediction_lst.append(model.predict(num_pred).tolist())\n",
    "            \n",
    "    else:\n",
    "        return \"Wrong model name\"\n",
    "    \n",
    "    # Define weight value\n",
    "    weight = list(range(1,num_models+1))\n",
    "    # Weighted the results from each Model\n",
    "    weighted_pred_lst = [[i*b for i in a] for a,b in zip(prediction_lst,weight)]\n",
    "    \n",
    "    # Aggregate the weighted predictions, then get Weighted value\n",
    "    partition_pred = [value / sum(weight) for value in map(sum,zip(*weighted_pred_lst))]\n",
    "    # Simple average \n",
    "    #partition_pred = [value / num_models for value in map(sum,zip(*prediction_lst))]\n",
    "    \n",
    "    return partition_pred"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "id": "spread-torture",
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Description: Build Vector ARIMA model on each partition\n",
    "Args:\n",
    "    partition_rdd: RDD of each partition\n",
    "    \n",
    "Return: \n",
    "    vector_arima: Vector ARIMA model\n",
    "'''\n",
    "def buildVectorARIMA(pandas_df):\n",
    "    \n",
    "    # Drop the column that don't need to predict\n",
    "    pandas_df.drop(['id'], axis=1, inplace=True)\n",
    "    pandas_df.set_index(\"Timestamp\", inplace=True)\n",
    "    start = time.time()\n",
    "    vector_arima = VAR(pandas_df)\n",
    "    end = time.time()\n",
    "    return (vector_arima, end-start)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "id": "broadband-update",
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Description: Calculate evaluation metrics\n",
    "Args:\n",
    "    y_test: Label of test data\n",
    "    y_pred: Prediction on test data\n",
    "    partition_num_train: Number of partition of Train data\n",
    "    partition_num_test: Number of partition of Test data\n",
    "    train_time: Time of training model\n",
    "    model_name: specify which model to make prediction\n",
    "    \n",
    "Return: \n",
    "    results: All the evaluation metrics in a dict\n",
    "'''\n",
    "def evaluationAssemble(y_test, y_pred, partition_num_train, partition_num_test, train_time, model_name):\n",
    "    # Explained variance score\n",
    "    exp_var = explained_variance_score(y_test,y_pred)\n",
    "\n",
    "    # Mean absolute error\n",
    "    mae = mean_absolute_error(y_test,y_pred)\n",
    "\n",
    "    # Root Mean squared error\n",
    "    rmse = mean_squared_error(y_test,y_pred,squared=False)\n",
    "\n",
    "    # Mean squared logarithmic error\n",
    "    msle = mean_squared_log_error(y_test,y_pred)\n",
    "\n",
    "    # Mean absolute percentage error\n",
    "    mape = mean_absolute_percentage_error(y_test,y_pred)\n",
    "\n",
    "    # R2 score, the coefficient of determination\n",
    "    r2 = r2_score(y_test,y_pred)\n",
    "\n",
    "    # Adjusted R2 score\n",
    "    n = len(y_pred)\n",
    "    if model_name == \"ARIMA\":\n",
    "        p = 1\n",
    "    elif model_name == \"VectorARIMA\":\n",
    "        p = len(feature_cols)\n",
    "        \n",
    "    adj_r2 = 1-(1-r2)*(n-1)/(n-p-1)\n",
    "\n",
    "    # Use dict to store each result\n",
    "    results = {\n",
    "        \"Model\": model_name,\n",
    "        \"P_train\": partition_num_train,\n",
    "        \"P_test\": partition_num_test,\n",
    "        \"Proportion\": proportion,\n",
    "        \"RMSE\": rmse,\n",
    "        \"MAPE\": mape,\n",
    "        \"MAE\": mae,\n",
    "        \"MSLE\": msle,\n",
    "        \"Variance\": exp_var,\n",
    "        \"R2\": r2,\n",
    "        \"Adjusted_R2\": adj_r2,\n",
    "        \"Time\": train_time,\n",
    "    }\n",
    "    return results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "id": "electric-singer",
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Description: Transform a Spark Row type list to pandas dataframe \n",
    "Args:\n",
    "    row_list: Data in pandas dataframe\n",
    "    column_names: Column names will display in pandas dataframe. The format need to be a list\n",
    "    \n",
    "Return: \n",
    "    pandas_df: Data in pandas dataframe\n",
    "'''\n",
    "def row2Pandasdf(row_list, column_names):\n",
    "    pandas_df = pd.DataFrame(columns = column_names)\n",
    "    \n",
    "    # each_row is Row() type in Spark\n",
    "    for each_row in row_list:\n",
    "        pandas_df = pandas_df.append(each_row.asDict(), ignore_index=True)\n",
    "    return pandas_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "id": "timely-schedule",
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Description: Local mode on Spark using Scikit-learn\n",
    "Args:\n",
    "    train_data: Train data in Spark datafram\n",
    "    test_data: Test data in Spark datafram\n",
    "    partition_num_train: Number of partition of Train data\n",
    "    partition_num_test: Number of partition of Test data\n",
    "    model_name: specify which model to make prediction\n",
    "    \n",
    "Return: \n",
    "    results: All the evaluation metrics in a dict\n",
    "'''\n",
    "def localMode(train_data, test_data, partition_num_train, partition_num_test, model_name):\n",
    "    \n",
    "    # Transform Train/Test to RDD type, manually set partition number\n",
    "    train_rdd = train_data.orderBy(\"id\").rdd.coalesce(partition_num_train)\n",
    "    test_rdd  = test_data.orderBy(\"id\").rdd.coalesce(partition_num_test)\n",
    "    \n",
    "    # Collect all the models which generated from each partition, to driver\n",
    "    if model_name == \"ARIMA\":\n",
    "        models = train_rdd.mapPartitions(partitionToPandas).map(buildARIMA).collect()\n",
    "    elif model_name == \"VectorARIMA\":\n",
    "        models = train_rdd.mapPartitions(partitionToPandas).map(buildVectorARIMA).collect()\n",
    "    else:\n",
    "        return \"Wrong model name\"\n",
    "    \n",
    "    train_time = max([model[1] for model in models])\n",
    "    models = [model[0] for model in models]\n",
    "    # broadcast models\n",
    "    broadcast_models = sc.broadcast(models)\n",
    "\n",
    "    # Transform each partition of test_rdd to pandas dataframe, then make prediction on each partition, then merge the results in a single list\n",
    "    y_pred = test_rdd.mapPartitions(partitionToPandas).map(lambda x: makePrediction(x, broadcast_models, model_name)).reduce(lambda x,y: x+y)\n",
    "\n",
    "    # Get the label of test data. (Row() type also works for calculating evaluation metrics)\n",
    "    y_test = test_data.select(\"NEXT_BTC_CLOSE\").collect()\n",
    "    \n",
    "    # Generate a pandas dataframe on predictions. Can help to plot graph easier later.\n",
    "    y_test_rows = test_data.select(\"Timestamp\",\"Close\").collect()\n",
    "    y_df = row2Pandasdf(y_test_rows, [\"Timestamp\",\"Close\"])\n",
    "    \n",
    "    # Add prediction to y_test_df\n",
    "    y_df[\"prediction\"] = y_pred\n",
    "    \n",
    "    # Plot the prediction\n",
    "    #line_plot(y_df.set_index(\"Timestamp\"), model_name)\n",
    "    \n",
    "    # Calculate evaluation metrics\n",
    "    results = evaluationAssemble(y_test, y_pred, partition_num_train, partition_num_test, train_time, model_name)\n",
    "    return results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "id": "split-broadway",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'Model': 'VectorARIMA',\n",
       " 'P_train': 3,\n",
       " 'P_test': 1,\n",
       " 'Proportion': 0.999,\n",
       " 'RMSE': 99.30026088094228,\n",
       " 'MAPE': 0.00146075831591763,\n",
       " 'MAE': 85.67083008310995,\n",
       " 'MSLE': 2.861754882841444e-06,\n",
       " 'Variance': -0.13429644874403057,\n",
       " 'R2': -2.8299854231302666,\n",
       " 'Adjusted_R2': -3.6185118337747335,\n",
       " 'Time': 0.009519815444946289}"
      ]
     },
     "execution_count": 88,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Only use VectorARIMA\n",
    "# model_name == \"ARIMA\" or \"VectorARIMA\"\n",
    "localMode(train_data, test_data, 3, 1, \"VectorARIMA\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "id": "administrative-annual",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'Model': 'ARIMA',\n",
       " 'P_train': 3,\n",
       " 'P_test': 1,\n",
       " 'Proportion': 0.999,\n",
       " 'RMSE': 854.7430334750338,\n",
       " 'MAPE': 0.014540466125094412,\n",
       " 'MAE': 853.2356479229529,\n",
       " 'MSLE': 0.00021529217008165193,\n",
       " 'Variance': -7.221118289546524e-06,\n",
       " 'R2': -282.77065439343886,\n",
       " 'Adjusted_R2': -289.8649207532748,\n",
       " 'Time': 0.49}"
      ]
     },
     "execution_count": 89,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Only use ARIMA\n",
    "# model_name == \"ARIMA\" or \"VectorARIMA\"\n",
    "localMode(train_data, test_data, 3, 1, \"ARIMA\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "atomic-harris",
   "metadata": {},
   "source": [
    "## 6. Time Series Cross Validation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "caroline-scale",
   "metadata": {},
   "outputs": [],
   "source": [
    "'''\n",
    "Description: Cross Validation on Time Series data\n",
    "Args:\n",
    "    dataSet: The dataSet which needs to be splited\n",
    "    cv_info: The type of Cross Validation\n",
    "    ml_model: The module to use\n",
    "    partition_num_train: Number of partition of Train data\n",
    "    partition_num_test: Number of partition of Test data\n",
    "Return: \n",
    "    tsCv_df: All the splits performance of each model in a pandas dataframe\n",
    "'''\n",
    "def tsCrossValidation(dataSet, ml_model, partition_num_train, partition_num_test, cv_info):\n",
    "    \n",
    "    # Get the number of samples\n",
    "    num = dataSet.count()\n",
    "    \n",
    "    # Save results in a list\n",
    "    result_lst = []\n",
    "        \n",
    "    # Identify the type of Cross Validation \n",
    "    if cv_info['cv_type'] == 'mulTs':\n",
    "        split_position_df = mulTsCrossValidation(num, cv_info['kSplits'])\n",
    "    elif cv_info['cv_type'] == 'blkTs':\n",
    "        split_position_df = blockedTsCrossValidation(num, cv_info['kSplits'])\n",
    "    elif cv_info['cv_type'] == 'wfTs':\n",
    "        split_position_df = wfTsCrossValidation(num, cv_info['min_obser'], cv_info['expand_window'])\n",
    "\n",
    "\n",
    "    for position in split_position_df.itertuples():\n",
    "        # Get the start/split/end position from a kind of Time Series Cross Validation\n",
    "        start = getattr(position, 'start')\n",
    "        splits = getattr(position, 'split')\n",
    "        end = getattr(position, 'end')\n",
    "        idx  = getattr(position, 'Index')\n",
    "\n",
    "        # Train/Test size\n",
    "        train_size = splits - start\n",
    "        test_size = end - splits\n",
    "\n",
    "        # Get training data and test data\n",
    "        train_data = dataSet.filter(F.col(\"id\").between(start, splits-1))\n",
    "        test_data = dataSet.filter(F.col(\"id\").between(splits, end-1))\n",
    "\n",
    "        # Cache it\n",
    "        train_data.cache()\n",
    "        test_data.cache()\n",
    "        \n",
    "        # train the local mode\n",
    "        if ml_model == \"VectorARIMA\":\n",
    "            results = localMode(train_data, test_data, partition_num_train, partition_num_test, \"VectorARIMA\")\n",
    "        elif ml_model == \"ARIMA\":\n",
    "            results = localMode(train_data, test_data, partition_num_train, partition_num_test, \"ARIMA\")\n",
    "        else:\n",
    "            return \"Wrong model name\"\n",
    "        \n",
    "        # Store each splits result\n",
    "        result_lst.append(results)\n",
    "            \n",
    "        # Release Cache\n",
    "        train_data.unpersist()\n",
    "        test_data.unpersist()\n",
    "\n",
    "    # Transform dict to pandas dataframe\n",
    "    tsCv_df = pd.DataFrame(result_lst)\n",
    "    return tsCv_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "insured-reading",
   "metadata": {},
   "outputs": [],
   "source": [
    "## Cross Validation Parameter\n",
    "# Multiple Splits Time Series Cross Validation\n",
    "mul_cv = {'cv_type':'mulTs',\n",
    "          'kSplits': 5}\n",
    "\n",
    "# Blocked Time Series Cross Validation\n",
    "blk_cv = {'cv_type':'blkTs',\n",
    "          'kSplits': 10}\n",
    "\n",
    "# Walk Forward Validation\n",
    "wf_cv = {'cv_type':'wfTs',\n",
    "         'min_obser': 41710,\n",
    "         'expand_window': 10}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "matched-guard",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Vector ARIMA CV\n",
    "arima_mul_cv = tsCrossValidation(dataset, \"VectorARIMA\", 3, 1, mul_cv)\n",
    "arima_blk_cv = tsCrossValidation(dataset, \"VectorARIMA\", 3, 1, blk_cv)\n",
    "arima_wf_cv = tsCrossValidation(dataset, \"VectorARIMA\", 3, 1, wf_cv)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "recognized-referral",
   "metadata": {},
   "outputs": [],
   "source": [
    "# ARIMA CV\n",
    "var_mul_cv = tsCrossValidation(dataset, \"ARIMA\", 3, 1, mul_cv)\n",
    "var_blk_cv = tsCrossValidation(dataset, \"ARIMA\", 3, 1, blk_cv)\n",
    "var_wf_cv = tsCrossValidation(dataset, \"ARIMA\", 3, 1, wf_cv)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "id": "wicked-underwear",
   "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>Model</th>\n",
       "      <th>RMSE</th>\n",
       "      <th>MAPE</th>\n",
       "      <th>MAE</th>\n",
       "      <th>Variance</th>\n",
       "      <th>R2</th>\n",
       "      <th>Adjusted_R2</th>\n",
       "      <th>Time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>VectorARIMA</td>\n",
       "      <td>3687.882986</td>\n",
       "      <td>0.057466</td>\n",
       "      <td>3230.708155</td>\n",
       "      <td>-0.715322</td>\n",
       "      <td>-2.758465</td>\n",
       "      <td>-2.762250</td>\n",
       "      <td>0.003124</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>VectorARIMA</td>\n",
       "      <td>1532.644759</td>\n",
       "      <td>0.024457</td>\n",
       "      <td>1335.276478</td>\n",
       "      <td>-0.902524</td>\n",
       "      <td>-3.814275</td>\n",
       "      <td>-3.854976</td>\n",
       "      <td>0.001555</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>VectorARIMA</td>\n",
       "      <td>65.436719</td>\n",
       "      <td>0.000900</td>\n",
       "      <td>52.796204</td>\n",
       "      <td>-0.034613</td>\n",
       "      <td>-1.622355</td>\n",
       "      <td>-10.800596</td>\n",
       "      <td>0.005222</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARIMA</td>\n",
       "      <td>3386.892678</td>\n",
       "      <td>0.051853</td>\n",
       "      <td>2890.069438</td>\n",
       "      <td>-0.163985</td>\n",
       "      <td>-2.034490</td>\n",
       "      <td>-2.034926</td>\n",
       "      <td>3.766000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARIMA</td>\n",
       "      <td>1710.349874</td>\n",
       "      <td>0.027724</td>\n",
       "      <td>1500.745420</td>\n",
       "      <td>-0.051543</td>\n",
       "      <td>-7.588432</td>\n",
       "      <td>-7.598730</td>\n",
       "      <td>0.371000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARIMA</td>\n",
       "      <td>809.841188</td>\n",
       "      <td>0.013782</td>\n",
       "      <td>808.719076</td>\n",
       "      <td>0.001849</td>\n",
       "      <td>-416.845531</td>\n",
       "      <td>-469.076223</td>\n",
       "      <td>4.280000</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "         Model         RMSE      MAPE          MAE  Variance          R2  \\\n",
       "0  VectorARIMA  3687.882986  0.057466  3230.708155 -0.715322   -2.758465   \n",
       "0  VectorARIMA  1532.644759  0.024457  1335.276478 -0.902524   -3.814275   \n",
       "0  VectorARIMA    65.436719  0.000900    52.796204 -0.034613   -1.622355   \n",
       "0        ARIMA  3386.892678  0.051853  2890.069438 -0.163985   -2.034490   \n",
       "0        ARIMA  1710.349874  0.027724  1500.745420 -0.051543   -7.588432   \n",
       "0        ARIMA   809.841188  0.013782   808.719076  0.001849 -416.845531   \n",
       "\n",
       "   Adjusted_R2      Time  \n",
       "0    -2.762250  0.003124  \n",
       "0    -3.854976  0.001555  \n",
       "0   -10.800596  0.005222  \n",
       "0    -2.034926  3.766000  \n",
       "0    -7.598730  0.371000  \n",
       "0  -469.076223  4.280000  "
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Define what model_info and evaluators in the Model Comparison Table\n",
    "model_info = ['Model']\n",
    "evaluator_lst = ['RMSE','MAPE','MAE','Variance','R2','Adjusted_R2','Time']\n",
    "\n",
    "# The the Cross Validation results would like to compare\n",
    "comparison_lst = [arima_mul_cv, arima_blk_cv, arima_wf_cv, var_mul_cv, var_blk_cv, var_wf_cv]\n",
    "pd.concat([modelComparison(cv_result,model_info,evaluator_lst) for cv_result in comparison_lst])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "editorial-blame",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}