diff --git a/bfxapi/websocket/client/bfx_websocket_bucket.py b/bfxapi/websocket/client/bfx_websocket_bucket.py index ef262b8..ca8625e 100644 --- a/bfxapi/websocket/client/bfx_websocket_bucket.py +++ b/bfxapi/websocket/client/bfx_websocket_bucket.py @@ -18,13 +18,14 @@ class BfxWebSocketBucket: MAXIMUM_SUBSCRIPTIONS_AMOUNT = 25 - def __init__(self, host, event_emitter, events_per_subscription): - self.host, self.event_emitter, self.events_per_subscription = host, event_emitter, events_per_subscription - self.websocket, self.subscriptions, self.pendings = None, {}, [] - self.condition = asyncio.locks.Condition() + def __init__(self, host, event_emitter): + self.host, self.websocket, self.event_emitter = \ + host, None, event_emitter - self.handler = PublicChannelsHandler(event_emitter=self.event_emitter, \ - events_per_subscription=self.events_per_subscription) + self.condition, self.subscriptions, self.pendings = \ + asyncio.locks.Condition(), {}, [] + + self.handler = PublicChannelsHandler(event_emitter=self.event_emitter) async def connect(self): async with websockets.connect(self.host) as websocket: @@ -45,11 +46,7 @@ class BfxWebSocketBucket: self.subscriptions[chan_id] = message - sub_id = message["subId"] - - if "subscribed" not in self.events_per_subscription.get(sub_id, []): - self.events_per_subscription.setdefault(sub_id, []).append("subscribed") - self.event_emitter.emit("subscribed", message) + self.event_emitter.emit("subscribed", message) elif message["event"] == "unsubscribed" and (chan_id := message["chanId"]): if message["status"] == "OK": del self.subscriptions[chan_id] diff --git a/bfxapi/websocket/client/bfx_websocket_client.py b/bfxapi/websocket/client/bfx_websocket_client.py index 916a722..515e11b 100644 --- a/bfxapi/websocket/client/bfx_websocket_client.py +++ b/bfxapi/websocket/client/bfx_websocket_client.py @@ -71,8 +71,6 @@ class BfxWebSocketClient: self.host, self.credentials, self.wss_timeout = host, credentials, wss_timeout - self.events_per_subscription = {} - self.event_emitter = AsyncIOEventEmitter() self.handler = AuthEventsHandler(event_emitter=self.event_emitter) @@ -102,7 +100,7 @@ class BfxWebSocketClient: "block the client with <429 Too Many Requests>.") for _ in range(connections): - self.buckets += [BfxWebSocketBucket(self.host, self.event_emitter, self.events_per_subscription)] + self.buckets += [BfxWebSocketBucket(self.host, self.event_emitter)] await self.__connect() diff --git a/bfxapi/websocket/handlers/public_channels_handler.py b/bfxapi/websocket/handlers/public_channels_handler.py index 6a4a919..4b09bf0 100644 --- a/bfxapi/websocket/handlers/public_channels_handler.py +++ b/bfxapi/websocket/handlers/public_channels_handler.py @@ -1,5 +1,5 @@ from typing import TYPE_CHECKING, \ - Union, Dict, List, Any, cast + Union, List, Any, cast from bfxapi.types import serializers @@ -28,11 +28,8 @@ class PublicChannelsHandler: "liquidation_feed_update" ] - def __init__(self, - event_emitter: "EventEmitter", - events_per_subscription: Dict[str, List[str]]) -> None: - self.__event_emitter, self.__events_per_subscription = \ - event_emitter, events_per_subscription + def __init__(self, event_emitter: "EventEmitter") -> None: + self.__event_emitter = event_emitter def handle(self, subscription: "Subscription", stream: List[Any]) -> None: def _strip(subscription: "Subscription", *args: str) -> "_NoHeaderSubscription": @@ -46,97 +43,101 @@ class PublicChannelsHandler: elif subscription["channel"] == "trades": self.__trades_channel_handler(cast("Trades", _subscription), stream) elif subscription["channel"] == "book": - self.__book_channel_handler(cast("Book", _subscription), stream) + _subscription = cast("Book", _subscription) + + if _subscription["prec"] != "R0": + self.__book_channel_handler(_subscription, stream) + else: + self.__raw_book_channel_handler(_subscription, stream) elif subscription["channel"] == "candles": self.__candles_channel_handler(cast("Candles", _subscription), stream) elif subscription["channel"] == "status": self.__status_channel_handler(cast("Status", _subscription), stream) - def __emit(self, event: str, subscription: "_NoHeaderSubscription", data: Any) -> None: - sub_id, should_emit_event = subscription["subId"], True - - if event in PublicChannelsHandler.ONCE_PER_SUBSCRIPTION_EVENTS: - if sub_id not in self.__events_per_subscription: - self.__events_per_subscription[sub_id] = [ event ] - elif event not in self.__events_per_subscription[sub_id]: - self.__events_per_subscription[sub_id] += [ event ] - else: should_emit_event = False - - if should_emit_event: - self.__event_emitter.emit(event, subscription, data) - - def __ticker_channel_handler(self, subscription: "Ticker", stream: List[Any]) -> None: + def __ticker_channel_handler(self, subscription: "Ticker", stream: List[Any]): if subscription["symbol"].startswith("t"): - return self.__emit("t_ticker_update", subscription, \ + return self.__event_emitter.emit("t_ticker_update", subscription, \ serializers.TradingPairTicker.parse(*stream[0])) if subscription["symbol"].startswith("f"): - return self.__emit("f_ticker_update", subscription, \ + return self.__event_emitter.emit("f_ticker_update", subscription, \ serializers.FundingCurrencyTicker.parse(*stream[0])) - def __trades_channel_handler(self, subscription: "Trades", stream: List[Any]) -> None: + def __trades_channel_handler(self, subscription: "Trades", stream: List[Any]): if (event := stream[0]) and event in [ "te", "tu", "fte", "ftu" ]: events = { "te": "t_trade_execution", "tu": "t_trade_execution_update", \ "fte": "f_trade_execution", "ftu": "f_trade_execution_update" } if subscription["symbol"].startswith("t"): - return self.__emit(events[event], subscription, \ + return self.__event_emitter.emit(events[event], subscription, \ serializers.TradingPairTrade.parse(*stream[1])) if subscription["symbol"].startswith("f"): - return self.__emit(events[event], subscription, \ + return self.__event_emitter.emit(events[event], subscription, \ serializers.FundingCurrencyTrade.parse(*stream[1])) if subscription["symbol"].startswith("t"): - return self.__emit("t_trades_snapshot", subscription, \ + return self.__event_emitter.emit("t_trades_snapshot", subscription, \ [ serializers.TradingPairTrade.parse(*sub_stream) \ for sub_stream in stream[0] ]) if subscription["symbol"].startswith("f"): - return self.__emit("f_trades_snapshot", subscription, \ + return self.__event_emitter.emit("f_trades_snapshot", subscription, \ [ serializers.FundingCurrencyTrade.parse(*sub_stream) \ for sub_stream in stream[0] ]) - def __book_channel_handler(self, subscription: "Book", stream: List[Any]) -> None: - symbol = subscription["symbol"] + def __book_channel_handler(self, subscription: "Book", stream: List[Any]): + if subscription["symbol"].startswith("t"): + if all(isinstance(sub_stream, list) for sub_stream in stream[0]): + return self.__event_emitter.emit("t_book_snapshot", subscription, \ + [ serializers.TradingPairBook.parse(*sub_stream) \ + for sub_stream in stream[0] ]) - is_raw_book = subscription["prec"] == "R0" + return self.__event_emitter.emit("t_book_update", subscription, \ + serializers.TradingPairBook.parse(*stream[0])) - serializer = { - "t": is_raw_book and serializers.TradingPairRawBook \ - or serializers.TradingPairBook, - "f": is_raw_book and serializers.FundingCurrencyRawBook \ - or serializers.FundingCurrencyBook - }[symbol[0]] + if subscription["symbol"].startswith("f"): + if all(isinstance(sub_stream, list) for sub_stream in stream[0]): + return self.__event_emitter.emit("f_book_snapshot", subscription, \ + [ serializers.FundingCurrencyBook.parse(*sub_stream) \ + for sub_stream in stream[0] ]) + return self.__event_emitter.emit("f_book_update", subscription, \ + serializers.FundingCurrencyBook.parse(*stream[0])) + + def __raw_book_channel_handler(self, subscription: "Book", stream: List[Any]): + if subscription["symbol"].startswith("t"): + if all(isinstance(sub_stream, list) for sub_stream in stream[0]): + self.__event_emitter.emit("t_raw_book_snapshot", subscription, \ + [ serializers.TradingPairRawBook.parse(*sub_stream) \ + for sub_stream in stream[0] ]) + + return self.__event_emitter.emit("t_raw_book_update", subscription, \ + serializers.TradingPairRawBook.parse(*stream[0])) + + if subscription["symbol"].startswith("f"): + if all(isinstance(sub_stream, list) for sub_stream in stream[0]): + return self.__event_emitter.emit("f_raw_book_snapshot", subscription, \ + [ serializers.FundingCurrencyRawBook.parse(*sub_stream) \ + for sub_stream in stream[0] ]) + + return self.__event_emitter.emit("f_raw_book_update", subscription, \ + serializers.FundingCurrencyRawBook.parse(*stream[0])) + + def __candles_channel_handler(self, subscription: "Candles", stream: List[Any]): if all(isinstance(sub_stream, list) for sub_stream in stream[0]): - event = symbol[0] + "_" + \ - (is_raw_book and "raw_book" or "book") + "_snapshot" - - return self.__emit(event, subscription, \ - [ serializer.parse(*sub_stream) \ - for sub_stream in stream[0] ]) - - event = symbol[0] + "_" + \ - (is_raw_book and "raw_book" or "book") + "_update" - - return self.__emit(event, subscription, \ - serializer.parse(*stream[0])) - - def __candles_channel_handler(self, subscription: "Candles", stream: List[Any]) -> None: - if all(isinstance(sub_stream, list) for sub_stream in stream[0]): - return self.__emit("candles_snapshot", subscription, \ + return self.__event_emitter.emit("candles_snapshot", subscription, \ [ serializers.Candle.parse(*sub_stream) \ for sub_stream in stream[0] ]) - return self.__emit("candles_update", subscription, \ + return self.__event_emitter.emit("candles_update", subscription, \ serializers.Candle.parse(*stream[0])) - def __status_channel_handler(self, subscription: "Status", stream: List[Any]) -> None: + def __status_channel_handler(self, subscription: "Status", stream: List[Any]): if subscription["key"].startswith("deriv:"): - return self.__emit("derivatives_status_update", subscription, \ + return self.__event_emitter.emit("derivatives_status_update", subscription, \ serializers.DerivativesStatus.parse(*stream[0])) if subscription["key"].startswith("liq:"): - return self.__emit("liquidation_feed_update", subscription, \ + return self.__event_emitter.emit("liquidation_feed_update", subscription, \ serializers.Liquidation.parse(*stream[0][0]))