我如何使用 Interactive Broker 的本机 Python API 获取 SPX 的历史柱数据？

Question

我只是想获取 SPX 的历史柱数据。有没有一种方法可以“找出”合适的合同，因为我似乎做不到。

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


class TestApp(EClient, EWrapper):
    def __init__(self):
        EClient.__init__(self, self)
        
    def nextValidId(self, orderId:int):
        mycontract = Contract()
        mycontract.symbol = 'SPX'
        mycontract.secType = 'IND'
        mycontract.exchange = 'CBOE'
        mycontract.currency = 'USD'
        
        #self.reqMarketDataType(4)
        self.reqHistoricalData(orderId, mycontract, "20230126-23:59:59", "1 D", "1 hour", "TRADES", 0, 1, 0, [])
        #self.reqMktData(orderId, mycontract, "", 0, 0, [])
        
    #def tickPrice(self, reqId, tickType, price, attrib):
    #   print(f"tickPrice reqId: {reqId}, ticktype: {TickTypeEnum.to_str(tickType)}, price: {price}, attrib: {attrib}")
        
    #def tickSize(self, reqId, tickType, size):
    #    print(f"tickSize reqId: {reqId}, ticktype: {TickTypeEnum.to_str(tickType)}, size: {size}")
    
    def historicalData(self, reqId, bar):
        print(f"Historical Data: {bar}")
        
    def historicalDataEnd(self, reqId, start, end):
        print(f"End of Historical Data")
        print(f"start: {start}, end: {end}")


app = TestApp()
app.connect("127.0.0.1", 7497, 1000)
app.run()

这是我最接近的结果 - 说我没有市场数据订阅。我想我不知道。有市场数据订阅的人可以确认这有效吗？

此外，是否有关于开始使用 Interactive Brokers Native Python API 的良好指南？我发现创建继承自 EClient 和 EWrapper 的应用程序非常不直观。也许这是另一个线程的主题。

Answer 1

您可以尝试使用 EUR.USD 合约来测试您的数据请求，因为该合约不需要订阅市场数据。

我认为来自本机 Interactive Brokers (IB) python API 的 EClient 和 EWrapper 类可能有点难以理解，尤其是因为它需要对类继承、异步过程、装饰器设计模式等有扎实的理解。如果您只是想使用来自 IB 的数据进行研究，然后可以尝试 ib-insync 包，这样您就不必覆盖 EWrapper 的大部分代码来处理从 EClient 接收的数据。

但如果你真的想使用 ibapi，那么这里就是你问题的答案：

您首先需要编写自己的类，该类继承自 EClient 和 EWrapper 类。为了进一步简化这一点，将 EClient 视为您要发送给 IB 的请求，并将 EWrapper 视为您将从 EClient 的请求中收到的消息。所以为了区分每一个请求，需要一个请求ID，即reqId。这是代码：

from datetime import datetime, timedelta, timezone
import threading, time

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
from ibapi.client import BarData
from ibapi.utils import iswrapper


class IB(EClient, EWrapper):

    def __init__(self):
        EClient.__init__(self, self)
        self.data = []

    @iswrapper
    def error(self, reqId, code, msg):
        '''Logging Error'''
        print(f"Error {reqId}, code: {code}, message: {msg}")

    @iswrapper
    def historicalData(self, reqId: int, bar: BarData) -> None:
        self.data.append([bar.date, bar.open, bar.high, bar.low, bar.close, bar.volume])

    @iswrapper
    def historicalDataEnd(self, reqId: int, start: str, end: str):
        start_time = datetime.strptime(start, '%Y%m%d %H:%M:%S')
        end_time = datetime.strptime(end, '%Y%m%d %H:%M:%S')
        super().historicalDataEnd(reqId, start, end)
        print("Historical data request with request ID: ", reqId, " for period between ",
              start_time, " and ", end_time, " is complete.")


def create_contract() -> Contract:
    ticker = Contract()
    ticker = Contract()
    ticker.symbol = 'SPY'
    ticker.secType = 'STK'
    ticker.exchange = 'SMART'
    ticker.currency = 'USD'
    return ticker


def run_loop(app):
    app.run()


if __name__ == '__main__':
    app = IB()
    app.connect(host='localhost', port=7496, clientId=1)
    time.sleep(2)
    api_thread = threading.Thread(target=run_loop, args=(app,))
    api_thread.start()

    ticker = create_contract()
    
    query_time = datetime(2023, 1, 1, tzinfo=timezone.utc).strftime('%Y%m%d %H:%M:%S %Z')
    app.reqHistoricalData(reqId=2,
                          contract=ticker,
                          endDateTime=query_time,
                          durationStr='2 W',
                          barSizeSetting='1 day',
                          whatToShow='TRADES',
                          useRTH=1,
                          formatDate=2,
                          keepUpToDate=False,
                          chartOptions=[])
    time.sleep(2)
    for bar in app.data:
        print(bar)
    app.disconnect()

historicalData 和historicalDataEnd 是属于 EWrapper 类的方法。因此，当我重写这些时，我将 @iswrapper 作为对自己的心理提示，这些方法已被覆盖。如果你密切关注@iswrapper 下的方法和下面的输出，我有办法处理来自我发出的请求的回调，即 error() 处理 app.connect()，historicalData() 处理接收到的数据并追加他们给 self.data 和 historicalDataEnd() 让我们知道，请求已经完成。也就是说，对于您使用 EClient 发出的每个请求，您都需要一种方法来处理在 EWrapper 中收到的消息。最后，我基本上只是从 TestApp 打印出数据。我希望这会帮助你开始！