如何在mplfinance实时K线图上绘制并实时更新对称三角形？

核心思路 #

tlines参数本身可行，但它是静态绘制的——每次调用plot()时都会覆盖之前的线条。实时更新的关键是：每次新K线进来后，重新识别对称三角形的关键顶点，生成新的tlines参数再传入绘图函数。

步骤实现 #

1. 识别对称三角形的关键顶点
   对称三角形需要至少四个符合条件的高低点：

• 左上方高点（阶段最高）
• 左下方低点（阶段最低）
• 右上方次高点（价格略低于左高点）
• 右下方次低点（价格略高于左低点）

可以用滚动窗口提取近期高低点，再加入判断逻辑筛选符合形态的点：

# 从实时更新的K线DataFrame中提取符合条件的高低点
if len(data) >= 10:
    # 滚动5根K线找近期高低点
    recent_highs = data['High'].rolling(5).max().dropna()
    recent_lows = data['Low'].rolling(5).min().dropna()
    
    # 筛选符合对称三角形的点（示例逻辑，可根据需求调整）
    peak1_idx, peak1_val = recent_highs.index[-10], recent_highs.iloc[-10]
    peak2_idx, peak2_val = recent_highs.index[-5], recent_highs.iloc[-5]
    trough1_idx, trough1_val = recent_lows.index[-10], recent_lows.iloc[-10]
    trough2_idx, trough2_val = recent_lows.index[-5], recent_lows.iloc[-5]
    
    # 验证形态：高点递减，低点递增
    valid_pattern = (peak1_val > peak2_val) and (trough1_val < trough2_val)
else:
    valid_pattern = False

2. 动态生成tlines参数
   每次绘图前，根据最新顶点生成线条字典列表，确保传入的tlines是最新的：

triangle_tlines = []
if valid_pattern:
    triangle_tlines = [
        # 三角形上边：左高点 → 右低点
        {'x1': peak1_idx, 'y1': peak1_val, 'x2': trough2_idx, 'y2': trough2_val, 
         'color':'#ff3333', 'linestyle':'--', 'linewidth':1.5},
        # 三角形下边：左低点 → 右高点
        {'x1': trough1_idx, 'y1': trough1_val, 'x2': peak2_idx, 'y2': peak2_val, 
         'color':'#ff3333', 'linestyle':'--', 'linewidth':1.5},
        # 可选：三角形底边（左高点→左低点）
        {'x1': peak1_idx, 'y1': peak1_val, 'x2': trough1_idx, 'y2': trough1_val, 
         'color':'#666666', 'linestyle':'-', 'linewidth':1}
    ]

3. 实时更新循环逻辑
   结合实时数据流（比如WebSocket推送），每次更新K线后重新计算顶点、生成线条，再调用plot()实现实时重绘：

import mplfinance as mpf
import pandas as pd
import time
import matplotlib.pyplot as plt

# 初始化K线数据容器
data = pd.DataFrame(columns=['Open', 'High', 'Low', 'Close'], index=pd.DatetimeIndex([]))
mpf_style = mpf.make_mpf_style(base_mpf_style='yahoo', gridstyle='')

# 模拟实时数据推送循环（替换为你的实际数据源）
while True:
    # 生成模拟新K线
    new_candle = pd.DataFrame({
        'Open': [data['Close'].iloc[-1] if len(data) > 0 else 25.0],
        'High': [25.3 + (0.2 * (pd.Timestamp.now().second % 5))],
        'Low': [24.7 + (0.2 * (pd.Timestamp.now().second % 5))],
        'Close': [25.0 + (0.2 * (pd.Timestamp.now().second % 5))]
    }, index=[pd.Timestamp.now()])
    
    # 更新数据，保留最近50根K线
    data = pd.concat([data, new_candle]).tail(50)
    
    # 计算对称三角形顶点和线条
    valid_pattern = False
    triangle_tlines = []
    if len(data) >= 10:
        recent_highs = data['High'].rolling(5).max().dropna()
        recent_lows = data['Low'].rolling(5).min().dropna()
        
        if len(recent_highs) >= 10 and len(recent_lows) >= 10:
            peak1_idx, peak1_val = recent_highs.index[-10], recent_highs.iloc[-10]
            peak2_idx, peak2_val = recent_highs.index[-5], recent_highs.iloc[-5]
            trough1_idx, trough1_val = recent_lows.index[-10], recent_lows.iloc[-10]
            trough2_idx, trough2_val = recent_lows.index[-5], recent_lows.iloc[-5]
            
            valid_pattern = (peak1_val > peak2_val) and (trough1_val < trough2_val)
            if valid_pattern:
                triangle_tlines = [
                    {'x1': peak1_idx, 'y1': peak1_val, 'x2': trough2_idx, 'y2': trough2_val, 
                     'color':'#ff3333', 'linestyle':'--', 'linewidth':1.5},
                    {'x1': trough1_idx, 'y1': trough1_val, 'x2': peak2_idx, 'y2': peak2_val, 
                     'color':'#ff3333', 'linestyle':'--', 'linewidth':1.5}
                ]
    
    # 实时重绘K线和三角形
    fig, axlist = mpf.plot(data, type='candle', style=mpf_style, 
                           tlines=triangle_tlines, title='实时K线+对称三角形',
                           returnfig=True, block=False)
    mpf.show()
    time.sleep(2)
    plt.close(fig)  # 关闭旧图释放内存

关键注意事项 #

• 形态识别逻辑要严谨：示例中的高低点筛选是简化版，实际可以结合TA-Lib的极值函数（如talib.PIVOT_HIGH）或自定义算法，确保识别到真正的对称三角形形态。
• 控制数据量：实时更新时只保留最近N根K线，避免计算和绘图卡顿。
• 每次重绘必须传最新tlines：mplfinance不会缓存之前的线条，每次调用plot()都要传入重新生成的tlines列表，否则旧线条会消失。

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