code/python_data_processing_walkthrough.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "source": [
    "# Moving Big Data Predict \n",
    "© Explore Data Science Academy\n",
    "\n",
    "## Data Processing Guide:  Historical Data Processing"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Overview \n",
    "\n",
    "This notebook is provided to help you complete the data processing component of you data pipeline, formed within the Moving Big Data predict. \n",
    "\n",
    "Several completed helper functions are initially provided, with the expectation that you will implement the main `data_processing` function. \n",
    "\n",
    "At the conclusion of testing, you are expected to convert this notebook into a `.py` script file that can be run whenever your pipeline is invoked.  "
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Imports\n",
    "\n",
    "We only make use of the `pandas` library in order to perform the data processing. You should ensure that the AMI and resulting EC2 instance used to run you data processing has this dependency met.  "
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "source": [
    "import pandas as pd"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Variables\n",
    "\n",
    "We initially define three path variables that are used within the data processing: \n",
    "1. Source data path: The path to the comany `.csv` files used within the data processing. \n",
    "2. Saved data path: Path to which the resulting aggregated `historical_stock_data.csv`should be saved. \n",
    "3. Path of index file: Path to the index file that defines companies whose data needs to be aggregated. Within the predict, we make use of the `top_companies.txt` file as this index. \n",
    "\n",
    "We declare these variables in two sets. The first, local, set of variables are defined to enable local testing of the functions formed. These paths can be adapted to suite testing needs. \n",
    "\n",
    "The second set of variables represents the direcory structure within the EC2 instance where data processing will occur during the pipeline's opperation. These paths should not be altered if at all possible, as doing so will require changes to other components to the predict. "
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "source": [
    "# Local file directories\n",
    "# These paths can be updated to meet your testing environment needs.\n",
    "source_path = \"../data/Stocks/\"\n",
    "save_path = \"../data/Output/\"\n",
    "index_file_path = '../data/top_companies.txt'\n",
    "\n",
    "# # EC2 file directories\n",
    "# Do not change these paths\n",
    "# source_path = \"/home/ec2-user/s3-drive/Stocks/\"\n",
    "# save_path = \"/home/ec2-user/s3-drive/Output/\"\n",
    "# index_file_path = \"/home/ec2-user/s3-drive/CompanyNames/top_companies.txt\""
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Helper Functions"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "Create a function that returns a list of all the companies (represented by their `csv` files) selected for inclusion within the data processing pipeline. "
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "source": [
    "def extract_companies_from_index(index_file_path):\n",
    "    \"\"\"Generate a list of company files that need to be processed. \n",
    "\n",
    "    Args:\n",
    "        index_file_path (str): path to index file\n",
    "\n",
    "    Returns:\n",
    "        list: Names of company names. \n",
    "    \"\"\"\n",
    "    company_file = open(index_file_path, \"r\")\n",
    "    contents = company_file.read()\n",
    "    contents = contents.replace(\"'\",\"\")\n",
    "    contents_list = contents.split(\",\")\n",
    "    cleaned_contents_list = [item.strip() for item in contents_list]\n",
    "    company_file.close()\n",
    "    return cleaned_contents_list"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "Create a function that attaches the source directory to each company csv file selected for processing."
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "source": [
    "def get_path_to_company_data(list_of_companies, source_data_path):\n",
    "    \"\"\"Creates a list of the paths to the company data\n",
    "       that will be processed\n",
    "\n",
    "    Args:\n",
    "        list_of_companies (list): Extracted `.csv` file names of companies whose data needs to be processed.\n",
    "        source_data_path (str): Path to where the company `.csv` files are stored. \n",
    "\n",
    "    Returns:\n",
    "        [type]: [description]\n",
    "    \"\"\"\n",
    "    path_to_company_data = []\n",
    "    for file_name in list_of_companies:\n",
    "        path_to_company_data.append(source_data_path + file_name)\n",
    "    return path_to_company_data"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "Create a function that saves a pandas dataframe in csv format"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "source": [
    "def save_table(dataframe, output_path, file_name, header):\n",
    "    \"\"\"Saves an input pandas dataframe as a CSV file according to input parameters.\n",
    "\n",
    "    Args:\n",
    "        dataframe (pandas.dataframe): Input dataframe.\n",
    "        output_path (str): Path to which the resulting `.csv` file should be saved. \n",
    "        file_name (str): The name of the output `.csv` file. \n",
    "        header (boolean): Whether to include column headings in the output file.\n",
    "    \"\"\"\n",
    "    print(f\"Path = {output_path}, file = {file_name}\")\n",
    "    dataframe.to_csv(output_path + file_name + \".csv\", index=False, header=header)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Data Processing Function"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "The students are now expected to form a function that takes as input an array of company names (formed from the `top_companies.txt` file, representing multiple `.csv` files), and output a single `.csv` file containing the combined collection of data from these files, represented as rows, along with two additional summary columns providing extra context for each company entry.    \n",
    "\n",
    "Instructions\n",
    "\n",
    "- The target output file should be named `historical_stock_data.csv`.\n",
    "- The output `csv` file must have the following schema:\n",
    "\n",
    "    |Column Name|DataType|\n",
    "    |---|---|\n",
    "    |stock_date|datetime|\n",
    "    |open_value|float64|\n",
    "    |high_value|float64|\n",
    "    |low_value|float64|\n",
    "    |close_value|float64|\n",
    "    |volume_traded|int64|\n",
    "    |daily_percent_change|float64|\n",
    "    |value_change|float64|\n",
    "    |company_name|string|\n",
    "\n",
    "    <br>\n",
    "\n",
    "- The output file should not contain any column headers.\n",
    "- Some input `csv` files may be corrupted. As such, be careful of exceptions when forming the processing function.\n",
    "- The `daily_percent_change` and `value_change` columns must be calculated based off the open_value and the close_value.\n",
    "  * `daily_percent_change` = ((`Close` - `Open`)/`Open`) * 100\n",
    "  * `value_change` = `Close` - `Open`\n",
    " "
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "source": [
    "def data_processing(file_paths, output_path):\n",
    "    \"\"\"Process and collate company csv file data for use within the data processing component of the formed data pipeline.  \n",
    "\n",
    "    Args:\n",
    "        file_paths (list[str]): A list of paths to the company csv files that need to be processed. \n",
    "        output_path (str): The path to save the resulting csv file to.  \n",
    "    \"\"\"\n",
    "\n",
    "    # INSERT YOUR CODE HERE\n"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Main Program Flow"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "With the data processing function completed, you need to assemble the provided functions together to form a script (standalone `.py` file) that can be run via a bash command called during your data pipeline's execution.  \n",
    "\n",
    "The following scaffolding is provided to help in this regard: "
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "if __name__ == \"__main__\":\n",
    "\n",
    "    # Get all file names in source data directory of companies whose data needs to be processed, \n",
    "    # This information is specified within the `top_companies.txt` file. \n",
    "    file_names = extract_companies_from_index(index_file_path)\n",
    "\n",
    "    # Update the company file names to include path information. \n",
    "    path_to_company_data = get_path_to_company_data(file_names, source_path)\n",
    "\n",
    "    # Process company data and create full data output\n",
    "    data_processing(path_to_company_data, save_path)"
   ],
   "outputs": [],
   "metadata": {
    "tags": [
     "outputPrepend"
    ]
   }
  }
 ],
 "metadata": {
  "kernelspec": {
   "name": "python3",
   "display_name": "Python 3.7.6 64-bit ('base': conda)"
  },
  "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.6"
  },
  "interpreter": {
   "hash": "f057f675914b11b512bf379bae07be6e1618e8fe1362d0973cc146d2f4f584aa"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}