We all know that in the last period cryptocurrencies have ushered the era of the next financial revolution. Having that in mind, I thought it'd be useful to know (using Python 3.6) when a certain cryptocurrency has a raise of \$X\$ percent in a period of time.
Details
To do this, I'm getting the data from this website and am using the binance module to make use of their API.
The logic of the program is pretty easy and straight-forward:
- Get the data from binance (\$Y\$s - gathering interval - can be adjusted)
- Store the data in a Postgres DB
- Process the data and print the values that have a raise \$\ge\$ than \$X\$ (this can be adjusted). The information printed should let the user know what was the raise during 5 mins: at 10 s / 1 min / 5 mins
I don't want the data to be extra-precise so I haven't used float numbers for percentages.
from time import localtime, sleep, strftime
from binance.client import Client
from binance.exceptions import BinanceAPIException
import psycopg2.extras
API_KEY = '<API_KEY>'
API_SECRET = '<API_SECRET>'
# How often the data should be gathered (in seconds)
DATA_GATHERING_SLEEP = 5
# The percentage (%) that sets when the user should be announced of a change
PERCENTAGE_THRESHOLD = 3.00
DB_NAME = 'binance'
DB_USER = 'postgres'
DB_PASS = 'postgres'
class Colors:
GREEN = '\033[92m'
WARNING = '\033[93m'
END = '\033[0m'
class BinanceStockerProcessor:
def __init__(self):
try:
self.client = Client(API_KEY, API_SECRET)
except BinanceAPIException as e:
print(e)
self.db_connection = psycopg2.connect(database=DB_NAME, user=DB_USER, password=DB_PASS)
self.db_cursor = self.db_connection.cursor(cursor_factory=psycopg2.extras.DictCursor)
self.tickers = self.client.get_all_tickers()
def calculate_percentage(self):
for ticker in self.tickers:
now = strftime("%Y-%m-%d %H:%M:%S", localtime())
# write currencies to DB
self.db_cursor.execute("""
INSERT INTO binance (date, symbol, price)
VALUES (%s,%s,%s)
""", (now, ticker['symbol'], ticker['price']))
self.db_connection.commit()
# get data from db
self.db_cursor.execute("""
SELECT * FROM binance
WHERE symbol=%s
ORDER BY date DESC LIMIT 1
""", (ticker['symbol'].upper(),))
rows_now = self.db_cursor.fetchall()
self.db_cursor.execute("""
SELECT * FROM binance
WHERE symbol=%s
AND date >= NOW() AT TIME ZONE 'EET' - INTERVAL '15 seconds'
ORDER BY date ASC
""", (ticker['symbol'].upper(),))
rows_10s = self.db_cursor.fetchall()
self.db_cursor.execute("""
SELECT * FROM binance
WHERE symbol=%s
AND date >= NOW() AT TIME ZONE 'EET' - INTERVAL '1 minutes'
ORDER BY date ASC
""", (ticker['symbol'].upper(),))
rows_1min = self.db_cursor.fetchall()
self.db_cursor.execute("""
SELECT * FROM binance
WHERE symbol=%s
AND date >= NOW() AT TIME ZONE 'EET' - INTERVAL '5 minutes'
ORDER BY date ASC
""", (ticker['symbol'].upper(),))
rows_5min = self.db_cursor.fetchall()
date_now, symbol_now, price_now = self.get_rows_data(rows_now)
date_10s, symbol_10s, price_10s = self.get_rows_data(rows_10s)
date_1min, symbol_1min, price_1min = self.get_rows_data(rows_1min)
date_5min, symbol_5min, price_5min = self.get_rows_data(rows_5min)
percentage_10s = self.get_percentage(price_now, price_10s)
percentage_1min = self.get_percentage(price_now, price_1min)
percentage_5min = self.get_percentage(price_now, price_5min)
if percentage_5min >= PERCENTAGE_THRESHOLD:
value = (f"{Colors.WARNING}{ticker['symbol'].upper()}{Colors.END}: "
f"{Colors.GREEN}{percentage_5min}%{Colors.END} (5min) | "
f"{Colors.GREEN}{percentage_1min}%{Colors.END} (1min) | "
f"{Colors.GREEN}{percentage_10s}%{Colors.END} (10s)")
print(value)
@staticmethod
def get_percentage(now, old):
if now == old:
return 0
try:
return ((now - old) / old) * 100.0
except ZeroDivisionError:
return 0
@staticmethod
def get_rows_data(data):
if len(data) >= 1:
return data[0][0], data[0][1], data[0][2]
else:
print('Error here! Will exit now...')
return
def main():
while True:
binance = BinanceStockerProcessor()
binance.calculate_percentage()
sleep(DATA_GATHERING_SLEEP)
if __name__ == '__main__':
try:
main()
except KeyboardInterrupt:
print('Exiting now...')
I know I don't have docstrings and that my API keys shouldn't be stored in my program (the same for DB credentials), so try to avoid these aspects when reviewing my code.
Setup
If someone wants to give this a try, these are the steps in order to make the above run:
requirements.txt
python-binance==0.5.11 psycopg2==2.7.3.2
pip install -r requirements.txt
Postgres:
- You have to install Postgres on your machine and make sure you use the credentials from the top of the program (feel free to modify those as you wish)
Then create a DB:
CREATE DATABASE binance;
Then create a table:
CREATE TABLE IF NOT EXISTS binance ( date TIMESTAMP, symbol VARCHAR(20), price REAL )
After this you should be all set and the program will run without issues.
Review
I'll add more functionality to this and probably try to DELETE
the data older than 3 weeks so that my DB won't be too loaded.
I'm sure that my program can be improved from the architecture/logic perspective so feel free to point out any improvements that come to mind.
One of the things that I personally don't like about this is that it prints duplicate data. For example, if the percentage for "ETHBTC" was greater than \$X\$% in the last 1 min, and it remains to the same value, that will be printed out each time.