Мне был необходим для личного удобства бот в Телеграм, который умеет выполнять следующие действия:
- Показать текущий размер активов в USD(T).
- Показать изменения по портфелю в USD(T) за неделю и за все время.
Мне критически важно не использовать никакое приложение вида «все биржи в одном месте» в качестве proxy — да, я не хочу никому предоставлять свои ключи.
Если нужен — давайте напишем.
Что нам потребуется?
В качестве примера в данной статье я буду использовать криптовалютную биржу Binance. Стоит отметить, что никаких ограничений на использование с другими биржами нет — весь каркас тот же самый. Разве что взаимодействие по API будет немного другим (зависит от биржи).
- Нам потребуется функционал Telegram (куда же без него).
- Нам потребуется Dropbox вместо базы данных (о, да).
- Нам потребуется инструмент, где будет крутиться наш бот. Лично я использую Heroku, но можно использовать и AWS.
Создаем бота в Телеграм
Данный этап был описан сотни раз в огромном числе статей, найти которые не составляет проблем. Поэтому здесь будет кратко.
Нам нужно найти в телеграмме контакт @BotFather и выполнить последовательность команд
/start
/newbot
# Нужно ввести приватное имя нового бота и его публичное имя
# Я использую непопулярное публичное имя, т.к. тут будут наши финансы
# Здесь же мы получим API token нашего бота, сохраним его
# Запомним ссылку вида t.me/alias_binance_bot
База из топора
Для начала мы подготовим Dropbox.
Перейдем по ссылке и получим свой API token для Dropbox, нажав на кнопку Create App. Я создам доступ на отдельную папку в Dropbox.
На текущей странице нам необходимо будет сгенерировать OAuth 2.0 для Dropbox:
После создания пройдем на вкладку Permissions и установим права на files.content.write
Теперь по ссылке у нас появилась папка APPS. Зайдем в нее, далее зайдем в поддерикторию с названием нашего бота. Туда нам необходимо поместить файл totalData.txt, содержащий только пустой список.
[]Взаимодействие с Binance
Нам необходимо создать наш API ключ на бирже Binance.
Данная активность происходит по этой ссылке.
Результатом данного действия для нас будет API Key и Secret Key. В нашем случае будет достаточно прав только на чтение.
В качестве следующего шага мы напишем код, который будем использовать в дальнейшем.
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 вы должны получить в ответ, например, такое сообщение:
Стоит отметить, что данного бота можно развивать и дальше до еще более прикладных применений.
Например, можно реализовать команду buy или sell. Можно подключить дополнительные аккаунты или сразу несколько бирж и мониторить все свои портфели в одном месте. Удобно? Удобно!
Для подключения дополнительных аккаунтов достаточно добавить еще один вызов в этом месте (и конечно же дополнительные api ключи — там они уже и так в list’е).
amountUsd, timenow = getBalance(1)Однако данные упражнения мы оставим дорогим читателям для самостоятельной работы. Всё же потребности у всех разные 🙂
В целом, все исходники уже представлены в статье, но на всякий случай — также они на GitHub.
Благодарю за внимание и буду рад ответить на ваши вопросы.
