Python Dash：如何下载图表所用的Pandas DataFrame为CSV文件

核心结论 #

不需要强制拆分两个@app.callback函数，既可以将下载功能整合到现有图表更新的callback中，也可以通过缓存数据的方式拆分两个callback（更推荐后者，避免重复计算）。

实现方案 #

方案1：整合到单个callback（紧凑但存在冗余计算） #

直接扩展现有callback的输出，通过判断触发源分别处理图表更新和下载请求：

1. 先在布局中添加下载组件和下载按钮：

# 布局部分新增
dcc.Download(id="download-dataframe-csv"),
html.Button("下载数据", id="download-button", n_clicks=0)

2. 修改现有callback：

import dash
from dash import Input, Output, State, callback_context, dcc, html
import plotly.express as px
import pandas as pd
import urllib.request
import json

@app.callback(
    [Output('dash_graph', 'figure'),
     Output('download-dataframe-csv', 'data')],
    [Input('button', 'n_clicks'),
     Input('download-button', 'n_clicks')],
    [State('ticker_input', 'value'),
     State('start_date_input', 'date'),
     State('end_date_input', 'date'),
     State('entry_price_input', 'value'),
     State('spread_input', 'value')]
)
def update_result_and_download(n_clicks_graph, n_clicks_download, ticker_input, start_date_input, end_date_input, entry_price_input, spread_input):
    ctx = callback_context
    if not ctx.triggered:
        return dash.no_update, dash.no_update
    
    # 获取触发组件ID
    trigger_id = ctx.triggered[0]['prop_id'].split('.')[0]
    
    # 统一处理API请求与数据清洗
    API_TOKEN = "demo"
    url = "your_api_url_here"  # 替换为实际API地址
    response = urllib.request.urlopen(url)
    eod_data = json.loads(response.read())
    eod_data_df = pd.json_normalize(eod_data)
    full_list = prep_df(eod_data_df)[0]
    date_list = prep_df(eod_data_df)[1]
    df = prep_graph(start_date_input, end_date_input, full_list, date_list, entry_price_input, entry_price_input, spread_input).copy()
    
    if trigger_id == 'button':
        # 触发图表更新
        chart = px.bar(
            data_frame=df,
            x='date',
            y='count',
            title='Time Series Chart' + '<br>' + '<span style="font-size: 10px;">{}  |  {} to {}  |  ${:.2f} to ${:.2f}  |  Spread ${}</span>'.format(ticker_input, start_date_input, end_date_input, entry_price_input, entry_price_input+spread_input, spread_input),
        )
        return chart, dash.no_update
    elif trigger_id == 'download-button':
        # 触发CSV下载
        return dash.no_update, dcc.send_data_frame(df.to_csv, f"{ticker_input}_data.csv", index=False)

方案2：用dcc.Store缓存数据（高效无冗余） #

通过dcc.Store组件缓存处理后的DataFrame，让图表更新和下载功能共享数据，避免重复调用API和清洗数据：

1. 布局中添加缓存组件、下载组件和下载按钮：

# 布局部分新增
dcc.Store(id='cached-dataframe'),  # 用于缓存DataFrame
dcc.Download(id="download-dataframe-csv"),
html.Button("下载数据", id="download-button", n_clicks=0)

2. 修改图表更新的callback，新增缓存输出：

@app.callback(
    [Output('dash_graph', 'figure'),
     Output('cached-dataframe', 'data')],
    [Input('button', 'n_clicks')],
    [State('ticker_input', 'value'),
     State('start_date_input', 'date'),
     State('end_date_input', 'date'),
     State('entry_price_input', 'value'),
     State('spread_input', 'value')]
)
def update_result(n_clicks, ticker_input, start_date_input, end_date_input, entry_price_input, spread_input):
    if n_clicks is None:
        return dash.no_update, dash.no_update

    API_TOKEN = "demo"
    url = "your_api_url_here"  # 替换为实际API地址
    response = urllib.request.urlopen(url)
    eod_data = json.loads(response.read())
    eod_data_df = pd.json_normalize(eod_data)
    full_list = prep_df(eod_data_df)[0]
    date_list = prep_df(eod_data_df)[1]
    df = prep_graph(start_date_input, end_date_input, full_list, date_list, entry_price_input, entry_price_input, spread_input).copy()

    chart = px.bar(
        data_frame=df,
        x='date',
        y='count',
        title='Time Series Chart' + '<br>' + '<span style="font-size: 10px;">{}  |  {} to {}  |  ${:.2f} to ${:.2f}  |  Spread ${}</span>'.format(ticker_input, start_date_input, end_date_input, entry_price_input, entry_price_input+spread_input, spread_input),
    )

    # 将DataFrame转为JSON格式存入缓存
    return chart, df.to_json(date_format='iso', orient='split')

3. 新增下载功能的callback：

@app.callback(
    Output('download-dataframe-csv', 'data'),
    [Input('download-button', 'n_clicks')],
    [State('cached-dataframe', 'data'),
     State('ticker_input', 'value')]
)
def download_data(n_clicks, cached_data, ticker_input):
    if n_clicks is None or cached_data is None:
        return dash.no_update
    
    # 从缓存中恢复DataFrame
    df = pd.read_json(cached_data, orient='split')
    # 返回CSV下载数据
    return dcc.send_data_frame(df.to_csv, f"{ticker_input}_data.csv", index=False)

方案对比 #

• 方案1：代码结构紧凑，但每次点击下载按钮都会重复执行API请求和数据清洗，适合数据量小、API响应快的场景。
• 方案2：通过缓存避免冗余计算，性能更优，适合数据量大或API响应较慢的场景，是更推荐的实现方式。

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