使用ib-insync获取标普500指数历史数据：区分股票与指数合约

from ib_insync import *
import pandas as pd
import time

# 连接到Interactive Brokers 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()

# 定义需要获取数据的标的列表，包含类型、交易所和货币信息
# 对于股票，通常使用'SMART'交易所
# 对于SPX指数，通常使用'CBOE'交易所
ticker_info = [
    {'symbol': 'TSLA', 'type': 'Stock', 'exchange': 'SMART', 'currency': 'USD'},
    {'symbol': 'SPX', 'type': 'Index', 'exchange': 'CBOE', 'currency': 'USD'}
]

def extract_historical_data(duration_period: str, bar_period: str) -> pd.DataFrame:
    """
    从Interactive Brokers API提取指定标的的历史数据。
    支持股票和指数。

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

    Returns:
        pd.DataFrame: 包含所有标的历史数据的DataFrame。
    """
    all_data = pd.DataFrame()

    for item in ticker_info:
        symbol = item['symbol']
        contract_type = item['type']
        exchange = item['exchange']
        currency = item['currency']

        print(f'-- 正在获取 {symbol} ({contract_type}) 的数据...')

        contract = None
        if contract_type == 'Stock':
            contract = Stock(symbol, exchange, currency)
        elif contract_type == 'Index':
            contract = Index(symbol, exchange, currency)
        else:
            print(f"未知合约类型: {contract_type} for {symbol}，跳过。")
            continue

        if not contract:
            continue

        # 可选：预先解析和验证合约，提高代码健壮性
        # try:
        #     resolved_contract = ib.qualifyContracts(contract)
        #     if not resolved_contract:
        #         print(f"无法解析合约: {symbol}. 请检查参数。")
        #         continue
        #     contract = resolved_contract[0] # 使用解析后的合约
        # except Exception as e:
        #     print(f"合约 {symbol} 解析失败: {e}")
        #     continue

        try:
            # reqHistoricalData的whatToShow参数对于指数通常是'TRADES'或'MIDPOINT'
            # useRTH对于指数数据通常设置为False，表示包含盘前盘后数据
            bars = ib.reqHistoricalData(
                contract,
                endDateTime='', # 留空表示当前时间
                durationStr=duration_period,
                barSizeSetting=bar_period,
                whatToShow='TRADES', # 对于指数，也可尝试'MIDPOINT'
                useRTH=False, # 对于指数，通常获取所有可用数据
                formatDate=1 # 1表示返回日期时间对象
            )

            if bars:
                df = util.df(bars)
                # 移除不必要的列，确保数据框结构一致性
                if 'average' in df.columns:
                    df = df.drop(['average'], axis=1)
                if 'barCount' in df.columns:
                    df = df.drop(['barCount'], axis=1)
                df['ticker'] = symbol
                all_data = pd.concat([all_data, df], ignore_index=True)
                print(f"成功获取 {symbol} 的 {len(df)} 条数据。")
            else:
                print(f"未获取到 {symbol} 的历史数据。")

        except Exception as e:
            print(f'获取 {symbol} 数据时发生错误: {e}')

        # 避免请求频率过快，TWS/Gateway有请求限制，建议每次请求后暂停
        time.sleep(1) 

    return all_data

# 调用函数获取数据
historical_df = extract_historical_data('1 D', '1 hour')
print("\n--- 获取到的历史数据 ---")
print(historical_df.head())
print(historical_df.tail())
print(f"\n成功获取数据的标的：\n{historical_df['ticker'].value_counts()}")

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