我如何在 Python 中接收来自 IBs API 的数据？
How do I receive the data coming from IBs API in Python?
Interactive Brokers 刚刚发布了他们 API 的 python 版本。我正在尝试获取数据。
我在 'Program.py' 中使用 'examples'，只是想获取帐户值。我只想知道帐户清算价值是多少，然后将其输入 python。这是 documentation. 这是创建和发送请求的代码：
app = TestApp()
app.connect("127.0.0.1", 4001, clientId=0)
print("serverVersion:%s connectionTime:%s" % (app.serverVersion(),
app.twsConnectionTime()))
app.reqAccountSummary(9004, 'All', '$LEDGER')
我可以使用 IB 网关，看到正在发送的请求，以及返回到 IB 网关的响应。我不知道如何将响应放入 Python。如果我正确阅读文档，我会看到：
Receiving
Summarised information is delivered via IBApi.EWrapper.accountSummary and IBApi.EWrapper.accountSummaryEnd
1 class TestWrapper(wrapper.EWrapper):
...
1 def accountSummary(self, reqId: int, account: str, tag: str, value: str,
2 currency: str):
3 super().accountSummary(reqId, account, tag, value, currency)
4 print("Acct Summary. ReqId:", reqId, "Acct:", account,
5 "Tag: ", tag, "Value:", value, "Currency:", currency)
6
...
1 def accountSummaryEnd(self, reqId: int):
2 super().accountSummaryEnd(reqId)
3 print("AccountSummaryEnd. Req Id: ", reqId)
我该怎么办？好像我调用这个函数来获取值，但这个函数需要我想要返回的值作为输入！我错过了什么！？？！
感谢任何人可以提供的帮助。
编辑：
这是我认为的'callback'：
@iswrapper
# ! [accountsummary]
def accountSummary(self, reqId: int, account: str, tag: str, value: str,
currency: str):
super().accountSummary(reqId, account, tag, value, currency)
print("Acct Summary. ReqId:", reqId, "Acct:", account,
"Tag: ", tag, "Value:", value, "Currency:", currency)
这就是我感到困惑的地方。这似乎需要一个帐户值（声明中的 'value: str'），这正是我要求它产生的值。我找不到我会说如下内容的地方：
myMonies = whateverTheHellGetsTheValue(reqID)
因此，'myMonies' 将持有账户价值，我可以继续我的快乐之路。
所以在你的情况下寻找正确的回调。例如，如果您请求一个选项（即 testbed/contractOperations_req）。结果进入 contractDetails (@iswrapper)，您可以在其中指定要执行的操作...也许 print(contractDetails.summary.symbol)等。所以只需找到帐户信息的相应回调，然后 print/return/etc。它回到你的程序。
我在这里回答了一个非常相似的问题。
这是一个程序，我在同一个 class 中子class EWrapper 和 EClient 并将其用于所有请求和接收回调。
调用EClient方法请求数据，通过EWrapper方法反馈。这些是带有 @iswrapper 符号的那些。
from ibapi import wrapper
from ibapi.client import EClient
from ibapi.utils import iswrapper #just for decorator
from ibapi.common 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
self.reqAccountSummary(9002, "All", "$LEDGER")
@iswrapper
def error(self, reqId:TickerId, errorCode:int, errorString:str):
print("Error. Id: " , reqId, " Code: " , errorCode , " Msg: " , errorString)
@iswrapper
def accountSummary(self, reqId:int, account:str, tag:str, value:str, currency:str):
print("Acct Summary. ReqId:" , reqId , "Acct:", account,
"Tag: ", tag, "Value:", value, "Currency:", currency)
@iswrapper
def accountSummaryEnd(self, reqId:int):
print("AccountSummaryEnd. Req Id: ", reqId)
# now we can disconnect
self.disconnect()
def main():
app = TestApp()
app.connect("127.0.0.1", 7497, clientId=123)
app.run()
if __name__ == "__main__":
main()
致其他像我这样的新手：
另请注意；我正在尝试：
print(self.reqHistoricalData(4102, ContractSamples.EurGbpFx(), queryTime,
"1 M", "1 day", "MIDPOINT", 1, 1, False, []))
或者获取 self.reqHistoricalData() 的 return。
正如布赖恩在上面提到的； EClient 发送请求，EWrapper 接收回信息。
所以看起来试图获取 self.reqHistoricalData() 的句柄不会给你任何东西（我得到 None 类型）
但是将请求添加到
@iswrapper
def nextValidId(self, orderId:int):
print("setting nextValidOrderId: %d", orderId)
self.nextValidOrderId = orderId
# here is where you start using api
self.reqAccountSummary(9002, "All", "$LEDGER")
self.reqHistoricalData(4102, ContractSamples.EurGbpFx(), queryTime,
"1 M", "1 day", "MIDPOINT", 1, 1, False, [])
contract = Contract()
contract.symbol = "AAPL"
contract.secType = "STK"
contract.currency = "USD"
contract.exchange = "SMART"
self.reqHistoricalData(4103, contract, queryTime,
"1 M", "1 day", "MIDPOINT", 1, 1, False, [])
足以让接收器 (EWrapper) 打印到控制台
2019-01-05 更新：
EWrapper 需要知道如何处理收到的消息。
为了让 EWrapper 为您提供句柄，例如打印到控制台；代码的编写者必须在 "# here is where you start using api"
行之外的代码中指定装饰器语句
举个例子：
如果代码包含此代码段：
@iswrapper
def nextValidId(self, orderId:int):
print("setting nextValidOrderId: %d", orderId)
#super().nextValidId(orderId)
self.nextValidOrderId = orderId
#here is where you start using api
queryTime = (datetime.datetime.today() - datetime.timedelta(days=5)).strftime("%Y%m%d %H:%M:%S")
self.reqHistoricalData(4102, ContractSamples.EurGbpFx(), queryTime,
"1 M", "1 day", "MIDPOINT", 1, 1, False, [])
@iswrapper
def historicalData(self, reqId:int, bar: BarData):
print("HistoricalData. ReqId:", reqId, "BarData.", bar)
然后我们将打印到控制台。
如果包装器装饰器方法被忽略，则不会打印到控制台。
例如在这段代码中：
@iswrapper
def nextValidId(self, orderId:int):
print("setting nextValidOrderId: %d", orderId)
#super().nextValidId(orderId)
self.nextValidOrderId = orderId
#here is where you start using api
queryTime = (datetime.datetime.today() - datetime.timedelta(days=5)).strftime("%Y%m%d %H:%M:%S")
self.reqHistoricalData(4102, ContractSamples.EurGbpFx(), queryTime,
"1 M", "1 day", "MIDPOINT", 1, 1, False, [])
#@iswrapper
#def historicalData(self, reqId:int, bar: BarData):
# print("HistoricalData. ReqId:", reqId, "BarData.", bar)
此脚本 (get-account.py) 将 return 代码行处的用户帐户“值”的值：
print(f"myfunds > {app.get_myfunds()}")
...（当然）可以将其分配给变量以进行进一步处理等
这里重要的是帐户“值”在代码的过程/脚本部分中作为对应用程序对象的方法调用可用。
我将脚本部分的某些部分置于 while 循环“time.sleep(1)”内休眠，以便让异步“EWrapper”回调方法有时间 return 来自 IB 的 API（在继续下一行之前）。
run_loop() 是从一个线程中调用的。我在 multi-threaded 编程方面相当陌生，所以我对它为什么有效没有任何真正的了解，但这似乎是脚本成功执行的关键。我尝试了 运行 没有多线程的脚本，它只是挂起迫使我终止进程>“kill -9 PID”。
from ibapi.client import EClient
from ibapi.wrapper import EWrapper
import threading
import time
class IBapi(EWrapper, EClient):
def __init__(self):
EClient.__init__(self, self)
# Custom attributes
self.myfunds = ''
self.connection_status = False
def get_myfunds(self):
""" Custom getter method that returns current value of custom attribute self.myfunds. """
return self.myfunds
def get_connection_status(self):
""" Custom getter method that returns current value of custom attribute self.connection_status. """
return self.connection_status
# @iswrapper
def accountSummary(self, reqId:int, account:str, tag:str, value:str, currency:str):
""" Returns the data from the TWS Account Window Summary tab in response to reqAccountSummary(). """
# Update value of custom attribute self.myfunds
self.myfunds = value
# @iswrapper
def accountSummaryEnd(self, reqId:int):
""" This method is called once all account summary data for a given request are received. """
self.done = True # This ends the messages loop
# @iswrapper
def connectAck(self):
""" Callback signifying completion of successful connection. """
# Update value of custom attribute self.connection_status
self.connection_status = True
def run_loop():
app.run()
app = IBapi()
# IP: ipconfig /all > Ethernet adapter Ethernet > IPv4 Address
# Production port: 7496 | Sandbox (paper account) port: 7497
# Client ID: 123 (used to identify this script to the API, can be any unique positive integer).
app.connect('127.0.0.1', 7497, 123)
# Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()
while app.get_connection_status() == False:
time.sleep(1) # Sleep interval to allow time for connection to server
print(f"Connection status > {app.get_connection_status()}")
app.reqAccountSummary(reqId = 2, groupName = "All", tags = "TotalCashValue")
while app.get_myfunds() == '':
time.sleep(1) # Sleep interval to allow time for incoming data
if app.get_myfunds() != '':
print(f"myfunds > {app.get_myfunds()}")
app.disconnect()
Interactive Brokers 刚刚发布了他们 API 的 python 版本。我正在尝试获取数据。
我在 'Program.py' 中使用 'examples'，只是想获取帐户值。我只想知道帐户清算价值是多少，然后将其输入 python。这是 documentation. 这是创建和发送请求的代码：
app = TestApp()
app.connect("127.0.0.1", 4001, clientId=0)
print("serverVersion:%s connectionTime:%s" % (app.serverVersion(),
app.twsConnectionTime()))
app.reqAccountSummary(9004, 'All', '$LEDGER')
我可以使用 IB 网关，看到正在发送的请求，以及返回到 IB 网关的响应。我不知道如何将响应放入 Python。如果我正确阅读文档，我会看到：
Receiving
Summarised information is delivered via IBApi.EWrapper.accountSummary and IBApi.EWrapper.accountSummaryEnd
1 class TestWrapper(wrapper.EWrapper):
...
1 def accountSummary(self, reqId: int, account: str, tag: str, value: str,
2 currency: str):
3 super().accountSummary(reqId, account, tag, value, currency)
4 print("Acct Summary. ReqId:", reqId, "Acct:", account,
5 "Tag: ", tag, "Value:", value, "Currency:", currency)
6
...
1 def accountSummaryEnd(self, reqId: int):
2 super().accountSummaryEnd(reqId)
3 print("AccountSummaryEnd. Req Id: ", reqId)
我该怎么办？好像我调用这个函数来获取值，但这个函数需要我想要返回的值作为输入！我错过了什么！？？！
感谢任何人可以提供的帮助。
编辑：
这是我认为的'callback'：
@iswrapper
# ! [accountsummary]
def accountSummary(self, reqId: int, account: str, tag: str, value: str,
currency: str):
super().accountSummary(reqId, account, tag, value, currency)
print("Acct Summary. ReqId:", reqId, "Acct:", account,
"Tag: ", tag, "Value:", value, "Currency:", currency)
这就是我感到困惑的地方。这似乎需要一个帐户值（声明中的 'value: str'），这正是我要求它产生的值。我找不到我会说如下内容的地方：
myMonies = whateverTheHellGetsTheValue(reqID)
因此，'myMonies' 将持有账户价值，我可以继续我的快乐之路。
所以在你的情况下寻找正确的回调。例如，如果您请求一个选项（即 testbed/contractOperations_req）。结果进入 contractDetails (@iswrapper)，您可以在其中指定要执行的操作...也许 print(contractDetails.summary.symbol)等。所以只需找到帐户信息的相应回调，然后 print/return/etc。它回到你的程序。
我在这里回答了一个非常相似的问题。
这是一个程序，我在同一个 class 中子class EWrapper 和 EClient 并将其用于所有请求和接收回调。
调用EClient方法请求数据，通过EWrapper方法反馈。这些是带有 @iswrapper 符号的那些。
from ibapi import wrapper
from ibapi.client import EClient
from ibapi.utils import iswrapper #just for decorator
from ibapi.common 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
self.reqAccountSummary(9002, "All", "$LEDGER")
@iswrapper
def error(self, reqId:TickerId, errorCode:int, errorString:str):
print("Error. Id: " , reqId, " Code: " , errorCode , " Msg: " , errorString)
@iswrapper
def accountSummary(self, reqId:int, account:str, tag:str, value:str, currency:str):
print("Acct Summary. ReqId:" , reqId , "Acct:", account,
"Tag: ", tag, "Value:", value, "Currency:", currency)
@iswrapper
def accountSummaryEnd(self, reqId:int):
print("AccountSummaryEnd. Req Id: ", reqId)
# now we can disconnect
self.disconnect()
def main():
app = TestApp()
app.connect("127.0.0.1", 7497, clientId=123)
app.run()
if __name__ == "__main__":
main()
致其他像我这样的新手：
另请注意；我正在尝试：
print(self.reqHistoricalData(4102, ContractSamples.EurGbpFx(), queryTime,
"1 M", "1 day", "MIDPOINT", 1, 1, False, []))
或者获取 self.reqHistoricalData() 的 return。
正如布赖恩在上面提到的； EClient 发送请求，EWrapper 接收回信息。
所以看起来试图获取 self.reqHistoricalData() 的句柄不会给你任何东西（我得到 None 类型）
但是将请求添加到
@iswrapper
def nextValidId(self, orderId:int):
print("setting nextValidOrderId: %d", orderId)
self.nextValidOrderId = orderId
# here is where you start using api
self.reqAccountSummary(9002, "All", "$LEDGER")
self.reqHistoricalData(4102, ContractSamples.EurGbpFx(), queryTime,
"1 M", "1 day", "MIDPOINT", 1, 1, False, [])
contract = Contract()
contract.symbol = "AAPL"
contract.secType = "STK"
contract.currency = "USD"
contract.exchange = "SMART"
self.reqHistoricalData(4103, contract, queryTime,
"1 M", "1 day", "MIDPOINT", 1, 1, False, [])
足以让接收器 (EWrapper) 打印到控制台
2019-01-05 更新：
EWrapper 需要知道如何处理收到的消息。
为了让 EWrapper 为您提供句柄，例如打印到控制台；代码的编写者必须在 "# here is where you start using api"
举个例子： 如果代码包含此代码段：
@iswrapper
def nextValidId(self, orderId:int):
print("setting nextValidOrderId: %d", orderId)
#super().nextValidId(orderId)
self.nextValidOrderId = orderId
#here is where you start using api
queryTime = (datetime.datetime.today() - datetime.timedelta(days=5)).strftime("%Y%m%d %H:%M:%S")
self.reqHistoricalData(4102, ContractSamples.EurGbpFx(), queryTime,
"1 M", "1 day", "MIDPOINT", 1, 1, False, [])
@iswrapper
def historicalData(self, reqId:int, bar: BarData):
print("HistoricalData. ReqId:", reqId, "BarData.", bar)
然后我们将打印到控制台。 如果包装器装饰器方法被忽略，则不会打印到控制台。 例如在这段代码中：
@iswrapper
def nextValidId(self, orderId:int):
print("setting nextValidOrderId: %d", orderId)
#super().nextValidId(orderId)
self.nextValidOrderId = orderId
#here is where you start using api
queryTime = (datetime.datetime.today() - datetime.timedelta(days=5)).strftime("%Y%m%d %H:%M:%S")
self.reqHistoricalData(4102, ContractSamples.EurGbpFx(), queryTime,
"1 M", "1 day", "MIDPOINT", 1, 1, False, [])
#@iswrapper
#def historicalData(self, reqId:int, bar: BarData):
# print("HistoricalData. ReqId:", reqId, "BarData.", bar)
此脚本 (get-account.py) 将 return 代码行处的用户帐户“值”的值：
print(f"myfunds > {app.get_myfunds()}")
...（当然）可以将其分配给变量以进行进一步处理等
这里重要的是帐户“值”在代码的过程/脚本部分中作为对应用程序对象的方法调用可用。
我将脚本部分的某些部分置于 while 循环“time.sleep(1)”内休眠，以便让异步“EWrapper”回调方法有时间 return 来自 IB 的 API（在继续下一行之前）。
run_loop() 是从一个线程中调用的。我在 multi-threaded 编程方面相当陌生，所以我对它为什么有效没有任何真正的了解，但这似乎是脚本成功执行的关键。我尝试了 运行 没有多线程的脚本，它只是挂起迫使我终止进程>“kill -9 PID”。
from ibapi.client import EClient
from ibapi.wrapper import EWrapper
import threading
import time
class IBapi(EWrapper, EClient):
def __init__(self):
EClient.__init__(self, self)
# Custom attributes
self.myfunds = ''
self.connection_status = False
def get_myfunds(self):
""" Custom getter method that returns current value of custom attribute self.myfunds. """
return self.myfunds
def get_connection_status(self):
""" Custom getter method that returns current value of custom attribute self.connection_status. """
return self.connection_status
# @iswrapper
def accountSummary(self, reqId:int, account:str, tag:str, value:str, currency:str):
""" Returns the data from the TWS Account Window Summary tab in response to reqAccountSummary(). """
# Update value of custom attribute self.myfunds
self.myfunds = value
# @iswrapper
def accountSummaryEnd(self, reqId:int):
""" This method is called once all account summary data for a given request are received. """
self.done = True # This ends the messages loop
# @iswrapper
def connectAck(self):
""" Callback signifying completion of successful connection. """
# Update value of custom attribute self.connection_status
self.connection_status = True
def run_loop():
app.run()
app = IBapi()
# IP: ipconfig /all > Ethernet adapter Ethernet > IPv4 Address
# Production port: 7496 | Sandbox (paper account) port: 7497
# Client ID: 123 (used to identify this script to the API, can be any unique positive integer).
app.connect('127.0.0.1', 7497, 123)
# Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()
while app.get_connection_status() == False:
time.sleep(1) # Sleep interval to allow time for connection to server
print(f"Connection status > {app.get_connection_status()}")
app.reqAccountSummary(reqId = 2, groupName = "All", tags = "TotalCashValue")
while app.get_myfunds() == '':
time.sleep(1) # Sleep interval to allow time for incoming data
if app.get_myfunds() != '':
print(f"myfunds > {app.get_myfunds()}")
app.disconnect()