如何使用 Python API 获取我在盈透证券的账户头寸？

Question

编辑：我找到了有关错误消息的解决方案 - 这是 IB 的 API 上的错误。我在下面作为答案显示的代码对于那些正在寻找从他们在 IB 的账户中读取头寸和资产净值的干净解决方案的人来说应该很有用。

最初的问题[见下面的解决方案; 在此处留下原始问题以获取上下文]：我正在尝试使用公司的 API 获取我在盈透证券 [IB] 的所有账户头寸。他们的文档虽然广泛，但非常令人困惑。示例代码充满了不必要的命令——我想要一些非常精简的东西。

我需要帮助：

1. 如何获取“每个帐户”的信息 [已解决]
2. 如何将变量带入 DataFrame [已解决]
3. 如何避免IB的API打印一系列错误信息[已解决]

到目前为止的代码：

from ibapi.client import EClient 
from ibapi.wrapper import EWrapper
from ibapi.common import * #for TickerId type
import pandas as pd

class ib_class(EWrapper, EClient): 
    def __init__(self): 
        EClient.__init__(self, self)
        self.all_positions = pd.DataFrame([], columns = ['Account','Symbol', 'Quantity', 'Average Cost'])

    def position(self, account, contract, pos, avgCost):
        index = str(account)+str(contract.symbol)
        self.all_positions.loc[index]=account,contract.symbol,pos,avgCost

    def error(self, reqId:TickerId, errorCode:int, errorString:str):
        if reqId > -1:
            print("Error. Id: " , reqId, " Code: " , errorCode , " Msg: " , errorString)

    def positionEnd(self):
        self.disconnect()

ib_api = ib_class() 
ib_api.connect("127.0.0.1", 7496, 0) 
ib_api.reqPositions()
current_positions = ib_api.all_positions
ib_api.run()

当我运行上面的代码时，我会得到一系列 (i) 帐号、(ii) 合约代码、(iii) 头寸和 (iv) 平均成本。因此，这回答了问题 #1。您可能想要“打印”这些值以查看 IB 的 API 如何向您发送信息。

我还能够定义一个 DataFrame 变量all_positions来接收我正在寻找的信息。看看下面的结果。请注意，我必须创建一个“索引”变量，它是 DataFrame 每一行的唯一标识符（作为帐号和符号的组合）。如果没有这个“索引”，我没有找到将信息“附加”到 DataFrame 的方法（欢迎任何关于如何做的想法）：

至于最后一个问题（错误消息）：

Brian 对“error”函数的建议（见下文）消除了“-1”错误。但我仍然得到以下信息：

EReader 线程 Traceback 中未处理的异常（最近一次调用最后一次）：文件“C:\Users\danil\Anaconda3\lib\site-packages\ibapi-9.76.1-py3.7.egg\ibapi\reader.py”，行34、在运行data = self.conn.recvMsg()文件“C:\Users\danil\Anaconda3\lib\site-packages\ibapi-9.76.1-py3.7.egg\ibapi\connection.py”，行99、在recvMsg buf = self._recvAllMsg()文件“C:\Users\danil\Anaconda3\lib\site-packages\ibapi-9.76.1-py3.7.egg\ibapi\connection.py”，第119行，在 _recvAllMsg buf = self.socket.recv(4096) OSError: [WinError 10038] 尝试对不是套接字的东西进行操作

Answer 1

经过大量实验，这是我编写的最终代码，我将其放置在名为 AB_API 的“.py”文件中（因此我可以将其“导入”为库）：

def read_positions(): #read all accounts positions and return DataFrame with information

    from ibapi.client import EClient
    from ibapi.wrapper import EWrapper
    from ibapi.common import TickerId
    from threading import Thread

    import pandas as pd
    import time

    class ib_class(EWrapper, EClient):

        def __init__(self):
            EClient.__init__(self, self)

            self.all_positions = pd.DataFrame([], columns = ['Account','Symbol', 'Quantity', 'Average Cost', 'Sec Type'])

        def error(self, reqId:TickerId, errorCode:int, errorString:str):
            if reqId > -1:
                print("Error. Id: " , reqId, " Code: " , errorCode , " Msg: " , errorString)

        def position(self, account, contract, pos, avgCost):
            index = str(account)+str(contract.symbol)
            self.all_positions.loc[index]= account, contract.symbol, pos, avgCost, contract.secType

    def run_loop():
        app.run()
    
    app = ib_class()
    app.connect('127.0.0.1', 7496, 0)
    #Start the socket in a thread
    api_thread = Thread(target=run_loop, daemon=True)
    api_thread.start()
    time.sleep(1) #Sleep interval to allow time for connection to server

    app.reqPositions() # associated callback: position
    print("Waiting for IB's API response for accounts positions requests...\n")
    time.sleep(3)
    current_positions = app.all_positions
    current_positions.set_index('Account',inplace=True,drop=True) #set all_positions DataFrame index to "Account"
    
    app.disconnect()

    return(current_positions)


def read_navs(): #read all accounts NAVs

    from ibapi.client import EClient
    from ibapi.wrapper import EWrapper
    from ibapi.common import TickerId
    from threading import Thread

    import pandas as pd
    import time

    class ib_class(EWrapper, EClient):

        def __init__(self):
            EClient.__init__(self, self)

            self.all_accounts = pd.DataFrame([], columns = ['reqId','Account', 'Tag', 'Value' , 'Currency'])

        def error(self, reqId:TickerId, errorCode:int, errorString:str):
            if reqId > -1:
                print("Error. Id: " , reqId, " Code: " , errorCode , " Msg: " , errorString)

        def accountSummary(self, reqId, account, tag, value, currency):
            index = str(account)
            self.all_accounts.loc[index]=reqId, account, tag, value, currency

    def run_loop():
        app.run()
    
    app = ib_class()
    app.connect('127.0.0.1', 7496, 0)
    #Start the socket in a thread
    api_thread = Thread(target=run_loop, daemon=True)
    api_thread.start()
    time.sleep(1) #Sleep interval to allow time for connection to server

    app.reqAccountSummary(0,"All","NetLiquidation")  # associated callback: accountSummary / Can use "All" up to 50 accounts; after that might need to use specific group name(s) created on TWS workstation
    print("Waiting for IB's API response for NAVs requests...\n")
    time.sleep(3)
    current_nav = app.all_accounts
    
    app.disconnect()

    return(current_nav)

我现在可以使用两个单行函数读取所有账户头寸和资产净值：

import IB_API

print("Testing IB's API as an imported library:")

all_positions = IB_API.read_positions()
all_navs = IB_API.read_navs()

如果要运行非常快速的测试（不导入/调用函数），请在 read_positions() 和 read_navs()下方执行以下操作：

print("Testing IB's API as an imported library:")

all_positions = read_positions()
print(all_positions)
print()
all_navs = read_navs()
print(all_navs)

我意识到一些用户对导入函数的概念感到困惑，因此上面的快速建议。

我希望盈透证券能够为客户提供一些简单的示例，让用户可以更轻松地测试一些想法并决定是否要使用他们的 API。