aljaz/bitfinex-api-py
1
0
Fork 0
mirror of https://github.com/aljazceru/bitfinex-api-py.git synced 2025-12-22 08:14:20 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add handlers in manager.py for CANDLES and STATUS channels. Add support for heartbeat data.

Browse Source
This commit is contained in:
Davide Casale
2022-11-08 18:16:24 +01:00
parent a37b7dda33
commit 7e416219e0
2 changed files with 19 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
bfxapi/websocket/channels.py
View File

@@ -3,4 +3,6 @@ from enum import Enum
class Channels(str, Enum):
TICKER = "ticker"
TRADES = "trades"
BOOK = "book"
BOOK = "book"
CANDLES = "candles"
STATUS = "status"

19
bfxapi/websocket/manager.py
View File

@@ -1,5 +1,7 @@
from .channels import Channels
HEARTBEAT = "hb"
class Manager(object):
def __init__(self, event_emitter):
self.event_emitter = event_emitter
@@ -7,11 +9,14 @@ class Manager(object):
self.__handlers = {
Channels.TICKER: self.__ticker_channel_handler,
Channels.TRADES: self.__trades_channel_handler,
Channels.BOOK: self.__book_channel_handler
Channels.BOOK: self.__book_channel_handler,
Channels.CANDLES: self.__candles_channel_handler,
Channels.STATUS: self.__status_channel_handler,
}
def handle(self, subscription, *parameters):
return self.__handlers[subscription["channel"]](subscription, *parameters)
if parameters[0] != HEARTBEAT:
self.__handlers[subscription["channel"]](subscription, *parameters)
def __ticker_channel_handler(self, subscription, *parameters):
self.event_emitter.emit("ticker", subscription, parameters[0])
@@ -26,4 +31,12 @@ class Manager(object):
def __book_channel_handler(self, subscription, *parameters):
if all(isinstance(element, list) for element in parameters[0]):
self.event_emitter.emit("book_snapshot", subscription, parameters[0])
else: self.event_emitter.emit("book_update", subscription, parameters[0])
else: self.event_emitter.emit("book_update", subscription, parameters[0])
def __candles_channel_handler(self, subscription, *parameters):
if all(isinstance(element, list) for element in parameters[0]):
self.event_emitter.emit("candles_snapshot", subscription, parameters[0])
else: self.event_emitter.emit("candles_update", subscription, parameters[0])
def __status_channel_handler(self, subscription, *parameters):
self.event_emitter.emit("status", subscription, parameters[0])