Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from datetime import datetime
- from time import localtime, sleep, strftime
- from binance.client import Client
- from binance.exceptions import BinanceAPIException
- import os
- import psycopg2.extras
- API_KEY = 'Ioa5gkMpSm4qnqj15WYS3QrTGDoaPjzRMRnRPyhWOkVVk4NBwsEPta1EUDrV3oyr'
- API_SECRET = 'SN8q3uLcQEYXacZjBsYnwb4qGInAxGtKmEDZjN8XefkDxrBThuJF6OSewgCKbEdh'
- # how often the data should be gathered -- seconds
- DATA_GATHERING_SLEEP = 5
- # the percentage when the user should be announced of a change
- PERCENTAGE_THRESHOLD = 3.00
- # database
- DB_NAME = 'binance'
- DB_USER = 'a'
- DB_PASS = '1q'
- 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()
- if not rows_now:
- continue
- 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()
- if not rows_10s:
- continue
- 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()
- if not rows_1min:
- continue
- 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()
- if not rows_5min:
- continue
- 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 = round(self.get_percentage(price_now, price_10s), 2)
- percentage_1min = round(self.get_percentage(price_now, price_1min), 2)
- percentage_5min = round(self.get_percentage(price_now, price_5min), 2)
- if percentage_5min >= PERCENTAGE_THRESHOLD and ticker['symbol'][:-3] is 'BTC':
- # print(
- # f"{ticker['symbol']}: "
- # f"({datetime.strftime(date_now, '%Y-%m-%d %H:%M:%S')} | {price_now}) "
- # f"({datetime.strftime(date_10s, '%Y-%m-%d %H:%M:%S')} | {percentage_10s} | {price_10s}) "
- # f"({datetime.strftime(date_1min, '%Y-%m-%d %H:%M:%S')} | {percentage_1min} | {price_1min}) "
- # f"({datetime.strftime(date_5min, '%Y-%m-%d %H:%M:%S')} | {percentage_5min} | {price_5min})"
- # )
- #os.system("firefox -new-tab -url https://www.binance.com/trade.html?symbol=" + ticker['symbol'][:3] + "_" + ticker['symbol'][-3:])
- print(
- f"🚀 Binance {ticker['symbol'][:3]}-BTC"
- f"Price: ({percentage_5min}%/5min)"
- f"({percentage_1min}%/1min)"
- f"({percentage_10s}%/10s)\n"
- )
- @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...')
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement