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

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")