Synchronizing IB python client to get historical data

ibapiexample_1.py

# Download API from http://interactivebrokers.github.io/#
#
# Install python API code /IBJts/source/pythonclient $ python3 setup.py install
#
# Note: The test cases, and the documentation refer to a python package called IBApi,
#    but the actual package is called ibapi. Go figure.
#
# Get the latest version of the gateway:
# https://www.interactivebrokers.com/en/?f=%2Fen%2Fcontrol%2Fsystemstandalone-ibGateway.php%3Fos%3Dunix
#    (for unix: windows and mac users please find your own version)
#
# Run the gateway
#
# user: edemo
# pwd: demo123
#
# Now I'll try and replicate the time telling example


import time
import datetime
from threading import Thread, Event
from ibapi.wrapper import EWrapper
from ibapi.client import EClient
from ibapi.contract import Contract
import pandas as pd

class TestWrapper(EWrapper):
    def __init__(self):
        setattr(self, "_my_data", dict())
        self._reqs = {}
        self._reqEvent = {}
    def connectAck(self):
        if self.async:
            self.startApi()
    ## error handling code
    def init_error(self):
        self._my_data['error']=None
    def iserror(self):
        an_error_if=self._my_data['error'] is not None
        return an_error_if
    def error(self, id, errorCode, errorString):
        ## Overriden method
        errormsg = "IB error id %d errorcode %d string %s" % (id, errorCode, errorString)
        self._my_data['error']=(id, errorCode, errorString)
        if(id in self._reqEvent):
            print(errormsg)
            self._reqEvent[id].set()
    ## Time telling code
    def init_time(self):
        self._my_data['time']=0
    def currentTime(self, time):
        ## Overriden method
        self._my_data['time']=int(time)
    def isValidTime(self):
        is_valid_time_if=not self._my_data['time']==0
        return is_valid_time_if
    def historicalData(self, reqId, date:str, open:float, high:float, 
                       low:float, close:float, volume:int, barCount:int, 
                        WAP:float, hasGaps:int):
        if(reqId in self._reqs):
            self._reqs[reqId].append({'reqId':reqId, 'date':date, 'open':open, 'high':high, 
                       'low':low, 'close':close, 'volume':volume, 'barCount':barCount, 
                        'WAP':WAP, 'hasGaps':hasGaps})
    def historicalDataEnd(self, reqId:int, start:str, end:str):
        self._reqEvent[reqId].set() 
     
class TestClient(EClient):
    def __init__(self, wrapper):
        EClient.__init__(self, wrapper)
        self.reqId=1
    def speaking_clock(self):
        start_time = time.time()
        self.wrapper.init_error()
        self.wrapper.init_time()
        error_occured = False
        finished = False
        print("Getting the time... ")
        self.reqCurrentTime()
        while not finished:
            current_time = self.wrapper._my_data['time']
            finished = self.wrapper.isValidTime()
            iserror = self.wrapper.iserror()
            MAX_WAIT=10
            if (time.time() - start_time) > MAX_WAIT:
                print("Exceeded maximum wait for wrapper to respond")
                finished = True
            if iserror:
                finished = True
        if iserror:
            print("Error happened")
            print(self._my_data['error'])
        return current_time
    def req_historical_data(self,symbol='EUR',secType='CASH',exchange='IDEALPRO',currency='USD',endDateTime='20161201 00:00:00 CET', duration='1 Y', barSize='15 mins',whatToShow='MIDPOINT', timeout=60):
        contract = Contract()
        contract.symbol = symbol
        contract.secType = secType
        contract.exchange = exchange
        contract.currency = currency
        self.reqId=self.reqId+1
        reqId=self.reqId
        self.wrapper._reqs[reqId]=[]
        self._reqEvent[reqId] = Event()
        self.reqHistoricalData(reqId, contract, endDateTime, duration, barSize, whatToShow, useRTH=1,formatDate=2,chartOptions=None)
        self._reqEvent[reqId].wait(timeout=timeout)
        print(len(self._reqs[reqId]))
        d=[]
        op=[]
        hi=[]
        lo=[]
        cl=[]
        for msg in self._reqs[reqId]:
            if(not msg['date'].startswith('finish')):
                d.append(datetime.datetime.fromtimestamp(int(msg['date'])))
                op.append(msg['open'])
                hi.append(msg['high'])
                lo.append(msg['low'])
                cl.append(msg['close'])
        df=pd.DataFrame(index=pd.DatetimeIndex(d),data={'Open':op,'High':hi,'Low':lo,'Close':cl})
        return df
    

class TestApp(TestWrapper, TestClient):
    def __init__(self):
        TestWrapper.__init__(self)
        TestClient.__init__(self, wrapper=self)
    def start(self,host="127.0.0.1", port=7496, clientId=10):
        self.connect(host, port, clientId)
        thread = Thread(target = self.run)
        thread.start()

app = TestApp()

app.start()
print("serverVersion:%s connectionTime:%s" % (app.serverVersion(), app.twsConnectionTime()))

queryTime = datetime.datetime.today().strftime("%Y%m%d %H:%M:%S")

eurjpy = app.req_historical_data(symbol='EUR',currency='JPY',endDateTime=queryTime + ' CET', duration='1 M', barSize='15 mins')
eurjpy.Close.plot()
app.disconnect()