使用 tkinter 显示 TWS API 报价

Question

可以使用盈透证券的TWS API获取实时数据，如(*1)所示，并可以显示在控制台上，如(*2)所示。

这里，我想垂直显示reqId1、reqId2和reqId3的实时值，水平显示BID和ASK，如tkinter(*3)所示。

例如，如果{'reqId': 1, 'tickType': 'BID', 'price': 0.658}，则reqId1的BID显示为价格的0.658，如果新的{'reqId': 1, 'tickType': 'BID', 'price': 0.6581} 出现，价格显示为 0.6581。

要完成上述操作，如何将 tkinter 代码 (*3) 引入到 TWS API (*1) 代码中？

特别是，如何修改函数 [def tickPrice]？

(*1)

    import time
    from ibapi.client import *
    from ibapi.wrapper import *

    global clientId
    clientId = 1001

    nextOrderId = None

    class TestApp(EClient, EWrapper):
        def __init__(self):
            EClient.__init__(self, self)

        def nextValidId(self, orderId: int):
            self.nextOrderId = orderId

            global nextOrderId
            nextOrderId = self.nextOrderId
            
            print("NextValidId:", orderId)

        def market_data(self, reqId, contract):
            self.reqMarketDataType(1)   # Live
            self.reqMktData(
                reqId=reqId,
                contract=contract,
                genericTickList="",
                snapshot=False,
                regulatorySnapshot=False,
                mktDataOptions=[])

        def tickPrice(self, reqId, tickType, price, attrib):
            result_dict = {
                'reqId':reqId, 
                'tickType':TickTypeEnum.to_str(tickType), 
                'price':price
            }
            print(result_dict)

        def stop(self):
            self.disconnect()

    def req_streaming_data(port, contract_dict):
        app = TestApp()
        app.connect("127.0.0.1", port, clientId)
        time.sleep(3)
        for key in contract_dict:
            app.market_data(reqId=key, contract=contract_dict[key])
        app.run()

    class CurrencyContract():
        def create_currency_contract(self, currency_code:str):
            symbol, currency = currency_code[:3], currency_code[3:]
            contract = Contract()
            contract.symbol = symbol
            contract.secType = 'CASH'
            contract.currency = currency
            contract.exchange = 'IDEALPRO'
            return contract    

    def main(port:int):
        C = CurrencyContract()
        contract_list = [
            C.create_currency_contract(currency_code='AUDUSD'),
            C.create_currency_contract(currency_code='GBPUSD'),
            C.create_currency_contract(currency_code='EURUSD')
            ]

        contract_dict = {i+1: value for i, value in enumerate(contract_list)}
        req_streaming_data(port, contract_dict)

    if __name__ == "__main__":
        PAPER_PORT = 7497
        main(PAPER_PORT)

(*2)

    {'reqId': 1, 'tickType': 'BID', 'price': 0.658}  
    {'reqId': 1, 'tickType': 'ASK', 'price': 0.6581}  
    ...  
    {'reqId': 1, 'tickType': 'BID', 'price': 0.6581}  
    {'reqId': 1, 'tickType': 'ASK', 'price': 0.65815}  
    {'reqId': 2, 'tickType': 'BID', 'price': 1.2889}  
    {'reqId': 2, 'tickType': 'ASK', 'price': 1.289}  
    {'reqId': 3, 'tickType': 'BID', 'price': 1.072}  
    ...

(*3)

    import tkinter as tk

    data = [
        {'reqId': 1, 'tickType': 'BID', 'price': 0.658}
        {'reqId': 1, 'tickType': 'ASK', 'price': 0.6581}
        ...
        {'reqId': 1, 'tickType': 'BID', 'price': 0.6581}
        {'reqId': 1, 'tickType': 'ASK', 'price': 0.65815}
        {'reqId': 2, 'tickType': 'BID', 'price': 1.2889}
        {'reqId': 2, 'tickType': 'ASK', 'price': 1.289}
        {'reqId': 3, 'tickType': 'BID', 'price': 1.072}
        ...    
    ]

    # Create GUI
    root = tk.Tk()
    root.title("realtime price display")

    # Create Label (3x2 Grid)
    labels = {}
    for req_id in range(1, 4):
        for tick_type in ['BID', 'ASK']:
            label = tk.Label(root, text="0.000", width=10)
            label.grid(row=req_id-1, column=tick_type == 'BID')
            labels[(req_id, tick_type)] = label

    # Data update function
    def update_labels(index=0):
        item = data[index]
        req_id = item['reqId']
        tick_type = item['tickType']
        price = item['price']
        if (req_id, tick_type) in labels:
            labels[(req_id, tick_type)].config(text=f"{price:.5f}")
        
        if index < len(data)-1:
            root.after(500, update_labels, index+1)  # recall after 500[mili sec]

    # initial display and start update
    update_labels()
    root.mainloop()

Answer 1

嗯，让我们试试这个：

    import time
from ibapi.client import *
from ibapi.wrapper import *
import tkinter as tk

global clientId
clientId = 1001
nextOrderId = None

class TestApp(EClient, EWrapper):
    def init(self, root):
        EClient.init(self, self)
        self.root = root
        self.labels = {}
        self.initialize_gui()

    def initialize_gui(self):
        # Create the GUI elements for displaying data
        self.root.title("Realtime Price Display")
        for req_id in range(1, 4):
            for tick_type in ['BID', 'ASK']:
                label = tk.Label(self.root, text="0.000", width=10)
                label.grid(row=req_id-1, column=0 if tick_type == 'BID' else 1)
                self.labels[(req_id, tick_type)] = label

    def nextValidId(self, orderId: int):
        self.nextOrderId = orderId
        global nextOrderId
        nextOrderId = self.nextOrderId
        print("NextValidId:", orderId)

    def market_data(self, reqId, contract):
        self.reqMarketDataType(1)  # Live
        self.reqMktData(
            reqId=reqId,
            contract=contract,
            genericTickList="",
            snapshot=False,
            regulatorySnapshot=False,
            mktDataOptions=[])

    def tickPrice(self, reqId, tickType, price, attrib):
        # Update the GUI label with the new data
        tick_type_str = TickTypeEnum.to_str(tickType)
        if (reqId, tick_type_str) in self.labels:
            self.labels[(reqId, tick_type_str)].config(text=f"{price:.5f}")

    def stop(self):
        self.disconnect()

def req_streaming_data(port, contract_dict, root):
    app = TestApp(root)
    app.connect("127.0.0.1", port, clientId)
    time.sleep(3)
    for key in contract_dict:
        app.market_data(reqId=key, contract=contract_dict[key])
    root.after(100, app.run)  # Non-blocking call to app.run() for GUI

class CurrencyContract():
    def create_currency_contract(self, currency_code: str):
        symbol, currency = currency_code[:3], currency_code[3:]
        contract = Contract()
        contract.symbol = symbol
        contract.secType = 'CASH'
        contract.currency = currency
        contract.exchange = 'IDEALPRO'
        return contract    

def main(port: int):
    root = tk.Tk()
    C = CurrencyContract()
    contract_list = [
        C.create_currency_contract(currency_code='AUDUSD'),
        C.create_currency_contract(currency_code='GBPUSD'),
        C.create_currency_contract(currency_code='EURUSD')
    ]
    contract_dict = {i+1: value for i, value in enumerate(contract_list)}
    req_streaming_data(port, contract_dict, root)
    root.mainloop()

if name == "main":
    PAPER_PORT = 7497
    main(PAPER_PORT)