IB Python API获取指定股票最近x秒实时价格 解决价格不更新问题

• 你当前代码在第一次收到卖价回调（tickType=2对应卖盘报价）时，就把当下的固定ask值传入了checkLogic函数，之后15秒的循环全程复用这个不变的数值，自然不会有价格更新。
• 回调函数中使用time.sleep会直接阻塞盈透API的事件驱动循环，后续新的报价推送根本无法进入tickPrice回调，就算想获取新价格也收不到消息。

• 新增实例变量存储实时更新的最新卖价，每次收到tickPrice推送时只更新这个变量，不做阻塞逻辑。
• 单独启动一个计时线程处理15秒的数据收集逻辑，不阻塞主API的消息接收循环。
• 计时结束后再统一导出CSV并断开连接。

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

# 时区可按需自定义，这里默认用本地时区，可替换为你原代码中的tz变量
tz = datetime.datetime.now().astimezone().tzinfo

symbol = "DAI"
contract = Contract()
contract.symbol = symbol
contract.secType = 'STK'
contract.exchange = 'SMART'
contract.currency = "EUR"
contract.primaryExchange = "SMART"

class TestApp(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)
        self.latest_ask = None # 存储实时更新的最新卖价
        self.collect_duration = 15 # 采集时长，可自行修改

    def nextValidId(self, orderId):
        self.nextOrderId = orderId
        self.start()

    def start(self):
        self.reqMarketDataType(3)
        self.reqMktData(1, contract, '', False, False, [])
        # 启动独立的采集计时线程，不阻塞API消息循环
        threading.Thread(target=self.collect_price_loop, daemon=True).start()

    def tickPrice(self, reqId, tickType, price, attrib):
        # 每次收到新的卖价推送就更新最新卖价
        if tickType == 2 and reqId == 1 and price > 0:
            self.latest_ask = price

    def collect_price_loop(self):
        # 等待第一次价格推送成功
        while self.latest_ask is None:
            time.sleep(0.1)
        
        prices = []
        seconds = []
        time_ = []
        seconds_counter = 0
        end_time = datetime.datetime.now() + timedelta(seconds=self.collect_duration)

        while datetime.datetime.now() <= end_time:
            current_time = datetime.datetime.now(tz)
            prices.append(self.latest_ask)
            seconds.append(seconds_counter)
            time_.append(str(current_time.time())[:8])
            time.sleep(1)
            seconds_counter += 1

        df = pd.DataFrame(np.column_stack([time_, seconds, prices]),
                          columns=['Time', 'Seconds', 'Prices'])
        df.to_csv(f"Prices/{symbol}.csv", index=False)

        self.disconnect()

def main():
    app = TestApp()
    app.connect('127.0.0.1', 7497, 24)
    app.run()

if __name__ == "__main__":
    main()

• 提前创建好Prices文件夹，否则导出CSV会报错
• 确保盈透客户端（TWS/IB Gateway）已开启API接入权限，端口和客户端ID对应正确
• 如果需要采集买价或者成交价，修改tickType的判断条件即可

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