IB TWS API实时数据存入Pandas DataFrame遇NameError：名称未定义

问题说明 #

尝试将IB TWS API接收的实时市场深度数据整理到Pandas DataFrame中，控制台可正常打印数据，但执行代码时触发NameError: name 'reqId' is not defined错误，无法完成数据存储。

错误原因 #

1. 主线程中直接使用reqId、position等变量，但这些变量仅在IBapi类的updateMktDepth回调方法内部有效，主线程未定义这些变量，导致名称未定义错误。
2. 数据存入DataFrame的逻辑位置错误：原代码在主线程提前执行DataFrame创建和追加操作，此时还未收到任何API数据，且变量不存在。
3. 未提前初始化全局的df变量，后续调用df.append会引发额外错误。

修正后的代码 #

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


class IBapi(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)
        self.data = []

    def error(self, reqId, errorCode, errorString):
        print("Error: ", reqId, " ", errorCode, " ", errorString)
    
    def updateMktDepth(self, reqId, position: int, operation: int,
                       side: int, price: float, size):
        global df
        super().updateMktDepth(reqId, position, operation, side, price, size)
        # 生成带时间戳的行数据
        current_time = datetime.datetime.now()
        row_data = [current_time, reqId, position, operation, side, price, size]
        self.data.append(row_data)
        # 将行数据转为DataFrame并追加到全局df
        new_row = pd.DataFrame([row_data], columns=df.columns)
        df = pd.concat([df, new_row], ignore_index=True)
        print("UpdateMarketDepth. ReqId:", reqId, "Position:", position, "Operation:",
              operation, "Side:", side, "Price:", price, "Size:", size) 
        
class DisconnectWrapper(EWrapper):
    def __init__(self):
        EWrapper.__init__(self)
        
    def error(self, reqId, errorCode, errorString):
        print(f"Error: {reqId} {errorCode} {errorString}")
        
        
def run_loop():
    app.run()

# 初始化全局空DataFrame，指定列名
cols = ['Time','ReqId','Position','Operation','Side','Price','Size']
df = pd.DataFrame(columns=cols)

app = IBapi()
app.connect('127.0.0.1', 7497, 13217)

#Start the socket in a thread
api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()

time.sleep(1) #Sleep interval to allow time for connection to server

#Create contract object
contract = Contract()
contract.symbol = "SPY"
contract.secType = "STK"
contract.exchange = "ARCA"
contract.currency = "USD"

#Request Market Depth
app.reqMktDepth(1, contract, 5, False, [])

time.sleep(5) #sleep to allow enough time for data to be returned

# 查看最终DataFrame内容
print("\n最终市场深度数据：")
print(df)

# 断开连接
client = EClient(DisconnectWrapper())
client.disconnect()
app.disconnect()

关键修正点 #

• 提前初始化全局df：在程序开始时创建指定列名的空DataFrame，避免后续追加时出错。
• 回调内处理数据：将DataFrame的更新逻辑移到updateMktDepth回调方法中，每次收到API推送的深度数据时，生成带时间戳的行数据并追加到全局DataFrame。
• 使用pd.concat替代df.append：df.append已被弃用，改用pd.concat完成数据追加，保证兼容性。

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