使用盈透证券交易平台 API - Python 作为对象/变量返回价格

Question

正如标题所示，我试图从 TWS API 获取给定证券的价格，并将其用作我在其他地方的程序中的变量。下面的代码（直接来自 Interactive Broker 的教程之一）将运行并在屏幕上打印价格，但我无法以一种可以创建包含价格的变量/对象的方式对其进行更改。该代码每十次尝试也只能运行一次，如果我在那里做错了什么，请告诉我。

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
from ibapi.ticktype import TickTypeEnum


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

    def error(self, reqId, errorCode, errorString):
        print('Error: ', reqId, ' ', errorCode, ' ', errorString)

    def tickPrice(self, reqId, tickType, price, attrib):
            print('Tick Price. Ticker Id:', reqId, 'tickType:', TickTypeEnum.to_str(tickType), 
            'Price:', price, end=' ')


def main():
    app = TestApp()

    app.connect('127.0.0.1', 7496, 0)

    contract = Contract()
    contract.symbol = 'AAPL'
    contract.secType = 'STK'
    contract.currency = 'USD'
    contract.exchange = 'SMART'
    
    app.reqMarketDataType(1)
    app.reqMktData(1, contract, '', False, False, [])

    app.run()


if __name__ == '__main__':
    main()

Answer 1

程序没有考虑到 api 的异步特性。

#here you are asking to connect, you must wait for a connection
app.connect('127.0.0.1', 7496, 0)

contract...
# you may not have a connection and you're not listeneing for responses yet.
app.reqMarketDataType(1)
app.reqMktData(1, contract, '', False, False, [])

# here is where you start listening for responses, that's what run() does
app.run()

我会这样写

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

class TestApp(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)
        # here you can add variables to the TestApp class, just use self.var in the class
        self.last = 0;
        
    # ib suggests waiting for this response to know that you have a connection
    def nextValidId(self, orderId:int):    
        self.reqMarketDataType(MarketDataTypeEnum.REALTIME) # or DELAYED
        contract = Contract()
        contract.symbol = "AAPL"
        contract.secType = "STK"
        contract.currency = "USD"
        contract.exchange = "SMART"
        self.reqMktData(1, contract, "", False, False, None)

    def error(self, reqId, errorCode, errorString):
        print('Error: ', reqId, ' ', errorCode, ' ', errorString)

    def tickPrice(self, reqId, tickType, price, attrib):
            print('Tick Price. Ticker Id:', reqId, 'tickType:', TickTypeEnum.to_str(tickType), 
            'Price:', price)
            if tickType == TickTypeEnum.LAST or tickType == TickTypeEnum.DELAYED_LAST :
                self.last = price;
                print("disconnecting")
                self.disconnect() # just for testing, normally the program would do something

def main():
    app = TestApp()
    app.connect('127.0.0.1', 7497, 123)
    app.run() # this blocks the program until disconnect()
    print("app.last:", app.last) # now you refer to the variable by class.var

if __name__ == '__main__':
    main()