不使用Ibpy的IB API Python示例
不使用Ibpy的IB API Python示例
提问于 2017-03-17 22:18:19
有人能帮我弄清楚如何使用来完成基本请求吗？(我正在使用最新的IB API，它似乎支持Python，因此不应该需要人们以前使用的Ibpy )。
我这样的代码可以简单地工作，并使它连接到TWS。问题是:我不知道如何“看到”从IB发送回来的信息。
from ibapi import wrapper
from ibapi.client import EClient
from ibapi.contract import *
w = wrapper.EWrapper()
myTWS = EClient(w)
myTWS.connect(host='localhost', port=7496, clientId=100)
print("serverVersion:%s connectionTime:%s" % (myTWS.serverVersion(),
myTWS.twsConnectionTime()))
myTWS.startApi()
c = Contract()
c.m_symbol = "AAPL"
c.m_secType = "STK"
c.m_exchange = "ISLAND"
c.m_currency = "USD"
myTWS.reqRealTimeBars(999, c, 5, "MIDPOINT", True, [])我知道，在使用IBPy之前，它是一种类似于注册表()的东西。我只是不知道如何在这个当前的IB原始python中做到这一点。有人能帮我举个简单的例子吗？提前谢谢。
回答 4
Stack Overflow用户
回答已采纳
发布于 2017-03-18 00:12:45
您必须子类/重写/实现wrapper.EWrapper。这就是您告诉EClient发送从TWS接收的数据的地方。
我从示例程序中删除了几乎所有的内容，并且运行。
from ibapi import wrapper
from ibapi.client import EClient
from ibapi.utils import iswrapper #just for decorator
from ibapi.common import *
from ibapi.contract import *
from ibapi.ticktype import *
class TestApp(wrapper.EWrapper, EClient):
def __init__(self):
wrapper.EWrapper.__init__(self)
EClient.__init__(self, wrapper=self)
@iswrapper
def nextValidId(self, orderId:int):
print("setting nextValidOrderId: %d", orderId)
self.nextValidOrderId = orderId
#here is where you start using api
contract = Contract()
contract.symbol = "AAPL"
contract.secType = "STK"
contract.currency = "USD"
contract.exchange = "SMART"
self.reqMktData(1101, contract, "", False, None)
@iswrapper
def error(self, reqId:TickerId, errorCode:int, errorString:str):
print("Error. Id: " , reqId, " Code: " , errorCode , " Msg: " , errorString)
@iswrapper
def tickPrice(self, reqId: TickerId , tickType: TickType, price: float,
attrib:TickAttrib):
print("Tick Price. Ticker Id:", reqId, "tickType:", tickType, "Price:", price)
#this will disconnect and end this program because loop finishes
self.done = True
def main():
app = TestApp()
app.connect("127.0.0.1", 7496, clientId=123)
print("serverVersion:%s connectionTime:%s" % (app.serverVersion(),
app.twsConnectionTime()))
app.run()
if __name__ == "__main__":
main()一旦您调用app.run()，程序将启动一个几乎无限循环读取消息，因此您需要一些其他方式来构造您的程序，因为必须启动该循环。
Stack Overflow用户
发布于 2017-10-19 20:04:24
有一个新项目简化了Python的工作。
它被称为insync，它允许同步和异步处理.对于TWS中的新手来说，它看起来非常棒。链接到项目页
使用IB-insync请求历史数据的示例：
from ib_insync import *
ib = IB()
ib.connect('127.0.0.1', 7497, clientId=1)
contract = Forex('EURUSD')
bars = ib.reqHistoricalData(contract, endDateTime='', durationStr='30 D',
barSizeSetting='1 hour', whatToShow='MIDPOINT', useRTH=True)
# convert to pandas dataframe:
df = util.df(bars)
print(df[['date', 'open', 'high', 'low', 'close']])Stack Overflow用户
发布于 2017-03-26 21:46:38
我一直在寻找如何处理app对象之外的请求序列的方法。
这是我对Brian代码的一点修改(感谢Brian介绍了如何使用它)，它获得了两个合同细节，首先它请求MSFT的合同细节，然后是IBM。
- 通过在app.run()方法中设置app.done = True，在接收所有contractDetails消息后完成contractDetailsEnd()
- 当app.done设置为True时，客户端在EClient.run()方法中断开连接。我不知道如何在不断开连接的情况下退出EClient.run()方法，因此更改了client.py EClient.run()方法中源代码中的退出： 最后：#self.disconnect() #Myk防止断开连接返回# Myk而不是断开连接返回
- 在未断开连接的情况下完成app.run()之后，可以再次调用，但必须先将app.done设置为False，否则run()方法退出
- 你必须在最后断开联系再见你自己
- 方法引发错误，但正如有人说的，似乎可以忽略它，特别是在代码结束时断开连接时。
如果有人在不改变API源代码的情况下知道更好的方法，我会很高兴你给我一个建议。
以下是代码：
from ibapi import wrapper
from ibapi.client import EClient
from ibapi.utils import iswrapper #just for decorator
from ibapi.common import *
from ibapi.contract import *
from ibapi.ticktype import *
class TestApp(wrapper.EWrapper, EClient):
def __init__(self):
wrapper.EWrapper.__init__(self)
EClient.__init__(self, wrapper=self)
self.reqIsFinished = True
self.started = False
self.nextValidOrderId = 0
@iswrapper
def nextValidId(self, orderId:int):
print("setting nextValidOrderId: %d", orderId)
self.nextValidOrderId = orderId
# we can start now
@iswrapper
def error(self, reqId:TickerId, errorCode:int, errorString:str):
print("Error. Id: " , reqId, " Code: " , errorCode , " Msg: " , errorString)
@iswrapper
# ! [contractdetails]
def contractDetails(self, reqId: int, contractDetails: ContractDetails):
super().contractDetails(reqId, contractDetails)
print("ContractDetails. ReqId:", reqId, contractDetails.summary.symbol,
contractDetails.summary.secType, "ConId:", contractDetails.summary.conId,
"@", contractDetails.summary.exchange)
# ! [contractdetails]
@iswrapper
# ! [contractdetailsend]
def contractDetailsEnd(self, reqId: int):
super().contractDetailsEnd(reqId)
print("ContractDetailsEnd. ", reqId, "\n")
self.done = True # This ends the messages loop
# ! [contractdetailsend]
def main():
app = TestApp()
app.connect("127.0.0.1", 4001, clientId=123)
print("serverVersion:%s connectionTime:%s" % (app.serverVersion(),
app.twsConnectionTime()))
print('MSFT contract details:')
contract = Contract()
contract.symbol = "MSFT"
contract.secType = "STK"
contract.currency = "USD"
contract.exchange = ""
app.reqContractDetails(210, contract)
app.run()
print('IBM contract details:')
contract.symbol = "IBM"
app.done = False # must be set before next run
app.reqContractDetails(210, contract)
app.run()
app.disconnect()
if __name__ == "__main__":
main()页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/42867933
复制相关文章
相似问题
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号