掌握IB API历史数据下载：解决异步断开连接问题

TabTab AI

首个全链路 Data Agent，让数据搜集、处理到深度分析一步到位。

292

查看详情

import threading import time # 引入time模块用于可能的延迟或超时处理 from ibapi.client import EClient from ibapi.wrapper import EWrapper from ibapi.contract import Contract from ibapi.common import Bar  # 定义一个继承EWrapper和EClient的类，用于处理IB API的事件和请求 class IBapi(EWrapper, EClient):     def __init__(self):         EClient.__init__(self, self)         # 初始化一个threading.Event对象，用于在数据接收后发出信号         self.data_received_event = threading.Event()         # 用于存储接收到的历史数据，可选         self.historical_data_storage = []       # 重写历史数据回调函数     def historicalData(self, reqId: int, bar: Bar):         # 打印接收到的历史数据         print(f"ReqId: {reqId}, Date: {bar.date}, High: {bar.high}, Low: {bar.low}, Volume: {bar.volume}")         # 将数据存储起来，如果需要进一步处理         self.historical_data_storage.append(bar)          # 在接收到数据后，设置事件，通知主线程数据已到达         # 注意：如果请求的是大量数据，此方法会在每条bar数据到达时被调用。         # 如果只想在所有数据都接收完毕后通知，需要更复杂的逻辑，         # 例如在historicalDataEnd回调中设置事件。         # 对于本例，我们假设一次请求的数据量不大，或者我们仅需确认数据开始接收。         # 更严谨的做法是在historicalDataEnd中设置：         # self.data_received_event.set()      # 重写历史数据结束回调函数，这是更推荐的设置事件的地方     def historicalDataEnd(self, reqId: int, start: str, end: str):         print(f"HistoricalDataEnd. ReqId: {reqId} from {start} to {end}")         # 在所有历史数据接收完毕后，设置事件，通知主线程         self.data_received_event.set()      # 错误处理回调     def Error(self, reqId: int, errorCode: int, errorString: str, advancedOrderRejectjson=''):         print(f"Error. Id: {reqId}, Code: {errorCode}, Msg: {errorString}")         if errorCode == 2104: # 2104是数据流已建立，不一定是错误             pass         # 如果是致命错误，也可以考虑设置事件或采取其他措施         # 例如，如果请求失败，我们可能不希望主线程一直等待         if errorCode in [162, 200, 502, 504]: # 常见错误码示例：162-历史数据请求失败, 200-合约无效             print("Request failed, unblocking main thread.")             self.data_received_event.set() # 即使失败也解除阻塞，防止死锁  # 实例化IBapi客户端 app = IBapi()  # 连接到IB TWS/gateway # 默认地址 '127.0.0.1'，默认端口 7497 (TWS) 或 7496 (gateway)，客户端ID 123 app.connect('127.0.0.1', 7497, 123)  # 启动一个独立的线程来运行IB API的事件循环 # daemon=True 确保主程序退出时，此线程也会随之退出 api_thread = threading.Thread(target=app.run, daemon=True) api_thread.start()  # 等待连接建立，给IB API客户端一些时间来初始化 # 实际应用中可能需要更复杂的连接状态检查 time.sleep(1)   # 定义合约信息 contract = Contract() contract.symbol = "VIX" contract.secType = "FUT" contract.exchange = "CFE" contract.currency = "USD" # 注意：对于期货，lastTradeDateOrContractMonth 应指定合约月份或最后交易日 # "20240117" 是一个具体的日期，确保该合约在IB系统中有数据 contract.lastTradeDateOrContractMonth = "202401" # 通常是yyYYMM格式 contract.multiplier = "1000" contract.includeExpired = True # 包含已过期合约，如果需要历史数据  # 重置事件，以防万一（尽管每次请求前应是未设置状态） app.data_received_event.clear()  # 请求历史数据 # reqId: 请求ID # contract: 合约对象 # endDateTime: 结束时间，格式 "YYYYMMDD HH:MM:SS TZ" 或 "" 表示现在 # durationStr: 持续时间，如 "1 M" (1个月), "1 W", "1 D", "1 Y" # barSizeSetting: K线周期，如 "1 min", "5 mins", "1 day" # whatToShow: 显示什么类型的数据，如 "TRADES", "BID", "ASK", "MIDPOINT" # useRTH: 0 (所有数据), 1 (常规交易时间) # formatDate: 1 (YYYYMMDD HH:MM:SS), 2 (YYYYMMDD) # keepUpToDate: 是否保持实时更新 # chartOptions: 额外图表选项 app.reqHistoricalData(1, contract, "", "1 M", "30 mins", "BID", 0, 1, False, [])  # 等待历史数据接收完毕的信号 # 可以添加timeout参数，例如 app.data_received_event.wait(timeout=30) # 如果在30秒内未收到数据，则继续执行，防止无限等待 print("Waiting for historical data...") data_received = app.data_received_event.wait(timeout=60) # 最多等待60秒  if data_received:     print("Historical data successfully received.")     # 在这里可以对 app.historical_data_storage 进行处理     # 例如：     # for bar in app.historical_data_storage:     #     print(f"Stored Bar: {bar.date}, {bar.high}") else:     print("Timeout: Historical data not received within the specified time.")     print("Please check contract details, connection, and TWS/Gateway logs.")  # 断开与IB服务器的连接 app.disconnect()  print("done")