from ibapi.client import *
from ibapi.wrapper import *
import time
import threading
from ibapi.ticktype import TickTypeEnum
# Default Ports:
# TWS Live Account: 7946
# TWS Paper Account: 7947
# IB Gateway Live Account: 4001
# IB Gateway Paper Account: 4002
port = 4002
class TestApp(EClient, EWrapper):
def __init__(self):
EClient.__init__(self, self)
def nextValidId(self, orderId: OrderId):
self.orderId = orderId
def nextId(self):
self.orderId += 1
return self.orderId
def error(self, reqId, errorTime, errorCode, errorString, advancedOrderReject=""):
print(
f"reqId: {reqId}, errorCode: {errorCode}, errorString: {errorString}, orderReject: {advancedOrderReject}"
)
def tickPrice(self, reqId, tickType, price, attrib):
print(
f"reqId: {reqId}, tickType: {TickTypeEnum.toStr(tickType)}, price: {price}, attrib: {attrib}"
)
def tickSize(self, reqId, tickType, size):
print(f"reqId: {reqId}, tickType: {TickTypeEnum.toStr(tickType)}, size: {size}")
app = TestApp()
app.connect("127.0.0.1", port, 0)
threading.Thread(target=app.run).start()
time.sleep(1)
mycontract = Contract()
mycontract.symbol = "AAPL"
mycontract.secType = "STK"
mycontract.exchange = "SMART"
mycontract.currency = "USD"
app.reqMarketDataType(3)
app.reqMktData(app.nextId(), mycontract, "", False, False, [])