请求协助：通过IBKR API获取EMINI期权链（Python实现）

以下是修正后的完整代码，可实现按到期日、行权价、期权类型获取合约ID、希腊值及价格，并将数据存入Pandas DataFrame：

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
from ibapi.ticktype import TickTypeEnum
import threading
import time
import pandas as pd

class IBApiApp(EWrapper, EClient):
    def __init__(self):
        EWrapper.__init__(self)
        EClient.__init__(self, wrapper=self)
        self.contracts_by_expiry = {}  # 按到期日存储期权合约
        self.option_data = {}          # 存储期权数据（key: reqId）
        self.current_req_id = 100      # 独立请求ID起始值，避免冲突

    def error(self, reqId, errorCode, errorString):
        print(f"错误. ID: {reqId}, 代码: {errorCode}, 信息: {errorString}")

    def contractDetails(self, reqId, contractDetails):
        """收集所有期权合约的详细信息"""
        contract = contractDetails.contract
        expiry = contract.lastTradeDateOrContractMonth
        
        if expiry not in self.contracts_by_expiry:
            self.contracts_by_expiry[expiry] = []
        
        self.contracts_by_expiry[expiry].append({
            "conId": contract.conId,
            "strike": contract.strike,
            "right": contract.right,
            "contract": contract
        })
        print(f"已收集合约: {contract.symbol} {expiry} {contract.right} {contract.strike}")

    def tickPrice(self, reqId, tickType, price, attrib):
        """捕获最新价格"""
        if reqId not in self.option_data:
            return
        if TickTypeEnum.to_str(tickType) == "LAST_PRICE":
            self.option_data[reqId]["price"] = price
            print(f"更新价格: ReqId {reqId}, 价格 = {price}")

    def tickGeneric(self, reqId, tickType, value):
        """捕获希腊值数据"""
        if reqId not in self.option_data:
            return
        
        tick_str = TickTypeEnum.to_str(tickType)
        field_map = {
            "DELTA": "delta",
            "GAMMA": "gamma",
            "THETA": "theta",
            "VEGA": "vega",
            "RHO": "rho"
        }
        
        if tick_str in field_map:
            self.option_data[reqId][field_map[tick_str]] = value
            print(f"更新希腊值: ReqId {reqId}, {tick_str} = {value}")

    def start(self):
        """启动连接并请求合约详情"""
        self.connect("127.0.0.1", 7496, clientId=1)
        
        thread = threading.Thread(target=self.run)
        thread.start()
        setattr(self, "_thread", thread)
        
        time.sleep(1)

        if not self.isConnected():
            print("连接失败")
            return

        # 定义EMINI标的合约（用来获取所有期权链）
        es_contract = Contract()
        es_contract.symbol = "ES"
        es_contract.secType = "OPT"
        es_contract.exchange = "GLOBEX"
        es_contract.currency = "USD"

        # 请求所有期权合约详情
        self.reqContractDetails(1, es_contract)

    def stop(self):
        """断开连接"""
        self.disconnect()
        self._thread.join()

    def request_option_data(self, expiry):
        """批量请求指定到期日的所有期权数据"""
        if expiry not in self.contracts_by_expiry:
            print(f"无到期日[{expiry}]的期权合约")
            return
        
        self.reqMarketDataType(1)  # 使用实时数据（可选：2=冻结，3=延迟，4=延迟冻结）
        
        for opt in self.contracts_by_expiry[expiry]:
            req_id = self.current_req_id
            self.current_req_id += 1
            
            # 初始化该期权的数据存储
            self.option_data[req_id] = {
                "conId": opt["conId"],
                "expiry": expiry,
                "strike": opt["strike"],
                "right": opt["right"],
                "price": None,
                "delta": None,
                "gamma": None,
                "theta": None,
                "vega": None,
                "rho": None
            }
            
            # 发起市场数据请求（包含希腊值）
            self.reqMktData(req_id, opt["contract"], "", False, False, [])
            print(f"发起请求: ReqId {req_id}, {opt['right']} {opt['strike']} {expiry}")

    def get_option_dataframe(self):
        """将收集的数据转换为Pandas DataFrame"""
        df = pd.DataFrame.from_dict(self.option_data, orient='index')
        # 重置索引并排序
        df = df.reset_index(drop=True)
        df = df.sort_values(by=["expiry", "strike", "right"])
        return df

def main():
    app = IBApiApp()
    app.start()

    # 等待合约详情收集完成（根据网络情况调整时长）
    time.sleep(15)

    # 选择前两个到期日的期权进行数据请求
    if len(app.contracts_by_expiry) >= 2:
        target_expiries = list(app.contracts_by_expiry.keys())[:2]
        for expiry in target_expiries:
            app.request_option_data(expiry)
    else:
        print("未收集到足够的到期日数据")

    # 等待数据推送完成
    time.sleep(25)

    # 生成并打印DataFrame
    df = app.get_option_dataframe()
    print("\n=== 期权链数据 ===")
    print(df)

    # 可选：保存到CSV文件
    # df.to_csv("emini_option_chain.csv", index=False)

    app.stop()

if __name__ == "__main__":
    main()

关键修改说明 #

• 请求ID管理：使用独立的current_req_id避免不同请求的数据混淆
• 合约自动收集：通过contractDetails回调获取EMINI的所有期权合约，无需硬编码行权价
• 完整数据捕获：实现tickPrice和tickGeneric回调，分别捕获期权价格与希腊值（Delta/Gamma/Theta/Vega/Rho）
• 结构化存储：用字典整理数据，最终转换为排序后的Pandas DataFrame
• 批量请求：支持一次性获取指定到期日下的所有行权价、类型的期权数据

使用注意事项 #

1. 确保IB Gateway或TWS已启动，端口设置为7496（可在代码中修改）
2. Spyder环境中需提前安装pandas和ibapi库
3. 根据网络延迟调整sleep时长，确保数据完全收集

内容的提问来源于stack exchange，提问作者Praveen Kumar