import time
import threading
from ibapi.client import *
from ibapi.wrapper import *
host = "127.0.0.1"
# 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)
self.nextOrderId = None
def nextValidId(self, orderId: OrderId):
self.nextOrderId = orderId
def contractDetails(self, reqId: int, contractDetails: ContractDetails):
print(contractDetails.contract)
self.placeOrder(myOrder.orderId, contractDetails.contract, myOrder)
def openOrder(
self, orderId: OrderId, contract: Contract, order: Order, orderState: OrderState
):
print(f"openOrder. orderId: {orderId}, contract: {contract}, order: {order}")
def orderStatus(
self,
orderId: OrderId,
status: str,
filled: Decimal,
remaining: Decimal,
avgFillPrice: float,
permId: int,
parentId: int,
lastFillPrice: float,
clientId: int,
whyHeld: str,
mktCapPrice: float,
):
print(
f"orderId: {orderId}, status: {status}, filled: {filled}, remaining: {remaining}, avgFillPrice: {avgFillPrice}, permId: {permId}, parentId: {parentId}, lastFillPrice: {lastFillPrice}, clientId: {clientId}, whyHeld: {whyHeld}, mktCapPrice: {mktCapPrice}"
)
def execDetails(self, reqId: int, contract: Contract, execution: Execution):
print(f"reqId: {reqId}, contract: {contract}, execution: {execution}")
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"
myOrder = Order()
myOrder.orderId = app.nextOrderId
myOrder.action = "BUY"
myOrder.tif = "GTC"
myOrder.orderType = "LMT"
myOrder.lmtPrice = 310.00
myOrder.totalQuantity = 10
app.myOrder = myOrder
app.reqContractDetails(app.nextOrderId, myContract)