Получение данных о ценах с биржи Bybit с использованием Python

GetCoder

GetCoder.ru

Получайте данные о ценах криптовалют с биржи Bybit через WebSocket API, используя Python.

В этой статье мы рассмотрим пример использования Python для получения данных о ценах с биржи Bybit. Мы создадим простое приложение, которое подключится к WebSocket API биржи Bybit, чтобы получать данные о текущих ценах криптовалют.

Введение

Биржа Bybit предоставляет API, которое позволяет разработчикам получать данные о рынке и торговать криптовалютами. Мы будем использовать WebSocket API Bybit для получения данных о текущих ценах криптовалют.

Настройка проекта

Для работы с WebSocket API Bybit нам понадобится установить библиотеку websocket-client. Вы можете установить ее с помощью pip:


pip install websocket-client

Код


#-*- coding: UTF-8 -*-
import time, websocket, json, threading
from functools import partial

class bybit():
	def __init__(self):
		self.ticker_socket = {}
	
	# Выключения сокета
	def on_error(self, _wsa, wsData):
		_wsa.close()
	
	# Данные от сокета
	def on_message(self, _wsa, wsData):
		try:
			mData = json.loads(wsData)
			if 'op' in mData:
				if mData['op'] == 'subscribe' and mData['success'] == False:
					_wsa.close()
			else:
				self.ticker_socket[mData['data']['symbol']] = mData['data']
		except Exception as e:
			print(str(e))
		
	# Открытие соединения
	def on_open(self, _wsa, args):
		try:
			_wsa.send(json.dumps({"op": "subscribe", "args": args}))
		except Exception as e:
			_wsa.close()
	
	# Запуск сокета
	def run_web_sock(self, pairs):
		while True:
			try:
				wss = 'wss://stream.bybit.com/v5/public/spot'
				wsa = websocket.WebSocketApp(wss, 
				on_message = self.on_message,
				on_open = partial(self.on_open, args = pairs), 
				on_error = self.on_error)
				
				textLog = u'Установлено соединение Web Socket'
				print(textLog)
				
				wsa.run_forever()
				raise Exception(u'Разрыв соединения Web Socket')
			except Exception as e:
				print(str(e))
				time.sleep(10)
	
	# Старт ws
	def start(self, pairs):
		ws = threading.Thread(target=self.run_web_sock, kwargs={'pairs': pairs})
		ws.setDaemon(True)
		ws.start()

# Проверка
class main():
	def __init__(self):
		# Инициализация класса / запуск потока
		pairs = ["tickers.BTCUSDT", "tickers.DOGEUSDT"]
		wb = bybit()
		wb.start(pairs)
		
		# Вывод данных
		while True:
			time.sleep(5)
			print(wb.ticker_socket)
		
		
if __name__ == "__main__":
	main()

Как это работает

Мы создаем класс Bybit, который содержит методы для обработки WebSocket соединения с биржей Bybit.
Метод run_web_sock устанавливает соединение с WebSocket API Bybit и подписывается на получение данных о котировках для указанных торговых пар.
При получении сообщений о котировках метод on_message обрабатывает их и сохраняет данные в атрибут ticker_socket.
Класс Main инициализирует экземпляр класса Bybit и запускает поток для получения данных о котировках. Затем он выводит полученные данные каждые 5 секунд.

Заключение

В этой статье мы рассмотрели пример использования Python для получения данных о ценах с биржи Bybit с использованием WebSocket API. Вы можете использовать этот код как отправную точку для создания собственных инструментов и приложлений для работы с данными рынка криптовалют.

Введение

Настройка проекта

Для работы с WebSocket API Bybit нам понадобится установить библиотеку websocket-client. Вы можете установить ее с помощью pip:


pip install websocket-client

Код


#-*- coding: UTF-8 -*-
import time, websocket, json, threading
from functools import partial

class bybit():
	def __init__(self):
		self.ticker_socket = {}
	
	# Выключения сокета
	def on_error(self, _wsa, wsData):
		_wsa.close()
	
	# Данные от сокета
	def on_message(self, _wsa, wsData):
		try:
			mData = json.loads(wsData)
			if 'op' in mData:
				if mData['op'] == 'subscribe' and mData['success'] == False:
					_wsa.close()
			else:
				self.ticker_socket[mData['data']['symbol']] = mData['data']
		except Exception as e:
			print(str(e))
		
	# Открытие соединения
	def on_open(self, _wsa, args):
		try:
			_wsa.send(json.dumps({"op": "subscribe", "args": args}))
		except Exception as e:
			_wsa.close()
	
	# Запуск сокета
	def run_web_sock(self, pairs):
		while True:
			try:
				wss = 'wss://stream.bybit.com/v5/public/spot'
				wsa = websocket.WebSocketApp(wss, 
				on_message = self.on_message,
				on_open = partial(self.on_open, args = pairs), 
				on_error = self.on_error)
				
				textLog = u'Установлено соединение Web Socket'
				print(textLog)
				
				wsa.run_forever()
				raise Exception(u'Разрыв соединения Web Socket')
			except Exception as e:
				print(str(e))
				time.sleep(10)
	
	# Старт ws
	def start(self, pairs):
		ws = threading.Thread(target=self.run_web_sock, kwargs={'pairs': pairs})
		ws.setDaemon(True)
		ws.start()

# Проверка
class main():
	def __init__(self):
		# Инициализация класса / запуск потока
		pairs = ["tickers.BTCUSDT", "tickers.DOGEUSDT"]
		wb = bybit()
		wb.start(pairs)
		
		# Вывод данных
		while True:
			time.sleep(5)
			print(wb.ticker_socket)
		
		
if __name__ == "__main__":
	main()

Как это работает

Мы создаем класс Bybit, который содержит методы для обработки WebSocket соединения с биржей Bybit.
Метод run_web_sock устанавливает соединение с WebSocket API Bybit и подписывается на получение данных о котировках для указанных торговых пар.
При получении сообщений о котировках метод on_message обрабатывает их и сохраняет данные в атрибут ticker_socket.
Класс Main инициализирует экземпляр класса Bybit и запускает поток для получения данных о котировках. Затем он выводит полученные данные каждые 5 секунд.