buggy night coding

src/comms/ibc/base_client_messaging.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 import logging
-
+import json
 
 from ib.ext.Contract import Contract
 from misc2.helpers import ContractHelper, ExecutionFilterHelper, OrderHelper
@@ -56,8 +56,8 @@ class GatewayCommandWrapper():
 #     def reqMktData(self, contract):
 #         self.producer.send_message('reqMktData', ContractHelper.object2kvstring(contract))
     def reqMktData(self, contract, snapshot=False):
-        self.producer.send_message('reqMktData', {'contract': ContractHelper.object2kvstring(contract), 
-                                                  'snapshot': snapshot})        
+        self.producer.send_message('reqMktData', json.dumps({'contract': ContractHelper.object2kvstring(contract), 
+                                                  'snapshot': snapshot}))        
         
     def reqAccountSummary(self, reqId, group, tags):
         self.producer.send_message('reqAccountSummary', self.producer.message_dumps([reqId, group, tags]))

src/comms/ibgw/base_messaging.py

@@ -21,6 +21,7 @@ from types import NoneType
 
 
 
+
 class Producer(threading.Thread):
     daemon = True
 
@@ -196,6 +197,7 @@ class BaseConsumer(threading.Thread, Publisher):
         return {'value': message.value, 'partition':message.partition, 'offset': message.offset}
         
     def extract_message_content(self, message):
+        logging.info('BaseConsumer: extract_message_content. %s %s' % (type(message), message))
         return json.loads(message.value)
     
     def set_stop(self):
@@ -310,7 +312,7 @@ class BaseConsumer(threading.Thread, Publisher):
                         if gap == 1:
                             # the message is valid for dispatching and not to be skipped
                             #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.debug('*** On first iteration: Gap=%d Dispatch this valid message to the listener <%s>' % (gap, message.value))
                         else: # gap exists
                             logging.info("*** On first iteration: [Topic:%s:Part:%d:Offset:%d]: Gap:%d Attempting to seek to latest message ..." 
@@ -340,7 +342,7 @@ class BaseConsumer(threading.Thread, Publisher):
                     # the "and" condition ensures that on a fresh start of kafka server this event is not triggered as
                     # both saved value in redis and current offset are both 0
                     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(BaseConsumer.KB_REACHED_LAST_OFFSET, self.extract_message_content(message))
                         #self.dispatch(message.topic, self.enrich_message(message))
                         self.dispatch(message.topic, self.extract_message_content(message))
                         logging.info('********************** reached the last message previously processed %s %d' % (message.topic, message.offset))

src/comms/ibgw/client_request_handler.py

@@ -82,11 +82,11 @@ class ClientRequestHandler(BaseMessageListener):
     
     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(contract, Contract))
-        except:
-            pass
+#         try:
+#             #self.contract_subscription_mgr.reqMktData(ContractHelper.kvstring2contract(sm_contract))
+#             self.contract_subscription_mgr.reqMktData(ContractHelper.kvstring2object(contract, Contract))
+#         except:
+#             pass
     
     def reqHistoricalData(self, event):
         contract = Contract()

src/comms/ibgw/subscription_manager.py

@@ -44,11 +44,12 @@ class SubscriptionManager(BaseMessageListener):
         # flag to indicate whether to save changes when persist_subscriptions is called       
         self.is_dirty = False
 
+        logging.warn('***** TEMPORARILY skip loading subscriptions from redis!!!')
         self.load_subscriptions()
         
     
     def get_contract_by_id(self, id):
-        return self.idContractMap['id_contract']
+        return self.idContractMap['id_contract'][id]
             
     def reset_subscriptions(self, reset_db):
         if reset_db:
@@ -150,7 +151,8 @@ class SubscriptionManager(BaseMessageListener):
 #         contract = ContractHelper.kvstring2object(message['contract'], Contract)
 #         snapshot = message['snapshot']
         #logging.info('SubscriptionManager: reqMktData')
-  
+
+        contract = ContractHelper.kvstring2object(contract, Contract)
         id = self.is_subscribed(contract)
         if id == -1: # not found
             

src/comms/ibgw/tws_event_handler.py

@@ -26,6 +26,7 @@ class TWS_event_handler(EWrapper):
             dict = self.tick_process_message(message, mapping)     
             if message == 'gw_subscriptions' or message == 'gw_subscription_changed':   
                 logging.info('TWS_event_handler: broadcast event: %s [%s]' % (dict['typeName'], dict))
+            logging.info('broadcast_event %s' % dict)
             self.producer.send_message(message, self.producer.message_dumps(dict))    
         except:
             logging.error('broadcast_event: exception while encoding IB event to client:  [%s]' % message)
@@ -73,8 +74,8 @@ 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})
+        self.broadcast_event('tickPrice', {'contract_key': self.subscription_manger.get_contract_by_id(tickerId), 
+                                           'field': field, 'price': price, 'canAutoExecute': canAutoExecute})
 
     def tickSize(self, tickerId, field, size):
         

src/config/tws_gateway.cfg

@@ -18,7 +18,7 @@ redis_db: 0
 #
 tws_host: 'localhost'
 tws_api_port: 7496
-tws_app_id: 74960
+tws_app_id: 74962
 #
 #
 #

src/misc2/observer.py

@@ -39,11 +39,12 @@ class Publisher:
     def dispatch(self, event, params=None):
         
         for subscriber, callback in self.get_subscribers(event).items():
+            callback(event, **params)
             #print 'observer:: subscriber**** %s' % subscriber
-            try:
-                callback(event, **params)
-            except TypeError:
-                logging.error (sys.exc_info()[0])
+#             try:
+#                 callback(event, **params)
+#             except TypeError:
+#                 logging.error (sys.exc_info()[0])
             
 #############################################################
 # Test classes to demo usage of Publisher and Subscriber