使用ib-insync获取标普500指数历史数据：正确配置合约类型

from ib_insync import *
import pandas as pd
import time

# 建立与IB TWS/Gateway的连接
ib = IB()
try:
    ib.connect('127.0.0.1', 7496, clientId=1)
    print("成功连接到IB TWS/Gateway。")
except Exception as e:
    print(f"连接失败: {e}")
    exit()

# 定义需要获取数据的标的列表及其类型和交易所
# 这是一个更灵活的方式来管理不同类型的合约
ticker_definitions = {
    'TSLA': {'type': 'Stock', 'exchange': 'SMART', 'currency': 'USD'},
    'SPX': {'type': 'Index', 'exchange': 'CBOE', 'currency': 'USD'}
}

def get_contract(symbol):
    """根据符号定义创建并返回对应的合约对象"""
    definition = ticker_definitions.get(symbol)
    if not definition:
        raise ValueError(f"未找到符号 '{symbol}' 的合约定义。")

    contract_type = definition['type']
    exchange = definition['exchange']
    currency = definition['currency']

    if contract_type == 'Stock':
        return Stock(symbol, exchange, currency)
    elif contract_type == 'Index':
        return Index(symbol, exchange, currency)
    else:
        raise ValueError(f"不支持的合约类型: {contract_type}")

def extract_historical_data(duration_period, bar_period):
    """
    从IB API提取指定标的的历史数据。

    参数:
    duration_period (str): 持续时间字符串，例如 '1 D', '5 D', '1 Y'。
    bar_period (str): K线周期字符串，例如 '1 hour', '1 day', '1 min'。

    返回:
    pandas.DataFrame: 包含所有标的历史数据的DataFrame。
    """
    all_data = pd.DataFrame()
    for ticker_symbol in ticker_definitions.keys():
        print(f'-- 正在处理: {ticker_symbol}')

        try:
            contract = get_contract(ticker_symbol)
            # 验证合约是否有效，这一步是可选但推荐的，可以帮助发现合约定义错误
            # qualified_contract = ib.qualifyContracts(contract)
            # if not qualified_contract:
            #     print(f"合约验证失败: {ticker_symbol}")
            #     continue
            # contract = qualified_contract[0] # 使用验证后的合约

            # 请求历史数据
            # 对于指数，whatToShow通常是'TRADES'或'MIDPOINT'，取决于交易所和数据类型
            # useRTH=False 表示获取所有交易时段的数据，包括盘前盘后
            bars = ib.reqHistoricalData(
                contract,
                endDateTime='',  # 空字符串表示当前时间
                durationStr=duration_period,
                barSizeSetting=bar_period,
                whatToShow='TRADES', # 对于指数，可能需要根据数据提供商选择 'TRADES' 或 'MIDPOINT'
                useRTH=False,
                formatDate=1
            )

            if bars:
                df = util.df(bars)
                # 删除不需要的列，并添加ticker列
                df = df.drop(['average', 'barCount'], axis=1, errors='ignore')
                df['ticker'] = ticker_symbol
                all_data = pd.concat([all_data, df], ignore_index=True)
                print(f'成功获取 {ticker_symbol} 数据。')
            else:
                print(f'未获取到 {ticker_symbol} 的历史数据。')
        except Exception as e:
            print(f'处理 {ticker_symbol} 时发生错误: {e}')

        # 为了避免触发API限速，每次请求后稍作停顿
        time.sleep(1) 

    return all_data

# 调用函数获取数据
historical_data_df = extract_historical_data('1 D', '1 hour')

# 打印结果
print("\n--- 获取到的历史数据 ---")
print(historical_data_df)

# 断开连接
ib.disconnect()
print("已断开与IB TWS/Gateway的连接。")