带有 Python 示例的基本 Interactive Brokers API 不起作用

Question

我正在尝试将 Interactive Brokers API 与 Python 一起使用。我试图通过完全按照他们在教学视频中所做的一切来实现他们的基本示例之一，但它似乎不起作用。下面是应该返回一些市场数据的代码，但是当我运行它时什么都没有打印出来。

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), end=' ')

    def tickSize(self, reqId, tickType, size):
        print("test")
        print("Tick Size. Ticker Id ", reqId, " tickType: ", TickTypeEnum.to_str(tickType), "Size: ", size)


def main():
    app = TestApp()

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

    contract = Contract()
    contract.symbol = "AAPL"
    contract.secType = "STK"
    contract.exchange = "SMART"
    contract.currency = "USD"
    contract.primaryExchange = "NASDAQ"

    app.reqMarketDataType(4)
    app.reqMktData(1, contract, "", False, False, [])
    app.run()


if __name__ == "__main__":
    main()

我不确定会发生什么，但我猜这与未正确覆盖 TestApp 函数或类似情况有关。感谢您提供的任何帮助或信息。

Answer 1

1. reMarketDataType()

我同意 Brian 的说法，reqMarketDataType(4) 只能在正常交易时间之外工作。它在 IB API 的文档中声明如下：

"2 - 冻结数据：冻结市场数据是收盘时记录的最后数据。在TWS中，冻结数据以灰色数字显示。当您将市场数据类型设置为冻结时，您是在要求TWS当前没有可用报价时发送最后一个可用报价。"

“3- 延迟数据：免费，延迟数据延迟 15 - 20 分钟。”

“4 - 延迟冻结数据：为未订阅市场数据的用户请求延迟“冻结”数据。”

[Market Data Types IB API Python documentation]

据我了解，“延迟冻结数据”既延迟了15-20分钟，又是收盘价，所以只能在收盘后申请。

1. Contract.primaryExchange

在初始化合约类时，如果交易所属性被定义为“SMART”，那么纳斯达克在primaryExchange字段中总是被定义为“ISLAND”：

contract.primaryExchange = "ISLAND"

1. 处理接收数据的新线程

同意 MatthewScarpino，避免发送请求和接收数据之间的中断：总是在后台启动一个新线程来运行 EWrapper 来处理返回的数据，然后在 API 连接启动后在 main() 中调用该函数。

def web_socket_thread(app: TestApp):
    websocket_thread = threading.Thread(target=app.run, daemon=True)
    websocket_thread.start()

def main():
    ...
    web_socket_thread(app)
    ...