PyQt5 pyqtgraph蜡烛图报错修复及按需动态刷新实现咨询

问题复盘 #

• 初始IndexError: index 1 is out of bounds for axis 0 with size 0报错根因：实例化CandlestickItem时错误传入了类属性dpoint_var.data2，该类属性始终为初始化时的空数组，而非实例dpoint调用new_dpoint()后生成的实例属性data2，访问空数组索引触发越界。
• 修复属性访问错误后图表不更新的根因：
  • Qt的QGraphicsItem体系不会自动监听绑定数据源的变化，底层numpy数组更新后，图元无法自动感知数据变更
  • 原有CandlestickItem仅在__init__阶段调用一次generatePicture()预渲染所有K线内容，后续数据变化后既没有更新预渲染的绘图缓存，也没有触发重绘事件
  • 原new_dpoint()方法每次生成全新的data2数组对象，CandlestickItem实例持有的还是旧数组的引用，即使触发重绘也读不到新数据
  • 原代码还存在类属性/实例属性混用的遗留bug：new_dpoint()末尾的dpoint.dpoint +=1引用的是全局实例对象，而非当前类实例的dpoint属性，多实例场景下会出现逻辑错误。

初始报错信息：

IndexError                                Traceback (most recent call last)
Input In [5], in <cell line: 99>()
     96 dpoint = dpoint_var()
     97 dpoint.new_dpoint()
---> 99 item = CandlestickItem(dpoint_var.data2)
    100 plt = pg.plot()
    101 plt.addItem(item)

Input In [5], in CandlestickItem.__init__(self, data)
     64 pg.GraphicsObject.__init__(self)
     65 self.data = data  ## data must have fields: time, open, close, min, max
---> 66 self.generatePicture()

Input In [5], in CandlestickItem.generatePicture(self)
     72 p = QtGui.QPainter(self.picture)
     73 p.setPen(pg.mkPen('w'))
---> 74 width = (self.data[1][0] - self.data[0][0]) / 3.
     75 for (t, open, close, min, max) in self.data:
     76     p.drawLine(QtCore.QPointF(t, min), QtCore.QPointF(t, max))

IndexError: index 1 is out of bounds for axis 0 with size 0

核心实现逻辑 #

不需要依赖QTimer轮询，完全通过信号槽机制在新数据到达时触发更新，同时支持增量渲染避免全量重绘，核心修改点如下：

• 改造CandlestickItem，新增数据更新接口，更新预渲染缓存后调用Qt原生的self.update()方法触发重绘，这是Qt图元刷新的标准API，无额外性能开销
• 保留原有QPicture预渲染机制做性能优化，追加新K线时仅绘制新增部分，不需要重绘所有历史K线
• 修正数据类的属性访问逻辑，避免类属性/实例属性混用
• 外部数据源产生新数据时，直接将数据到达信号绑定到图元的更新方法即可，无需轮询

可运行实现代码 #

import pyqtgraph as pg
import numpy as np
from PyQt5 import QtWidgets, QtCore, QtGui
import sys 
from random import uniform

class DpointStore:
    def __init__(self, init_count=20):
        self.dpoint = init_count
        self.data2 = np.empty((0,5), dtype=float)
        # 初始化初始K线数据
        self._gen_batch_data(init_count)

    def _gen_single_kline(self, t):
        # 生成单条符合OHLC结构的测试K线
        open_p = uniform(0.0005, 0.01)
        close_p = open_p + uniform(-0.002, 0.002)
        min_p = min(open_p, close_p) - uniform(0, 0.001)
        max_p = max(open_p, close_p) + uniform(0, 0.001)
        return np.array([t, open_p, close_p, min_p, max_p])

    def _gen_batch_data(self, count):
        # 批量生成count条K线
        start_t = len(self.data2)
        new_data = np.array([self._gen_single_kline(start_t + i) for i in range(count)])
        self.data2 = np.vstack([self.data2, new_data]) if len(self.data2) >0 else new_data
        self.dpoint = len(self.data2)
        return new_data

    def add_new_dpoint(self):
        # 对外暴露的新增单条K线方法，返回新增的数据
        new_kline = self._gen_batch_data(1)
        return new_kline

class CandlestickItem(pg.GraphicsObject):
    def __init__(self, data):
        pg.GraphicsObject.__init__(self)
        self.data = data
        self.bar_width = 1/3. # K线宽度固定
        self.picture = QtGui.QPicture()
        # 初始化时全量渲染初始数据
        self._draw_range(0, len(self.data))

    def _draw_range(self, start_idx, end_idx):
        # 绘制指定索引区间的K线，支持增量绘制
        p = QtGui.QPainter(self.picture)
        p.setPen(pg.mkPen('w'))
        for i in range(start_idx, end_idx):
            t, open_p, close_p, min_p, max_p = self.data[i]
            p.drawLine(QtCore.QPointF(t, min_p), QtCore.QPointF(t, max_p))
            if open_p > close_p:
                p.setBrush(pg.mkBrush('r'))
            else:
                p.setBrush(pg.mkBrush('g'))
            p.drawRect(QtCore.QRectF(t-self.bar_width, open_p, self.bar_width*2, close_p-open_p))
        p.end()

    def append_data(self, new_data):
        # 追加新数据的接口，仅绘制新增部分，不全量重绘
        old_len = len(self.data)
        self.data = np.vstack([self.data, new_data])
        # 仅绘制新增的K线段
        self._draw_range(old_len, len(self.data))
        # 触发重绘
        self.update()

    def full_update(self, new_full_data):
        # 全量替换数据的接口，全量重绘
        self.data = new_full_data
        self.picture = QtGui.QPicture()
        self._draw_range(0, len(self.data))
        self.update()

    def paint(self, p, *args):
        p.drawPicture(0, 0, self.picture)

    def boundingRect(self):
        # QPicture自动计算边界，无需手动处理
        return QtCore.QRectF(self.picture.boundingRect())

if __name__ == '__main__':
    app = QtWidgets.QApplication(sys.argv)
    # 初始化数据存储
    store = DpointStore(init_count=20)
    # 初始化蜡烛图
    item = CandlestickItem(store.data2)
    plt = pg.plot()
    plt.addItem(item)
    plt.setWindowTitle('pyqtgraph 动态蜡烛图')
    # --------------------------
    # 以下QTimer仅为测试用，实际使用时删除
    # 实际场景下将外部数据源的新数据信号连接到item.append_data即可
    def test_add():
        new_k = store.add_new_dpoint()
        item.append_data(new_k)
    timer = QtCore.QTimer()
    timer.timeout.connect(test_add)
    timer.start(1000)
    # --------------------------
    sys.exit(app.exec_())

关键说明 #

• 重绘触发：所有Qt图元内容更新后，调用self.update()即可通知视图层重绘，这是最轻量的刷新方式，不会造成阻塞
• 性能优化：append_data方法仅绘制新增的K线，历史K线的绘制结果会被QPicture缓存，即使上万条K线也能保持流畅
• 信号触发：实际使用时不需要定时器，只要外部数据源（比如行情接口、数据采集线程）产生新数据，直接调用item.append_data(新K线数据)即可，完全符合事件驱动的更新逻辑，没有轮询开销
• 边界处理：每次更新QPicture后，boundingRect()会自动获取新的绘制边界，不会出现渲染裁剪、残影问题

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