IB API Python客户端历史数据获取教程：异步处理与等待机制

import threading
from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
import time # 引入time模块用于演示

class IBapi(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)
        # 初始化一个Event对象，用于同步历史数据接收
        self.data_received_event = threading.Event()
        self.historical_data_bars = [] # 用于存储接收到的历史数据

    def historicalData(self, reqId, bar):
        # 当接收到历史数据时，打印并存储
        print(f"ReqId: {reqId}, Date: {bar.date}, High: {bar.high}, Low: {bar.low}, Volume: {bar.volume}")
        self.historical_data_bars.append(bar)

    def historicalDataEnd(self, reqId, start, end):
        # 历史数据接收完毕的回调
        print(f"HistoricalDataEnd. ReqId: {reqId} from {start} to {end}")
        self.data_received_event.set() # 设置Event，通知主线程数据已接收完毕

    def error(self, reqId, errorCode, errorString, advancedOrderRejectJson=''):
        # 错误处理回调
        print(f"Error. Id: {reqId}, Code: {errorCode}, Msg: {errorString}")
        # 对于历史数据请求，如果返回错误（如无数据），也应释放等待
        if errorCode == 162: # No historical data for this contract
            print("No historical data found for the specified contract and period.")
            self.data_received_event.set() # 即使没有数据，也应该释放等待
        # 可以根据其他错误码进行更精细的处理

# 主程序入口
if __name__ == "__main__":
    app = IBapi()
    # 连接到IB TWS/Gateway
    app.connect('127.0.0.1', 7497, 123)

    # 启动API线程。daemon=True 确保主程序退出时线程也会终止
    api_thread = threading.Thread(target=app.run, daemon=True)
    api_thread.start()

    # 等待连接建立，通常需要一小段时间
    time.sleep(1) # 给予IB API连接和初始化足够的时间

    # 定义合约
    contract = Contract()
    contract.symbol = "VIX"
    contract.secType = "FUT"
    contract.exchange = "CFE"
    contract.currency = "USD"
    contract.lastTradeDateOrContractMonth = "20240117"
    contract.multiplier = "1000"
    contract.includeExpired = True

    # 清除之前的Event状态，确保重新等待（如果多次请求）
    app.data_received_event.clear()
    # 请求历史数据
    # reqId, contract, endDateTime, durationStr, barSizeSetting, whatToShow, useRTH, formatDate, keepUpToDate, chartOptions
    app.reqHistoricalData(1, contract, "", "1 M", "30 mins", "BID", 0, 1, False, [])

    print("请求已发送，等待历史数据...")
    # 阻塞主线程，直到data_received_event被设置（即historicalDataEnd被调用）
    # 可以添加超时机制，防止无限等待：app.data_received_event.wait(timeout=30)
    app.data_received_event.wait()
    print("历史数据接收完毕或超时。")

    # 打印接收到的数据量
    print(f"共接收到 {len(app.historical_data_bars)} 条历史数据。")

    # 断开连接
    app.disconnect()
    print("程序执行完毕。")