Crypto Stats and Tweets Data Pipeline using Airflow

Overview

Crypto Stats and Tweets Data Pipeline using Airflow

Introduction

Project Overview

This project was brought upon through Udacity's nanodegree program.

For the capstone project within the nanodegree, the ultimate goal is to build a data pipeline that uses the technologies and applications covered in the the program.

With the recent rise of crypto currency interests and the evolution of crypto twitter into the media spotlight, revolving my capstone project around these two areas seemed like a good idea.

The ultimate goal of this project is to create both crypto statistics and crypto tweets datasets that can be used in downstream applications.

That goal was accomplished through this project. However, I have further goals for this project, which will be discussed later.

Project Requirements

At least 2 data sources

  • twitter.com accessed through snscrape tweets libary
  • coingecko public API resulting in crypto currency statistical data starting in 2015.

More than 1 million lines of data.

  • The snscrape_tweets_hist dataset has over 1.5 million rows
  • The coin_stats_hist has over 250k rows.

At least two data sources/formats (csv, api, json)

  • Stored in S3 (mkgpublic)
    • mkgpublic/capstone/tweets/tweets.parquet
    • mkgpublic/capstone/crypto/cg_hourly.csv

Data Ingestion Process

Tweets

The original data ingestion process ran into few snafus. As I decided to use the twitter API to get the tweets side of the data at first; however, due to limitations within the twitter API, I couldn't get more than 1000 tweets per call.

Thus, I decided to use the snscrape tweets python library instead, which provided a much easier method to get a ton of tweets in a reasonable amount of time.

Through using the snscrape tweets python library, the tweets were gathered running a library function.

The tweets were than stored in a MongoDB database as an intermediary storage solution.

Data was continuously ingested using this process until enough tweets about various crypto currencies was gathered.

After storing the tweets in MongoDB the tweets were then pulled from the MongoDB database, stored in a pandas dataframe and written to the mkgpublic s3 bucket as a parquet file.

Crypto

Using the coingecko api, crypto currency statistical data was pulled and stored in a pandas dataframe.

After storing the data in the pandas df, the data was written to the MongoDB database used for tweets.

Data is continously ingested through this process until enough statistical data about various crypto currencies was stored.

Finally the crypto currency statistical data is pulled from the MongoDB database, stored in a pandas dataframe and written to the mkgpublic s3 bucket as a CSV. *** Note *** I stored the data as a CSV because two sets of data formats were requested. I originally choose to store the crypto stats data as a json file, but even when partitioning the file into several JSON files, the files were too big for airflow to handle. Thus, I went with the csv format.

Crypto Stats and Tweets ELT

Now we get into the udacity capstone data ingestion and processing part of this project.

Ultimately, I choose to follow a similar process to what is in the mkg_airflow repository where I am using airflow to run a sequence of tasks.

Main Scripts

  • dags/tweets_and_crypto_etl.py
  • plugins/helpers/sql_queries.py
  • plugins/operators/stage_redshift.py
  • plugins/operators/load_dimension.py
  • plugins/operators/load_fact.py
  • plugins/helpers/analysis.py
  • plugins/operators/data_quality.py

Data Model

Udacity Capstone Project Data Model
  1. Data is loaded into the staging tables cg_coin_list_stg, snscrape_tweets_stg, and cg_hourly_stg on a Redshift Cluster from the S3 bucket
  2. Date information is loaded into Date Dim
  3. Data is loaded into the cg_coin_list table from cg_coin_list_stg
  4. Data is loaded into coin_stats_hist using a join between date_dim, cg_hourly_stg, and cg_coin_list using date_keys and coin names as parameters to get foreign key allocation
  5. Data is loaded into snscrape_tweets_hist using a join between date_dim, snscrape_tweets_stg and cg_coin_list using date_keys and coin names as parameters to get foreign key allocation

Ultimately, this data model was chosen as the end state will be combining crypto price action with tweet sentiment to determine how the market reacts to price action. So, we need a relationship between the crypto and tweets datasets in order to one day achieve this future state result.

Steps

Airflow Udacity Capstone Dag
  1. Create Redshift Cluster
  2. Create Crypto, Tweets, and Dim Schemas
  3. Create Crypto/Tweets staging and Dim Tables
  4. Staging
  5. Stage Coingecko Token List Mapping Table
  6. Stage Coingecko hourly crypto currency statistical table
  7. Stage snscrape tweets crypto twitter table
  8. Load Dimensions
  9. Load Coingecko Token List Mapping Table
  10. Load Date Dim with date information from Coingecko hourly crypto currency statistical staging table
  11. Load Date Dim with date information from Stage snscrape tweets crypto twitter staging table
  12. Create Fact Tables
  13. Load Fact Tables
  14. Load crypto currency statistics history table
  15. Load snscrape tweets history table
  16. Run Data Quality Checks
  17. Select Statements that make sure data is actually present
  18. Build an Aggregate table with min statistic and max statistic values per month from the coin_stats_hist table
  19. Store resulting dim, fact and aggregate tables in S3
  20. Delete Redshift Cluster

Future Work and Final Thoughts

Some questions for future work:

  • What if the data was increased by 100x.
    • I would use a spark emr cluster to process the data as that would speed up both the data ingestion and the processing parts of the project.
    • This is likely going to happen in my future steps for this project, so ultimately this will be added in future versions.
  • What if the pipelines would be run on a daily basis by 7 am every day.
    • I need a way to get the first part of this process easier. The issue is sometimes either the coingecko or the snscrape tweets api breaks. Thus, if this pipeline would need to be run every day at 7am I would need to fix the initial data ingestion into my S3 bucket, as in, making the process more automated.
    • Nonetheless, if we are just referring to the S3-->Redshift-->S3 part of the process, then I would set airflow to run the current elt process daily as the initial api --> MongoDB --> S3 part of the process would be taken care of.
    • I would also need to add in an extra step so that the pipeline combines the data that is previously stored in the S3 bucket with the new data added.
  • What if the database needed to be accessed by 100+ people.
    • If the database needs to be accessed by 100+ people than I would need to either:
      • constantly run a redshift cluster with the tables stored in said cluster (this requires additional IAM configuration and security protocols)
      • store the results in MongoDB so everyone can just pull from that database using pandas (requires adding everyones IP to the MongoDB Network)
      • have users simply pull from the mkgpublic S3 Bucket (just need the S3 URI) and using a platform like Databricks for users to run analysis

Future Work

Ultimately, I want to use these datasets as the backend to a dashboard hosted on a website.

I want to incoporate reddit data as well into the mix. Afterwards, I want to run sentiment analysis on both the tweets and reddit thread datasets to determine the current crypto market sentiment.

Work will be done over the next few months on the above tasks.

Owner
Matthew Greene
Backend Engineer
Matthew Greene
CertPy is a high level toolkit for generating x509 (e.g. SSL/TLS/HTTPS) certificates in Python.

CertPy CertPy is a high level toolkit for generating x509 (e.g. SSL/TLS/HTTPS) certificates in Python. Certificate “profiles” are implemented as Pytho

Ryan Castellucci 4 Feb 21, 2022
SVSHI - Secure and Verified Smart Home Infrastructure

The SVSHI (Secure and Verified Smart Home Infrastructure) (pronounced like "sushi") project is a platform/runtime/toolchain for developing and running formally verified smart infrastructures, such as

Dependable Systems Laboratory 3 Oct 28, 2022
Challenge2022 - A backend of a Chia project donation platform

Overview This is a backend of a Chia project donation platform. People can publi

Kronus91 2 Feb 04, 2022
This project aims to assist in the search for leaked passwords while maintaining a high level of privacy using the k-anonymity method.

To achieve this, the APIs of different services are used, sending only a part of the Hash of the password we want to check, for example, the first 5 characters.

Telefónica 36 Jul 06, 2022
Random Password Generator With Python

Random_Password_Generator example output length

Mahdi Rostami Pooya 2 Dec 22, 2021
Marketplace but with cryptocurrencies only.

MoneroMarket Marketplace but with cryptocurrencies only. MoneroMarket was created as a way to be able to use cryptocurrencies as an actual currency to

Janoher 35 Jan 01, 2023
dashboard to track crypto prices and change via the coinmarketcap APIs

crypto-dashboard Dashboard to track crypto prices and change via the coinmarketcap APIs. Uses chart.js and ag-grid. Requirements: python 3 (was writte

4 Nov 09, 2021
This folder contains all the assignment of the course COL759 : Cryptography & Computer Security

Cryptography This folder contains all the assignment of the course COL759 : "Cryptography & Computer Security" Assignment 1 : Encyption, Decryption &

0 Jan 21, 2022
PyBeacon is a collection of scripts for dealing with Cobalt Strike's encrypted traffic.

PyBeacon is a collection of scripts for dealing with Cobalt Strike's encrypted traffic. It can encrypt/decrypt beacon metadata, as well as pa

NCC Group Plc 162 Dec 21, 2022
Message Encrypt and decrypt software // allows you to encrypt the secrete message and decrypt Another Encryption Message. |

Message-Encrypy-Decrypt-App Message Encrypt and decrypt software // allows you to encrypt the secrete message and decrypt Another Encryption Message.

Abdulrahman-Haji 2 Dec 16, 2021
Simple python crypto bot to trade crypto on Binance based on RSI. Utilizing web sockets to get real-time prices

Py Crypto Bot Using Binance WebSocket API to get real-time price data for cryptocurrencies. Using the TA-Lib library to calculate the RSI and execute

Kennedy Ngugi Mwaura 15 Jan 04, 2023
Linear encryption software programmed with python

Echoder linear encryption software programmed with python How does it work? The text in the text section runs a function with two keys entered keys mu

Emre Orhan 4 Dec 20, 2021
A tool that can encrypt python2 or python3 code with the given password and can reuse with that password

A tool that can encrypt python2 or python3 code with the given password and can reuse with that password

Md Rasel Bhuyan 3 Feb 28, 2022
cryptography is a package designed to expose cryptographic primitives and recipes to Python developers.

pyca/cryptography cryptography is a package which provides cryptographic recipes and primitives to Python developers. Our goal is for it to be your "c

Python Cryptographic Authority 5.2k Dec 30, 2022
Modern(-ish) password hashing for your software and your servers

bcrypt Good password hashing for your software and your servers Installation To install bcrypt, simply: $ pip install bcrypt Note that bcrypt should b

Python Cryptographic Authority 947 Dec 28, 2022
Persian caesar and rot16 encryptor and decryptor

persian caesar and rot16 encrypt and decrypt how to install if you use windows python -m venv .venv .\.venv\Script\activate python -m pip install -r r

Mehdi Radfar 5 Oct 28, 2022
GreenDoge is a modern community-centric green cryptocurrency based on a proof-of-space-and-time consensus algorithm.

GreenDoge Blockchain Download GreenDoge blockchain GreenDoge is a modern community-centric green cryptocurrency based on a proof-of-space-and-time con

40 Sep 11, 2022
An BlockChain Based solution for storing the medical records

Blockchain-based Medical Records 📄 Abstract Blockchain has the ability to keep an incorruptible, decentralized, and transparent log of all patient da

Yuvraj Singh Deora 3 Jan 14, 2022
基于python的一款 加解密工具

基于python的一款 加解密工具 加密: SHA序列: sha1 , sha2 , sha224 , sha256 , sha384 , sha512 , sha512-256 , sha3-224 , sha3-256 , sha3-384 , sha3-512 MD序列: md4 , md5

3 May 05, 2022
A crypto bot that checks the price movement in the markets and creates buy and sell signals

Booter bot Purpose The purpose of this bot is to check the price fluctuations in a given market in binance and create the idealistic signals based on

2 Oct 09, 2022