Add filter parameter to BfxWebsocketClient's __init__.py. Rewrite .notify coroutine adding new feature. Add Notifications channel handlers in handlers.py. Add Notification serializer in serializers.py.

bfxapi/websocket/BfxWebsocketClient.py

@@ -30,10 +30,10 @@ class BfxWebsocketClient(object):
         *AuthenticatedChannelsHandler.EVENTS
     ]
 
-    def __init__(self, host, buckets=5, log_level = "INFO", API_KEY=None, API_SECRET=None):
+    def __init__(self, host, buckets=5, log_level = "INFO", API_KEY=None, API_SECRET=None, filter=None):
         self.host, self.websocket, self.event_emitter = host, None, AsyncIOEventEmitter()
 
-        self.API_KEY, self.API_SECRET, self.authentication = API_KEY, API_SECRET, False
+        self.API_KEY, self.API_SECRET, self.filter, self.authentication = API_KEY, API_SECRET, filter, False
 
         self.handler = AuthenticatedChannelsHandler(event_emitter=self.event_emitter)
 
@@ -45,7 +45,7 @@ class BfxWebsocketClient(object):
         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))
+            tasks.append(self.__connect(self.API_KEY, self.API_SECRET, self.filter))
 
         await asyncio.gather(*tasks)
 
@@ -113,6 +113,10 @@ class BfxWebsocketClient(object):
 
         return wrapper
 
+    @__require_websocket_authentication
+    async def notify(self, info, MESSAGE_ID=None, **kwargs):
+        await self.websocket.send(json.dumps([ 0, "n", MESSAGE_ID, { "type": "ucm-test", "info": info, **kwargs } ]))
+
     @__require_websocket_authentication
     async def new_order(self, data):
         await self.websocket.send(json.dumps([ 0, "on", None, data ]))

bfxapi/websocket/handlers.py

@@ -146,12 +146,15 @@ class AuthenticatedChannelsHandler(object):
         ("bu",): serializers.BalanceInfo
     }
 
-    EVENTS = list(__abbreviations.values())
+    EVENTS = [ "notification", *list(__abbreviations.values()) ]
 
     def __init__(self, event_emitter, strict = False):
         self.event_emitter, self.strict = event_emitter, strict
 
     def handle(self, type, stream):
+        if type == "n":
+            return self.__notification(stream)
+
         for types, serializer in AuthenticatedChannelsHandler.__serializers.items():
             if type in types:
                 event = AuthenticatedChannelsHandler.__abbreviations[type]
@@ -162,4 +165,7 @@ class AuthenticatedChannelsHandler(object):
                 return self.event_emitter.emit(event, serializer.parse(*stream))
         
         if self.strict == True:
-            raise BfxWebsocketException(f"Event of type <{type}> not found in self.__handlers.")
+            raise BfxWebsocketException(f"Event of type <{type}> not found in self.__handlers.")
+    
+    def __notification(self, stream):
+        return self.event_emitter.emit("notification", serializers.Notification.parse(*stream))

bfxapi/websocket/serializers.py

@@ -305,4 +305,19 @@ BalanceInfo = _Serializer("BalanceInfo", labels=[
     "AUM_NET", 
 ])
 
+#endregion
+
+#region Serializers definition for Notifications channel
+
+Notification = _Serializer("Notification", labels=[
+    "MTS",
+    "TYPE",
+    "MESSAGE_ID",
+    "_PLACEHOLDER",
+    "NOTIFY_INFO",
+    "CODE",
+    "STATUS",
+    "TEXT"
+])
+
 #endregion