Cryptocurrency notifier - let's get rich (or at least let's try)

Question

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.

Answer 1

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.

Just mentioning even though you've asked not to. Let the API keys and DB values be fetched from environment variables. For DB vars, I prefer creating a dict like:

DB_DICT = {
    'database': os.getenv('DB_NAME'),
    'user': os.getenv('DB_USER'),
}

so that I can later simply unpack those values:

self.db_connection = psycopg2.connect(**DB_DICT)

---

You are in any case fetching a single row at any moment of time. Use the cursor.fetchone() method over fetchall. I also notice that you're specifying the following (for cursor initialisation):

cursor_factory=psycopg2.extras.DictCursor

but not actually making use of the dict type objects returned.

---

The following class declaration is overkill:

class Colors:
    GREEN = '\033[92m'
    WARNING = '\033[93m'
    END = '\033[0m'

and can be replaced with a simplistic collections.namedtuple. But, the whole declaration can be removed, because of next point.

---

Instead of using print statements, make use of the logging module. It has some helper packages (on PyPI), or you can write your own for coloured output.

---

def main():
    while True:
        binance = BinanceStockerProcessor()
        binance.calculate_percentage()
        sleep(DATA_GATHERING_SLEEP)

Instead of creating new objects in every iteration. Create it just once, and have a try-except block where you simply refresh your binance.Client connection or the psycopg2 connection in case of errors. As it currently stands, it doesn't appear that you are actually making use of any OOP features.

---

try:
    return ((now - old) / old) * 100.0
except ZeroDivisionError:
    return 0

The chance that it will raise a ZeroDivisionError is only and only when your old is \$ 0 \$. Just add that condition in the prior if statement:

if now == old or old == 0:
    return 0

and you won't need the try-except block anymore.

---

(ticker['symbol'].upper(),))

You keep calling upper on the accessed key of hashmap repeatedly. Simply store it in a local variable.

---

As a developer, you (and your application) should always know what columns your program needs from the DB, and specifically fetch those selectively. Using

SELECT *

seems like your application is unaware of the DB layer. You only need the price column right now. Fetch only that.

---

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(),))

The query is repeated 3 times. Only the INTERVAL clause changes. Pass that as another parameter? Store responses in a list. Comprehend the response from your price_now value and you'll have a single list containing all 3 values needed at any moment. You could also combine the whole query using UNIONs and LIMITs. I'd have to play around with some data-points before I provide you with a working code, but I'm thinking along the following lines:

SELECT price
FROM (
    SELECT price
    FROM binance
    WHERE symbol = %s
        AND date >= NOW() AT TIME ZONE 'EET' - INTERVAL '15 seconds'
    ORDER BY DATE ASC
    LIMIT 1
) current
UNION
SELECT price
FROM (
    SELECT price
    FROM binance
    WHERE symbol = %s
        AND date >= NOW() AT TIME ZONE 'EET' - INTERVAL '60 seconds'
    ORDER BY DATE ASC
    LIMIT 1
) minute_ago

---

As you are only making use of the ISO formatted time string, why not rely on the datetime's isoformat method?

---

Caching can be achieved if you keep a counter on iterations, or maintain a clock for program. The prices retrieved right now, will be used again 15 seconds later, then 1 minute later and yet again 5 minutes later. You won't need to make extra SQL calls (if you're not going to use a single query call).

---

Instead of defining get_percentage as a staticmethod to class, define this inside the calculate body itself. You do not need it as a method on class object in any case.