如何用IBAPI持续获取历史K线并转换为DataFrame？求代码示例

一、原生IBAPI改进版 #

问题分析 #

你的原有代码存在几个关键问题：

• app.run()是阻塞调用，外部的while True无法循环执行
• 历史数据请求的时间范围固定，无法动态获取“过去一周”的数据
• 重复使用同一个reqId会导致API报错
• 缺少定时重复请求的逻辑

改进后的代码 #

import time
from datetime import datetime, timedelta
from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
import pandas as pd
import mplfinance as mpf

class TestApp(EClient, EWrapper):
    def __init__(self):
        EClient.__init__(self, self)
        self.histbars = []
        self.next_order_id = 1
        self.cnadles_plot = 10000
        self.hours_change = -5

    def nextValidId(self, orderId: int):
        self.next_order_id = orderId
        # 首次请求后启动定时循环
        self.schedule_historical_data_request()

    def schedule_historical_data_request(self):
        while True:
            self.histbars = []  # 清空之前的数据
            self.request_historical_data()
            time.sleep(15)  # 每15秒请求一次

    def request_historical_data(self):
        # 创建合约
        mycontract = Contract()
        mycontract.symbol = "TSLA"
        mycontract.secType = "STK"
        mycontract.exchange = "SMART"
        mycontract.currency = "USD"

        # 设置请求的时间范围：过去一周
        end_time = datetime.now().strftime("%Y%m%d-%H:%M:%S")
        duration_str = "7 D"

        # 用新的reqId发起请求
        self.reqHistoricalData(
            self.next_order_id,
            mycontract,
            end_time,
            duration_str,
            "1 min",
            "TRADES",
            0,
            1,
            False,
            []
        )
        self.next_order_id += 1  # 自增reqId避免冲突

    def historicalData(self, reqId: int, bar):
        bardict = {
            "Date": bar.date,
            "Open": bar.open,
            "High": bar.high,
            "Low": bar.low,
            "Close": bar.close,
            "Volume": bar.volume,
            "Count": bar.barCount
        }
        self.histbars.append(bardict)

    def historicalDataEnd(self, reqId: int, start: str, end: str):
        print(f"请求完成 | 开始时间: {start}, 结束时间: {end}")
        
        # 转换为DataFrame
        df = pd.DataFrame.from_records(self.histbars)
        df["Date"] = pd.to_datetime(df["Date"].str.split().str[:2].str.join(' '))
        df["Date"] = df["Date"] + timedelta(hours=self.hours_change)
        df.set_index("Date", inplace=True)
        df["Volume"] = pd.to_numeric(df["Volume"])

        # 计算VWAP
        def vwap(df_group):
            avg_price = (df_group["High"] + df_group["Low"]) / 2
            df_group["vwap"] = (avg_price * df_group["Volume"]).cumsum() / df_group["Volume"].cumsum()
            return df_group
        
        df = df.groupby(df.index.date, group_keys=False).apply(vwap)

        # 输出和可视化
        print(df.tail(self.cnadles_plot))
        print(df.dtypes)
        apdict = mpf.make_addplot(df['vwap'])
        mpf.plot(df, type="candle", volume=True, tight_layout=True, show_nontrading=True, addplot=apdict)

if __name__ == "__main__":
    app = TestApp()
    app.connect("127.0.0.1", 7496, 1000)
    app.run()

关键改进点 #

• 将定时请求逻辑放在客户端内部，避免app.run()阻塞外部循环
• 动态计算时间范围，每次请求获取过去一周的数据
• 自动递增reqId，避免重复ID导致的API错误
• 每次请求前清空历史数据列表，避免数据累积混乱

二、ib_insync简便版 #

ib_insync是基于IBAPI的异步封装库，代码更简洁，适合新手快速实现需求：

安装ib_insync #

pip install ib_insync

实现代码 #

import time
from datetime import timedelta
import pandas as pd
import mplfinance as mpf
from ib_insync import IB, Stock, util

def fetch_and_process_data(ib):
    # 创建合约
    stock = Stock("TSLA", "SMART", "USD")
    ib.qualifyContracts(stock)

    # 获取过去一周的1分钟K线数据
    bars = ib.reqHistoricalData(
        contract=stock,
        endDateTime="",
        durationStr="7 D",
        barSizeSetting="1 min",
        whatToShow="TRADES",
        useRTH=False,
        keepUpToDate=False,
        formatDate=1
    )

    # 转换为DataFrame
    df = util.df(bars)
    df["date"] = pd.to_datetime(df["date"])
    df["date"] = df["date"] + timedelta(hours=-5)  # 时区调整
    df.set_index("date", inplace=True)

    # 计算VWAP
    def vwap(df_group):
        avg_price = (df_group["high"] + df_group["low"]) / 2
        df_group["vwap"] = (avg_price * df_group["volume"]).cumsum() / df_group["volume"].cumsum()
        return df_group
    
    df = df.groupby(df.index.date, group_keys=False).apply(vwap)

    # 输出和可视化
    print(df.tail(10000))
    print(df.dtypes)
    apdict = mpf.make_addplot(df['vwap'])
    mpf.plot(df, type="candle", volume=True, tight_layout=True, show_nontrading=True, addplot=apdict)

if __name__ == "__main__":
    ib = IB()
    ib.connect("127.0.0.1", 7496, clientId=1000)

    try:
        while True:
            fetch_and_process_data(ib)
            time.sleep(15)  # 每15秒请求一次
    finally:
        ib.disconnect()

ib_insync的优势 #

• 无需手动处理回调函数，直接返回数据
• 内置DataFrame转换工具util.df()，无需手动构造字典
• 代码结构更直观，减少新手容易出错的细节

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