buggy commit

src/comms/ibc/base_client_messaging.py

@@ -80,7 +80,7 @@ class AbstractGatewayListener(BaseMessageListener):
         BaseMessageListener.__init__(self, name)
         
     
-    def tickPrice(self, event, message_value):  # tickerId, field, price, canAutoExecute):
+    def tickPrice(self, event, contract_key, field, price, canAutoExecute):  # tickerId, field, price, canAutoExecute):
         """ generated source for method tickPrice """
         raise NotImplementedException
    

src/comms/ibc/gw_ex_request_exit.py

@@ -47,7 +47,9 @@ def test_client(kwargs):
     
     cm.add_listener_topics(cl, kwargs['topics'])
     cm.start_manager()
-    cm.gw_message_handler.send_message('ae_req_tds_internal', '')
+    contractTuples = [('HSI', 'FUT', 'HKFE', 'HKD', '20170330', 0, '')]#
+    
+    cm.reqMktData(ContractHelper.makeContract(contractTuples[0]), True)
     try:
         logging.info('TWS_gateway:main_loop ***** accepting console input...')
         while not cm.is_stopped(): 
@@ -78,9 +80,10 @@ if __name__ == '__main__':
       'tws_app_id': 38868,
       'group_id': 'EX_REQUEST',
       'session_timeout_ms': 10000,
-      'clear_offsets':  False,
+      'clear_offsets':  True,
       'logconfig': {'level': logging.INFO},
-      'topics': ['positionEnd']
+      'topics': ['tickPrice'],
+      'seek_to_end': ['tickPrice']
       }
 
     usage = "usage: %prog [options]"
@@ -96,7 +99,7 @@ if __name__ == '__main__':
         if value <> None:
             kwargs[option] = value
             
-    #print kwargs    
+    print kwargs    
       
     logconfig = kwargs['logconfig']
     logconfig['format'] = '%(asctime)s %(levelname)-8s %(message)s'    

src/comms/ibc/tws_client_lib.py

@@ -112,8 +112,9 @@ class GatewayMessageListener(AbstractGatewayListener):
     def __init__(self, name):
         AbstractGatewayListener.__init__(self, name)
              
-    def tickPrice(self, event, message_value):  # tickerId, field, price, canAutoExecute):
-        logging.info('GatewayMessageListener:%s. val->[%s]' % (event, message_value))
+    #def tickPrice(self, event, message_value):  # tickerId, field, price, canAutoExecute):
+    def tickPrice(self, event, contract_key, field, price, canAutoExecute):  # tickerId, field, price, canAutoExecute):    
+        logging.info('GatewayMessageListener:%s. val->[%s]' % (event, vars()))
 
     def tickSize(self, event, message_value):  # tickerId, field, price, canAutoExecute):
         logging.info('GatewayMessageListener:%s. val->[%s]' % (event, message_value))

src/comms/ibgw/base_messaging.py

@@ -196,7 +196,7 @@ class BaseConsumer(threading.Thread, Publisher):
         return {'value': message.value, 'partition':message.partition, 'offset': message.offset}
         
     def extract_message_content(self, message):
-        return message.value
+        return json.loads(message.value)
     
     def set_stop(self):
         self.done = True
@@ -342,13 +342,13 @@ class BaseConsumer(threading.Thread, Publisher):
                     if self.my_topics[message.topic][str(message.partition)] == message.offset and message.offset <> 0:
                         self.dispatch(BaseConsumer.KB_REACHED_LAST_OFFSET, self.enrich_message(message))
                         #self.dispatch(message.topic, self.enrich_message(message))
-                        self.dispatch(message.topic, **self.extract_message_content(message))
+                        self.dispatch(message.topic, self.extract_message_content(message))
                         logging.info('********************** reached the last message previously processed %s %d' % (message.topic, message.offset))
                     else:
                         self.persist_offsets(message.topic, message.partition, message.offset)
                         #self.dispatch(BaseConsumer.KB_EVENT, {'message': message})
                         #self.dispatch(message.topic, self.enrich_message(message))
-                        self.dispatch(message.topic, **self.extract_message_content(message))
+                        self.dispatch(message.topic, self.extract_message_content(message))
             except StopIteration:
                 logging.debug('BaseConsumer:run StopIteration Caught. No new message arriving...')
                 continue

src/comms/ibgw/client_request_handler.py

@@ -80,11 +80,11 @@ class ClientRequestHandler(BaseMessageListener):
         self.tws_connect.requestFA(1)
     
     
-    def reqMktData(self, event, sm_contract):
-        logging.info('ClientRequestHandler received reqMktData request: %s' % sm_contract)
+    def reqMktData(self, event, contract, snapshot):
+        logging.info('ClientRequestHandler received reqMktData request: %s' % contract)
         try:
             #self.contract_subscription_mgr.reqMktData(ContractHelper.kvstring2contract(sm_contract))
-            self.contract_subscription_mgr.reqMktData(ContractHelper.kvstring2object(sm_contract, Contract))
+            self.contract_subscription_mgr.reqMktData(ContractHelper.kvstring2object(contract, Contract))
         except:
             pass
     

src/comms/ibgw/subscription_manager.py

@@ -144,10 +144,11 @@ class SubscriptionManager(BaseMessageListener):
         return self.idContractMap['next_id']
 
             
-    def reqMktData(self, event, message):
+    #def reqMktData(self, event, message):
+    def reqMktData(self, event, contract, snapshot):        
                   
-        contract = ContractHelper.kvstring2object(message['contract'], Contract)
-        snapshot = message['snapshot']
+#         contract = ContractHelper.kvstring2object(message['contract'], Contract)
+#         snapshot = message['snapshot']
         #logging.info('SubscriptionManager: reqMktData')
   
         id = self.is_subscribed(contract)

src/comms/ibgw/tws_event_handler.py

@@ -4,7 +4,7 @@ from time import strftime
 import logging
 import traceback
 from ib.ext.EWrapper import EWrapper
-from comms.ibgw import subscription_manager
+
 
         
 class TWS_event_handler(EWrapper):
@@ -12,12 +12,13 @@ class TWS_event_handler(EWrapper):
     TICKER_GAP = 1000
     producer = None
     
-    def __init__(self, producer, subscription_manager):        
+    def __init__(self, producer):        
         self.producer = producer
-        self.subscription_manger = subscription_manager
+        
  
  
-
+    def set_subscription_manager(self, subscription_manager):
+        self.subscription_manger = subscription_manager
  
     def broadcast_event(self, message, mapping):
 
@@ -71,7 +72,7 @@ class TWS_event_handler(EWrapper):
                 
     
     def tickPrice(self, tickerId, field, price, canAutoExecute):
-        
+        logging.info('TWS_event_handler:tickPrice. %d<->%s' % (tickerId,self.subscription_manger.get_contract_by_id(tickerId) ))
         self.broadcast_event('tickPrice', {self.subscription_manger.get_contract_by_id(tickerId),
                                            field, price, canAutoExecute})
 

src/comms/ibgw/tws_gateway.py

@@ -126,7 +126,7 @@ class TWS_gateway():
                                                              self.get_redis_conn(), self.kwargs)
         
         
-
+        self.tws_event_handler.set_subscription_manager(self.contract_subscription_mgr)
 
     def initialize_redis(self):
 

src/docs/ticker_handling.txt

@@ -20,7 +20,7 @@ Case 1: Request market data snapshot
 - GatewayCommandWrapper -> producer.send_message('reqMktData', json_message_str)
 
 3. Process request
-- prosumer/ BaseConsumer -> enrich message value in message['value'] key
+- prosumer/ BaseConsumer baseconsumer: self.dispatch(message.topic, **self.extract_message_content(message))
 - ClientRequestHandler(BaseMessageListener) -> reqMktData -> subscription_mgr.reqMktData(contract_object, snapshot)
 
 4. map id:key 

src/misc2/observer.py

@@ -1,4 +1,6 @@
 import json
+import sys
+import logging
 
 class NotImplementedException(Exception):
     def __init__(self, value):
@@ -38,7 +40,10 @@ class Publisher:
         
         for subscriber, callback in self.get_subscribers(event).items():
             #print 'observer:: subscriber**** %s' % subscriber
-            callback(event, params)
+            try:
+                callback(event, **params)
+            except TypeError:
+                logging.error (sys.exc_info()[0])
             
 #############################################################
 # Test classes to demo usage of Publisher and Subscriber