Plotly实现事件触发式蜡烛图模拟器：按需追加数据

要实现按需追加数据点更新蜡烛图，Plotly的静态绘图模式无法满足同一窗口实时更新的需求，改用Dash框架是最优方案——它能在浏览器中运行交互式应用，支持通过回调函数响应点击/计时器事件，实时更新图表数据。

实现思路 #

• 用Dash搭建基础应用框架，包含图表区域和触发控件（按钮或计时器）
• 维护一份动态更新的数据集，每次触发事件时追加新数据
• 通过回调函数更新OHLC蜡烛图和EMA曲线的数据源，实现无刷新更新

完整代码示例 #

import dash
from dash import dcc, html, Input, Output, State
import plotly.graph_objects as go
import pandas as pd

# 模拟生成初始K线和EMA数据（替换成你的真实数据逻辑）
def generate_initial_data():
    data = {
        'Open': [100],
        'High': [102],
        'Low': [98],
        'Close': [101],
        'ema13': [100.5]
    }
    return pd.DataFrame(data, index=pd.date_range(start='2024-01-01', periods=1))

# 基于上一根K线生成新数据（替换成你的真实数据生成/获取逻辑）
def generate_new_data(last_close, last_ema):
    open_price = last_close
    high_price = open_price + pd.np.random.uniform(0, 3)
    low_price = open_price - pd.np.random.uniform(0, 3)
    close_price = low_price + pd.np.random.uniform(0, high_price - low_price)
    # 计算新的EMA（用标准EMA公式）
    new_ema = (close_price * (2/(13+1))) + (last_ema * (1 - 2/(13+1)))
    return pd.DataFrame({
        'Open': [open_price],
        'High': [high_price],
        'Low': [low_price],
        'Close': [close_price],
        'ema13': [new_ema]
    }, index=[df.index[-1] + pd.Timedelta(hours=1)])

# 初始化应用和初始数据
app = dash.Dash(__name__)
df = generate_initial_data()

# 应用布局
app.layout = html.Div([
    html.Button('添加下一根K线', id='add-candle-btn', n_clicks=0),
    # 可选：添加自动更新计时器，默认关闭
    dcc.Interval(
        id='auto-update-interval',
        interval=3*1000,  # 3秒触发一次
        n_intervals=0,
        disabled=True
    ),
    dcc.Graph(id='candlestick-graph')
])

# 回调函数：响应触发事件并更新图表
@app.callback(
    Output('candlestick-graph', 'figure'),
    [Input('add-candle-btn', 'n_clicks'),
     Input('auto-update-interval', 'n_intervals')],
)
def update_graph(n_clicks, n_intervals):
    global df
    # 仅在按钮点击或计时器触发时更新数据
    if n_clicks > 0 or n_intervals > 0:
        last_close = df['Close'].iloc[-1]
        last_ema = df['ema13'].iloc[-1]
        new_row = generate_new_data(last_close, last_ema)
        df = pd.concat([df, new_row])
    
    # 重新生成图表
    fig = go.Figure()
    fig.add_trace(go.Ohlc(
        x=df.index,
        open=df['Open'],
        high=df['High'],
        low=df['Low'],
        close=df['Close'],
        name='OHLC'
    ))
    fig.add_trace(go.Scatter(
        x=df.index,
        y=df['ema13'],
        opacity=0.7,
        line=dict(color='blue', width=2),
        name='EMA13'
    ))
    fig.update_layout(xaxis_rangeslider_visible=False)
    return fig

if __name__ == '__main__':
    app.run_server(debug=True)

关键说明 #

• 触发模式：支持两种更新方式——点击按钮手动添加K线，或者解除计时器的disabled=True配置，实现自动定时更新
• 数据维护：通过全局变量df动态存储数据集，每次触发事件时追加新数据行
• 无刷新更新：Dash的回调机制会自动在浏览器中更新图表，不会弹出新窗口
• 适配真实数据：替换generate_initial_data和generate_new_data函数，接入你的真实数据源（如API拉取、本地数据库读取等）

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