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

Rewrite reconnection system with numerous fixes.

Browse Source
This commit is contained in:
Davide Casale
2023-02-15 21:48:34 +01:00
parent 99f58ddb04
commit fa9bdfc333
9 changed files with 98 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
bfxapi/websocket/client/bfx_websocket_bucket.py
View File

@@ -24,27 +24,35 @@ class BfxWebsocketBucket(object):
MAXIMUM_SUBSCRIPTIONS_AMOUNT = 25
def __init__(self, host, event_emitter, __bucket_open_signal):
self.host, self.event_emitter, self.__bucket_open_signal = host, event_emitter, __bucket_open_signal
def __init__(self, host, event_emitter, on_open_event):
self.host, self.event_emitter, self.on_open_event = host, event_emitter, on_open_event
self.websocket, self.subscriptions, self.pendings = None, dict(), list()
self.handler = PublicChannelsHandler(event_emitter=self.event_emitter)
async def _connect(self, index):
reconnection = False
async for websocket in websockets.connect(self.host):
self.websocket = websocket
self.__bucket_open_signal(index)
if reconnection == True or (reconnection := False):
for pending in self.pendings:
await self.websocket.send(json.dumps(pending))
for _, subscription in self.subscriptions.items():
await self._subscribe(**subscription)
self.subscriptions.clear()
self.on_open_event.set()
try:
async for message in websocket:
message = json.loads(message)
if isinstance(message, dict) and message["event"] == "info" and "version" in message:
if BfxWebsocketBucket.VERSION != message["version"]:
raise OutdatedClientVersion(f"Mismatch between the client version and the server version. Update the library to the latest version to continue (client version: {BfxWebsocketBucket.VERSION}, server version: {message['version']}).")
elif isinstance(message, dict) and message["event"] == "subscribed" and (chanId := message["chanId"]):
if isinstance(message, dict) and message["event"] == "subscribed" and (chanId := message["chanId"]):
self.pendings = [ pending for pending in self.pendings if pending["subId"] != message["subId"] ]
self.subscriptions[chanId] = message
self.event_emitter.emit("subscribed", message)
@@ -55,8 +63,9 @@ class BfxWebsocketBucket(object):
self.event_emitter.emit("wss-error", message["code"], message["msg"])
elif isinstance(message, list) and (chanId := message[0]) and message[1] != _HEARTBEAT:
self.handler.handle(self.subscriptions[chanId], *message[1:])
except websockets.ConnectionClosedError: continue
finally: await self.websocket.wait_closed(); break
except websockets.ConnectionClosedError: self.on_open_event.clear(); reconnection = True; continue
await self.websocket.wait_closed(); break
@_require_websocket_connection
async def _subscribe(self, channel, subId=None, **kwargs):
@@ -64,11 +73,11 @@ class BfxWebsocketBucket(object):
raise TooManySubscriptions("The client has reached the maximum number of subscriptions.")
subscription = {
**kwargs,
"event": "subscribe",
"channel": channel,
"subId": subId or str(uuid.uuid4()),
**kwargs
}
self.pendings.append(subscription)

79
bfxapi/websocket/client/bfx_websocket_client.py
View File

@@ -2,13 +2,17 @@ import traceback, json, asyncio, hmac, hashlib, time, websockets
from typing import cast
from collections import namedtuple
from datetime import datetime
from pyee.asyncio import AsyncIOEventEmitter
from .bfx_websocket_bucket import _HEARTBEAT, F, _require_websocket_connection, BfxWebsocketBucket
from .bfx_websocket_inputs import BfxWebsocketInputs
from ..handlers import PublicChannelsHandler, AuthenticatedChannelsHandler
from ..exceptions import WebsocketAuthenticationRequired, InvalidAuthenticationCredentials, EventNotSupported
from ..exceptions import WebsocketAuthenticationRequired, InvalidAuthenticationCredentials, EventNotSupported, OutdatedClientVersion
from ...utils.JSONEncoder import JSONEncoder
@@ -16,7 +20,7 @@ from ...utils.logger import Formatter, CustomLogger
def _require_websocket_authentication(function: F) -> F:
async def wrapper(self, *args, **kwargs):
if self.authentication == False:
if hasattr(self, "authentication") and self.authentication == False:
raise WebsocketAuthenticationRequired("To perform this action you need to authenticate using your API_KEY and API_SECRET.")
await _require_websocket_connection(function)(self, *args, **kwargs)
@@ -26,7 +30,7 @@ def _require_websocket_authentication(function: F) -> F:
class BfxWebsocketClient(object):
VERSION = BfxWebsocketBucket.VERSION
MAXIMUM_BUCKETS_AMOUNT = 20
MAXIMUM_CONNECTIONS_AMOUNT = 20
EVENTS = [
"open", "subscribed", "authenticated", "wss-error",
@@ -34,10 +38,12 @@ class BfxWebsocketClient(object):
*AuthenticatedChannelsHandler.EVENTS
]
def __init__(self, host, API_KEY = None, API_SECRET = None, filter = None, buckets = 5, log_level = "WARNING"):
def __init__(self, host, API_KEY = None, API_SECRET = None, filter = None, log_level = "WARNING"):
self.host, self.websocket, self.event_emitter = host, None, AsyncIOEventEmitter()
self.API_KEY, self.API_SECRET, self.filter, self.authentication = API_KEY, API_SECRET, filter, False
self.API_KEY, self.API_SECRET, self.filter = API_KEY, API_SECRET, filter
self.inputs = BfxWebsocketInputs(handle_websocket_input=self.__handle_websocket_input)
self.handler = AuthenticatedChannelsHandler(event_emitter=self.event_emitter)
@@ -48,36 +54,58 @@ class BfxWebsocketClient(object):
str().join(traceback.format_exception(type(exception), exception, exception.__traceback__))[:-1])
)
if buckets > BfxWebsocketClient.MAXIMUM_BUCKETS_AMOUNT:
self.logger.warning(f"It is not safe to use more than {BfxWebsocketClient.MAXIMUM_BUCKETS_AMOUNT} buckets from the same \
connection ({buckets} in use), the server could momentarily block the client with <429 Too Many Requests>.")
def run(self, connections = 5):
return asyncio.run(self.start(connections))
self.buckets = [ BfxWebsocketBucket(self.host, self.event_emitter, self.__bucket_open_signal) for _ in range(buckets) ]
async def start(self, connections = 5):
if connections > BfxWebsocketClient.MAXIMUM_CONNECTIONS_AMOUNT:
self.logger.warning(f"It is not safe to use more than {BfxWebsocketClient.MAXIMUM_CONNECTIONS_AMOUNT} buckets from the same " +
f"connection ({connections} in use), the server could momentarily block the client with <429 Too Many Requests>.")
self.inputs = BfxWebsocketInputs(self.__handle_websocket_input)
self.on_open_events = [ asyncio.Event() for _ in range(connections) ]
def run(self):
return asyncio.run(self.start())
self.buckets = [
BfxWebsocketBucket(self.host, self.event_emitter, self.on_open_events[index])
for index in range(connections)
]
async def start(self):
tasks = [ bucket._connect(index) for index, bucket in enumerate(self.buckets) ]
if self.API_KEY != None and self.API_SECRET != None:
tasks.append(self.__connect(self.API_KEY, self.API_SECRET, self.filter))
tasks.append(self.__connect(self.API_KEY, self.API_SECRET, self.filter))
await asyncio.gather(*tasks)
async def __connect(self, API_KEY, API_SECRET, filter=None):
Reconnection = namedtuple("Reconnection", ["status", "code", "timestamp"])
reconnection = Reconnection(status=False, code=0, timestamp=None)
async for websocket in websockets.connect(self.host):
self.websocket = websocket
await self.__authenticate(API_KEY, API_SECRET, filter)
self.websocket, self.authentication = websocket, False
if (await asyncio.gather(*[ on_open_event.wait() for on_open_event in self.on_open_events ])):
self.event_emitter.emit("open")
if self.API_KEY != None and self.API_SECRET != None:
await self.__authenticate(API_KEY=API_KEY, API_SECRET=API_SECRET, filter=filter)
try:
async for message in websocket:
if reconnection.status == True:
self.logger.warning(f"Reconnect Attempt Successful (error <{reconnection.code}>): The " +
f"client has been offline for a total of {datetime.now() - reconnection.timestamp} " +
f"(first reconnection attempt: {reconnection.timestamp:%d-%m-%Y at %H:%M:%S}).")
reconnection = Reconnection(status=False, code=0, timestamp=None)
message = json.loads(message)
if isinstance(message, dict) and message["event"] == "auth":
if isinstance(message, dict) and message["event"] == "info" and "version" in message:
if BfxWebsocketClient.VERSION != message["version"]:
raise OutdatedClientVersion(f"Mismatch between the client version and the server version. " +
f"Update the library to the latest version to continue (client version: {BfxWebsocketClient.VERSION}, " +
f"server version: {message['version']}).")
elif isinstance(message, dict) and message["event"] == "auth":
if message["status"] == "OK":
self.event_emitter.emit("authenticated", message); self.authentication = True
else: raise InvalidAuthenticationCredentials("Cannot authenticate with given API-KEY and API-SECRET.")
@@ -85,8 +113,13 @@ class BfxWebsocketClient(object):
self.event_emitter.emit("wss-error", message["code"], message["msg"])
elif isinstance(message, list) and (chanId := message[0]) == 0 and message[1] != _HEARTBEAT:
self.handler.handle(message[1], message[2])
except websockets.ConnectionClosedError: continue
finally: await self.websocket.wait_closed(); break
except websockets.ConnectionClosedError as error:
self.logger.error(f"Connection terminated due to an error (status code: <{error.code}>) -> {str(error)}. Attempting to reconnect...")
reconnection = Reconnection(status=True, code=error.code, timestamp=datetime.now());
continue
if reconnection.status == False:
await self.websocket.wait_closed(); break
async def __authenticate(self, API_KEY, API_SECRET, filter=None):
data = { "event": "auth", "filter": filter, "apiKey": API_KEY }
@@ -130,10 +163,6 @@ class BfxWebsocketClient(object):
async def __handle_websocket_input(self, input, data):
await self.websocket.send(json.dumps([ 0, input, None, data], cls=JSONEncoder))
def __bucket_open_signal(self, index):
if all(bucket.websocket != None and bucket.websocket.open == True for bucket in self.buckets):
self.event_emitter.emit("open")
def on(self, *events, callback = None):
for event in events:
if event not in BfxWebsocketClient.EVENTS:

18
bfxapi/websocket/client/bfx_websocket_inputs.py
View File

@@ -6,15 +6,15 @@ from .. enums import OrderType, FundingOfferType
from ... utils.JSONEncoder import JSON
class BfxWebsocketInputs(object):
def __init__(self, __handle_websocket_input):
self.__handle_websocket_input = __handle_websocket_input
def __init__(self, handle_websocket_input):
self.handle_websocket_input = handle_websocket_input
async def submit_order(self, type: OrderType, symbol: str, amount: Union[Decimal, float, str],
price: Optional[Union[Decimal, float, str]] = None, lev: Optional[int] = None,
price_trailing: Optional[Union[Decimal, float, str]] = None, price_aux_limit: Optional[Union[Decimal, float, str]] = None, price_oco_stop: Optional[Union[Decimal, float, str]] = None,
gid: Optional[int] = None, cid: Optional[int] = None,
flags: Optional[int] = 0, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None):
await self.__handle_websocket_input("on", {
await self.handle_websocket_input("on", {
"type": type, "symbol": symbol, "amount": amount,
"price": price, "lev": lev,
"price_trailing": price_trailing, "price_aux_limit": price_aux_limit, "price_oco_stop": price_oco_stop,
@@ -26,7 +26,7 @@ class BfxWebsocketInputs(object):
cid: Optional[int] = None, cid_date: Optional[str] = None, gid: Optional[int] = None,
flags: Optional[int] = 0, lev: Optional[int] = None, delta: Optional[Union[Decimal, float, str]] = None,
price_aux_limit: Optional[Union[Decimal, float, str]] = None, price_trailing: Optional[Union[Decimal, float, str]] = None, tif: Optional[Union[datetime, str]] = None):
await self.__handle_websocket_input("ou", {
await self.handle_websocket_input("ou", {
"id": id, "amount": amount, "price": price,
"cid": cid, "cid_date": cid_date, "gid": gid,
"flags": flags, "lev": lev, "delta": delta,
@@ -34,12 +34,12 @@ class BfxWebsocketInputs(object):
})
async def cancel_order(self, id: Optional[int] = None, cid: Optional[int] = None, cid_date: Optional[str] = None):
await self.__handle_websocket_input("oc", {
await self.handle_websocket_input("oc", {
"id": id, "cid": cid, "cid_date": cid_date
})
async def cancel_order_multi(self, ids: Optional[List[int]] = None, cids: Optional[List[Tuple[int, str]]] = None, gids: Optional[List[int]] = None, all: bool = False):
await self.__handle_websocket_input("oc_multi", {
await self.handle_websocket_input("oc_multi", {
"ids": ids, "cids": cids, "gids": gids,
"all": int(all)
})
@@ -47,14 +47,14 @@ class BfxWebsocketInputs(object):
async def submit_funding_offer(self, type: FundingOfferType, symbol: str, amount: Union[Decimal, float, str],
rate: Union[Decimal, float, str], period: int,
flags: Optional[int] = 0):
await self.__handle_websocket_input("fon", {
await self.handle_websocket_input("fon", {
"type": type, "symbol": symbol, "amount": amount,
"rate": rate, "period": period,
"flags": flags
})
async def cancel_funding_offer(self, id: int):
await self.__handle_websocket_input("foc", { "id": id })
await self.handle_websocket_input("foc", { "id": id })
async def calc(self, *args: str):
await self.__handle_websocket_input("calc", list(map(lambda arg: [arg], args)))
await self.handle_websocket_input("calc", list(map(lambda arg: [arg], args)))

2
bfxapi/websocket/enums.py
View File

@@ -1,6 +1,6 @@
from .. enums import *
class Channels(str, Enum):
class Channel(str, Enum):
TICKER = "ticker"
TRADES = "trades"
BOOK = "book"

2
bfxapi/websocket/handlers/public_channels_handler.py
View File

@@ -28,7 +28,7 @@ class PublicChannelsHandler(object):
_clear = lambda dictionary, *args: { key: value for key, value in dictionary.items() if key not in args }
if (channel := subscription["channel"]) and channel in self.__handlers.keys():
return self.__handlers[channel](_clear(subscription, "event", "channel", "subId"), *stream)
return self.__handlers[channel](_clear(subscription, "event", "channel", "chanId"), *stream)
if self.strict:
raise HandlerNotFound(f"No handler found for channel <{subscription['channel']}>.")

14
bfxapi/websocket/subscriptions.py
View File

@@ -1,4 +1,4 @@
from typing import TypedDict, Union, Optional
from typing import TypedDict, Union, Literal, Optional
__all__ = [
"Subscription",
@@ -10,22 +10,22 @@ __all__ = [
"Status"
]
_Header = TypedDict("_Header", { "event": str, "channel": str, "subId": str })
_Header = TypedDict("_Header", { "event": Literal["subscribed"], "channel": str, "chanId": int })
Subscription = Union["Ticker", "Trades", "Book", "Candles", "Status"]
class Ticker(TypedDict):
chanId: int; symbol: str
subId: str; symbol: str
pair: Optional[str]
currency: Optional[str]
class Trades(TypedDict):
chanId: int; symbol: str
subId: str; symbol: str
pair: Optional[str]
currency: Optional[str]
class Book(TypedDict):
chanId: int
subId: str
symbol: str
prec: str
freq: str
@@ -33,9 +33,9 @@ class Book(TypedDict):
pair: str
class Candles(TypedDict):
chanId: int
subId: str
key: str
class Status(TypedDict):
chanId: int
subId: str
key: str

4
examples/websocket/order_book.py
View File

@@ -7,7 +7,7 @@ from typing import List
from bfxapi import Client, Constants
from bfxapi.websocket import subscriptions
from bfxapi.websocket.enums import Channels, Error
from bfxapi.websocket.enums import Channel, Error
from bfxapi.websocket.types import TradingPairBook
class OrderBook(object):
@@ -47,7 +47,7 @@ def on_wss_error(code: Error, msg: str):
@bfx.wss.on("open")
async def on_open():
for symbol in SYMBOLS:
await bfx.wss.subscribe(Channels.BOOK, symbol=symbol)
await bfx.wss.subscribe(Channel.BOOK, symbol=symbol)
@bfx.wss.on("subscribed")
def on_subscribed(subscription):

4
examples/websocket/raw_order_book.py
View File

@@ -7,7 +7,7 @@ from typing import List
from bfxapi import Client, Constants
from bfxapi.websocket import subscriptions
from bfxapi.websocket.enums import Channels, Error
from bfxapi.websocket.enums import Channel, Error
from bfxapi.websocket.types import TradingPairRawBook
class RawOrderBook(object):
@@ -47,7 +47,7 @@ def on_wss_error(code: Error, msg: str):
@bfx.wss.on("open")
async def on_open():
for symbol in SYMBOLS:
await bfx.wss.subscribe(Channels.BOOK, symbol=symbol, prec="R0")
await bfx.wss.subscribe(Channel.BOOK, symbol=symbol, prec="R0")
@bfx.wss.on("subscribed")
def on_subscribed(subscription):

4
examples/websocket/ticker.py
View File

@@ -3,7 +3,7 @@
from bfxapi import Client, Constants
from bfxapi.websocket import subscriptions
from bfxapi.websocket.enums import Channels
from bfxapi.websocket.enums import Channel
from bfxapi.websocket.types import TradingPairTicker
bfx = Client(WSS_HOST=Constants.PUB_WSS_HOST)
@@ -16,6 +16,6 @@ def on_t_ticker_update(subscription: subscriptions.Ticker, data: TradingPairTick
@bfx.wss.once("open")
async def open():
await bfx.wss.subscribe(Channels.TICKER, symbol="tBTCUSD")
await bfx.wss.subscribe(Channel.TICKER, symbol="tBTCUSD")
bfx.wss.run()