merge subscription

Author: mvberg

.github/workflows/publish-to-pypi.yml

@@ -12,20 +12,12 @@ jobs:
       uses: actions/setup-python@v1
       with:
         python-version: 3.7
-    - name: Install pep517
-      run: >-
-        python -m
-        pip install
-        pep517
-        --user
     - name: Build a binary wheel and a source tarball
       run: >-
         python -m
-        pep517.build
-        --source
-        --binary
-        --out-dir dist/
-        .
+        setup.py 
+        sdist 
+        bdist_wheel
     - name: Publish package to Test PyPI
       uses: pypa/gh-action-pypi-publish@master
       with:

README.md

@@ -2,7 +2,9 @@
 
 This SDK encapuslates the openfeed proto objects and faciliates client connections to openfeed servers.
 
-![PyPI and TestPyPI](https://github.com/openfeed-org/sdk-python/workflows/PyPI%20and%20TestPyPI/badge.svg)
+![Build](https://github.com/openfeed-org/sdk-python/workflows/PyPI%20and%20TestPyPI/badge.svg)
+
+![PyPI](https://img.shields.io/pypi/v/openfeed?label=PyPI%20)
 
 ## Installation
 

openfeed/__init__.py

@@ -8,4 +8,4 @@
 
 from .openfeed_client import OpenfeedClient
 
-VERSION = '1.0.3'
+VERSION = '1.0.5'

openfeed/openfeed_client.py

@@ -1,3 +1,4 @@
+import traceback
 import sys
 import inspect
 from generated import openfeed_api_pb2
@@ -52,7 +53,8 @@ def add_symbol_subscription(self, symbol, callback):
             self.symbol_handlers[symbol] = []
 
             if self.token is not None:
-                self._send_message(self.__create_symbol_request([symbol]))
+                self._send_message(
+                    self.__create_subscription_request(None, [symbol]))
 
         self.symbol_handlers[symbol].append(callback)
 
@@ -61,17 +63,23 @@ def add_exchange_subscription(self, exchange, callback):
             self.exchange_handlers[exchange] = []
 
             if self.token is not None:
-                self._send_message(self.__create_exchange_request([exchange]))
+                self._send_message(
+                    self.__create_subscription_request([exchange], None))
 
         self.exchange_handlers[exchange].append(callback)
 
+    def get_instrument_definitions(self):
+        return self.instrument_definitions
+
     def get_instrument_definition(self, id):
         return self.instrument_definitions[id]
 
     def get_instrument_definition_by_symbol(self, symbol):
         return self.instruments_by_symbol[symbol]
 
     def _send_message(self, msg):
+        if self.debug:
+            print("Sending:", msg)
         self.ws.send(msg.SerializeToString(), websocket.ABNF.OPCODE_BINARY)
 
     def __reset(self):
@@ -89,13 +97,9 @@ def handleLogin(msg):
             self.token = msg.loginResponse.token
 
             # sub to all existing interest
-            self._send_message(self.__create_symbol_request(
-                self.symbol_handlers.keys()))
-
-            # sub to all existing exchanges
-            self._send_message(self.__create_exchange_request(
-                self.exchange_handlers.keys()))
-
+            self._send_message(self.__create_subscription_request(
+                self.exchange_handlers.keys(), self.symbol_handlers.keys()))
+                
             return msg
 
         def handleHeartbeat(msg):
@@ -134,7 +138,10 @@ def handleMarketSnapshot(msg):
             inst = self.instrument_definitions[msg.marketSnapshot.marketId].instrumentDefinition
 
             self.__notify_exchange_listeners(inst.barchartExchangeCode, msg)
-            self.__notify_symbol_listeners(inst.symbol, msg)
+
+            # TODO review symbology handling, subbing by one and keying off the other can create unexpected results
+            # for example subscribing to "ZCYAIA40.CM" will come back with OF symbol (less the suffix) in `instrument.symbol`
+            self.__notify_symbol_listeners(inst.symbols[0].symbol, msg)
 
             return msg
 
@@ -167,7 +174,7 @@ def on_message(ws, message):
         def on_error(ws, error):
             if self.debug:
                 print("WS Error: ", error)
-
+                traceback.print_exc()
             self.__callback(self.on_error, error)
 
         def on_close(ws):
@@ -225,30 +232,22 @@ def __notify_heartbeat_listeners(self, msg):
                     print("Failed to notify `heartbeat` callback", e)
                 self.__callback(self.on_error, e)
 
-    def __create_symbol_request(self, symbols):
+    def __create_subscription_request(self, exchanges, symbols):
         requests = []
 
-        for sym in symbols:
-            requests.append(openfeed_api_pb2.SubscriptionRequest.Request(
-                symbol=sym,
-            ))
+        if len(exchanges) > 0:
+            for exch in exchanges:
+                requests.append(openfeed_api_pb2.SubscriptionRequest.Request(
+                    exchange=exch,
+                    subscriptionType=[
+                        openfeed_api_pb2.SubscriptionType.Value("QUOTE")]
+                ))
 
-        of_req = openfeed_api_pb2.OpenfeedGatewayRequest(
-            subscriptionRequest=openfeed_api_pb2.SubscriptionRequest(
-                token=self.token,
-                service=openfeed_pb2.Service.Value("REAL_TIME"),
-                requests=requests
-            ))
-
-        return of_req
-
-    def __create_exchange_request(self, exchanges):
-        requests = []
-
-        for exch in exchanges:
-            requests.append(openfeed_api_pb2.SubscriptionRequest.Request(
-                exchange=exch,
-            ))
+        if len(symbols) > 0:
+            for sym in symbols:
+                requests.append(openfeed_api_pb2.SubscriptionRequest.Request(
+                    symbol=sym,
+                ))
 
         of_req = openfeed_api_pb2.OpenfeedGatewayRequest(
             subscriptionRequest=openfeed_api_pb2.SubscriptionRequest(
@@ -282,7 +281,6 @@ 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="FOREX", callback=handle_message)
     of_client.add_heartbeat_subscription(callback=handle_heartbeat)
@@ -292,7 +290,8 @@ def handle_heartbeat(msg):
     of_client.on_connected = lambda x: print("of-client: connected")
 
     # blocking mode
-    of_client.start(blocking=True)
+    of_client.start(blocking=False)
 
-    # while True:
-    #    time.sleep(3)
+    while True:
+        print("Number of Instruments:", len(of_client.get_instrument_definitions()))
+        time.sleep(10)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "openfeed"
-version = "1.0.3"
+version = "1.0.5"
 description = "Python SDK for Openfeed"
 authors = ["Barchart <[email protected]>"]
 license = "MIT"

setup.py

@@ -17,7 +17,7 @@
 
 setup(
     name='openfeed',
-    version='1.0.3',
+    version='1.0.5',
     author='Barchart',
     author_email='[email protected]',
     license='MIT',