ib_insync获取SP500指数历史数据：正确配置合约类型与交易所

from ib_insync import *
import pandas as pd

ib = IB()
# 假设已连接到IB Gateway/TWS，此处为示例，实际运行时需确保连接
# ib.connect('127.0.0.1', 7496, clientId=1)

ticker_list = ['TSLA', 'SPX']

def extract_data_incorrect(duration_period, bar_period):
    full_data = pd.DataFrame()
    for i in ticker_list:
        print(f'-- 尝试获取: {i}')
        # 错误：将SPX定义为Stock合约，且交易所可能不匹配
        contract = Stock(i, 'SMART', 'USD')
        try:
            bars = ib.reqHistoricalData(
                contract,
                endDateTime='',
                durationStr=duration_period,
                barSizeSetting=bar_period,
                whatToShow='TRADES',
                useRTH=False,
                formatDate=1
            )
            df = util.df(bars)
            df = df.drop(['average', 'barCount'], axis=1)
            df['ticker'] = i
            full_data = pd.concat([full_data, df])
        except Exception as e:
            print(f'获取 {i} 数据时发生错误: {e}')
            # 针对SPX，会收到类似 Error 200, reqId X: No security definition has been found for the request
    return full_data

# 注意：此函数在未连接IB API时无法运行
# extract_data_incorrect('1 D', '1 hour')
# print(full_data)

from ib_insync import *
import pandas as pd
import datetime

# 初始化IB对象
ib = IB()

# 连接到IB Gateway或TWS
# 注意：确保IB Gateway/TWS已运行并登录，且API端口为7496（默认）
try:
    ib.connect('127.0.0.1', 7496, clientId=1)
    print("成功连接到Interactive Brokers API。")
except Exception as e:
    print(f"连接失败: {e}")
    # 在实际应用中，这里可能需要更复杂的错误处理或重试机制
    exit() # 连接失败则退出脚本

# 定义要获取数据的金融工具及其类型和交易所
# 使用字典来管理不同类型的合约，增强代码的可读性和可维护性
ticker_configs = {
    'TSLA': {'type': 'Stock', 'exchange': 'SMART', 'currency': 'USD'},
    'SPX': {'type': 'Index', 'exchange': 'CBOE', 'currency': 'USD'} # SPX作为指数，指定CBOE交易所
}

def extract_historical_data(duration_period, bar_period, configs):
    """
    从Interactive Brokers API获取指定金融工具的历史K线数据。

    Args:
        duration_period (str): 数据持续时间，例如 '1 D', '1 W', '1 M'。
        bar_period (str): K线周期，例如 '1 hour', '1 day'。
        configs (dict): 包含金融工具符号、类型、交易所和货币的字典。

    Returns:
        pd.DataFrame: 包含所有金融工具历史数据的DataFrame。
    """
    all_historical_data = pd.DataFrame()
    for ticker_symbol, config in configs.items():
        print(f'-- 正在获取 {ticker_symbol} 的数据...')

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

        # 可选但推荐：验证合约是否有效
        # details = ib.reqContractDetails(contract)
        # if not details:
        #     print(f"未能找到 {ticker_symbol} 的合约详情，请检查符号、类型和交易所。跳过。")
        #     continue
        # else:
        #     print(f"合约 {ticker_symbol} 详情已确认。")

        try:
            # reqHeadTimeStamp 可用于获取最早可用数据的时间戳，这里暂时省略
            # ib.reqHeadTimeStamp(contract, whatToShow='TRADES', useRTH=True)

            bars = ib.reqHistoricalData(
                contract,
                endDateTime='', # 空字符串表示当前时间
                durationStr=duration_period,
                barSizeSetting=bar_period,
                whatToShow='TRADES', # 对于指数，通常是TRADES或MIDPOINT
                useRTH=False, # 是否只使用常规交易时间，True为常规交易时间，False为所有时间
                formatDate=1 # 返回日期格式为YYYYMMDD HH:MM:SS
            )

            if bars:
                df = util.df(bars)
                # 某些数据类型可能没有'average'或'barCount'，这里进行安全删除
                if 'average' in df.columns:
                    df = df.drop('average', axis=1)
                if 'barCount' in df.columns:
                    df = df.drop('barCount', axis=1)
                df['ticker'] = ticker_symbol
                all_historical_data