Python Interactive brokers IB API 非常非常慢
Python Interactive brokers IB API very very slow
我正在试用新的 Python Interactive Broker API，但我在第一步遇到了一些严重的速度问题...
以下代码（见下文）次
0:00:08.832813直到数据接收完毕
0:00:36.000785 直到应用程序完全断开...
为什么这么慢？
加快速度的最佳方法是什么？
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 *
import datetime
from datetime import timedelta
class DataApp(wrapper.EWrapper, EClient):
def __init__(self):
wrapper.EWrapper.__init__(self)
EClient.__init__(self, wrapper=self)
@iswrapper
def historicalData(self, reqId: TickerId, date: str, open: float, high: float,
low: float, close: float, volume: int, barCount: int,
WAP: float, hasGaps: int):
super().historicalData(reqId, date, open, high, low, close, volume,
barCount, WAP, hasGaps)
print("HistoricalData. ", reqId, " Date:", date, "Open:", open,
"High:", high, "Low:", low, "Close:", close, "Volume:", volume)
@iswrapper
def historicalDataEnd(self, reqId: int, start: str, end: str):
super().historicalDataEnd(reqId, start, end)
print("HistoricalDataEnd ", reqId, "from", start, "to", end)
print(datetime.datetime.now()-startime)
self.done = True # This ends the messages loop - this was not in the example code...
def get_data(self):
self.connect("127.0.0.1", 4002, clientId=10)
print("serverVersion:%s connectionTime:%s" % (self.serverVersion(),
self.twsConnectionTime()))
cont = Contract()
cont.symbol = "ES"
cont.secType = "FUT"
cont.currency = "USD"
cont.exchange = "GLOBEX"
cont.lastTradeDateOrContractMonth = "201706"
self.reqHistoricalData(1, cont, datetime.datetime.now().strftime("%Y%m%d %H:%M:%S"),
"1800 S", "30 mins", "TRADES", 0, 1, [])
self.run()
self.disconnect()
print(datetime.datetime.now()-startime)
global starttime
startime = datetime.datetime.now()
DA = DataApp()
DA.get_data()
我也尝试不断地运行它是后台，以便只用
即时提交请求
def runMe():
app.run() # where run() has be removed from the class definition
import threading
thread = threading.Thread(target = runMe)
thread.start()
但它也非常慢。
任何建议表示赞赏
app.run()是一个无限循环，而app.done == False，但当app.done设置为True时，它不会立即停止。 （不知道为什么）
我所做的是编写一个新方法而不是使用 app.run()。
这是我的解决方案：
import time
from ibapi import (decoder, reader, comm)
并将此功能放入您的客户端 Class。
def getMessage(self, wait=3):
# wait 3 secs for response to come in
time.sleep(wait)
# get everything in app.msg_queue
while not self.msg_queue.empty():
text = self.msg_queue.get(block=True, timeout=0.2)
fields = comm.read_fields(text)
self.decoder.interpret(fields)
用法很简单。只需使用 app.getMessage() 而不是 app.run()
我建议您在 ibapi 模块中修改连接 class 中的连接套接字锁。推荐来自于github上的何世铭；如果您有权访问私人交互式经纪人存储库，则可以在此处访问讨论 https://github.com/InteractiveBrokers/tws-api/issues/464
我这样做了，它显着提高了性能。
和世铭建议您减少每次发送或接收消息时都会调用的套接字锁对象的超时时间。
修改套接字锁，进入ibapi的site-packages文件夹，修改connection.py内的connect函数，将"self.socket.settimeout(1)"改为"self.socket.settimeout(0.01)"。这是我的版本 connection.py 中的第 48 行。
如果你看不到和士铭的post，我已经把它放在这篇post的底部了。
备选方案：另一个有趣的解决方案是将 asyncio 用于异步事件循环。我没有这样做，但看起来很有希望。请参阅 Ewald 放在一起的示例 https://github.com/erdewit/tws_async
和士铭评论：
The implementation of Connection /ibapi/connection.py
has a Lock object shared in both sendMsg and recvMsg. Since connect,
self.socket.settimeout(1) is called, therefore the underlying
self.socket.recv(4096) only times out once per second.
Such implementation creates a performance problem. Since the lock is
shared, the socket cannot send data while receiving. In the scenario
where the message received is less than 4k bytes long, the recvMsg
function will wait for 1 second before releasing the lock, making
subsequent sendMsg wait. In my experiment, most messages appear to be
shorter than 4k bytes. In other words, this imposes a cap of one
recvMsg per second.
There are couple strategies to mitigate this. One can reduce the
receive buffer to a number much less than 4k, or reduce the socket
timeout to something like 0.001 second to make it block less.
Or according to
, the socket itself is actually thread-safe. Thus no locks are
necessary.
I tried all three strategies. Removing the lock works the best. And
reducing the timeout to 0.001 works in similar ways.
I can only vouch for linux/unix platforms, and I haven't tried it on
Windows. Would you consider to change the implementation to improve
this?
当前的 API（2019 年 5 月）通过移除 recvMsg 中的锁定提高了速度，正如@BenM 在他的回答中提出的那样。但是，根据请求的类型，它仍然可能很慢。
删除大部分日志记录 有所帮助，大概是因为某些消息类型非常大。我最初尝试使用 logging 库将其过滤掉，但完全删除代码在速度方面效果更好，我将其归结为在传递给 [= 之前生成更大的字符串所需的额外处理11=].
用双端队列替换队列：用collections.deque替换client.py中的Queue可能也值得，因为deque 快了 10 倍以上。我已经在其他地方完成了此操作以提高速度，但还没有做到这一点。 deque 应该是线程安全的，没有锁，就像这里使用的那样："Deques support thread-safe, memory efficient appends and pops from either side of the deque"（来自文档）。
Queue vs deque speed comparison
我正在试用新的 Python Interactive Broker API，但我在第一步遇到了一些严重的速度问题...
以下代码（见下文）次
0:00:08.832813直到数据接收完毕
0:00:36.000785 直到应用程序完全断开...
为什么这么慢？ 加快速度的最佳方法是什么？
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 *
import datetime
from datetime import timedelta
class DataApp(wrapper.EWrapper, EClient):
def __init__(self):
wrapper.EWrapper.__init__(self)
EClient.__init__(self, wrapper=self)
@iswrapper
def historicalData(self, reqId: TickerId, date: str, open: float, high: float,
low: float, close: float, volume: int, barCount: int,
WAP: float, hasGaps: int):
super().historicalData(reqId, date, open, high, low, close, volume,
barCount, WAP, hasGaps)
print("HistoricalData. ", reqId, " Date:", date, "Open:", open,
"High:", high, "Low:", low, "Close:", close, "Volume:", volume)
@iswrapper
def historicalDataEnd(self, reqId: int, start: str, end: str):
super().historicalDataEnd(reqId, start, end)
print("HistoricalDataEnd ", reqId, "from", start, "to", end)
print(datetime.datetime.now()-startime)
self.done = True # This ends the messages loop - this was not in the example code...
def get_data(self):
self.connect("127.0.0.1", 4002, clientId=10)
print("serverVersion:%s connectionTime:%s" % (self.serverVersion(),
self.twsConnectionTime()))
cont = Contract()
cont.symbol = "ES"
cont.secType = "FUT"
cont.currency = "USD"
cont.exchange = "GLOBEX"
cont.lastTradeDateOrContractMonth = "201706"
self.reqHistoricalData(1, cont, datetime.datetime.now().strftime("%Y%m%d %H:%M:%S"),
"1800 S", "30 mins", "TRADES", 0, 1, [])
self.run()
self.disconnect()
print(datetime.datetime.now()-startime)
global starttime
startime = datetime.datetime.now()
DA = DataApp()
DA.get_data()
我也尝试不断地运行它是后台，以便只用
即时提交请求def runMe():
app.run() # where run() has be removed from the class definition
import threading
thread = threading.Thread(target = runMe)
thread.start()
但它也非常慢。 任何建议表示赞赏
app.run()是一个无限循环，而app.done == False，但当app.done设置为True时，它不会立即停止。 （不知道为什么）
我所做的是编写一个新方法而不是使用 app.run()。
这是我的解决方案：
import time
from ibapi import (decoder, reader, comm)
并将此功能放入您的客户端 Class。
def getMessage(self, wait=3):
# wait 3 secs for response to come in
time.sleep(wait)
# get everything in app.msg_queue
while not self.msg_queue.empty():
text = self.msg_queue.get(block=True, timeout=0.2)
fields = comm.read_fields(text)
self.decoder.interpret(fields)
用法很简单。只需使用 app.getMessage() 而不是 app.run()
我建议您在 ibapi 模块中修改连接 class 中的连接套接字锁。推荐来自于github上的何世铭；如果您有权访问私人交互式经纪人存储库，则可以在此处访问讨论 https://github.com/InteractiveBrokers/tws-api/issues/464
我这样做了，它显着提高了性能。
和世铭建议您减少每次发送或接收消息时都会调用的套接字锁对象的超时时间。 修改套接字锁，进入ibapi的site-packages文件夹，修改connection.py内的connect函数，将"self.socket.settimeout(1)"改为"self.socket.settimeout(0.01)"。这是我的版本 connection.py 中的第 48 行。
如果你看不到和士铭的post，我已经把它放在这篇post的底部了。
备选方案：另一个有趣的解决方案是将 asyncio 用于异步事件循环。我没有这样做，但看起来很有希望。请参阅 Ewald 放在一起的示例 https://github.com/erdewit/tws_async
和士铭评论：
The implementation of Connection /ibapi/connection.py has a Lock object shared in both sendMsg and recvMsg. Since connect, self.socket.settimeout(1) is called, therefore the underlying self.socket.recv(4096) only times out once per second.
Such implementation creates a performance problem. Since the lock is shared, the socket cannot send data while receiving. In the scenario where the message received is less than 4k bytes long, the recvMsg function will wait for 1 second before releasing the lock, making subsequent sendMsg wait. In my experiment, most messages appear to be shorter than 4k bytes. In other words, this imposes a cap of one recvMsg per second.
There are couple strategies to mitigate this. One can reduce the receive buffer to a number much less than 4k, or reduce the socket timeout to something like 0.001 second to make it block less.
Or according to , the socket itself is actually thread-safe. Thus no locks are necessary.
I tried all three strategies. Removing the lock works the best. And reducing the timeout to 0.001 works in similar ways.
I can only vouch for linux/unix platforms, and I haven't tried it on Windows. Would you consider to change the implementation to improve this?
当前的 API（2019 年 5 月）通过移除 recvMsg 中的锁定提高了速度，正如@BenM 在他的回答中提出的那样。但是，根据请求的类型，它仍然可能很慢。
删除大部分日志记录 有所帮助，大概是因为某些消息类型非常大。我最初尝试使用 logging 库将其过滤掉，但完全删除代码在速度方面效果更好，我将其归结为在传递给 [= 之前生成更大的字符串所需的额外处理11=].
用双端队列替换队列：用collections.deque替换client.py中的Queue可能也值得，因为deque 快了 10 倍以上。我已经在其他地方完成了此操作以提高速度，但还没有做到这一点。 deque 应该是线程安全的，没有锁，就像这里使用的那样："Deques support thread-safe, memory efficient appends and pops from either side of the deque"（来自文档）。
Queue vs deque speed comparison