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, LoggerNoBaseMessagingFilter  
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
from comms.ibgw.tws_gateway_restapi import WebConsole 
from comms.ibgw.order_manager import OrderManager
from ormdapi.v2.quote_handler import QuoteRESTHandler
from ormdapi.v2.position_handler import AccountPositionTracker
from ormdapi.v2.contract_handler import ContractHandler
from ormdapi.v2.ws.ws_api_server import ApiSocketServer
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
            4a. start a bunch of services for the REST API
            5. start the prosumer 
            6. run web console
        
        '''

        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('initialize order_id_manager and quote_handler for REST API...')
        self.initialize_order_quote_manager()
        


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

        logging.info('start web console...')
        self.start_web_console()

        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)


    '''
        this group of objects are for handling rest API requests
    '''
    def initialize_order_quote_manager(self):
#         self.order_id_mgr = OrderIdManager(self.tws_connection)
#         self.tws_event_handler.set_order_id_manager(self.order_id_mgr)
#         self.order_id_mgr.start()
        self.order_manager = OrderManager('order_manager', self, self.kwargs)
        self.order_manager.start_order_manager()
        self.quote_manager = QuoteRESTHandler('quote_manager', self)
        self.pos_manager = AccountPositionTracker('acctpos_manager', self)
        self.contract_info_manager = ContractHandler('contract_info_mgr', self)
        self.ws_manager = ApiSocketServer('api_ws_manager', self, self.kwargs['restapi.web_socket_host'], self.kwargs['restapi.web_socket_port'])
        t = threading.Thread(name='websocket server', target=self.ws_manager.run_forever)
        t.setDaemon(True)
        t.start()
        
    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 start_web_console(self):
        
        def start_flask():
            w = WebConsole(self)
            
            # tell tws_event_handler that WebConsole  
            # is interested to receive all messages
            for e in TWS_event_handler.PUBLISH_TWS_EVENTS:
                self.tws_event_handler.register(e, w)
            
            w.add_resource()
            w.app.run(host=self.kwargs['webconsole.host'], port=self.kwargs['webconsole.port'],
                      debug=self.kwargs['webconsole.debug'], use_reloader=self.kwargs['webconsole.auto_reload'])
            
        t_webApp = threading.Thread(name='Web App', target=start_flask)
        t_webApp.setDaemon(True)
        t_webApp.start()
                
                
    def get_order_id_manager(self):
        return self.order_manager.get_order_id_mgr()

    def get_order_manager(self):
        return self.order_manager
    
    def get_tws_connection(self):
        return self.tws_connection
    
    def get_tws_event_handler(self):
        return self.tws_event_handler
    
    def get_subscription_manager(self):
        return self.contract_subscription_mgr
    
    def get_quote_manager(self):
        return self.quote_manager
    
    def get_pos_manager(self):
        return self.pos_manager
    
    def get_contract_info_manager(self):
        return self.contract_info_manager
    
    def get_ws_manager(self):
        return self.ws_manager
    
    def get_redis_conn(self):
        return self.rs

    def get_config(self):
        return self.kwargs
    
    def get_ib_conn_status(self):
        return self.ib_conn_status

    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 shutdown_all(self):
        sleep(1)
        logging.info('shutdown_all sequence started....')
        self.gw_message_handler.set_stop()
        self.gw_message_handler.join()
        self.ibh.shutdown()
        self.menu_loop_done = True
        self.get_order_id_manager().set_stop()
        
        sys.exit(0)
        

    def post_shutdown(self):
        th = threading.Thread(target=self.shutdown_all)
        th.daemon = True
        th.start()               

        

    def main_loop(self):
        def print_menu():
            menu = {}
            menu['1']="Dump subscription manager content to log" 
            menu['2']=""
            menu['3']="Start up configuration"
            menu['4']=""
            menu['9']="Exit"
    
            choices=menu.keys()
            choices.sort()
            for entry in choices: 
                print entry, menu[entry]                             
            
        def get_user_input(selection):
                logging.info('TWS_gateway:main_loop ***** accepting console input...')
                print_menu()
                while 1:
                    resp = sys.stdin.readline()
                    response[0] = resp.strip('\n')        
        try:
            
            response = [None]
            user_input_th = threading.Thread(target=get_user_input, args=(response,))
            user_input_th.daemon = True
            user_input_th.start()               
            self.pcounter = 0
            self.menu_loop_done = False
            while not self.menu_loop_done: 
                
                sleep(.5)
                self.persist_subscription_table()
                if response[0] is not None:
                    selection = response[0]
                    if selection =='1':
                        self.contract_subscription_mgr.dump()
                    elif selection == '3':
                        print '\n'.join('[%s]:%s' % (k,v) for k,v in self.kwargs.iteritems())
                    elif selection == '9': 
                        self.shutdown_all()
                        sys.exit(0)
                        break
                    else: 
                        pass                        
                    response[0] = None
                    print_menu()                
                
        except (KeyboardInterrupt, SystemExit):
                logging.error('TWS_gateway: caught user interrupt. Shutting down...')
                self.shutdown_all()
                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.getLogger().addFilter(LoggerNoBaseMessagingFilter())
    logging.info('config settings: %s' % kwargs)
    
    app = TWS_gateway(kwargs)