[05.02.20] Main set up, exchange connections, price calculations and database insertion

README.md

@@ -1,2 +1,2 @@
-# price-collector
+# Cryptosky Price Collector
-Cryptocurrency price collector
+Projects Price Collector service that collects the: High, Low, Open, Close prices, Volume and calculates average price for the hour.

configuration/coinbase.env

@@ -0,0 +1,2 @@
+API_KEY=""
+API_SECRET=""

src/main.py

@@ -1,6 +1,11 @@
 #!/usr/bin/env python
 
-from utils.databaseConnect import connect
+from threading import Thread
+from pricing.collector import collector
 
 if __name__=='__main__':
-    connect()
+    # Dynamically create new child for each currency
+    currencies = [ "btc_gbp" ]
+
+    for i in range(len(currencies)):
+        Thread(target = collector("btc_gbp")).start()

src/pricing/bitfinex.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python
+
+import requests, json, sys
+
+def bitfinexPublicTicker(type):
+
+    try:
+        uri = "https://api.bitfinex.com/v2/tickers?symbols=" + "t"+type.toLower().replace('_', '')
+
+        response = requests.request("GET", uri)
+        response = json.loads(response.text)
+
+        price = (float(response[0][1])+ float(response[0][3]) + float(response[0][7]))/3
+        price = round(price, 3)
+        return price
+    except KeyError as e:
+        print("Error: %s" % str(e))
+        sys.stdout.flush()
+        price = 0
+        return price
+
+def bitfinexHighLowVol(type):
+
+    try:
+        uri = "https://api.bitfinex.com/v2/tickers?symbols=" + "t"+type.toLower().replace('_', '')
+
+        response = requests.request("GET", uri)
+        response = json.loads(response.text)
+
+        high = round(float(response[0][9]), 3)
+        low = round(float(response[0][10]), 3)
+        vol = round(float(response[0][8]), 3)
+
+        return high, low, vol
+    except KeyError as e:
+        print("Error: %s" % str(e))
+        sys.stdout.flush()
+        return 0, 0, 0

src/pricing/coinbase.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+
+import sys, os
+
+from coinbase.wallet.client import Client
+
+from dotenv import load_dotenv
+from pathlib import Path  # python3 only
+env_path = Path('.') / 'configuration/coinbase.env'
+load_dotenv(dotenv_path=env_path)
+
+class keys():
+
+    def __init__(self):
+        self.api_key = os.getenv('API_KEY')
+        self.api_secret = os.getenv("API_SECRET")
+
+def coinbasePublicTicker(type):
+
+    api_key = keys().api_key
+    api_secret = keys().api_secret
+
+    type = type.toUpper().replace('_', '-')
+
+    try:
+        client = Client(api_key, api_secret)
+        repsonse = client.get_spot_price(currency_pair = type)
+        price = (float(repsonse['amount']))
+        price = round(price, 3)
+        return price
+    except KeyError as e:
+        print("Error: %s" % str(e))
+        sys.stdout.flush()
+        price = 0
+        return price

src/pricing/collector.py

@@ -0,0 +1,106 @@
+#!/usr/bin/env python
+
+import sys, json, os
+from datetime import datetime, timedelta
+
+from pricing.bitfinex import bitfinexPublicTicker, bitfinexHighLowVol
+from pricing.coinbase import coinbasePublicTicker
+from pricing.gemini import geminiPublicTicker, geminiHighLowVol, geminiOpenClose
+
+from util.databaseConnect import connect, closeConnection
+
+btc_usd="resources/sql/V1_INSERT_NEW_PRICE_RECORD_BTC.sql"
+
+def getInsertForType(type):
+    if type == "btc_usd":
+        with open(btc_usd, 'r') as s:
+            sql = s.read()
+            s.close()
+            return sql
+    elif type == "":
+        return ""
+
+def averager(type):
+
+    # FORMAT FOR RFC-3339
+    timestamp = datetime.now()# + timedelta(hours=1)
+
+    coinbase_P = coinbasePublicTicker(type)
+    bitfinex_P = bitfinexPublicTicker(type)
+    gemini_P = geminiPublicTicker(type)
+
+    if coinbase_P == 0 or bitfinex_P == 0 or gemini_P == 0:
+        if coinbase_P and bitfinex_P == 0:
+            averagePrice = gemini_P
+            return
+        elif coinbase_P and gemini_P == 0:
+            averagePrice = bitfinex_P
+            return
+        elif bitfinex_P and gemini_P == 0:
+            averagePrice = coinbase_P
+            return
+        averagePrice = (coinbase_P + bitfinex_P + gemini_P)/2
+    else:
+        averagePrice = (coinbase_P + bitfinex_P + gemini_P)/3
+
+    averagePrice = round(averagePrice, 3)
+
+    print("Price: ", averagePrice)
+
+    return averagePrice, timestamp
+
+def getHighLowVol(type):
+    bH, bL, bV = bitfinexHighLowVol(type)
+    gH, gL, gV = geminiHighLowVol(type)
+
+    if ( bH == 0 or bL == 0 or bV == 0 ) or ( gH == 0 or gL == 0 or gL == 0):
+        if bH == 0:
+            high = gH
+        elif gH == 0:
+            high = bH
+        if bL == 0:
+            low = gL
+        elif gL == 0:
+            low = bL
+        if bV == 0:
+            vol = gV
+        elif gV == 0:
+            vol = bV
+    else:
+        high = (bH + gH)/2
+        low = (bL + gL)/2
+        vol = (bV + gV)/2
+
+    return high, low, vol
+
+def saveToDatabase(type, timestamp, av_price, high, low, vol, open, close):
+    try:
+        cur = connect()
+
+        sql = getInsertForType(type)
+
+        cur.execute(sql, (type, timestamp, av_price, high, low, open, close, vol))
+
+        closeConnection(cur)
+        return True
+    except BaseException as exception:
+        print("Error: %s" % str(exception))
+        sys.stdout.flush()
+        return False
+
+def getOpenClose(type):
+    open, close = geminiOpenClose()
+    return open, close
+
+
+# Dynamically Spin up Child process for each type wanting to track
+def collector(type):
+    print("Console: ", "== Historical Price Collector ==")
+
+    while True:
+        sleep(3600)
+        av_price, timestamp = averager(type)
+        high, low, vol = getHighLowVol(type)
+        open, close = getOpenClose(type)
+
+        saveToDatabase(type, timestamp, av_price, high, low, vol, open, close)

src/pricing/gemini.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python
+
+import requests, json, sys
+
+def geminiPublicTicker(type):
+
+    try:
+        uri = "https://api.gemini.com/v1/pubticker/" + type.toUpper().replace('_', '')
+        response = requests.request("GET", uri)
+        response = json.loads(response.text)
+
+        price = (float(response['last']) + float(response['ask']) + float(response['bid']))/3
+        price = round(price, 3)
+        return price
+    except KeyError as e:
+        print("Error: %s" % str(e))
+        sys.stdout.flush()
+        price = 0
+        return price
+
+def geminiHighLowVol(type):
+    try:
+        uri = "https://api.gemini.com/v2/ticker/" + type.toUpper().replace('_', '')
+        response = requests.request("GET", uri)
+        response = json.loads(response.text)
+
+        high = float(response['high'])
+        low = float(response['low'])
+        vol = float(response['volo'])
+
+        return high, low, vol
+    except KeyError as e:
+        print("Error: %s" % str(e))
+        sys.stdout.flush()
+        return 0, 0, 0
+
+def geminiOpenClose(type):
+
+    try:
+        uri = "https://api.gemini.com/v2/ticker/" + type.toUpper().replace('_', '')
+        response = requests.request("GET", uri)
+        response = json.loads(response.text)
+
+        open = float(response['open'])
+        close = float(response['close'])
+
+        return open, close
+    except KeyError as e:
+        print("Error: %s" % str(e))
+        sys.stdout.flush()
+        return 0, 0

src/resources/sql/V1_INSERT_NEW_PRICE_RECORD_BTC.sql

@@ -0,0 +1,2 @@
+INSERT INTO btc_price(timestamp, symbol, av_price, h_price, l_price, o_price, c_price, volume)
+VALUES(%s, %s, %f, %f, %f, %f, %f, %f);

src/utils/databaseConnect.py

@@ -26,10 +26,14 @@ def connect():
         print(db_version)
 
         # close the communication with the PostgreSQL
-        cur.close()
+        # cur.close()
+        return cur;
     except (Exception, psycopg2.DatabaseError) as error:
         print(error)
     finally:
         if conn is not None:
             conn.close()
             print('Database connection closed.')
+
+def closeConnection(cur):
+    cur.close()