使用 Python 的 TWS 交互式经纪人 API。通过多种方法接收实时数据时出现问题答案

【问题标题】：TWS interactive brokers API with Python. Trouble putting live data together when received by several methods methods使用 Python 的 TWS 交互式经纪人 API。通过多种方法接收实时数据时出现问题
【发布时间】：2021-12-13 12:24:37
【问题描述】：

提供有关我的问题的更多背景信息：

我正在使用 python 构建一个连接到交互式经纪商 TWS 的 API。我设法构建了一些功能性的东西，并且能够使用 IB 文档中给出的方法从合约中获取实时数据。现在我想使用所有数据来构建其他并行系统，我遇到了组织来自 IB 服务器的数据的问题。

我的程序循环一个包含 30 个符号的列表以从中获取实时数据，然后我想将每个符号的数据（例如“HIGH”、“LOW”、“CLOSE”、“VWAP”等）放在一起一个数据框来计算指标，并从那里提出一个基于这些指标的警报系统。

这个目标我已经完成了，整个程序只使用一个符号。很容易将数据存储在实例或变量中，将其传递给 DataFrame，然后计算以设置警报。

现在，当循环 30 个值的列表并接收所有这些值的数据时，我一直在努力尝试将每个符号的数据存储在一起，然后计算并设置警报。特别是当我必须使用几种方法来接收数据时（例如，我对某些数据使用tickPrice，对其他一些数据使用tickString）这几种方法一个接一个地执行，但它们不一定同时拥有所有数据，有些值比其他值需要更多时间才能显示出来。

我将展示我的代码示例，以提供更多关于我的目标的上下文：

这是我的 EWrapper 类：

class IBApi(EWrapper, EClient):
    def __init__(self):
        self.syms = ['EN', 'DG', 'AI', 'ORA', 'RI', 'ENGI', 'AC', 'VIV', 'KER', 'CA', 'BN', 'WLN', 'OR', 'VIE',
                     'LR', 'ML', 'SGO', 'CAP', 'MC', 'ACA', 'ATO', 'UG', 'SU', 'HO', 'BNP', 'GLE', 'SAN', 'SW', 'AIR', 'TTE']
        EClient.__init__(self, self)
    # Reciving Real Time Data
    def tickString(self, reqId, tickType, value):
        super().tickString(reqId, tickType, value)
        try:
            if reqId == 0:
                reqId = self.syms[0]
            if reqId == 1:
                reqId = self.syms[1]
            if reqId == 2:
                reqId = self.syms[2]
            if reqId == 3:
                reqId = self.syms[3]
            if reqId == 4:
                reqId = self.syms[4]
            if reqId == 5:
                reqId = self.syms[5]
            if reqId == 6:
                reqId = self.syms[6]
            if reqId == 7:
                reqId = self.syms[7]
            if reqId == 8:
                reqId = self.syms[8]
            if reqId == 9:
                reqId = self.syms[9]
            if reqId == 10:
                reqId = self.syms[10]
            if reqId == 11:
                reqId = self.syms[11]
            if reqId == 12:
                reqId = self.syms[12]
            if reqId == 13:
                reqId = self.syms[13]
            if reqId == 14:
                reqId = self.syms[14]
            if reqId == 15:
                reqId = self.syms[15]
            if reqId == 16:
                reqId = self.syms[16]
            if reqId == 17:
                reqId = self.syms[17]
            if reqId == 18:
                reqId = self.syms[18]
            if reqId == 19:
                reqId = self.syms[19]
            if reqId == 20:
                reqId = self.syms[20]
            if reqId == 21:
                reqId = self.syms[21]
            if reqId == 22:
                reqId = self.syms[22]
            if reqId == 23:
                reqId = self.syms[23]
            if reqId == 24:
                reqId = self.syms[24]
            if reqId == 25:
                reqId = self.syms[25]
            if reqId == 26:
                reqId = self.syms[26]
            if reqId == 27:
                reqId = self.syms[27]
            if reqId == 28:
                reqId = self.syms[28]
            if reqId == 29:
                reqId = self.syms[29]
            if reqId == 30:
                reqId = self.syms[30]
            if tickType == 48 != 0.0:
                rtVolume = value.split(";")
                vwap = float(rtVolume[4])
                self.myData(reqId, TickTypeEnum.to_str(tickType), vwap)
        except Exception as e:
            print(e)

    def tickPrice(self, reqId, tickType, price, attrib):
        super().tickPrice(reqId, tickType, price, attrib)
        try:
            if reqId == 0:
                reqId = self.syms[0]
            if reqId == 1:
                reqId = self.syms[1]
            if reqId == 2:
                reqId = self.syms[2]
            if reqId == 3:
                reqId = self.syms[3]
            if reqId == 4:
                reqId = self.syms[4]
            if reqId == 5:
                reqId = self.syms[5]
            if reqId == 6:
                reqId = self.syms[6]
            if reqId == 7:
                reqId = self.syms[7]
            if reqId == 8:
                reqId = self.syms[8]
            if reqId == 9:
                reqId = self.syms[9]
            if reqId == 10:
                reqId = self.syms[10]
            if reqId == 11:
                reqId = self.syms[11]
            if reqId == 12:
                reqId = self.syms[12]
            if reqId == 13:
                reqId = self.syms[13]
            if reqId == 14:
                reqId = self.syms[14]
            if reqId == 15:
                reqId = self.syms[15]
            if reqId == 16:
                reqId = self.syms[16]
            if reqId == 17:
                reqId = self.syms[17]
            if reqId == 18:
                reqId = self.syms[18]
            if reqId == 19:
                reqId = self.syms[19]
            if reqId == 20:
                reqId = self.syms[20]
            if reqId == 21:
                reqId = self.syms[21]
            if reqId == 22:
                reqId = self.syms[22]
            if reqId == 23:
                reqId = self.syms[23]
            if reqId == 24:
                reqId = self.syms[24]
            if reqId == 25:
                reqId = self.syms[25]
            if reqId == 26:
                reqId = self.syms[26]
            if reqId == 27:
                reqId = self.syms[27]
            if reqId == 28:
                reqId = self.syms[28]
            if reqId == 29:
                reqId = self.syms[29]
            if reqId == 30:
                reqId = self.syms[30]
            self.myData(reqId, TickTypeEnum.to_str(tickType), price)
            time.sleep(0.5)
        except Exception as e:
            print(e)
    @staticmethod
    def myData(reqId, type, price):
        if type == 'RT_VOLUME':
            values = {
                'SYMBOL': [reqId],
                'TYPE': [type],
                'VWAP': [price]
            }
            print(values)
        else:
            values = {
                'SYMBOL': [reqId],
                'TYPE': [type],
                'PRICE': [price]
            }
            print(values)
    def error(self, id, errorCode, errorMsg):
        print(errorCode)
        print(errorMsg)

然后我有我的应用类：

class App:
    ib = None
    def __init__(self):
        self.ib = IBApi()
        self.ib.connect("127.0.0.1", 7496, 88)
        ib_thread = threading.Thread(target=self.run_loop, daemon=True)
        ib_thread.start()
        time.sleep(0.5)


        for sym in self.ib.syms:
            self.marketData(self.ib.syms.index(sym), self.symbolsForData(sym))

    def symbolsForData(self, mySymbol, sec_type='STK', currency='EUR', exchange='SBF'):
        contract1 = Contract()
        contract1.symbol = mySymbol.upper()
        contract1.secType = sec_type
        contract1.currency = currency
        contract1.exchange = exchange
        return contract1

    def marketData(self, req_num, contract1):
        self.ib.reqMktData(reqId=req_num,
                      contract=contract1,
                      genericTickList='233',
                      snapshot=False,
                      regulatorySnapshot=False,
                      mktDataOptions=[])

    def run_loop(self):
        self.ib.run()

# Start App
App()

正如您在 EWrapper 类中看到的那样，我有两种方法来接收数据并将符号分配给每个 reqId，然后我传递一个静态方法，它将接收到的数据放在一起，还有列表值循环低谷。然后在我的 App 类中有连接、构建合同的方法、从 IB 保存 reqMktData 的方法以及获取我需要的参数的参数，以及使用 EWrapper 类中的列表执行 reqMktData 的循环。

以这种方式一切正常，我有这样正确输入的数据：

问题

因此，我的数据到达方式对我来说并没有真正有用，以设置警报系统，因为我没有将每个合同的所有数据放在一起，而且我不能只制作条件使用不同的值并提出警报。有一次我要么只有“HIGH”，要么只有“LOW”，或者只有“VWAP”，但我很难弄清楚如何将每个符号放在一起，因为我没有一次拥有所有东西，而且数据每次都不断出现我就是找不到路。

我想澄清一下，我是编程新手和使用 python 的新手。对不起我的菜鸟代码，可能是“明显”的问题。 但是，如果有人可以帮助我解决这个问题，我将不胜感激。如有任何其他意见，我们将不胜感激。

亲切的问候

马里奥

【问题讨论】：

标签： python pandas dataframe interactive-brokers ib-api

【解决方案1】：

这是我前一阵子写的，看看 tkinter 处理数据网格的能力如何。我刚刚添加了一个带有示例警报功能的数据框。

import tkinter as tk
from tkinter import ttk
import threading

from io import StringIO
import pandas as pd 

from ibapi import wrapper
from ibapi.client import EClient
from ibapi.common import *
from ibapi.ticktype import *
from ibapi.contract import Contract

specs: StringIO = StringIO("""
Sym,Mo,Yr,Exch,Type,Curr,BID,LAST,ASK,OPEN,HIGH,LOW,CLOSE,VWAP    
ES,12,2021,GLOBEX,FUT,USD,
EUR,12,2021,GLOBEX,FUT,USD
JPY,12,2021,GLOBEX,FUT,USD
cl,12,2021,nymex,FUT,USD
USD,,,IDEALPRO,CASH,CAD
""")

class Window(tk.Tk):
    def __init__(self):
        super().__init__()
        self.protocol("WM_DELETE_WINDOW", self.close)
        self.df = pd.read_csv(specs, index_col=0, dtype=str,  na_filter= False)
        self.df.columns = self.df.columns.str.strip()
        self.title('Quotes')
        self.geometry(f'{75*self.df.shape[1]+100}x{25*self.df.shape[0]+100}')
        self.tree = ttk.Treeview(self)
        cols = list(self.df.columns)
        self.tree["columns"] = cols
        self.tree.column('#0', width=75)
        self.tree.heading('#0', text='Sym', anchor='w')
        self.tree.tag_configure('alert', background='#773333')
        
        for col in cols:
            self.tree.column(col, anchor="w", width=75)
            self.tree.heading(col, text=col, anchor='w')

        for index, row in self.df.iterrows():
            self.tree.insert("",tk.END , text=index, values=list(row))
            
        self.tree.pack()

        self.client = Client(self)
        self.client.connect("127.0.0.1", 7497, clientId=123)
        thread = threading.Thread(daemon=True, target = self.client.run)
        thread.start()
        
    def start(self):
        self.client.reqData(self.df)
    
    def update(self, reqId, col, val):
        try:
            item = self.tree.get_children()[reqId]
            self.tree.set(item, column=col, value=val)
            row = self.df.iloc[reqId] # the row is the reqId
            row.at[col]=val 
            if self.alert(row):
                self.tree.selection_set(item)
        except:
            pass
            
    def alert(self,row):
        if row.at['LAST'] > row.at['VWAP']:
            return True
        return False
            
    def close(self):
        print(self.df)
        try:
            self.client.quit()
        except: 
            pass
        finally:
            self.destroy()

class Client(wrapper.EWrapper, EClient):        
    
    def __init__(self, wdow):
        self.wdow = wdow
        self.nextValidOrderId = 0
        wrapper.EWrapper.__init__(self)
        EClient.__init__(self, wrapper=self)
        
    def quit(self):
        self.disconnect()
        
    def reqData(self, df):
        reqId = 0
        for idx, row in df.iterrows():
            cont = Contract()
            cont.symbol = idx
            cont.secType = row['Type']
            cont.currency = row['Curr']
            cont.exchange = row['Exch']
            cont.lastTradeDateOrContractMonth = row['Yr']+row['Mo']
            self.reqMktData(reqId, cont, "233", False, False, None)
            reqId += 1 # the row is the reqId
        
    def cancelMktData(self, reqId:TickerId):
        self.cancelMktData(reqId)
        
    def nextValidId(self, orderId:int):
        self.nextValidOrderId = orderId
        self.wdow.start()

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

    def tickString(self, reqId:TickerId, tickType:TickType, value:str):
        if tickType == TickTypeEnum.RT_VOLUME:
            rtVolume = value.split(";")
            vwap = float(rtVolume[4])
            self.wdow.update(reqId, 'VWAP', vwap)
    
    def tickPrice(self, reqId: TickerId, tickType: TickType, price: float, attrib: TickAttrib):
        self.wdow.update(reqId, TickTypeEnum.to_str(tickType), price)#price,size,time
    
if __name__ == '__main__':
    wdow = Window()
    wdow.mainloop()

【讨论】：

这是个好东西。 tkinter 树，我什至没有想到。我有一个版本的程序，在 tkinter 中使用消息框发出警报，它的功能性但值仍然只在终端中，我无法按照你在那里的方式构建 DF。非常感谢您向我展示您的方法。我现在可以看到如何使用 tkinter 中的树以及构造 DF 的函数非常棒。再次为布赖恩欢呼，马里奥
DF 是通过调用pd.read_csv() 生成的。这是从文件中读取（只需传递文件名）。你看到的是一个黑客，因为你无法读取我的文件。实际上，我将specs 作为一个名为 specs.csv 的文件放在与程序相同的目录中，通常只会加载它。
是的，我看到了，非常迷人。 StringIO 为此做了适当的工作。从现在开始，我将直接使用 Treeview 和 StringIo 来完成我必须做的事情。我制作的大多数产品都需要实时数据。我制作的第一个原型仅使用一个符号，并且值显示在终端中。但这对我来说是革命性的。谢谢你告诉我
嗨，Brian，您认为在进行其他一些计算后使用您的方法可以过滤行吗？通过使用 HIGH 和 LOW 计算幅度，然后根据幅度只丢弃一些不符合标准的行？我试过用 Treeview 过滤，但不是很方便。我曾尝试使用分离来摆脱那些不符合标准但最终程序分离所有内容的程序。提前感谢您的任何建议。马里奥
@MarioChacon 我不知道。除了尝试新事物，我不使用 python。如果您使用 python，PyQt5 可能是高级 UI 的更好选择。请注意，如果您从树视图中删除一行，则 reqId 将不再与该行匹配该符号。

【解决方案2】：

从包含列表的 Python 字典创建 Pandas 数据框很容易。例如，以下代码创建一个包含股票代码、买入价和卖出价的字典：

ticker_dict = {}
ticker_dict['SYMBOL'] = []
for sym in self.ib.syms:
    ticker_dict['SYMBOL'].append(sym)
ticker_dict['BID'] = [0.0] * len(self.ib.syms)
ticker_dict['ASK'] = [0.0] * len(self.ib.syms)
...

现在假设调用tickPrice，reqId 为 5。您可以使用以下代码设置第五个投标价格：

ticker_dict['BID'][5] = price

收集完所有数据后，您可以通过调用 from_dict 将 dict 转换为数据帧。

您可能希望使用空的 NumPy 数组而不是零列表。然后在收集到数据后，就可以将数组转换为列表了。

Algorithmic Trading with Interactive Brokers

【讨论】：

感谢您的提示。这是我上周有点挣扎的事情，这是正确构建 DF 的数据。我会牢记这一点，并越来越多地练习。谢谢你的帮助。马里奥
这本书看起来很棒。我可能会明白，这对我目前正在做的事情非常有用。干杯