feature_engineering.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "floral-breath",
   "metadata": {},
   "source": [
    "# Feature Engineering"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cleared-syracuse",
   "metadata": {},
   "source": [
    "* **Description**: COMP4103(Big Data)--Group Project\n",
    "* **Author**: Aaron\n",
    "* **Version**: 0.1"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "about-request",
   "metadata": {},
   "source": [
    "## 1. load packages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "frozen-greensboro",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Apache Spark\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.functions import monotonically_increasing_id\n",
    "from pyspark.sql.window import Window"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "sitting-mexican",
   "metadata": {},
   "source": [
    "## 2. Create a Spark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "prescription-syndrome",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start a SparkSession\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .master(\"local[*]\") \\\n",
    "    .appName(\"data preprocessing\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "nervous-obligation",
   "metadata": {},
   "source": [
    "## 3. Data combination"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ordered-contribution",
   "metadata": {},
   "outputs": [],
   "source": [
    "# DataSet\n",
    "bitcoin_data = \"bitcoin_10y_1min_interpolate.csv\"\n",
    "blockChain_data = \"blockChain_10y_1min_interpolate.csv\"\n",
    "\n",
    "df = spark.read.format(\"csv\") \\\n",
    "          .option(\"inferSchema\",'True') \\\n",
    "          .option(\"header\",True) \\\n",
    "          .load(bitcoin_data) \\\n",
    "          .withColumn(\"id\", F.row_number().over(Window.orderBy(F.monotonically_increasing_id()))-1)\n",
    "\n",
    "blockChain_df = spark.read.format(\"csv\") \\\n",
    "                     .option(\"inferSchema\",'True') \\\n",
    "                     .option(\"header\",True) \\\n",
    "                     .load(blockChain_data) \\\n",
    "                     .withColumn(\"id\", F.row_number().over(Window.orderBy(F.monotonically_increasing_id()))-1) \n",
    "\n",
    "# join data\n",
    "df = df.join(blockChain_df, on=['id','Timestamp'], how='inner')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "stock-fusion",
   "metadata": {},
   "source": [
    "## 4. Generate the label column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "engaged-atlantic",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Add a NEXT_BTC_CLOSE represent next step bitcoin price as the label column\n",
    "# https://sparkbyexamples.com/pyspark/pyspark-window-functions/\n",
    "df = df.withColumn(\"NEXT_BTC_CLOSE\", F.lag(\"Close\", offset=-1) \\\n",
    "       .over(Window.orderBy(\"id\"))) \\\n",
    "       .dropna()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "comfortable-convergence",
   "metadata": {},
   "source": [
    "## 5. Generate financial indicators"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "champion-violence",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generate additional valuable features\n",
    "\n",
    "# Rate of Change allows investors to spot security momentum and other trends\n",
    "# Typically a 12-day Rate-of-Change is used but for simplicity, I used it for every 30-min interval\n",
    "df = df.withColumn(\"Rate_of_Change\", (F.col(\"NEXT_BTC_CLOSE\") / F.col(\"Previous_close\") - 1) * 100)\n",
    "\n",
    "# computing Simple Moving Averages \n",
    "# Adapted from: https://stackoverflow.com/questions/45806194/pyspark-rolling-average-using-timeseries-data\n",
    "def simple_moving_average(df, period, col=\"NEXT_BTC_CLOSE\", orderby=\"id\"):\n",
    "    \n",
    "    df = df.withColumn(f\"SMA_{period}\", F.avg(col) \\\n",
    "           .over(Window.orderBy(orderby) \\\n",
    "           .rowsBetween(-period,0))) \n",
    "    return df\n",
    "\n",
    "#MA number 5/7/10/20/50/100/200 days;\n",
    "MA5 = 60 * 24 * 5\n",
    "MA7 = 60 * 24 * 7\n",
    "MA10 = 60 * 24 * 10\n",
    "MA20 = 60 * 24 * 20\n",
    "MA50 = 60 * 24 * 50\n",
    "MA100 = 60 * 24 * 100\n",
    "\n",
    "# periods selected based on this article: \n",
    "# https://www.investopedia.com/ask/answers/122414/what-are-most-common-periods-used-creating-moving-average-\n",
    "# ma-lines.asp#:~:text=Traders%20and%20market%20analysts%20commonly,averages%20are%20the%20most%20common.\n",
    "\n",
    "# to analyze short-term trends\n",
    "df = simple_moving_average(df, MA5) # these might have to be 240 - 1 actually\n",
    "df = simple_moving_average(df, MA7)\n",
    "df = simple_moving_average(df, MA10)\n",
    "df = simple_moving_average(df, MA20)\n",
    "df = simple_moving_average(df, MA50)\n",
    "# to analyze long-term trends\n",
    "df = simple_moving_average(df, MA100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "tracked-times",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Save the complete data to a CSV file\n",
    "df.write.option(\"header\",True).csv(\"complete_10y_1min_interpolate.csv\")"
   ]
  }
 ],
 "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
}