如何实现Python DataFrame数据在图表中定时自动更新？

首先明确：invoke_long_callback主打处理耗时后台任务（比如复杂计算、大数量级数据拉取），但定期刷新数据并同步更新图表的场景，用dcc.Interval触发普通回调更直接；如果必须用长回调，要确保回调逻辑正确绑定输出并传递数据。

以下是两种可行方案：

方案1：dcc.Interval+普通回调（推荐，适配常规刷新场景） #

适合固定频率刷新数据的场景，逻辑简洁：

1. 定义基础布局，包含图表组件和定时触发器：

import dash
from dash import dcc, html, Input, Output
import pandas as pd
import plotly.express as px

app = dash.Dash(__name__)

# 模拟你的数据刷新逻辑，替换成实际的数据源拉取/更新代码
def get_latest_sales_data():
    return pd.DataFrame({
        "时间": pd.date_range(start="2024-01-01", periods=10, freq="H"),
        "销售额": [120, 150, 90, 200, 180, 220, 160, 190, 210, 230]
    })

app.layout = html.Div([
    dcc.Graph(id="real-time-sales-chart"),
    # 每5秒触发一次刷新，单位为毫秒，可按需调整
    dcc.Interval(
        id="data-refresh-trigger",
        interval=5*1000,
        n_intervals=0
    )
])

2. 编写回调实现数据与图表同步：

@app.callback(
    Output("real-time-sales-chart", "figure"),
    Input("data-refresh-trigger", "n_intervals")
)
def update_sales_chart(_):
    # 获取最新数据
    latest_df = get_latest_sales_data()
    # 生成更新后的折线图
    fig = px.line(latest_df, x="时间", y="销售额", title="实时销售数据")
    return fig

3. 运行应用：

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

方案2：正确使用invoke_long_callback（适配耗时数据刷新） #

如果你的数据刷新逻辑本身耗时（比如从数据库拉取海量数据、复杂数据清洗），用长回调避免阻塞主线程，注意通过回调返回值传递数据，不要直接修改全局变量：

from dash.long_callback import DiskcacheLongCallbackManager
import diskcache

# 初始化长回调缓存管理器
cache = diskcache.Cache("./callback_cache")
long_callback_manager = DiskcacheLongCallbackManager(cache)

app = dash.Dash(__name__, long_callback_manager=long_callback_manager)

app.layout = html.Div([
    dcc.Graph(id="sales-chart-long-callback"),
    # 每10秒触发一次长回调
    dcc.Interval(id="long-callback-trigger", interval=10*1000, n_intervals=0),
    # 用dcc.Store存储序列化后的最新数据，避免重复计算
    dcc.Store(id="latest-sales-data-store")
])

# 长回调：耗时数据刷新，结果存入dcc.Store
@app.long_callback(
    Output("latest-sales-data-store", "data"),
    Input("long-callback-trigger", "n_intervals"),
    prevent_initial_call=False
)
def fetch_latest_data(_):
    # 这里替换成你的耗时数据刷新逻辑
    latest_df = get_latest_sales_data()
    # 将DataFrame序列化为JSON格式存入Store
    return latest_df.to_json(date_format="iso", orient="split")

# 普通回调：从Store取数据更新图表
@app.callback(
    Output("sales-chart-long-callback", "figure"),
    Input("latest-sales-data-store", "data")
)
def update_chart_from_store(data):
    if not data:
        return {}
    # 从JSON反序列化为DataFrame
    df = pd.read_json(data, orient="split")
    fig = px.line(df, x="时间", y="销售额", title="实时销售数据（长回调版）")
    return fig

常见问题排查 #

如果你之前用invoke_long_callback失败，大概率是以下原因：

• 未正确绑定输出目标（比如没关联到图表的figure属性）
• 长回调返回的数据不可序列化（比如直接返回DataFrame而非JSON格式）
• 缺少触发长回调的输入组件（比如没加dcc.Interval或按钮）
• 数据刷新函数get_latest_sales_data()没有返回最新数据（比如读取的是缓存文件而非实时数据源）

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