impl sub by exch

allow setup.py and "pep517" to co-exist for now

Author: mvberg

openfeed/openfeed_client.py

@@ -22,6 +22,7 @@ def __init__(self, username, password, server="openfeed.aws.barchart.com", debug
 
         self.instrument_definitions = {}
         self.instruments_by_symbol = {}
+
         self.symbol_handlers = {}
         self.exchange_handlers = {}
         self.heartbeat_handlers = []
@@ -106,50 +107,61 @@ def handleSubscriptionResponse(msg):
             if msg.subscriptionResponse.status.result > 1:
                 raise Exception("Subscription has failed", msg)
 
-            self.__notify_symbol_listeners(
-                msg.subscriptionResponse.symbol, msg)
+            if len(msg.subscriptionResponse.symbol) > 0:
+                self.__notify_symbol_listeners(
+                    msg.subscriptionResponse.symbol, msg)
+            else:
+                self.__notify_exchange_listeners(
+                    msg.subscriptionResponse.exchange, msg)
+
             return msg
 
         def handleInstrumentDefinition(msg):
             self.instrument_definitions[msg.instrumentDefinition.marketId] = msg
             self.instruments_by_symbol[msg.instrumentDefinition.symbol] = msg
+
             return msg
 
         def handleMarketUpdate(msg):
-            symbol = self.instrument_definitions[msg.marketUpdate.marketId].instrumentDefinition.symbol
+            inst = self.instrument_definitions[msg.marketUpdate.marketId].instrumentDefinition
 
-            self.__notify_symbol_listeners(symbol, msg)
+            self.__notify_exchange_listeners(inst.barchartExchangeCode, msg)
+            self.__notify_symbol_listeners(inst.symbol, msg)
 
             return msg
 
         def handleMarketSnapshot(msg):
-            symbol = self.instrument_definitions[msg.marketSnapshot.marketId].instrumentDefinition.symbol
+            inst = self.instrument_definitions[msg.marketSnapshot.marketId].instrumentDefinition
+
+            self.__notify_exchange_listeners(inst.barchartExchangeCode, msg)
+            self.__notify_symbol_listeners(inst.symbol, msg)
 
-            self.__notify_symbol_listeners(symbol, msg)
             return msg
 
+        handlers = {
+            "loginResponse": handleLogin,
+            "heartBeat": handleHeartbeat,
+            "subscriptionResponse": handleSubscriptionResponse,
+            "instrumentDefinition": handleInstrumentDefinition,
+            "marketSnapshot": handleMarketSnapshot,
+            "marketUpdate": handleMarketUpdate,
+        }
+
         def on_message(ws, message):
 
             msg = openfeed_api_pb2.OpenfeedGatewayMessage()
             msg.ParseFromString(message)
 
-            handlers = {
-                "loginResponse": handleLogin,
-                "heartBeat": handleHeartbeat,
-                "subscriptionResponse": handleSubscriptionResponse,
-                "instrumentDefinition": handleInstrumentDefinition,
-                "marketSnapshot": handleMarketSnapshot,
-                "marketUpdate": handleMarketUpdate,
-            }
+            msg_type = msg.WhichOneof("data")
 
-            handler = handlers.get(msg.WhichOneof(
-                "data"), lambda x: print("Unhandled Message: ", x))
+            handler = handlers.get(
+                msg_type, lambda x: print("Unhandled Message: ", x))
 
             try:
                 handler(msg)
             except Exception as e:
                 if self.debug:
-                    print("Failed to handling incoming message:", e)
+                    print("Failed handling incoming message:", msg_type, e)
                 self.__callback(self.on_error, e)
 
         def on_error(ws, error):
@@ -181,6 +193,9 @@ def on_open(ws):
         self.ws.run_forever()
 
     def __notify_symbol_listeners(self, symbol, msg):
+        if symbol not in self.symbol_handlers:
+            return
+
         for cb in self.symbol_handlers[symbol]:
             try:
                 cb(msg)
@@ -189,6 +204,18 @@ def __notify_symbol_listeners(self, symbol, msg):
                     print("Failed to notify `symbol` callback", e)
                 self.__callback(self.on_error, e)
 
+    def __notify_exchange_listeners(self, exchange, msg):
+        if exchange not in self.exchange_handlers:
+            return
+
+        for cb in self.exchange_handlers[exchange]:
+            try:
+                cb(msg)
+            except Exception as e:
+                if self.debug:
+                    print("Failed to notify `exchange` callback", e)
+                self.__callback(self.on_error, e)
+
     def __notify_heartbeat_listeners(self, msg):
         for cb in self.heartbeat_handlers:
             try:
@@ -256,8 +283,8 @@ def handle_heartbeat(msg):
     of_client = OpenfeedClient("username", "password", debug=False)
 
     of_client.add_symbol_subscription(symbol="AAPL", callback=handle_message)
-    # of_client.add_exchange_subscription(
-    #    exchange="NYSE", callback=handle_message)
+    of_client.add_exchange_subscription(
+        exchange="FOREX", callback=handle_message)
     of_client.add_heartbeat_subscription(callback=handle_heartbeat)
 
     of_client.on_error = lambda x: print("of-client: something went wrong:", x)

pyproject.toml

@@ -3,6 +3,7 @@ name = "openfeed"
 version = "1.0.2"
 description = "Python SDK for Openfeed"
 authors = ["Barchart <[email protected]>"]
+license = "MIT"
 
 [tool.poetry.dependencies]
 python = "*"
@@ -13,5 +14,5 @@ protobuf = "*"
 pytest = "^3.4"
 
 [build-system]
-requires = ["poetry>=0.12"]
-build-backend = "poetry.masonry.api"
+requires = ["setuptools>=41.0", "setuptools-scm", "wheel"]
+build-backend = "setuptools.build_meta"