Last active
June 9, 2018 13:48
-
-
Save yakneens/5320aac8c3d3aa2c87f8ef7ae88044ff to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from ibapi.wrapper import EWrapper | |
from ibapi.client import EClient | |
from ibapi.contract import Contract as IBcontract | |
from threading import Thread | |
import queue | |
import datetime | |
import time | |
DEFAULT_HISTORIC_DATA_ID=50 | |
DEFAULT_GET_CONTRACT_ID=43 | |
## marker for when queue is finished | |
FINISHED = object() | |
STARTED = object() | |
TIME_OUT = object() | |
class finishableQueue(object): | |
def __init__(self, queue_to_finish): | |
self._queue = queue_to_finish | |
self.status = STARTED | |
def get(self, timeout): | |
""" | |
Returns a list of queue elements once timeout is finished, or a FINISHED flag is received in the queue | |
:param timeout: how long to wait before giving up | |
:return: list of queue elements | |
""" | |
contents_of_queue=[] | |
finished=False | |
while not finished: | |
try: | |
current_element = self._queue.get(timeout=timeout) | |
if current_element is FINISHED: | |
finished = True | |
self.status = FINISHED | |
else: | |
contents_of_queue.append(current_element) | |
## keep going and try and get more data | |
except queue.Empty: | |
## If we hit a time out it's most probable we're not getting a finished element any time soon | |
## give up and return what we have | |
finished = True | |
self.status = TIME_OUT | |
return contents_of_queue | |
def timed_out(self): | |
return self.status is TIME_OUT | |
class TestWrapper(EWrapper): | |
""" | |
The wrapper deals with the action coming back from the IB gateway or TWS instance | |
We override methods in EWrapper that will get called when this action happens, like currentTime | |
Extra methods are added as we need to store the results in this object | |
""" | |
def __init__(self): | |
self._my_contract_details = {} | |
self._my_historic_data_dict = {} | |
self._my_timestamps = {} | |
## error handling code | |
def init_error(self): | |
error_queue=queue.Queue() | |
self._my_errors = error_queue | |
def get_error(self, timeout=5): | |
if self.is_error(): | |
try: | |
return self._my_errors.get(timeout=timeout) | |
except queue.Empty: | |
return None | |
return None | |
def is_error(self): | |
an_error_if=not self._my_errors.empty() | |
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_errors.put(errormsg) | |
## get contract details code | |
def init_contractdetails(self, reqId): | |
contract_details_queue = self._my_contract_details[reqId] = queue.Queue() | |
return contract_details_queue | |
def init_timestamps(self, reqId): | |
timestamp_queue = self._my_timestamps[reqId] = queue.Queue() | |
return timestamp_queue | |
def headTimestamp(self, reqId, headTimestamp): | |
if reqId not in self._my_timestamps.keys(): | |
self.init_timestamps(reqId) | |
self._my_contract_details[reqId].put(headTimestamp) | |
def contractDetails(self, reqId, contractDetails): | |
## overridden method | |
if reqId not in self._my_contract_details.keys(): | |
self.init_contractdetails(reqId) | |
self._my_contract_details[reqId].put(contractDetails) | |
def contractDetailsEnd(self, reqId): | |
## overriden method | |
if reqId not in self._my_contract_details.keys(): | |
self.init_contractdetails(reqId) | |
self._my_contract_details[reqId].put(FINISHED) | |
## Historic data code | |
def init_historicprices(self, tickerid): | |
historic_data_queue = self._my_historic_data_dict[tickerid] = queue.Queue() | |
return historic_data_queue | |
def historicalData(self, tickerid , bar): | |
## Overriden method | |
## Note I'm choosing to ignore barCount, WAP and hasGaps but you could use them if you like | |
bardata=(bar.date, bar.open, bar.high, bar.low, bar.close, bar.volume) | |
historic_data_dict=self._my_historic_data_dict | |
## Add on to the current data | |
if tickerid not in historic_data_dict.keys(): | |
self.init_historicprices(tickerid) | |
historic_data_dict[tickerid].put(bardata) | |
def historicalDataEnd(self, tickerid, start, end): | |
## overriden method | |
if tickerid not in self._my_historic_data_dict.keys(): | |
self.init_historicprices(tickerid) | |
self._my_historic_data_dict[tickerid].put(FINISHED) | |
def historicalTicks(self, reqId, ticks, done): | |
#this_tick_data=IBtick(self.get_time_stamp(),tickType, price) | |
historic_data_dict=self._my_historic_data_dict | |
## Add on to the current data | |
if reqId not in historic_data_dict.keys(): | |
self.init_historicprices(reqId) | |
for tick in ticks: | |
print("Historical Tick. Req Id: ", reqId, ", time: ", tick.time, ", price: ", tick.price, ", size: ", tick.size) | |
historic_data_dict[reqId].put(tick) | |
class TestClient(EClient): | |
""" | |
The client method | |
We don't override native methods, but instead call them from our own wrappers | |
""" | |
def __init__(self, wrapper): | |
## Set up with a wrapper inside | |
EClient.__init__(self, wrapper) | |
def get_head_time_stamp(self, ibcontract, whatToShow, useRTH, formatDate,reqId=DEFAULT_GET_CONTRACT_ID,): | |
timestamp_queue = finishableQueue(self.init_timestamps(reqId)) | |
self.reqHeadTimeStamp(reqId, ibcontract, whatToShow, useRTH, formatDate) | |
## Run until we get a valid contract(s) or get bored waiting | |
MAX_WAIT_SECONDS = 30 | |
new_timestamp = timestamp_queue.get(timeout = MAX_WAIT_SECONDS) | |
while self.wrapper.is_error(): | |
print(self.get_error()) | |
if timestamp_queue.timed_out(): | |
print("Exceeded maximum wait for wrapper to confirm finished - seems to be normal behaviour") | |
if len(new_timestamp)==0: | |
print("Failed to get additional contract details: returning unresolved contract") | |
return ibcontract | |
if len(new_timestamp)>1: | |
print("got multiple contracts using first one") | |
new_timestamp=new_timestamp[0] | |
#resolved_ibcontract=new_timestamp.summary | |
return new_timestamp | |
def resolve_ib_contract(self, ibcontract, reqId=DEFAULT_GET_CONTRACT_ID): | |
""" | |
From a partially formed contract, returns a fully fledged version | |
:returns fully resolved IB contract | |
""" | |
## Make a place to store the data we're going to return | |
contract_details_queue = finishableQueue(self.init_contractdetails(reqId)) | |
print("Getting full contract details from the server... ") | |
self.reqContractDetails(reqId, ibcontract) | |
## Run until we get a valid contract(s) or get bored waiting | |
MAX_WAIT_SECONDS = 30 | |
new_contract_details = contract_details_queue.get(timeout = MAX_WAIT_SECONDS) | |
while self.wrapper.is_error(): | |
print(self.get_error()) | |
if contract_details_queue.timed_out(): | |
print("Exceeded maximum wait for wrapper to confirm finished - seems to be normal behaviour") | |
if len(new_contract_details)==0: | |
print("Failed to get additional contract details: returning unresolved contract") | |
return ibcontract | |
if len(new_contract_details)>1: | |
print("got multiple contracts using first one") | |
new_contract_details=new_contract_details[0] | |
resolved_ibcontract=new_contract_details.summary | |
return resolved_ibcontract | |
def get_IB_tick_data(self, resolved_ibcontract, startDate, endDate, numTicks, whatToShow, useRth,tickerid=DEFAULT_HISTORIC_DATA_ID): | |
""" | |
Kick off market data streaming | |
:param resolved_ibcontract: a Contract object | |
:param tickerid: the identifier for the request | |
:return: tickerid | |
""" | |
historic_data_queue = finishableQueue(self.init_historicprices(tickerid)) | |
self.reqHistoricalTicks(tickerid, resolved_ibcontract, startDate, endDate, numTicks, whatToShow, useRth, ignoreSize=False, miscOptions="null") | |
## Wait until we get a completed data, an error, or get bored waiting | |
MAX_WAIT_SECONDS = 30 | |
print("Getting historical data from the server... could take %d seconds to complete " % MAX_WAIT_SECONDS) | |
historic_data = historic_data_queue.get(timeout = MAX_WAIT_SECONDS) | |
while self.wrapper.is_error(): | |
print(self.get_error()) | |
if historic_data_queue.timed_out(): | |
print("Exceeded maximum wait for wrapper to confirm finished - seems to be normal behaviour") | |
self.cancelHistoricalData(tickerid) | |
return historic_data | |
def get_IB_historical_data(self, ibcontract, durationStr="1 M", barSizeSetting="1 hour", | |
tickerid=DEFAULT_HISTORIC_DATA_ID): | |
""" | |
Returns historical prices for a contract, up to today | |
ibcontract is a Contract | |
:returns list of prices in 4 tuples: Open high low close volume | |
""" | |
## Make a place to store the data we're going to return | |
historic_data_queue = finishableQueue(self.init_historicprices(tickerid)) | |
# Request some historical data. Native method in EClient | |
self.reqHistoricalData( | |
tickerid, # tickerId, | |
ibcontract, # contract, | |
datetime.datetime.today().strftime("%Y%m%d %H:%M:%S %Z"), # endDateTime, | |
durationStr, # durationStr, | |
barSizeSetting, # barSizeSetting, | |
"TRADES", # whatToShow, | |
1, # useRTH, | |
1, # formatDate | |
False, # KeepUpToDate <<==== added for api 9.73.2 | |
[] ## chartoptions not used | |
) | |
## Wait until we get a completed data, an error, or get bored waiting | |
MAX_WAIT_SECONDS = 10 | |
print("Getting historical data from the server... could take %d seconds to complete " % MAX_WAIT_SECONDS) | |
historic_data = historic_data_queue.get(timeout = MAX_WAIT_SECONDS) | |
while self.wrapper.is_error(): | |
print(self.get_error()) | |
if historic_data_queue.timed_out(): | |
print("Exceeded maximum wait for wrapper to confirm finished - seems to be normal behaviour") | |
self.cancelHistoricalData(tickerid) | |
return historic_data | |
class TestApp(TestWrapper, TestClient): | |
def __init__(self, ipaddress, portid, clientid): | |
TestWrapper.__init__(self) | |
TestClient.__init__(self, wrapper=self) | |
self.connect(ipaddress, portid, clientid) | |
thread = Thread(target = self.run) | |
thread.start() | |
setattr(self, "_thread", thread) | |
self.init_error() | |
#if __name__ == '__main__': | |
app = TestApp("127.0.0.1", 7496, 1) | |
ibcontract = IBcontract() | |
ibcontract.secType = "FOP" | |
ibcontract.symbol="ES" | |
ibcontract.exchange="GLOBEX" | |
#ibcontract.primaryExchange = "CME" | |
ibcontract.currency = "USD" | |
ibcontract.lastTradeDateOrContractMonth = "20180608" | |
ibcontract.strike = 2775 | |
ibcontract.right = "C" | |
ibcontract.multiplier = "50" | |
ibcontract.conId = 317012336 | |
ibcontract.localSymbol = "EW2M8 C2775" | |
resolved_ibcontract=app.resolve_ib_contract(ibcontract) | |
time.sleep(10) | |
print("Contract: " + str(resolved_ibcontract)) | |
#earliest = app.reqHeadTimeStamp(4103, resolved_ibcontract, "TRADES", 0, 1) | |
#print("Head timestamp: " + str(earliest)) | |
#historic_data = app.get_IB_historical_data(resolved_ibcontract) | |
historic_data = app.get_IB_tick_data(resolved_ibcontract, startDate="20180507 13:17:21", endDate="", numTicks=1000, whatToShow="TRADES", useRth=1) | |
print(historic_data) | |
app.disconnect() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment