finopt/src/comms/ibgw/tws_gateway.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import copy
from time import sleep, strftime
import logging
import json

from ib.ext.Contract import Contract
from ib.ext.EClientSocket import EClientSocket

from misc2.helpers import ContractHelper, ConfigMap
from optparse import OptionParser
from comms.ibgw.base_messaging import Prosumer
from comms.ibgw.tws_event_handler import TWS_event_handler
from comms.ibgw.ib_heartbeat import IbHeartBeat
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
from threading import Lock
         
class TWS_gateway():

    
    # monitor IB connection / heart beat
#     ibh = None
#     tlock = None
#     ib_conn_status = None
    TWS_GW_DEFAULT_CONFIG = {
      'name': 'tws_gateway_server',
      'bootstrap_host': 'localhost',
      'bootstrap_port': 9092,
      'redis_host': 'localhost',
      'redis_port': 6379,
      'redis_db': 0,
      'tws_host': 'localhost',
      'tws_api_port': 8496,
      'tws_app_id': 38888,
      'group_id': 'TWS_GW',
      'session_timeout_ms': 10000,
      'clear_offsets':  False,
      'order_transmit': False,
      'topics': list(TWS_Protocol.topicMethods),
      'reset_db_subscriptions': False
      }
               
    
    def __init__(self, kwargs):
        

             
       
        temp_kwargs = copy.copy(kwargs)
        self.kwargs = copy.copy(TWS_gateway.TWS_GW_DEFAULT_CONFIG)
        for key in self.kwargs:
            if key in temp_kwargs:
                self.kwargs[key] = temp_kwargs.pop(key)        
        self.kwargs.update(temp_kwargs)    

        '''
            TWS_gateway start up sequence
            
            1. establish redis connection
            2. initialize prosumer instance - gateway message handler
            3. establish TWS gateway connectivity
            
            4. initialize listeners: ClientRequestHandler and SubscriptionManager
            5. start the prosumer 
        
        '''

        logging.info('starting up TWS_gateway...')
        self.ib_order_transmit = self.kwargs['order_transmit']
        logging.info('Order straight through (no-touch) flag = %s' % ('True' if self.ib_order_transmit == True else 'False'))
        
        
        logging.info('establishing redis connection...')
        self.initialize_redis()
        
        logging.info('starting up gateway message handler - kafka Prosumer...')        
        self.gw_message_handler = Prosumer(name='tws_gw_prosumer', kwargs=self.kwargs)
        
        logging.info('initializing TWS_event_handler...')        
        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)

        
        
        
        logging.info('establishing TWS gateway connectivity...')
        if self.connect_tws() == False:
            logging.error('TWS_gateway: unable to establish connection to IB %s:%d' % 
                          (self.kwargs['tws_host'], self.kwargs['tws_api_port']))
            self.disconnect_tws()
            sys.exit(-1)
        else:
            # start heart beat monitor
            logging.info('starting up IB heart beat monitor...')
            self.tlock = Lock()
            self.ibh = IbHeartBeat(self.kwargs)
            self.ibh.register_listener([self.on_ib_conn_broken])
            self.ibh.run()  


        logging.info('instantiating listeners...cli_req_handler')        
        self.cli_req_handler = ClientRequestHandler('client_request_handler', self)
        logging.info('instantiating listeners subscription manager...')
        self.initialize_subscription_mgr()
        logging.info('registering messages to listen...')
        self.gw_message_handler.add_listeners([self.cli_req_handler])
        self.gw_message_handler.add_listener_topics(self.contract_subscription_mgr, self.kwargs['subscription_manager.topics'])

        logging.info('start TWS_event_handler. Start prosumer processing loop...')
        self.gw_message_handler.start_prosumer()


        logging.info('**** Completed initialization sequence. ****')
        
        self.main_loop()
        

    def initialize_subscription_mgr(self):
        
        
        self.contract_subscription_mgr = SubscriptionManager(self.kwargs['name'], self.tws_connection, 
                                                             self.gw_message_handler, 
                                                             self.get_redis_conn(), self.kwargs)
        
        
        self.tws_event_handler.set_subscription_manager(self.contract_subscription_mgr)

    def initialize_redis(self):

        self.rs = redis.Redis(self.kwargs['redis_host'], self.kwargs['redis_port'], self.kwargs['redis_db'])
        try:
            self.rs.client_list()
        except redis.ConnectionError:
            logging.error('TWS_gateway: unable to connect to redis server using these settings: %s port:%d db:%d' % 
                          (self.kwargs['redis_host'], self.kwargs['redis_port'], self.kwargs['redis_db']))
            logging.error('aborting...')
            sys.exit(-1)
            
    def get_redis_conn(self):
        return self.rs

    def connect_tws(self):
        if type(self.kwargs['tws_app_id']) <> int:
            logging.error('TWS_gateway:connect_tws. tws_app_id must be of int type but detected %s!' % str(type(kwargs['tws_app_id'])))
            sys.exit(-1)
            
        logging.info('TWS_gateway - eConnect. Connecting to %s:%d App Id: %d...' % 
                     (self.kwargs['tws_host'], self.kwargs['tws_api_port'], self.kwargs['tws_app_id']))
        self.tws_connection.eConnect(self.kwargs['tws_host'], self.kwargs['tws_api_port'], self.kwargs['tws_app_id'])
        
        return self.tws_connection.isConnected()

    def disconnect_tws(self, value=None):
        sleep(2)
        self.tws_connection.eDisconnect()




    def on_ib_conn_broken(self, msg):
        logging.error('TWS_gateway: detected broken IB connection!')
        self.ib_conn_status = 'ERROR'
        self.tlock.acquire() # this function may get called multiple times
        try:                 # block until another party finishes executing
            if self.ib_conn_status == 'OK': # check status
                return                      # if already fixed up while waiting, return 
            
            #self.disconnect_tws()
            self.connect_tws()
            while not self.tws_connection.isConnected():
                logging.error('TWS_gateway: attempt to reconnect...')
                self.connect_tws()
                sleep(2)
            
            # we arrived here because the connection has been restored
            # resubscribe tickers again!
            self.ib_conn_status = 'OK'
            logging.info('TWS_gateway: IB connection restored...resubscribe contracts')
            self.contract_subscription_mgr.force_resubscription()             
            
            
        finally:
            self.tlock.release()          
        
        
    
    def persist_subscription_table(self):
        self.pcounter = (self.pcounter + 1) % 10
        if (self.pcounter >= 8):
            self.contract_subscription_mgr.persist_subscriptions()
           
        

    def main_loop(self):
        try:
            logging.info('TWS_gateway:main_loop ***** accepting console input...')
            
            
            self.pcounter = 0
            while True: 
                
                sleep(.5)
                self.persist_subscription_table()
                
                
        except (KeyboardInterrupt, SystemExit):
                logging.error('TWS_gateway: caught user interrupt. Shutting down...')
                self.gw_message_handler.set_stop()
                self.gw_message_handler.join()
                self.ibh.shutdown()
                logging.info('TWS_gateway: Service shut down complete...')
                sys.exit(0)        



    

        
    
if __name__ == '__main__':


    usage = "usage: %prog [options]"
    parser = OptionParser(usage=usage)
    parser.add_option("-c", "--clear_offsets", action="store_true", dest="clear_offsets",
                      help="delete all redis offsets used by this program")
    parser.add_option("-r", "--reset_db_subscriptions", action="store_true", dest="reset_db_subscriptions",
                      help="delete subscriptions entries in redis used by this program")
    
    parser.add_option("-f", "--config_file",
                      action="store", dest="config_file", 
                      help="path to the config file")
    
    (options, args) = parser.parse_args()
    
    kwargs = ConfigMap().kwargs_from_file(options.config_file)
    for option, value in options.__dict__.iteritems():
        
        if value <> None:
            kwargs[option] = value


    logconfig = kwargs['logconfig']
    logconfig['format'] = '%(asctime)s %(levelname)-8s %(message)s'    
    logging.basicConfig(**logconfig)        

    logging.info('config settings: %s' % kwargs)
    
    app = TWS_gateway(kwargs)