Worker.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"images/splash3.png\">\n",
    "Requests data for the past hour and inserts this into SQL. To be ran hourly.\n",
    "\n",
    "\n",
    "<p><b>What you were doing:</b></p>\n",
    "<p>&nbsp; &nbsp; possible issue: only fetches tweets with 140+ character length? </p>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sqlite3 as lite\n",
    "import sys\n",
    "import requests\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import datetime\n",
    "import json\n",
    "import tweepy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Twitter stuff\n",
    "CONSUMER_KEY = 'hiZmnSOV63GdqWPqW5MN533T1'# TWITTER_CONSUMER_TOKEN\n",
    "CONSUMER_SECRET = 'GJa7snhSw8jDQxTVhvoHJyIyJZBjLQikAeuNL0Y3o0Aw61zuOr' # TWITTER_CONSUMER_SECRET\n",
    "ACCESS_TOKEN = '933287551212904448-uWflGB32dNF8rbgmBxJk1ql5AqemO14' # TWITTER_ACCESS_TOKEN\n",
    "ACCESS_SECRET = 'boUb7syKFxrCmaJN3T3SHsWu6rDQQfdPxertNIEnXR6X9' # TWITTER_ACCESS_SECRET"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Class & function definitions\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Coin:\n",
    "    def __init__(self, symbol, comparison_symbol, limit, aggregate):\n",
    "        self.symbol = symbol\n",
    "        self.comparison_symbol = comparison_symbol\n",
    "        self.limit = limit\n",
    "        self.aggregate = aggregate\n",
    "        \n",
    "    def fetch_marketdata(self):    \n",
    "        url = 'https://min-api.cryptocompare.com/data/histohour?fsym={}&tsym={}&limit={}&aggregate={}'\\\n",
    "                .format(self.symbol.upper(), self.comparison_symbol.upper(), self.limit, self.aggregate)       \n",
    "        page = requests.get(url)\n",
    "        data = page.json()['Data']\n",
    "        df = pd.DataFrame(data)\n",
    "        df['mid'] = df[[\"high\", \"low\"]].mean(axis=1)\n",
    "        df['timestamp'] = [datetime.datetime.fromtimestamp(d) for d in df.time]\n",
    "        return df\n",
    "    \n",
    "    def fetch_tweets(self):\n",
    "        # Create extractor object\n",
    "        extractor = twitter_setup()\n",
    "        # Create a tweet list\n",
    "        #tweets = extractor.search(q =\"${}\".format(coin_instances[coin].symbol), count=20, tweet_mode=\"extended\", lang=\"en\")\n",
    "        #print(\"[coin] = \", coin)\n",
    "        #print(\"Searching for coin_instances[coin].symbol: \", coin_instances[coin].symbol)\n",
    "        tweets = extractor.search(q =\"${} {}\".format(coin_instances[coin].symbol, \"-url:binance -url:kucoin -url:ku-coin -url:discord\"), \n",
    "                                  count=100, \n",
    "                                  tweet_mode=\"extended\", lang=\"en\")\n",
    "        \n",
    "        def fulltext(tweet):\n",
    "            try:\n",
    "                text = tweet.retweeted_status.full_text\n",
    "                return(text)\n",
    "            except:\n",
    "                text = tweet.full_text\n",
    "                return(text)\n",
    "\n",
    "        df = pd.DataFrame(data=[fulltext(tweet) for tweet in tweets], columns=['Tweets'])\n",
    "        \n",
    "        df['time'] = np.array([tweet.created_at for tweet in tweets])\n",
    "        df['time'] = df['time'].astype(np.int64) // 10**9 # Convert to unix\n",
    "        df['ID']   = np.array([tweet.id for tweet in tweets])\n",
    "        df['Len']  = np.array([len(tweet.full_text) for tweet in tweets])\n",
    "        df['Reply_To_Status'] = np.array([tweet.in_reply_to_user_id for tweet in tweets])\n",
    "        df['Reply_To_Name'] = np.array([tweet.in_reply_to_screen_name for tweet in tweets])\n",
    "        df['Coordinates'] = np.array([tweet.coordinates for tweet in tweets])\n",
    "        df['Retweet_Count'] = np.array([tweet.retweet_count for tweet in tweets])\n",
    "        df['User_Verified'] = np.array([tweet.user.verified for tweet in tweets])\n",
    "        df['User_Location'] = np.array([tweet.user.location for tweet in tweets])\n",
    "        df['User_Followers_Count'] = np.array([tweet.user.followers_count for tweet in tweets])\n",
    "        df['User_Favorite_Count'] = np.array([tweet.user.favourites_count for tweet in tweets])\n",
    "        df['Source'] = np.array([tweet.source for tweet in tweets])\n",
    "        df['Likes']  = np.array([tweet.favorite_count for tweet in tweets])\n",
    "        df['Favorite_Count'] = np.array([tweet.favorite_count for tweet in tweets])\n",
    "        df['RTs'] = np.array([tweet.retweet_count for tweet in tweets])        \n",
    "        return df\n",
    "        \n",
    "def twitter_setup():\n",
    "    \"\"\"\n",
    "    Utility function to setup the Twitter's API\n",
    "    with our access keys provided.\n",
    "    \"\"\"\n",
    "    # Authentication and access using keys\n",
    "    auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)\n",
    "    auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET)\n",
    "\n",
    "    # Return API with authentication\n",
    "    api = tweepy.API(auth)\n",
    "    return api    \n",
    "    \n",
    "    \n",
    "def market_to_db(market_data):\n",
    "    con = None\n",
    "    con = lite.connect('/hdd/db/crypto.db') #Connect to DB\n",
    "\n",
    "    \n",
    "    with con: #Dump to SQL\n",
    "        try:\n",
    "            date = pd.read_sql_query(\"SELECT time FROM {} ORDER BY time DESC LIMIT 1\".format(coin_instances[coin].symbol), con)\n",
    "            latest_db_entry = date['time'].values[0]\n",
    "        except Exception as e:\n",
    "            print(\"Skipping coin\", coin_instances[coin].symbol, \", Error: \")\n",
    "            print(e)\n",
    "            latest_db_entry = 0 # Probably no table exists, set \n",
    "            \n",
    "        if latest_db_entry < market_data.iloc[0]['time']:\n",
    "            #print(latest_db_entry, \"is lower than\", market_data.iloc[0]['time'], \"<- Time to write!\") # Debug\n",
    "            market_data.to_sql(name=coin_instances[coin].symbol, con=con, if_exists='append', index=True)\n",
    "            #print(\"wrote to DB for coin\", coin_instances[coin].symbol) # Debug \n",
    "        else:\n",
    "            pass\n",
    "            #print(latest_db_entry, \"(latest db entry) is the same or higher than\", market_data.iloc[0]['time'], \"(fetched data), skipping write for\", coinlist[coin])\n",
    "            \n",
    "def tweets_to_db(df_tweets):\n",
    "    con = None\n",
    "    con = lite.connect('/hdd/db/crypto.db') #Connect to DB\n",
    "    \n",
    "    with con: #Dump to SQL\n",
    "        try:\n",
    "            date = pd.read_sql_query(\"SELECT time FROM {}_tweets ORDER BY time DESC LIMIT 1\".format(coin_instances[coin].symbol), con)\n",
    "            latest_db_entry = date['time'].values[0] #Get last date\n",
    "        except Exception as e:\n",
    "            print(\"Skipping coin\", coin_instances[coin].symbol, \", Error: \")\n",
    "            print(e)\n",
    "            latest_db_entry = 0 # Probably no table exists, set \n",
    "        \n",
    "        df_tweets = df_tweets.loc[df_tweets['time'] > latest_db_entry] #Slice df to only include new tweets\n",
    "        \n",
    "        if(len(df_tweets) > 0): # If there is something to write \n",
    "            df_tweets.to_sql(name=\"{}_tweets\".format(coin_instances[coin].symbol), con=con, if_exists='append', index=True)\n",
    "            #print(\"Going to write {} tweets to {}_tweets\".format(len(df_tweets), coin_instances[coin].symbol))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Fetch data & store in db"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Skipping coin BTCP , Error: \n",
      "Execution failed on sql 'SELECT time FROM BTCP ORDER BY time DESC LIMIT 1': no such table: BTCP\n",
      "Skipping coin BTCP , Error: \n",
      "Execution failed on sql 'SELECT time FROM BTCP_tweets ORDER BY time DESC LIMIT 1': no such table: BTCP_tweets\n",
      "Skipping coin LOOM , Error: \n",
      "Execution failed on sql 'SELECT time FROM LOOM ORDER BY time DESC LIMIT 1': no such table: LOOM\n",
      "Skipping coin LOOM , Error: \n",
      "Execution failed on sql 'SELECT time FROM LOOM_tweets ORDER BY time DESC LIMIT 1': no such table: LOOM_tweets\n",
      "Skipping coin WAN , Error: \n",
      "Execution failed on sql 'SELECT time FROM WAN ORDER BY time DESC LIMIT 1': no such table: WAN\n",
      "Skipping coin WAN , Error: \n",
      "Execution failed on sql 'SELECT time FROM WAN_tweets ORDER BY time DESC LIMIT 1': no such table: WAN_tweets\n",
      "Skipping coin MITH , Error: \n",
      "Execution failed on sql 'SELECT time FROM MITH ORDER BY time DESC LIMIT 1': no such table: MITH\n",
      "Skipping coin MITH , Error: \n",
      "Execution failed on sql 'SELECT time FROM MITH_tweets ORDER BY time DESC LIMIT 1': no such table: MITH_tweets\n",
      "Skipping coin CTXC , Error: \n",
      "Execution failed on sql 'SELECT time FROM CTXC ORDER BY time DESC LIMIT 1': no such table: CTXC\n",
      "Skipping coin CTXC , Error: \n",
      "Execution failed on sql 'SELECT time FROM CTXC_tweets ORDER BY time DESC LIMIT 1': no such table: CTXC_tweets\n",
      "Skipping coin XMR , Error: \n",
      "Execution failed on sql 'SELECT time FROM XMR ORDER BY time DESC LIMIT 1': no such table: XMR\n",
      "Skipping coin XMR , Error: \n",
      "Execution failed on sql 'SELECT time FROM XMR_tweets ORDER BY time DESC LIMIT 1': no such table: XMR_tweets\n",
      "Skipping coin BCN , Error: \n",
      "Execution failed on sql 'SELECT time FROM BCN ORDER BY time DESC LIMIT 1': no such table: BCN\n",
      "Skipping coin BCN , Error: \n",
      "Execution failed on sql 'SELECT time FROM BCN_tweets ORDER BY time DESC LIMIT 1': no such table: BCN_tweets\n",
      "Skipping coin BTS , Error: \n",
      "Execution failed on sql 'SELECT time FROM BTS ORDER BY time DESC LIMIT 1': no such table: BTS\n",
      "Skipping coin BTS , Error: \n",
      "Execution failed on sql 'SELECT time FROM BTS_tweets ORDER BY time DESC LIMIT 1': no such table: BTS_tweets\n",
      "Skipping coin XLM , Error: \n",
      "Execution failed on sql 'SELECT time FROM XLM ORDER BY time DESC LIMIT 1': no such table: XLM\n",
      "Skipping coin XLM , Error: \n",
      "Execution failed on sql 'SELECT time FROM XLM_tweets ORDER BY time DESC LIMIT 1': no such table: XLM_tweets\n",
      "Skipping coin XRP , Error: \n",
      "Execution failed on sql 'SELECT time FROM XRP ORDER BY time DESC LIMIT 1': no such table: XRP\n",
      "Skipping coin XRP , Error: \n",
      "Execution failed on sql 'SELECT time FROM XRP_tweets ORDER BY time DESC LIMIT 1': no such table: XRP_tweets\n",
      "Skipping coin SYS , Error: \n",
      "Execution failed on sql 'SELECT time FROM SYS ORDER BY time DESC LIMIT 1': no such table: SYS\n",
      "Skipping coin SYS , Error: \n",
      "Execution failed on sql 'SELECT time FROM SYS_tweets ORDER BY time DESC LIMIT 1': no such table: SYS_tweets\n",
      "Skipping coin XVG , Error: \n",
      "Execution failed on sql 'SELECT time FROM XVG ORDER BY time DESC LIMIT 1': no such table: XVG\n",
      "Skipping coin XVG , Error: \n",
      "Execution failed on sql 'SELECT time FROM XVG_tweets ORDER BY time DESC LIMIT 1': no such table: XVG_tweets\n",
      "Skipping coin DOGE , Error: \n",
      "Execution failed on sql 'SELECT time FROM DOGE ORDER BY time DESC LIMIT 1': no such table: DOGE\n",
      "Skipping coin DOGE , Error: \n",
      "Execution failed on sql 'SELECT time FROM DOGE_tweets ORDER BY time DESC LIMIT 1': no such table: DOGE_tweets\n",
      "Skipping coin USDT , Error: \n",
      "Execution failed on sql 'SELECT time FROM USDT ORDER BY time DESC LIMIT 1': no such table: USDT\n",
      "Skipping coin USDT , Error: \n",
      "Execution failed on sql 'SELECT time FROM USDT_tweets ORDER BY time DESC LIMIT 1': no such table: USDT_tweets\n",
      "Skipping coin XEM , Error: \n",
      "Execution failed on sql 'SELECT time FROM XEM ORDER BY time DESC LIMIT 1': no such table: XEM\n",
      "Skipping coin XEM , Error: \n",
      "Execution failed on sql 'SELECT time FROM XEM_tweets ORDER BY time DESC LIMIT 1': no such table: XEM_tweets\n",
      "Skipping coin DNA , Error: \n",
      "Execution failed on sql 'SELECT time FROM DNA ORDER BY time DESC LIMIT 1': no such table: DNA\n",
      "Skipping coin DNA , Error: \n",
      "Execution failed on sql 'SELECT time FROM DNA_tweets ORDER BY time DESC LIMIT 1': no such table: DNA_tweets\n"
     ]
    }
   ],
   "source": [
    "fh = open(\"/hdd/db/tracked_coins.json\", 'r')\n",
    "coinlist = json.load(fh)\n",
    "coin_instances = [Coin(coinlist[coin], 'BTC','1','1') for coin in range(len(coinlist))] # Make Coin objects. Take all coins in coinlist, convert to BTC, poll 1 entry and aggregate per 1-hour period.\n",
    "\n",
    "#Get data\n",
    "for coin in range(len(coin_instances)):\n",
    "    try:\n",
    "        market_data = coin_instances[coin].fetch_marketdata() # Get coindata\n",
    "        df_tweets = coin_instances[coin].fetch_tweets() # Get tweets\n",
    "        market_data = market_data.drop(market_data.index[len(market_data)-1]) # Only keep the latest timestamped column\n",
    "        market_to_db(market_data)\n",
    "        tweets_to_db(df_tweets)\n",
    "    except Exception as e:\n",
    "        print(\"Skipping coin\", coin_instances[coin].symbol, \", Error: \")\n",
    "        print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 305,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Coin:  DOGE\n",
      "          close          high           low          open        time  \\\n",
      "0  3.900000e-07  4.000000e-07  3.900000e-07  3.900000e-07  1531123200   \n",
      "1  3.900000e-07  3.900000e-07  3.900000e-07  3.900000e-07  1531126800   \n",
      "\n",
      "    volumefrom  volumeto           mid           timestamp  \n",
      "0  11980491.43      4.65  3.950000e-07 2018-07-09 10:00:00  \n",
      "1         0.00      0.00  3.900000e-07 2018-07-09 11:00:00  \n"
     ]
    }
   ],
   "source": [
    "# ### JUST SOME PRINT LINES ###\n",
    "# coin = 95\n",
    "# print(\"Coin: \", coinlist[95])\n",
    "\n",
    "# market_data = coin_instances[95].fetch_marketdata() # Get coindata\n",
    "# print(market_data)\n",
    "\n",
    "# df_tweets = coin_instances[95].fetch_tweets() # Get tweets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 306,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      " ------------- Random tweet -----------------:\n",
      "\n",
      "this channel called $DGB before it made 6x. they only share high reward, low risk signals. join:\n",
      "\n",
      "https://t.co/T5egWYl250\n",
      "\n",
      "$BTC $ETH $ETC $BCH $LTC $XRP $DASH $BTG $XLM $XMR $ZEC $ADA $SNT $NEO $NXT $OMG $POWR $VTC $VOX $XEM $LSK $DGB $DOGE $XVG $GRS $AMP $ZCL $DRGN $DCR 79548\n"
     ]
    }
   ],
   "source": [
    "# print(\"\\n ------------- Random tweet -----------------:\\n\")\n",
    "# print(df_tweets['Tweets'].iloc[1])"
   ]
  }
 ],
 "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.5.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}