Игра Яндекс Практикума
Игра Яндекс Практикума
Игра Яндекс Практикума

Создаём Telegram-бот для трекинга вашего криптопортфеля

Создаём бот в Телеграме для Binance, который показывает баланс в USD(T) и изменения по портфелю в USD(T) за неделю и за все время.

10К открытий11К показов

Мне был необходим для личного удобства бот в Телеграм, который умеет выполнять следующие действия:

  1. Показать текущий размер активов в USD(T).
  2. Показать изменения по портфелю в USD(T) за неделю и за все время.

Мне критически важно не использовать никакое приложение вида «все биржи в одном месте» в качестве proxy — да, я не хочу никому предоставлять свои ключи.

Если нужен — давайте напишем.

Что нам потребуется?

В качестве примера в данной статье я буду использовать криптовалютную биржу Binance. Стоит отметить, что никаких ограничений на использование с другими биржами нет — весь каркас тот же самый. Разве что взаимодействие по API будет немного другим (зависит от биржи).

  1. Нам потребуется функционал Telegram (куда же без него).
  2. Нам потребуется Dropbox вместо базы данных (о, да).
  3. Нам потребуется инструмент, где будет крутиться наш бот. Лично я использую Heroku, но можно использовать и AWS.

Создаем бота в Телеграм

Данный этап был описан сотни раз в огромном числе статей, найти которые не составляет проблем. Поэтому здесь будет кратко.

Нам нужно найти в телеграмме контакт @BotFather и выполнить последовательность команд

			/start
/newbot
		

 

			# Нужно ввести приватное имя нового бота и его публичное имя
# Я использую непопулярное публичное имя, т.к. тут будут наши финансы
# Здесь же мы получим API token нашего бота, сохраним его
# Запомним ссылку вида t.me/alias_binance_bot
		

База из топора

Для начала мы подготовим Dropbox.

Перейдем по ссылке и получим свой API token для Dropbox, нажав на кнопку Create App. Я создам доступ на отдельную папку в Dropbox.

Создаём Telegram-бот для трекинга вашего криптопортфеля 1

На текущей странице нам необходимо будет сгенерировать OAuth 2.0 для Dropbox:

Создаём Telegram-бот для трекинга вашего криптопортфеля 2

После создания пройдем на вкладку Permissions и установим права на files.content.write

Теперь по ссылке у нас появилась папка APPS. Зайдем в нее, далее зайдем в поддерикторию с названием нашего бота. Туда нам необходимо поместить файл totalData.txt, содержащий только пустой список.

			[]
		

Взаимодействие с Binance

Нам необходимо создать наш API ключ на бирже Binance.

Данная активность происходит по этой ссылке.

Результатом данного действия для нас будет API Key и Secret Key. В нашем случае будет достаточно прав только на чтение.

Создаём Telegram-бот для трекинга вашего криптопортфеля 3

В качестве следующего шага мы напишем код, который будем использовать в дальнейшем.

			import json
import time
import logging
import os
import binance
import dropbox
from binance.client import Client

logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)

logger = logging.getLogger(__name__)

binanceKey = ['BinanceKey']
binanceSecret = ['BinanceSecret']

dropboxToken = 'DropboxKey'

SLEEP_TIMEOUT = 1 * 60

def getBalance(idx):
    client = Client(binanceKey[idx], binanceSecret[idx])
    balances = client.get_account()['balances']
    balanceUsd = 0
    prices = client.get_all_tickers()
    for b in balances:
        curB = float(b['free']) + float(b['locked'])
        asset = b['asset']
        if curB < 0.000001:
            continue
        if asset == "USDT":
            balanceUsd += curB
        prices = sorted(prices, key=lambda p: p['symbol'])
        for p in prices:
            if p['symbol'] == str(asset) + "USDT":
                balanceUsd += float(curB) * float(p['price'])
    balanceUsd = float("%.2f" % balanceUsd)
    curt = time.time()
    return balanceUsd, curt

def getAccountInfo():
        amountUsd, timenow = getBalance(0)
        return {'usd': amountUsd, 'ts': timenow}

def loadJsonFromDropbox(dbx):
    for i in range(1):
        try:
            meta, resp = dbx.files_download('/totalData.txt')
            print(meta, resp)
            body = resp.content.decode('utf-8')
            resp.close()
            return json.loads(body)
        except Exception as ex:
            time.sleep(0.5 * (2 ** i))
            print(ex)

def saveJsonToDropbox(dbx, content):
    jsonBytes = json.dumps(content, indent=4).encode('utf-8')
    dbx.files_upload(jsonBytes, '/totalData.txt', mode=dropbox.files.WriteMode('overwrite'))

def addInfoPointToDropbox(dbx):
        content = loadJsonFromDropbox(dbx)
        content += [getAccountInfo()]
        saveJsonToDropbox(dbx, content)

def main():
    dbx = dropbox.Dropbox(dropboxToken)
    while True:
        addInfoPointToDropbox(dbx)
        time.sleep(SLEEP_TIMEOUT)
    amountUsd, timenow = getBalance(0)
    print(amountUsd)
    print(timenow)

if __name__ == '__main__':
	main()
		

Для начала попробуем запустить данный код локально. Если все сделано правильно — код будет исполняться каждые 60 секунд и спустя некоторое время файл totalData.txt должен выглядеть как-то так:

			[
    {
        "usd": 2.81,
        "ts": 1670699696.930476
    },
    {
        "usd": 2.82,
        "ts": 1670699760.437554
    },
    {
        "usd": 2.84,
        "ts": 1670699823.819883
    },
    {
        "usd": 2.86,
        "ts": 1670700537.611635
    },
    {
        "usd": 2.88,
        "ts": 1670700600.6501918
    }
]
		

Еще немного кода. Как считать diff

Далее я приведу пример кода, с помощью которого мы будем получать сами изменения портфеля, а так же этот же код должен запускаться ботом.Как я уже писал выше — я использую Heroku. К тому же до недавнего времени данный сервис был бесплатным, но времена меняются и уже можно задуматься — стоит ли использовать Heroku или присягнуть AWS. Если вы, как и решите выбрать Heroku — я могу посоветовать данную статью.

Сам бот будет иметь одну ключевую команду — stats.

API ключей от Binance здесь уже не потребуется. Только token полученный при регистрации бота в Telegram и Dropbox token (вы же помните, Dropbox заменяет нам базу данных?). Для подсчета информации по неделе мы просто генерим список по балансам за неделю. При необходимости код несложно изменить и считать diff за любой временной срез.

			import logging
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters
import os
import dropbox
import time
import json
import datetime

PORT = int(os.environ.get('PORT', 5000))

# Enable logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)

logger = logging.getLogger(__name__)
TOKEN = os.environ.get('TELEGRAM_TOKEN', None)

dropboxToken = ''

def start(update, context):
    update.message.reply_text('Hi!')

def loadJsonFromDropbox(dbx):
	meta, resp = dbx.files_download('/totalData.txt')
	body = resp.content.decode('utf-8')
	resp.close()
	return json.loads(body)

def getHistory():
	dbx = dropbox.Dropbox(dropboxToken)
	prices = loadJsonFromDropbox(dbx)

	timeNow = time.time()
	dtNow = datetime.datetime.now()
	dtToday = dtNow.replace(hour=0, minute=0, second=1)
	dtWeek = dtToday - datetime.timedelta(days=dtToday.weekday())
	dtAllTime = dtNow - datetime.timedelta(days=100000)

	stats = {
		'this week': {
			'since': dtWeek.timestamp(),
			'till': dtNow.timestamp(),
			'prices': []
		},
		'all time': {
			'since': dtAllTime.timestamp(),
			'till': dtNow.timestamp(),
			'prices': []
		}
	}

	for item in prices:
		for stat in stats:
			if stats[stat]['since'] < item['ts'] < stats[stat]['till']:
				stats[stat]['prices'].append(item)

	text = ''
	totalBalance = 0.
	totalBalanceUsd = 0.
	for stat in stats:
		usdt = [p['usd'] for p in stats[stat]['prices']]

		if len(usdt) >= 1:
			u1 = usdt[-1]
			u2 = usdt[0]
			valueUsd = '{:+.2f} USD'.format(u1 - u2)
		else:
		    values = 'n/a'
		text += '{}: {}n'.format(stat, valueUsd)
		if stat == 'all time':
			totalBalanceUsd = u1
				
	dt = datetime.datetime.fromtimestamp(prices[-1]['ts'])
	text += 'nLast update: {:%A %H:%M:%S} UTC+0n'.format(dt)

	return update.message.reply_text(text, parse_mode='markdown')

def main():
    updater = Updater(TOKEN, use_context=True)
    dp = updater.dispatcher

    dp.add_handler(CommandHandler("start", start))
    dp.add_handler(CommandHandler("stats", getHistory))
    dp.add_handler(MessageHandler(Filters.text, echo))
    dp.add_error_handler(error)

    updater.start_webhook(listen="0.0.0.0", port=int(PORT), url_path=TOKEN)
    print(TOKEN + ' <- TOKEN | ' + str(PORT) + ' <- PORT')
    updater.bot.setWebhook('https://ваш_хероку_апп.com/' + TOKEN)

    updater.idle()

if __name__ == '__main__':
    main()
		

Вместо заключения

В результате применения команды stats вы должны получить в ответ, например, такое сообщение:

Создаём Telegram-бот для трекинга вашего криптопортфеля 4

Стоит отметить, что данного бота можно развивать и дальше до еще более прикладных применений.

Например, можно реализовать команду buy или sell. Можно подключить дополнительные аккаунты или сразу несколько бирж и мониторить все свои портфели в одном месте. Удобно? Удобно!

Для подключения дополнительных аккаунтов достаточно добавить еще один вызов в этом месте (и конечно же дополнительные api ключи — там они уже и так в list’е).

			amountUsd, timenow = getBalance(1)
		

Однако данные упражнения мы оставим дорогим читателям для самостоятельной работы. Всё же потребности у всех разные ?

В целом, все исходники уже представлены в статье, но на всякий случай — также они на GitHub.

Благодарю за внимание и буду рад ответить на ваши вопросы.

Следите за новыми постами
Следите за новыми постами по любимым темам
10К открытий11К показов