以AI量化为生：18.实时K线图表系统开发

唐耀星

334人浏览 · 2026-04-10 07:30:00

唐耀星 · 2026-04-10 07:30:00 发布

本文是《以AI量化为生》系列的第18篇，我们将把EnhancedChartWidget接入vnpy的ChartWizard模块，实现实时tick数据更新。从tick-to-bar合成逻辑到成交量增量计算，从实时价格线显示到光标标签修复，解决实盘图表的各种诡异问题。

实时K线图表

快速上手

第一步：连接CTP

启动程序后，先在"系统"菜单中选择"连接CTP"。

连接成功后，状态栏会显示"已连接"。只有连接成功后，才能订阅实时tick数据。

第二步：打开K线图表

菜单栏会出现"K线图表"入口：

K线图表入口

点击进入后，输入合约代码（如rb2605.SHFE），点击"新建图表"：

新建图表

系统会自动：

查询最近10000条历史分钟K线数据
订阅实时tick数据
根据当前周期合成K线并更新图表

图表支持多周期切换（1分钟、5分钟、15分钟、1小时、日线），所有技术指标会自动重新计算。

多周期切换

写在前面

最近有读者私信问：图表看起来挺完整了，但实盘的时候能不能实时更新K线？还有成交量怎么总是跳来跳去，一会儿特别大一会儿又是0？

这确实是个绕不开的问题。回测的时候数据都是历史K线，直接加载就完事了。但实盘就不一样了，tick数据不停地过来，你得把tick合成到K线上，还得保证成交量算得对。听起来简单，实际做起来坑可不少。

这篇文章就讲讲怎么把ChartWizard接入主系统，让图表支持实时tick更新，还有怎么解决成交量计算和光标显示的各种诡异问题。

ChartWizard是什么

vnpy自带一个ChartWizard模块，就是专门用来看实时K线的。之前我们一直在回测系统里用自己写的EnhancedChartWidget，回测是没问题，但实盘缺少实时更新的入口。ChartWizard正好填补了这个空白。

ChartWizard的工作原理很简单：你输入合约代码，它去查历史数据，然后订阅实时tick，把tick合成到K线上。听起来很完美，但原版的ChartWizard用的是基础的ChartWidget，没有我们那些技术指标和交互功能。

所以第一步就是把ChartWizard的图表组件换成EnhancedChartWidget。

集成EnhancedChartWidget

改起来其实不复杂。ChartWizard的UI代码在vnpy_chartwizard/ui/widget.py，核心就是ChartWizardWidget这个类。

最关键的两行改动：

# 导入增强版图表组件
from core.charts.enhanced_chart_widget import EnhancedChartWidget

def create_chart(self) -> EnhancedChartWidget:
    """创建增强版图表对象"""
    chart: EnhancedChartWidget = EnhancedChartWidget()
    return chart

本来是创建基础的ChartWidget，现在换成EnhancedChartWidget。这样一来，所有的技术指标、参数配置、双击专注模式这些功能就都有了。

然后在main.py里启用ChartWizard：

from vnpy_chartwizard import ChartWizardApp

main_engine.add_app(ChartWizardApp)

启动程序，菜单栏就多了个"K线图表"入口，点进去输入合约代码，图表就出来了。

tick数据处理的三个关键问题

图表出来了，但tick更新的时候有几个明显的问题：

每次tick到来，图表自动滚到最右边，你正在看的历史区域一下子就跳走了
tick明明是9:33的数据，却创建了新的K线，而不是更新9:30那根15分钟K线
看不到最新价格在哪个位置

这三个问题背后其实是tick-to-bar合成逻辑没处理好。

问题1：取消自动滚动

vnpy的ChartWidget有个自动滚动逻辑，每次更新bar都会判断：如果你现在看的区域接近最右边，就强制滚到最右边。这在回测的时候挺好，但实盘就很烦，你想看前面的K线结构，结果每次tick一来就被拉回去。

解决办法就是重写update_bar方法，把自动滚动的逻辑注释掉：

def update_bar(self, bar: BarData) -> None:
    """更新单个K线数据"""
    self._manager.update_bar(bar)

    for item in self._items.values():
        item.update_bar(bar)

    self._update_plot_limits()

    # 不再自动滚动到最右边，让用户自由控制
    # if self._right_ix >= (self._manager.get_count() - self._bar_count / 2):
    #     self.move_to_right()

问题2：智能tick-to-bar合成

这个问题更核心。原来的逻辑是用vnpy的BarGenerator来合成K线，但BarGenerator不知道你现在显示的是什么周期的图表。你在看15分钟图，它还是按1分钟合成，结果tick一来就创建了新K线。

正确的做法是：根据当前图表的显示周期，判断tick应该更新哪根K线。

比如现在是15分钟图，最后一根K线是9:30，tick时间是9:33。我们得计算出9:33属于哪个15分钟区间：

# 15分钟周期
bar_minute = (last_bar.datetime.minute // 15) * 15  # 9:30
bar_start = last_bar.datetime.replace(minute=bar_minute, second=0)
bar_end = bar_start + timedelta(minutes=15)  # 9:30 - 9:45

# tick在这个范围内，更新当前K线
if bar_start <= tick_time < bar_end:
    updated_bar = BarData(
        open_price=last_bar.open_price,  # 开盘价不变
        high_price=max(last_bar.high_price, tick.last_price),
        low_price=min(last_bar.low_price, tick.last_price),
        close_price=tick.last_price,
        volume=new_volume,  # 成交量单独处理
        ...
    )
    self.update_bar(updated_bar)
else:
    # tick超出范围，创建新K线
    self._create_new_bar_from_tick(tick, bar_end)

这样就能保证tick正确地更新到对应周期的K线上，而不是无脑创建新K线。

问题3：实时价格线

这个需求比较直观：看盘的时候总想知道最新价格在什么位置。解决办法是用PyQtGraph的InfiniteLine画一条横线：

def _init_price_line(self):
    """初始化实时价格线"""
    self.price_line = pg.InfiniteLine(
        pos=0,
        angle=0,  # 水平线
        pen=pg.mkPen(color=(255, 165, 0), width=1,
                     style=QtCore.Qt.PenStyle.DashLine),
        movable=False
    )
    candle_plot.addItem(self.price_line)

    # 价格标签
    self.price_label = pg.TextItem(
        anchor=(0, 0.5),
        color=(255, 165, 0)
    )
    candle_plot.addItem(self.price_label)

def _update_price_line(self, price: float):
    """更新价格线位置"""
    self.price_line.setPos(price)
    self.price_label.setText(f" {price:.2f} ")
    # 标签位置稍微往左偏移，避免被Y轴挡住
    x_pos = view_range[0][1] - view_width * 0.05
    self.price_label.setPos(x_pos, price)

每次tick更新后调用_update_price_line(tick.last_price)，价格线就会跟着移动。

成交量计算的坑

tick数据能更新K线了，但成交量又出问题了。表现是：刚开始成交量显示正常，但每次切换周期或者新建图表，第一个tick到来的时候成交量就暴涨，显示好几万手。

问题根源：累计成交量vs增量成交量

调试了半天才发现，tick数据里的volume字段不是这个tick的成交量，而是当日累计成交量。比如开盘到现在总共成交了5000手，tick.volume就是5000。

但我们的K线需要的是增量成交量：这个tick相比上个tick增加了多少手。如果你直接把tick.volume加到K线上，那就相当于把全天的成交量都加上去了，自然暴涨。

解决方案：增量计算

正确的做法是记录上一次tick的累计成交量，然后计算增量：

# 初始化时添加成交量追踪
self._last_tick_volume = 0  # 上一次tick的累计成交量

def update_tick(self, tick) -> None:
    """更新tick数据"""
    if hasattr(tick, 'volume') and tick.volume > 0:
        if self._last_tick_volume == 0:
            # 第一次收到tick，不计算增量（避免暴涨）
            new_volume = last_bar.volume
            self._last_tick_volume = tick.volume
        else:
            # 后续tick，计算增量
            volume_delta = tick.volume - self._last_tick_volume
            if volume_delta > 0:
                new_volume = last_bar.volume + volume_delta
            else:
                new_volume = last_bar.volume
            self._last_tick_volume = tick.volume

关键点有两个：

第一次收到tick时，只记录_last_tick_volume，不计算增量（避免把历史累计量全加上）
后续tick才计算增量并累加到K线

创建新K线时的成交量

还有个细节：创建新K线的时候，成交量应该初始化为0，而不是tick.volume：

new_bar = BarData(
    open_price=tick.last_price,
    high_price=tick.last_price,
    low_price=tick.last_price,
    close_price=tick.last_price,
    volume=0,  # 新K线成交量初始化为0
    ...
)

这样后续的tick更新就能正确累加增量了。

光标x轴标签的诡异bug

还有个隐蔽的问题：切换副图指标的时候，十字光标的x轴时间标签会消失。比如你勾选了wavetrend指标，然后取消勾选，光标的时间标签就不见了。

问题根源

调试发现，vnpy的ChartWidget在初始化光标时，会把x_label添加到最后一个plot上。但当你隐藏副图时，x_label所在的plot被隐藏了，标签自然就看不见了。

解决方案

正确的做法是：每次副图可见性改变时，把x_label重新定位到最后一个可见的plot上。

def _relocate_cursor_x_label(self):
    """将光标的x轴标签重新定位到最后一个可见的plot上"""
    if not self._cursor or not hasattr(self._cursor, '_x_label'):
        return
        
    # 找到最后一个可见的plot
    visible_plots = [
        name for name in self._plots.keys() 
        if self._plots[name].isVisible()
    ]
    
    if not visible_plots:
        return
        
    bottom_plot_name = visible_plots[-1]
    bottom_plot = self._plots[bottom_plot_name]
    x_label = self._cursor._x_label
    
    # 从旧的plot中移除（用plot.removeItem而不是scene.removeItem）
    current_label_plot = getattr(self._cursor, '_x_label_plot_name', None)
    if current_label_plot and current_label_plot in self._plots:
        old_plot = self._plots[current_label_plot]
        if x_label in old_plot.items:
            old_plot.removeItem(x_label)
    
    # 添加到新的plot
    if x_label not in bottom_plot.items:
        bottom_plot.addItem(x_label, ignoreBounds=True)
    
    self._cursor._x_label_plot_name = bottom_plot_name
    x_label.setZValue(1000)
    x_label.show()

关键点是用plot.removeItem()而不是scene().removeItem()，这样才能正确维护PlotItem的items列表。

然后在切换副图指标的方法里调用这个修复：

def _toggle_sub_indicator(self, name: str, state: int):
    """切换副图指标"""
    # ... 原有的显示/隐藏逻辑 ...
    
    # 重新定位x_label到正确的可见plot上
    self._relocate_cursor_x_label()

这样不管怎么切换副图，光标的时间标签都能正确显示。

交易时段支持

还有最后一个细节：1小时K线的聚合。之前第13篇文章讲过，期货的交易时段不是自然小时，比如中国期货是09:00-09:59、10:00-11:14、11:15-14:14这样的。

为了保证小时K线按实际交易时段聚合，需要在ChartWizard加载历史数据后设置交易时段：

from config.trading_sessions_config import get_trading_session_by_symbol

def process_history_event(self, event: Event) -> None:
    """处理历史数据事件"""
    history: list[BarData] = event.data
    bar: BarData = history[0]
    chart: EnhancedChartWidget = self.charts[bar.vt_symbol]

    # 根据合约设置交易时段
    trading_session = get_trading_session_by_symbol(
        bar.symbol,
        bar.exchange.value
    )
    chart.trading_session = trading_session

    chart.update_history(history)