preprocess_bitcoin_pyspark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Apache Spark\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions as F\n",
    "\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import VectorAssembler,StandardScaler\n",
    "from pyspark.ml.regression import LinearRegression, GeneralizedLinearRegression, DecisionTreeRegressor, RandomForestRegressor, GBTRegressor\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "\n",
    "# Python\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from itertools import product\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start a SparkSession\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .master(\"local[*]\") \\\n",
    "    .appName(\"Bitcoin Prediction\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read csv file\n",
    "filename = \"bitstampUSD_1-min_data_2012-01-01_to_2021-03-31.csv\"\n",
    "dataset = spark.read.format(\"csv\") \\\n",
    "          .option(\"inferSchema\",'True') \\\n",
    "          .option(\"header\",True) \\\n",
    "          .load(filename)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----+----+----+-----+------------+------------------+--------------+\n",
      "| Timestamp|Open|High| Low|Close|Volume_(BTC)| Volume_(Currency)|Weighted_Price|\n",
      "+----------+----+----+----+-----+------------+------------------+--------------+\n",
      "|1325317920|4.39|4.39|4.39| 4.39|  0.45558087|2.0000000193000003|          4.39|\n",
      "|1325317980|null|null|null| null|        null|              null|          null|\n",
      "|1325318040|null|null|null| null|        null|              null|          null|\n",
      "|1325318100|null|null|null| null|        null|              null|          null|\n",
      "|1325318160|null|null|null| null|        null|              null|          null|\n",
      "+----------+----+----+----+-----+------------+------------------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dataset.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### fill NaN values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Open',\n",
       " 'High',\n",
       " 'Low',\n",
       " 'Close',\n",
       " 'Volume_(BTC)',\n",
       " 'Volume_(Currency)',\n",
       " 'Weighted_Price']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# select cols to fill NaN\n",
    "na_cols = dataset.columns\n",
    "na_cols = na_cols[1:]\n",
    "na_cols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Window"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### interpolation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "def interpolation_fill(dataset,order_col, value_col):\n",
    "    # fill not nan value with null to fetch the last/fisrt not null val\n",
    "    #dataset = dataset.withColumn(value_col,F.when(F.isnan(F.col(value_col)),None).otherwise(F.col(value_col)))\n",
    "    # idx_not_na -> not null value column's idx number\n",
    "    dataset=dataset.withColumn('idx_not_na',F.when(F.col(value_col).isNotNull(),F.col(\"idx\")))\n",
    "    \n",
    "    w_start = Window.orderBy(order_col).rowsBetween(Window.unboundedPreceding,-1)\n",
    "    # start_val -> last not null value\n",
    "    dataset = dataset.withColumn('start_val',F.last(value_col,True).over(w_start))\n",
    "    # start_idx -> last not null value idx\n",
    "    dataset = dataset.withColumn('start_idx',F.last('idx_not_na',True).over(w_start))\n",
    "    \n",
    "    w_end = Window.orderBy(order_col).rowsBetween(0,Window.unboundedFollowing)\n",
    "    # end_val -> next not null value\n",
    "    dataset = dataset.withColumn('end_val',F.first(value_col,True).over(w_end))\n",
    "    # end_idx -> next not null value idx\n",
    "    dataset = dataset.withColumn('end_idx',F.first('idx_not_na',True).over(w_end))\n",
    "    \n",
    "    # diff_idx -> numbers of missing value columns\n",
    "    dataset = dataset.withColumn('diff_idx',F.col('end_idx')-F.col('start_idx'))\n",
    "    dataset = dataset.withColumn('curr_idx',F.col('diff_idx')-(F.col('end_idx')-F.col('idx')))\n",
    "    \n",
    "    #fill the null values with the function\n",
    "    lin_interp_func = (F.col('start_val')+(F.col('end_val')-F.col('start_val'))/F.col('diff_idx')*F.col('curr_idx'))\n",
    "    dataset = dataset.withColumn(value_col,F.when(F.col(value_col).isNull(),lin_interp_func).otherwise(F.col(value_col)))\n",
    "    \n",
    "    # dropping all the unnecessary columns\n",
    "    dataset = dataset.select([order_col]+na_cols+['idx'])\n",
    "    return dataset\n",
    "\n",
    "#add an idx column 9th\n",
    "w = Window.orderBy('Timestamp')\n",
    "dataset = dataset.withColumn('idx',F.row_number().over(w))\n",
    "\n",
    "for value_col in na_cols:\n",
    "    dataset = interpolation_fill(dataset, 'Timestamp', value_col)\n",
    "dataset = dataset.drop('idx')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### forward fill (out final method)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "def fwfill(dataset,order_col, value_col):\n",
    "    w_start = Window.orderBy(order_col).rowsBetween(Window.unboundedPreceding,-1)\n",
    "    dataset = dataset.withColumn(value_col,F.when(F.col(value_col).isNull(),F.last(value_col,True).over(w_start)).otherwise(F.col(value_col)))\n",
    "    dataset = dataset.select([order_col]+na_cols)\n",
    "    return dataset\n",
    "     \n",
    "for value_col in na_cols:\n",
    "    dataset = fwfill(dataset, 'Timestamp', value_col)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### backward fill"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def bwfill(dataset,order_col, value_col):\n",
    "    w_end = Window.orderBy(order_col).rowsBetween(0,Window.unboundedFollowing)\n",
    "    dataset = dataset.withColumn(value_col,F.when(F.col(value_col).isNull(),F.first(value_col,True).over(w_end)).otherwise(F.col(value_col)))\n",
    "    dataset = dataset.select([order_col]+na_cols)\n",
    "    return dataset\n",
    "    \n",
    "for value_col in na_cols:\n",
    "    dataset = bwfill(dataset, 'Timestamp', value_col)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### fill with mean values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def fill_mean(dataset,order_col, value_col):\n",
    "    w_start = Window.orderBy(order_col).rowsBetween(Window.unboundedPreceding,-1)\n",
    "    # start_val -> last not null value\n",
    "    dataset = dataset.withColumn('start_val',F.last(value_col,True).over(w_start))\n",
    "    \n",
    "    w_end = Window.orderBy(order_col).rowsBetween(0,Window.unboundedFollowing)\n",
    "    # end_val -> next not null value\n",
    "    dataset = dataset.withColumn('end_val',F.first(value_col,True).over(w_end))\n",
    "    \n",
    "    #fill the null values with the function\n",
    "    mean_func = (F.col('start_val')+F.col('end_val'))/2\n",
    "    dataset = dataset.withColumn(value_col,F.when(F.col(value_col).isNull(),mean_func).otherwise(F.col(value_col)))\n",
    "    \n",
    "    dataset = dataset.select([order_col]+na_cols)\n",
    "    return dataset\n",
    "     \n",
    "for value_col in na_cols:\n",
    "    dataset = fill_mean(dataset, 'Timestamp', value_col)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### select time period"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col\n",
    "dataset = dataset.withColumn(\"Timestamp\", col('Timestamp').cast(\"timestamp\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# select the time interval\n",
    "from pyspark.sql.functions import unix_timestamp,from_unixtime\n",
    "start = F.unix_timestamp(F.lit('2011-12-31 08:00:00')).cast('timestamp')\n",
    "\n",
    "dataset = dataset.filter(F.col(\"Timestamp\") >= start)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----+----+----+-----+------------------+-----------------+--------------+\n",
      "|          Timestamp|Open|High| Low|Close|      Volume_(BTC)|Volume_(Currency)|Weighted_Price|\n",
      "+-------------------+----+----+----+-----+------------------+-----------------+--------------+\n",
      "|2011-12-31 08:00:00|4.39|4.39|4.39| 4.39|1.2513033658995818|5.493221776299164|          4.39|\n",
      "|2011-12-31 08:01:00|4.39|4.39|4.39| 4.39|1.3507686778870294|5.929874495924059|          4.39|\n",
      "|2011-12-31 08:02:00|4.39|4.39|4.39| 4.39| 1.450233989874477|6.366527215548954|          4.39|\n",
      "|2011-12-31 08:03:00|4.39|4.39|4.39| 4.39|1.5496993018619247| 6.80317993517385|          4.39|\n",
      "|2011-12-31 08:04:00|4.39|4.39|4.39| 4.39|1.6491646138493725|7.239832654798745|          4.39|\n",
      "+-------------------+----+----+----+-----+------------------+-----------------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dataset.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### resample 30min"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "#add a new column for re-sample purpose\n",
    "from pyspark.sql import Window\n",
    "seconds = 60*30\n",
    "seconds_window = F.from_unixtime(F.unix_timestamp('Timestamp') - F.unix_timestamp('Timestamp') % seconds)\n",
    "dataset = dataset.withColumn('30_min_window', seconds_window)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Open',\n",
       " 'High',\n",
       " 'Low',\n",
       " 'Close',\n",
       " 'Volume_(BTC)',\n",
       " 'Volume_(Currency)',\n",
       " 'Weighted_Price']"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# for renaming purpose\n",
    "cols = dataset.columns\n",
    "cols = cols[1:-1]\n",
    "cols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "#agg and mean all values\n",
    "dataset = dataset.groupBy('30_min_window').mean()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- 30_min_window: string (nullable = true)\n",
      " |-- avg(Open): double (nullable = true)\n",
      " |-- avg(High): double (nullable = true)\n",
      " |-- avg(Low): double (nullable = true)\n",
      " |-- avg(Close): double (nullable = true)\n",
      " |-- avg(Volume_(BTC)): double (nullable = true)\n",
      " |-- avg(Volume_(Currency)): double (nullable = true)\n",
      " |-- avg(Weighted_Price): double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dataset.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset = dataset.withColumn('Timestamp',F.to_timestamp('30_min_window'))\n",
    "dataset = dataset.orderBy('Timestamp')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+---------+---------+--------+----------+------------------+----------------------+-------------------+-------------------+\n",
      "|      30_min_window|avg(Open)|avg(High)|avg(Low)|avg(Close)| avg(Volume_(BTC))|avg(Volume_(Currency))|avg(Weighted_Price)|          Timestamp|\n",
      "+-------------------+---------+---------+--------+----------+------------------+----------------------+-------------------+-------------------+\n",
      "|2011-12-31 08:00:00|     4.39|     4.39|    4.39|      4.39| 2.693550389717573|    11.824686210860147|               4.39|2011-12-31 08:00:00|\n",
      "|2011-12-31 08:30:00|     4.39|     4.39|    4.39|      4.39| 5.677509749341007|    24.924267799607012|               4.39|2011-12-31 08:30:00|\n",
      "|2011-12-31 09:00:00|     4.39|     4.39|    4.39|      4.39| 8.661469108964436|     38.02384938835389|               4.39|2011-12-31 09:00:00|\n",
      "|2011-12-31 09:30:00|     4.39|     4.39|    4.39|      4.39|11.645428468587864|     51.12343097710073|               4.39|2011-12-31 09:30:00|\n",
      "|2011-12-31 10:00:00|     4.39|     4.39|    4.39|      4.39|14.629387828211296|      64.2230125658476|               4.39|2011-12-31 10:00:00|\n",
      "+-------------------+---------+---------+--------+----------+------------------+----------------------+-------------------+-------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dataset.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from functools import reduce\n",
    "\n",
    "old_cols = dataset.columns\n",
    "old_cols = old_cols[1:-2]\n",
    "\n",
    "dataset = reduce(lambda dataset, idx: dataset.withColumnRenamed(old_cols[idx], cols[idx]),\\\n",
    "                 range(len(old_cols)), dataset)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+----+-----+------------------+------------------+-------------------+-------------------+\n",
      "|Open|High| Low|Close|      Volume_(BTC)| Volume_(Currency)|avg(Weighted_Price)|          Timestamp|\n",
      "+----+----+----+-----+------------------+------------------+-------------------+-------------------+\n",
      "|4.39|4.39|4.39| 4.39| 2.693550389717573|11.824686210860147|               4.39|2011-12-31 08:00:00|\n",
      "|4.39|4.39|4.39| 4.39| 5.677509749341007|24.924267799607012|               4.39|2011-12-31 08:30:00|\n",
      "|4.39|4.39|4.39| 4.39| 8.661469108964436| 38.02384938835389|               4.39|2011-12-31 09:00:00|\n",
      "|4.39|4.39|4.39| 4.39|11.645428468587864| 51.12343097710073|               4.39|2011-12-31 09:30:00|\n",
      "|4.39|4.39|4.39| 4.39|14.629387828211296|  64.2230125658476|               4.39|2011-12-31 10:00:00|\n",
      "+----+----+----+-----+------------------+------------------+-------------------+-------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dataset = dataset.drop('30_min_window')\n",
    "dataset.show(5)"
   ]
  }
 ],
 "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}