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

Apply pylint's linting rules to bfxapi/websocket/handlers/*.py.

Browse Source
This commit is contained in:
Davide Casale
2023-03-06 17:53:54 +01:00
parent a7f5aa684a
commit 7e627dd239
5 changed files with 62 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
bfxapi/websocket/handlers/authenticated_channels_handler.py
View File

@@ -1,18 +1,19 @@
from .. import serializers
from .. types import *
from .. serializers import _Notification
from .. exceptions import HandlerNotFound
class AuthenticatedChannelsHandler(object):
class AuthenticatedChannelsHandler:
__abbreviations = {
"os": "order_snapshot", "on": "order_new", "ou": "order_update", "oc": "order_cancel",
"ps": "position_snapshot", "pn": "position_new", "pu": "position_update", "pc": "position_close",
"te": "trade_executed", "tu": "trade_execution_update",
"fos": "funding_offer_snapshot", "fon": "funding_offer_new", "fou": "funding_offer_update", "foc": "funding_offer_cancel",
"fcs": "funding_credit_snapshot", "fcn": "funding_credit_new", "fcu": "funding_credit_update", "fcc": "funding_credit_close",
"fls": "funding_loan_snapshot", "fln": "funding_loan_new", "flu": "funding_loan_update", "flc": "funding_loan_close",
"ws": "wallet_snapshot", "wu": "wallet_update",
"os": "order_snapshot", "on": "order_new", "ou": "order_update",
"oc": "order_cancel", "ps": "position_snapshot", "pn": "position_new",
"pu": "position_update", "pc": "position_close", "te": "trade_executed",
"tu": "trade_execution_update", "fos": "funding_offer_snapshot", "fon": "funding_offer_new",
"fou": "funding_offer_update", "foc": "funding_offer_cancel", "fcs": "funding_credit_snapshot",
"fcn": "funding_credit_new", "fcu": "funding_credit_update", "fcc": "funding_credit_close",
"fls": "funding_loan_snapshot", "fln": "funding_loan_new", "flu": "funding_loan_update",
"flc": "funding_loan_close", "ws": "wallet_snapshot", "wu": "wallet_update",
"bu": "balance_update",
}
@@ -27,43 +28,42 @@ class AuthenticatedChannelsHandler(object):
("bu",): serializers.Balance
}
EVENTS = [
EVENTS = [
"notification",
"on-req-notification", "ou-req-notification", "oc-req-notification",
"oc_multi-notification",
"fon-req-notification", "foc-req-notification",
*list(__abbreviations.values())
*list(__abbreviations.values())
]
def __init__(self, event_emitter, strict = True):
self.event_emitter, self.strict = event_emitter, strict
def __init__(self, event_emitter):
self.event_emitter = event_emitter
def handle(self, type, stream):
if type == "n":
def handle(self, abbrevation, stream):
if abbrevation == "n":
return self.__notification(stream)
for types, serializer in AuthenticatedChannelsHandler.__serializers.items():
if type in types:
event = AuthenticatedChannelsHandler.__abbreviations[type]
for abbrevations, serializer in AuthenticatedChannelsHandler.__serializers.items():
if abbrevation in abbrevations:
event = AuthenticatedChannelsHandler.__abbreviations[abbrevation]
if all(isinstance(substream, list) for substream in stream):
return self.event_emitter.emit(event, [ serializer.parse(*substream) for substream in stream ])
return self.event_emitter.emit(event, serializer.parse(*stream))
if self.strict:
raise HandlerNotFound(f"No handler found for event of type <{type}>.")
raise HandlerNotFound(f"No handler found for event of type <{abbrevation}>.")
def __notification(self, stream):
type, serializer = "notification", serializers._Notification(serializer=None)
event, serializer = "notification", _Notification(serializer=None)
if stream[1] == "on-req" or stream[1] == "ou-req" or stream[1] == "oc-req":
type, serializer = f"{stream[1]}-notification", serializers._Notification(serializer=serializers.Order)
event, serializer = f"{stream[1]}-notification", _Notification(serializer=serializers.Order)
if stream[1] == "oc_multi-req":
type, serializer = f"{stream[1]}-notification", serializers._Notification(serializer=serializers.Order, iterate=True)
event, serializer = f"{stream[1]}-notification", _Notification(serializer=serializers.Order, iterate=True)
if stream[1] == "fon-req" or stream[1] == "foc-req":
type, serializer = f"{stream[1]}-notification", serializers._Notification(serializer=serializers.FundingOffer)
event, serializer = f"{stream[1]}-notification", _Notification(serializer=serializers.FundingOffer)
return self.event_emitter.emit(type, serializer.parse(*stream))
return self.event_emitter.emit(event, serializer.parse(*stream))