Python调用IB TWS API execDetails存成交明细到DataFrame报错

报错由3个核心写法错误导致：

• 构造待插入数据时写法错误：你写的{"ExecDetails. ReqId:", reqId, "Symbol:", contract.symbol, ... , execution}不符合Python字典语法，字典要求每个元素为键:值的键值对格式，仅用逗号分隔值的大括号结构会被Python识别为集合（set），这就是报错提示无法拼接set类型的直接原因。
• 结果取值逻辑错误：reqExecutions是异步发送请求的方法，本身没有返回值；execDetails是接收数据的回调函数，不是存储数据的属性，直接将函数赋值给变量无法拿到存好的成交数据。
• 额外兼容性问题：DataFrame.append()方法在pandas 1.4及以上版本已被废弃，运行会产生警告，更推荐用列表暂存记录后一次性转DataFrame，性能更好也更稳定。

import time
import pandas as pd
from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.execution import ExecutionFilter

class TradingApp(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)
        # 初始化存储成交记录的列表，比逐行追加DataFrame效率高很多
        self.order_exec_records = []
        # 定义DataFrame列名，和返回字段一一对应
        self.exec_columns = [
            'ReqId', 'Symbol', 'SecType', 'Currency', 'ExecId', 'Time', 'Account', 
            'Exchange', 'Side', 'Shares', 'Price', 'PermId', 'ClientId', 'OrderId', 
            'Liquidation', 'CumQty', 'AvgPrice', 'OrderRef', 'EvRule', 'EvMultiplier', 
            'ModelCode', 'LastLiquidity'
        ]
        self.exec_fetch_done = False

    def execDetails(self, reqId, contract, execution):
        super().execDetails(reqId, contract, execution)
        # 打印调试信息
        print(f"ExecDetails. ReqId:{reqId}, Symbol:{contract.symbol}, SecType:{contract.secType}, Currency:{contract.currency}, ExecId:{execution.execId}")
        # 按列名对应构造单条记录字典，所有字段从contract/execution对象取对应属性
        single_record = {
            'ReqId': reqId,
            'Symbol': contract.symbol,
            'SecType': contract.secType,
            'Currency': contract.currency,
            'ExecId': execution.execId,
            'Time': execution.time,
            'Account': execution.acctNumber,
            'Exchange': execution.exchange,
            'Side': execution.side,
            'Shares': execution.shares,
            'Price': execution.price,
            'PermId': execution.permId,
            'ClientId': execution.clientId,
            'OrderId': execution.orderId,
            'Liquidation': execution.liquidation,
            'CumQty': execution.cumQty,
            'AvgPrice': execution.avgPrice,
            'OrderRef': execution.orderRef,
            'EvRule': execution.evRule,
            'EvMultiplier': execution.evMultiplier,
            'ModelCode': execution.modelCode,
            'LastLiquidity': execution.lastLiquidity
        }
        self.order_exec_records.append(single_record)

    def execDetailsEnd(self, reqId):
        super().execDetailsEnd(reqId)
        print(f"ReqId {reqId} 成交记录拉取完成")
        self.exec_fetch_done = True

    def get_exec_df(self):
        # 将暂存的记录批量转成DataFrame
        return pd.DataFrame(self.order_exec_records, columns=self.exec_columns)

# 调用前置步骤：先完成连接、启动后台消息监听线程
# app = TradingApp()
# app.connect("127.0.0.1", 7497, clientId=1)
# import threading
# threading.Thread(target=app.run, daemon=True).start()
# time.sleep(1) # 等待连接建立完成

# 发送成交记录拉取请求
app.reqExecutions(10001, ExecutionFilter())
# 等待拉取完成，最长等待5秒，比固定sleep1秒更可靠
wait_count = 0
while not app.exec_fetch_done and wait_count < 50:
    time.sleep(0.1)
    wait_count += 1

# 取出最终结果
orderExec_df = app.get_exec_df()
print(orderExec_df)

• 运行前必须先完成TWS/IB网关的配置：开启API端口、勾选允许本地连接，且必须启动后台消息监听线程，否则回调无法正常触发。
• 如果需要筛选特定时间段、特定品种的成交记录，可以给ExecutionFilter传入对应筛选参数，不需要全量拉取。

内容的提问来源于stack exchange，提问作者Kurtis Chua