alpha-beta version - overhaul market data tick management for the nth

time.

src/comms/ibc/base_client_messaging.py

@@ -84,7 +84,7 @@ class AbstractGatewayListener(BaseMessageListener):
         """ generated source for method tickPrice """
         raise NotImplementedException
    
-    def tickSize(self, event, message_value):  # tickerId, field, size):
+    def tickSize(self, event, contract_key, tickerId, field, size):
         """ generated source for method tickSize """
         raise NotImplementedException
    
@@ -220,11 +220,11 @@ class AbstractGatewayListener(BaseMessageListener):
         """ generated source for method commissionReport """
         raise NotImplementedException
    
-    def position(self, event, message_value):  # account, contract, pos, avgCost):
+    def position(self, event, account, contract, pos, avgCost):
         """ generated source for method position """
         raise NotImplementedException
    
-    def positionEnd(self, event, message_value):
+    def positionEnd(self, event): #, message_value):
         """ generated source for method positionEnd """
         raise NotImplementedException
    
@@ -243,7 +243,7 @@ class AbstractGatewayListener(BaseMessageListener):
     def gw_subscriptions(self, event, message_value):
         raise NotImplementedException        
       
-    def error(self, event, message_value):
+    def error(self, id=None, errorCode=None, errorMsg=None):
         raise NotImplementedException
     
     def on_kb_reached_last_offset(self, event, message_value):  # event, items):

src/comms/ibc/gw_ex_request_exit.py

@@ -3,6 +3,7 @@
 from time import sleep, strftime
 import logging
 import json
+import sys
 
 from ib.ext.Contract import Contract
 from optparse import OptionParser
@@ -10,22 +11,26 @@ from misc2.helpers import ContractHelper
 from comms.ibgw.base_messaging import Prosumer
 from comms.tws_protocol_helper import TWS_Protocol
 from comms.ibc.tws_client_lib import TWS_client_manager, AbstractGatewayListener
+from QuantLib._QuantLib import VanillaOption_priceCurve
+from rethink.tick_datastore import TickDataStore
+from finopt.instrument import Symbol
 
          
 class MessageListener(AbstractGatewayListener):   
-    def __init__(self, name, parent):
+    def __init__(self, name, tick_ds):
         AbstractGatewayListener.__init__(self, name)
-        self.parent = parent
+        self.tick_ds = tick_ds
 
-    def position(self, event, message_value):  # account, contract, pos, avgCost):
-        logging.info('MessageListener:%s. val->[%s]' % (event, message_value))
+    def position(self, event, account, contract, pos, avgCost):
+        #logging.info('MessageListener:%s. val->[%s]' % (event, vars()))
+        logging.info('MessageListener: %s %s %d %8.2f' % (account, ContractHelper.kv2contract(contract).m_symbol, pos, avgCost))
    
-    def positionEnd(self, event, message_value):
-        logging.info('MessageListener:%s. val->[%s]' % (event, message_value))
+    def positionEnd(self, event):
+        logging.info('MessageListener:%s. val->[%s]' % (event, vars()))
         #self.parent.stop_manager()
         
-    def error(self, event, message_value):
-        logging.info('MessageListener:%s. val->[%s]' % (event, message_value))  
+    def error(self, event, id, errorCode, errorMsg):
+        logging.info('MessageListener:%s. val->[%s]' % (event, vars()))  
 
 
     def gw_subscriptions(self, event, message_value):
@@ -37,24 +42,43 @@ class MessageListener(AbstractGatewayListener):
         
 
     def tickPrice(self, event, contract_key, field, price, canAutoExecute):
-        logging.info('MessageListener: %s' % vars())
+        #logging.info('MessageListener:%s. %s %d %8.2f' % (event, contract_key, field, price))
+        self.tick_ds.set_symbol_tick_price(contract_key, field, price, canAutoExecute)
+
+    def tickSize(self, event, contract_key, field, size):
+        self.tick_ds.set_symbol_tick_price(contract_key, field, size, 0)
+        #logging.info('MessageListener:%s. %s: %d %8.2f' % (event, contract_key, field, size))
+        
+
 
 
 def test_client(kwargs):
 
+    ts = TickDataStore(kwargs['name'])
     cm = TWS_client_manager(kwargs)
-    cl = MessageListener('gw_client_message_listener', cm)
+    cl = MessageListener('gw_client_message_listener', ts)
     
     cm.add_listener_topics(cl, kwargs['topics'])
     cm.start_manager()
-    contractTuples = [('HSI', 'FUT', 'HKFE', 'HKD', '20170330', 0, '')]#
-    
-    cm.reqMktData(ContractHelper.makeContract(contractTuples[0]), True)
+    contractTuples = [('HSI', 'FUT', 'HKFE', 'HKD', '20170330', 0, ''),
+                      ('USD', 'CASH', 'IDEALPRO', 'JPY', '', 0, ''),
+                      ('AUD', 'CASH', 'IDEALPRO', 'USD', '', 0, ''),
+                      ('QQQ', 'STK', 'SMART', 'USD', '', 0, ''),
+                      ]
+                          
+                              
+    map(lambda x: cm.reqMktData(ContractHelper.makeContract(x), False), contractTuples)
+    syms = map(lambda x: Symbol(ContractHelper.makeContract(x)), contractTuples)
+    map(lambda x: ts.add_symbol(x), syms)
+    #cm.reqPositions()
+    #cm.reqMktData(ContractHelper.makeContract(contractTuples[1]), False)
     try:
         logging.info('TWS_gateway:main_loop ***** accepting console input...')
         while not cm.is_stopped(): 
         
             sleep(.45)
+            read_ch = raw_input("Enter command:")
+            ts.dump()
         
     except (KeyboardInterrupt, SystemExit):
         logging.error('TWS_client_manager: caught user interrupt. Shutting down...')
@@ -80,10 +104,10 @@ if __name__ == '__main__':
       'tws_app_id': 38868,
       'group_id': 'EX_REQUEST',
       'session_timeout_ms': 10000,
-      'clear_offsets':  True,
+      'clear_offsets':  False,
       'logconfig': {'level': logging.INFO},
-      'topics': ['tickPrice'],
-      'seek_to_end': ['tickPrice']
+      'topics': ['tickSize', 'tickPrice',  'position', 'positionEnd'],
+      'seek_to_end': ['tickPrice', 'tickSize','position', 'positionEnd'],
       }
 
     usage = "usage: %prog [options]"
@@ -93,8 +117,17 @@ if __name__ == '__main__':
     parser.add_option("-g", "--group_id",
                       action="store", dest="group_id", 
                       help="assign group_id to this running instance")
+    parser.add_option("-n", "--name",
+                      action="store", dest="name", 
+                      help="assign an identifier to this running instance")
+    
+    
     
     (options, args) = parser.parse_args()
+    if options.name == None or options.group_id == None:
+        print "Name or Group id was not specified. Use -h to see all options. Exiting..."
+        sys.exit()
+        
     for option, value in options.__dict__.iteritems():
         if value <> None:
             kwargs[option] = value

src/comms/ibc/tws_client_lib.py

@@ -67,7 +67,7 @@ class TWS_client_manager(GatewayCommandWrapper):
         self.initialize_redis()
         
         logging.info('starting up gateway message handler - kafka Prosumer...')        
-        self.gw_message_handler = Prosumer(name='tws_cli_prosumer', kwargs=self.kwargs)
+        self.gw_message_handler = Prosumer(name=self.kwargs['name'], kwargs=self.kwargs)
         GatewayCommandWrapper.__init__(self, self.gw_message_handler)        
 
 

src/comms/ibgw/base_messaging.py

@@ -197,8 +197,12 @@ 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)
+        #logging.info('BaseConsumer: extract_message_content. %s %s' % (type(message), message))
+        try:
+            return json.loads(message.value)
+        except ValueError:
+            logging.info('extract_message_content exception: %s' % message)
+            return {}
     
     def set_stop(self):
         self.done = True
@@ -263,7 +267,7 @@ class BaseConsumer(threading.Thread, Publisher):
             try:
                 message = consumer.next()
                 
-                
+
                 # the next if block is there to serve information purpose only
                 # it may be useful to detect slow consumer situation
                 if message.offset % BaseConsumer.SLOW_CONSUMER_QUALIFY_NUM == 0:

src/comms/ibgw/subscription_manager.py

@@ -36,20 +36,32 @@ class SubscriptionManager(BaseMessageListener):
             id_contract and contract_id are dict and reverse dict that store the index of id 
             to contarct and vice versa
             
-            id_contract: {<int>, <Contract>}
-            contract_id: {<kvs_contract>, <int>}
+            id_contract: {<int>, <contract-redis-key>}
+            contract_id: {<contract-redis-key>, <int>}
             
         '''
         self.idContractMap ={'next_id': 0, 'id_contract':{},'contract_id':{}}       
         # flag to indicate whether to save changes when persist_subscriptions is called       
         self.is_dirty = False
 
-        logging.warn('***** TEMPORARILY skip loading subscriptions from redis!!!')
+        #logging.warn('***** TEMPORARILY skip loading subscriptions from redis!!!')
         self.load_subscriptions()
         
     
     def get_contract_by_id(self, id):
-        return self.idContractMap['id_contract'][id]
+        
+        try:
+            logging.debug('get_contract_by_id %d' % id)
+            #self.dump()
+            return self.idContractMap['id_contract'][id]
+                
+        except (KeyError, ):
+            
+            if (id  >= TWS_event_handler.TICKER_GAP):
+                return self.idContractMap['id_contract'][id  - TWS_event_handler.TICKER_GAP]
+
+        
+        raise
             
     def reset_subscriptions(self, reset_db):
         if reset_db:
@@ -89,7 +101,7 @@ class SubscriptionManager(BaseMessageListener):
             # rebuild the internal data map
             for ic in ic_list:
                 self.idContractMap['id_contract'][ic[0]] = ic[1]
-                self.idContractMap['contract_id'][ContractHelper.makeRedisKeyEx(ic[1])] = ic[0]        
+                self.idContractMap['contract_id'][ic[1]] = ic[0]        
             
             # derive the next id by finding the max id
             max_id = reduce(lambda x,y: max(x,y), self.idContractMap['id_contract'].keys())
@@ -99,8 +111,9 @@ class SubscriptionManager(BaseMessageListener):
             # subscribe market data, first call is normal subscription,
             # first for snapshot, then subscribe for the latest
             logging.info('SubscriptionManager:load_subscription. request market data for: %s' % (ic_list))
-            map(lambda ic: self.request_market_data(ic[0], ic[1], snapshot=True), ic_list)
-            map(lambda ic: self.request_market_data(ic[0], ic[1], snapshot=False), ic_list) 
+            
+            map(lambda ic: self.request_market_data(ic[0], ContractHelper.makeContractfromRedisKeyEx(ic[1]), snapshot=True), ic_list)
+            map(lambda ic: self.request_market_data(ic[0], ContractHelper.makeContractfromRedisKeyEx(ic[1]), snapshot=False), ic_list)
             
         else:
             logging.warn('SubscriptionManager:load_subscription. No saved id:contracts found in redis.')
@@ -108,12 +121,14 @@ class SubscriptionManager(BaseMessageListener):
         logging.info('SubscriptionManager:load_subscription. Complete populating stored map into idContract dict.')
     
     def request_market_data(self, id, contract, snapshot=False):
+        contract.m_conId = 0
         if snapshot:
             # the call to TWS will return a snapshot follow 
             # by the subscription being cancelled. Add 1000 to avoid clashing 
             # with other subscription ids.  
-            print 'request_market_data: %d' % (id + TWS_event_handler.TICKER_GAP)
+            logging.info( 'request_market_data: %d %s' % (id + TWS_event_handler.TICKER_GAP, ContractHelper.printContract(contract)))
             self.tws_connect.reqMktData(id + TWS_event_handler.TICKER_GAP, contract, '', True)
+            
         else:
             self.tws_connect.reqMktData(id, contract, '', False)
 #
@@ -129,7 +144,7 @@ class SubscriptionManager(BaseMessageListener):
         try:
             return self.idContractMap['contract_id'][ckey]
         except KeyError:
-            logging.debug('is_subscribed: key not found %s' % ckey)
+            logging.warn('is_subscribed: key not found %s' % ckey)
             return -1
 
     def add_subscription(self, contract):
@@ -137,12 +152,13 @@ class SubscriptionManager(BaseMessageListener):
         # structure of idContractMap ={'next_id': -1, 'id_contract':{}, 'contract_id':{}}
         #
         id = self.idContractMap['next_id']
-        self.idContractMap['id_contract'][id] = contract
-        logging.debug('add_subscription %s' % ContractHelper.makeRedisKeyEx(contract))
-        self.idContractMap['contract_id'][ContractHelper.makeRedisKeyEx(contract)] = id        
+        key = ContractHelper.makeRedisKeyEx(contract)
+        self.idContractMap['id_contract'][id] = key
+        logging.info('add_subscription %s' % key)
+        self.idContractMap['contract_id'][key] = id        
         self.idContractMap['next_id'] = id + 1
   
-        return self.idContractMap['next_id']
+        return id
 
             
     #def reqMktData(self, event, message):
@@ -153,23 +169,27 @@ class SubscriptionManager(BaseMessageListener):
         #logging.info('SubscriptionManager: reqMktData')
 
         contract = ContractHelper.kvstring2object(contract, Contract)
+        
+        
         id = self.is_subscribed(contract)
+        logging.info('reqMktData subscription_manager: id %d' % id)
+        
         if id == -1: # not found
             
             id = self.add_subscription(contract)
             #
             # the conId must be set to zero when calling TWS reqMktData
             # otherwise TWS will fail to subscribe the contract
-            contract.m_conId = 0
+            
             self.request_market_data(id, contract, True)
-            self.request_market_data(id, contract, False) 
+            self.request_market_data(id, contract, False)
             self.is_dirty = True
                 
-            logging.info('SubscriptionManager:reqMktData. Requesting market data, id = %d, contract = %s' % (id, ContractHelper.makeRedisKeyEx(contract)))
+            logging.info('SubscriptionManager:reqMktData. New request: id = %d, contract = %s' % (id, ContractHelper.makeRedisKeyEx(contract)))
         
         else:  
-            self.request_market_data(id, contract, snapshot)
-            logging.info('SubscriptionManager:reqMktData. Request id: %d, contract = %s snapshot=%s' % 
+            self.request_market_data(id, contract, True)
+            logging.info('SubscriptionManager:reqMktData. Existing request get snapshot. id: %d, contract = %s snapshot=%s' % 
                          (id, ContractHelper.makeRedisKeyEx(contract), snapshot))
         #self.dump()
 
@@ -203,10 +223,8 @@ class SubscriptionManager(BaseMessageListener):
             try:
                 id_contracts = json.loads(self.rs.get(self.subscription_key))
                 
-                def utf2asc(x):
-                    return x if isinstance(x, unicode) else x
-                
-                return map(lambda x: (x[0], ContractHelper.kvstring2contract(utf2asc(x[1]))), id_contracts)
+
+                return id_contracts
             except TypeError:
                 logging.error('SubscriptionManager:get_id_contracts. Exception when trying to get id_contracts from redis ***')
                 return None
@@ -223,7 +241,7 @@ class SubscriptionManager(BaseMessageListener):
 
         if self.is_dirty:
             # for each id:contract pair in idContractMap['id_contract'] dict, map to a list of (id, kvs_contract) values
-            ic = json.dumps(self.get_id_kvs_contracts(db=False))
+            ic = json.dumps(self.get_id_contracts(db=False))
             self.rs.set(self.subscription_key, ic)
             self.is_dirty = False
 
@@ -233,7 +251,7 @@ class SubscriptionManager(BaseMessageListener):
     def dump(self):
 
         logging.info('subscription manager table:---------------------\n')
-        logging.info(''.join ('\n[%s]:[%s]' % (str(ic[0]).rjust(4), ic[1]) for ic in self.get_id_kvs_contracts(db=False)))
+        logging.info(''.join ('\n[%s]:[%s]' % (str(ic[0]).rjust(4), ic[1]) for ic in self.get_id_contracts(db=False)))
         logging.info(''.join ('\n[%s]:[%d]' % (k.rjust(20), self.idContractMap['contract_id'][k]) 
                                for k in sorted(self.idContractMap['contract_id'])))       
         logging.info( 'Number of instruments subscribed: %d' % self.idContractMap['next_id'])
@@ -249,7 +267,7 @@ class SubscriptionManager(BaseMessageListener):
         except:
             from_id = '<empty_sender_id>'
             
-        ic = self.get_id_kvs_contracts(db=False)
+        ic = self.get_id_contracts(db=False)
     
         
         self.producer.send_message('gw_subscriptions', self.producer.message_dumps({'subscriptions': ic , 'sender_id':self.name, 'target_id':from_id}))

src/comms/ibgw/tws_event_handler.py

@@ -75,25 +75,32 @@ 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', {'contract_key': self.subscription_manger.get_contract_by_id(tickerId), 
-                                           'field': field, 'price': price, 'canAutoExecute': canAutoExecute})
-
+                                          'field': field, 'price': price, 'canAutoExecute': canAutoExecute})
+        #pass
+    
     def tickSize(self, tickerId, field, size):
-        
-        self.broadcast_event('tickSize', vars()) #vars())
-
+         logging.info('TWS_event_handler:tickSize. %d<->%s' % (tickerId,self.subscription_manger.get_contract_by_id(tickerId) ))
+         self.broadcast_event('tickSize', {'contract_key': self.subscription_manger.get_contract_by_id(tickerId), 
+                                            'field': field, 'size': size})
+        #pass
+    
+    
     def tickOptionComputation(self, tickerId, field, impliedVol, delta, optPrice, pvDividend, gamma, vega, theta, undPrice):
         
         #self.broadcast_event('tickOptionComputation', self.tick_process_message(vars())) #vars())
         pass
 
     def tickGeneric(self, tickerId, tickType, value):
-        self.broadcast_event('tickGeneric', vars()) 
+        #self.broadcast_event('tickGeneric', vars())
+        pass 
 
     def tickString(self, tickerId, tickType, value):
-        self.broadcast_event('tickString', vars()) 
+        #self.broadcast_event('tickString', vars())
+        pass 
 
     def tickEFP(self, tickerId, tickType, basisPoints, formattedBasisPoints, impliedFuture, holdDays, futureExpiry, dividendImpact, dividendsToExpiry):
-        self.broadcast_event('tickEFP', vars())
+        #self.broadcast_event('tickEFP', vars())
+        pass
 
     def orderStatus(self, orderId, status, filled, remaining, avgFillPrice, permId, parentId, lastFillPrice, clientId, whyHeId):
         self.broadcast_event('orderStatus', vars())
@@ -140,7 +147,9 @@ class TWS_event_handler(EWrapper):
     def error(self, id=None, errorCode=None, errorMsg=None):
         try:
             logging.error(self.tick_process_message('error', vars()))
-            self.broadcast_event('error', vars())
+            self.broadcast_event('error', {'id': id, 
+                                           'errorCode': errorCode, 'errorMsg': errorMsg})
+
         except:
             pass
 
@@ -208,7 +217,9 @@ class TWS_event_handler(EWrapper):
 
 
     def position(self, account, contract, pos, avgCost):
+        #self.broadcast_event('position', vars())
         self.broadcast_event('position', vars())
+        
 
     def positionEnd(self):
         self.broadcast_event('positionEnd', vars())

src/comms/ibgw/tws_gateway.py

@@ -18,6 +18,7 @@ from comms.ibgw.client_request_handler import ClientRequestHandler
 from comms.ibgw.subscription_manager import SubscriptionManager
 from comms.tws_protocol_helper import TWS_Protocol 
 import redis
+import threading
          
 class TWS_gateway():
 
@@ -84,7 +85,27 @@ class TWS_gateway():
         self.tws_event_handler = TWS_event_handler(self.gw_message_handler)
         
         logging.info('starting up IB EClientSocket...')
+         
+        
         self.tws_connection = EClientSocket(self.tws_event_handler)
+#         class th(threading.Thread):
+#             def __init__(self, evet ):
+#                 threading.Thread.__init__(self)
+#                 
+#                 self.tc = EClientSocket(evet)
+#                 
+#             def get_conn(self):
+#                 return self.tc
+#             def run(self):
+#                 while 1:
+#                     sleep(0.05)
+#                     
+#         
+#         ti = th(self.tws_event_handler)
+#         self.tws_connection = ti.get_conn()
+#         ti.start()
+        
+        
         
         logging.info('establishing TWS gateway connectivity...')
         if not self.connect_tws():
@@ -115,6 +136,7 @@ class TWS_gateway():
 
 
         logging.info('**** Completed initialization sequence. ****')
+        
         self.main_loop()
         
 

src/comms/test/quick_test_ib.py

@@ -188,13 +188,14 @@ class Wrapger(EWrapper):
 def test_IB():
     ew = Wrapger()
     es = EClientSocket(ew)
-    es.eConnect('localhost', 4001, 5555)
+    es.eConnect('localhost', 7496, 5555)
     print es.isConnected()
     
     contractTuple = ('HSI', 'FUT', 'HKFE', 'HKD', '20170330', 0, '')
     contract = ContractHelper.makeContract(contractTuple)
-    es.reqMktData(0, contract, '', False) 
-                
+    es.reqMktData(1000, contract, '', True) 
+    es.reqMktData(0, contract, '', True)
+
     sleep(5)
     print 'disconnecting...'
     es.eDisconnect()

src/config/tws_client_lib.cfg

@@ -16,7 +16,7 @@ redis_db: 0
 #
 group_id: 'TWS_CLI'
 session_timeout_ms: 10000
-topics:['tickSize', 'tickPrice', 'error']
+topics:['tickSize', 'tickPrice', ]
 seek_to_end:['tickSize', 'tickPrice']
 logconfig: { 'filemode': 'w', 'filename': '/tmp/tws_client_lib.log',  'level': logging.INFO}
 #logconfig: {'level': logging.INFO}

src/config/tws_gateway.cfg

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

src/misc2/helpers.py

@@ -186,6 +186,24 @@ class ContractHelper(BaseHelper):
     def is_equal(c1, c2):
         return ContractHelper.makeRedisKeyEx(c1) == ContractHelper.makeRedisKeyEx(c2) 
 
+    @staticmethod
+    def makeContractfromRedisKeyEx(key):
+        
+        def utf2asc(x):
+            return x.encode('ascii') if isinstance(x, unicode) else x
+                
+                #return map(lambda x: (x[0], ContractHelper.kvstring2contract(utf2asc(x[1]))), id_contracts)
+        
+        toks = utf2asc(key).split('-')
+        c = Contract()
+        c.m_symbol = toks[0]
+        c.m_expiry = toks[1]
+        c.m_strike = float(toks[2])
+        c.m_right = toks[3]
+        c.m_secType = toks[4]
+        c.m_currency = toks[5]
+        c.m_exchange = toks[6]               
+        return c 
     
 # def str2dict(s):
 #     return ast.literal_eval(s)

src/rethink/tick_datastore.py

@@ -5,6 +5,8 @@ from misc2.observer import Publisher
 from misc2.observer import NotImplementedException
 from misc2.helpers import ContractHelper
 from comms.ibc.base_client_messaging import AbstractGatewayListener
+from numpy import disp
+import symbol
 
 class TickDataStore(Publisher):
     """
@@ -44,15 +46,16 @@ class TickDataStore(Publisher):
     
 
     EVENT_TICK_UPDATED = 'tds_event_tick_updated'
-    EVENT_NEW_SYMBOL_ADDED = 'tds_event_new_symbol_added'
-    TDS_EVENTS = [EVENT_TICK_UPDATED, EVENT_NEW_SYMBOL_ADDED] 
+    EVENT_SYMBOL_ADDED = 'tds_event_symbol_added'
+    EVENT_SYMBOL_DELETED = 'tds_event_symbol_deleted'
+    TDS_EVENTS = [EVENT_TICK_UPDATED, EVENT_SYMBOL_ADDED, EVENT_SYMBOL_DELETED] 
 
     
     def __init__(self, name):
         
-
-        self.tickers = {}
         self.symbols = {}
+        
+        
         self.lock = RLock()
         Publisher.__init__(self, TickDataStore.TDS_EVENTS)
         self.first_run = True
@@ -62,65 +65,116 @@ class TickDataStore(Publisher):
         map(lambda e: self.register(e, l, getattr(l, e)), TickDataStore.TDS_EVENTS)
 
     def dump(self):
-            # print ', '.join('[%s:%s]' % (k, v['ticker_id'])) 
-        logging.info('TickDataStore-symbols:\nkey : ticker : object cnt---->\n%s' % ('\n'.join('[%s :  %d : %d]' % 
-                                                                                                (k, v['ticker_id'], len(v['syms'])) for k, v in self.symbols.iteritems())))
-        logging.info('TickDataStore-tickers:\nticker: object\n%s' % ('\n'.join('%s:%s' % (str(k).ljust(4), v) for k, v in self.tickers.iteritems())
-                                                                   ))
     
         
-     
+        def format_tick_val(val, fmt):
+            if val == None:
+                length = len(fmt % (0))
+                return ' ' * length
+            
+            return fmt % (val) 
+        
+        # last, bidq, bid, ask, askq, imvol, delta, theta
+        fmt_spec = '%8.2f'
+        fmt_spec2 = '%8.4f'
+        fmt_specq = '%8d'
+        
+        
+        def get_field(sym, fld_id):
+            try:
+                return sym[0].get_tick_value(fld_id)
+            except:
+                return ''
+
+        
+        fmt_sym = map(lambda x: (x[0], '%s,%s,%s,%s,%s' % (
+                                            format_tick_val(get_field(x[1]['syms'],4), fmt_spec),
+                                            format_tick_val(get_field(x[1]['syms'],0), fmt_specq),                                                                                                                  
+                                            format_tick_val(get_field(x[1]['syms'],1), fmt_spec),
+                                            format_tick_val(get_field(x[1]['syms'],2), fmt_spec), 
+                                            format_tick_val(get_field(x[1]['syms'],3), fmt_specq),
+                                            
+                                            )), [(k,v) for k, v in self.symbols.iteritems()])        
+        
+
+        for e in fmt_sym:
+            print('[%s]%s' % (e[0].ljust(50), e[1]))
+
+    def is_symbol_in_list(self, symbol, list):
+    
+        for s in list:
+            if s is symbol:
+                return True
+        
+        return False
+    
+    
     
     def add_symbol(self, symbol):
         try:
-            dispatch = False
+            dispatch = True
             self.lock.acquire()
             key = symbol.get_key()
             if key not in self.symbols:
-                self.symbols[key] = {'ticker_id':-1, 'syms': []}
-                dispatch = True
+                self.symbols[key] = {'syms': [symbol]}
+
             else:
-                ticker_id = self.symbols[key]['ticker_id']
-                self.tickers[ticker_id] = key
-            self.symbols[key]['syms'].append(symbol)        
+                if not self.is_symbol_in_list(symbol, self.symbols[key]['syms']): 
+                    self.symbols[key]['syms'].append(symbol)        
     
-            # defer the dispatch at the end of this method        
-            if dispatch:
-                self.dispatch(TickDataStore.EVENT_NEW_SYMBOL_ADDED, symbol)
         except KeyError:
-            logging.error('TickDataStore: add_symbol. Exception when adding symbol:%s' % ContractHelper.makeRedisKeyEx(symbol.get_contract()))
+            dispatch = False
+            logging.error('TickDataStore: add_symbol. Exception when adding symbol:%s' % key)
         finally:            
-            self.lock.release()
+            self.lock.release()        
+            if dispatch:
+                self.dispatch(TickDataStore.EVENT_SYMBOL_ADDED, symbol)
+            
             
     def del_symbol(self, symbol):
-        raise NotImplementedException     
+           
+        try:
+            dispatch = True
+            self.lock.acquire()
+            key = symbol.get_key()
+            if key not in self.symbols:
+                return
+            else:
+                for s in self.symbols[key]['syms']:
+                    if s is symbol:
+                        self.symbols[key]['syms'].remove(s)
+                    
+        except KeyError:
+            dispatch = False
+            logging.error('TickDataStore: del_symbol. Exception when deleting symbol:%s' % key)
+        finally:            
+            self.lock.release()   
+            if dispatch:
+               self.dispatch(TickDataStore.EVENT_SYMBOL_DELETED, symbol)                 
+                                    
     
        
         
-    def set_symbol_price(self, items):   
+    def set_symbol_tick_price(self, contract_key, field, price, canAutoExecute):   
         
         # message_value: dict: '{"tickerId": 0, "size": 3, "field": 3}'
         
         
-        tid = items['tickerId']
+        
         logging.debug('set_symbol_price: -------------------')
         try:
             self.lock.acquire()
-            contract_key = self.tickers[tid]
-            if (tid == 10):
-                logging.info('set_symbol_price %s' % items)
-            logging.debug('set_symbol_price: -------------------tick id:%d symbol list length=%d %s' % (tid, len(self.symbols[contract_key]['syms']), 
-                                                                                            contract_key))
-            map(lambda e: e.set_tick_value(items['field'], items['price']), self.symbols[contract_key]['syms'])
+            if contract_key in self.symbols:
+                map(lambda e: e.set_tick_value(field, price), self.symbols[contract_key]['syms'])
             
-        except KeyError:
+        except:
             # contract not set up in the datastore, ignore message
-            logging.error('set_symbol_price: KeyError: %d' % tid)
+            logging.error('set_symbol_price: exception occured to: %s' % contract_key)
             #self.dump()
             pass
         finally:
             self.lock.release()
-            self.dispatch(TickDataStore.EVENT_TICK_UPDATED, items)
+            self.dispatch(TickDataStore.EVENT_TICK_UPDATED, vars())
             
 
 

src/sh/gw_ex_request_exit.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+
+
+HOST=$(hostname)
+echo $HOST
+if [ $HOST == 'hkc-larryc-vm1' ]; then
+	FINOPT_HOME=~/ironfly-workspace/finopt/src
+else
+	FINOPT_HOME=~/l1304/workspace/finopt-ironfly/finopt/src
+fi
+export PYTHONPATH=$FINOPT_HOME:$PYTHONPATH
+python $FINOPT_HOME/comms/ibc/gw_ex_request_exit.py -g AAA -n dumpty 

src/sh/start_twsgw.sh

@@ -10,4 +10,5 @@ else
 fi
 export PYTHONPATH=$FINOPT_HOME:$PYTHONPATH
 #python $FINOPT_HOME/comms/ibgw/tws_gateway.py -r -c -f $FINOPT_HOME/config/tws_gateway.cfg 
-python $FINOPT_HOME/comms/ibgw/tws_gateway.py  -r -f $FINOPT_HOME/config/tws_gateway.cfg 
+#python $FINOPT_HOME/comms/ibgw/tws_gateway.py  -r -f $FINOPT_HOME/config/tws_gateway.cfg 
+python $FINOPT_HOME/comms/ibgw/tws_gateway.py   -f $FINOPT_HOME/config/tws_gateway.cfg