如何通过IB API获取当前价格并将其存储到变量中

因为IB API采用异步回调的机制接收行情数据，最优方案是直接在IBapi类的实例属性中新增字段存储目标价格，具体修改如下：

1 改造IBapi类，新增存储属性 #

在初始化方法中定义用来存卖价的变量，回调触发时直接更新该变量：

from ibapi.wrapper import EWrapper
from ibapi.client import EClient
from ibapi.contract import Contract
import time
import threading

class IBapi(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)
        # 新增自定义变量存储AAPL最新卖价，初始值设为None
        self.aapl_ask_price = None

    def tickPrice(self, reqId, tickType, price, attrib):
        # tickType=2对应卖价，reqId=1对应你申请的AAPL行情
        if tickType == 2 and reqId == 1:
            print('The current ask price is: ', price)
            # 拿到价格后直接赋值给实例属性
            self.aapl_ask_price = price

2 新增异步等待逻辑，确保拿到数据后再访问变量 #

IB API的行情推送是异步的，请求行情后需要等待回调触发才能拿到有效数值，不能请求之后立刻访问变量，完整调用示例如下：

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

# 创建AAPL合约
apple_contract = Contract()
apple_contract.symbol = 'AAPL'
apple_contract.secType = 'STK'
apple_contract.exchange = 'SMART'
apple_contract.currency = 'USD'

# 请求行情
app.reqMktData(1, apple_contract, '', False, False, [])

# 把API回调逻辑放到单独子线程运行，避免阻塞主线程的变量读取逻辑
api_thread = threading.Thread(target=app.run, daemon=True)
api_thread.start()

# 轮询等待价格推送完成
while app.aapl_ask_price is None:
    print("等待AAPL卖价推送...")
    time.sleep(0.5)

# 此时变量已经赋值完成，可以在任意业务逻辑中调用
print(f"存储到自定义变量的AAPL卖价为：{app.aapl_ask_price}")

# 用完记得断开连接
app.disconnect()

扩展说明 #

如果你需要同时存储多个标的、多种行情类型（买价、卖价、最新成交价），可以把存储属性改成字典，用reqId或者标的代码当key即可，扩展性更好。

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