Dash 动态更新图表（实时数据可视化）

本指南展示如何使用 Dash + Plotly + WebSocket / 轮询 实现 DASH 加密货币数据的实时动态图表，支持：

• 价格实时线图
• 交易量柱状图
• K 线图 + 技术指标
• 主节点数量监控
• InstantSend 交易流

---

核心技术栈

bash

pip install dash plotly pandas requests websocket-client

组件	作用
Dash	Web 框架
Plotly	交互图表
dcc.Interval	定时刷新
WebSocket	实时推送（推荐）
Dash Platform / Insight API	数据源

---

示例 1：实时价格线图（每 5 秒更新）

python

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

app = dash.Dash(__name__)

# 初始数据
fig = go.Figure()
fig.add_trace(go.Scatter(x=[], y=[], mode='lines+markers', name='DASH Price'))
fig.update_layout(
    title='DASH 实时价格 (USD)',
    xaxis_title='时间',
    yaxis_title='价格',
    xaxis_range=[datetime.now(), datetime.now()],
    yaxis_range=[20, 40]
)

app.layout = html.Div([
    html.H1("DASH 实时价格监控", style={'textAlign': 'center'}),
    dcc.Graph(id='live-price', figure=fig),
    dcc.Interval(id='interval-component', interval=5*1000, n_intervals=0)  # 每5秒
])

# 回调：更新图表
@app.callback(
    Output('live-price', 'figure'),
    Input('interval-component', 'n_intervals')
)
def update_price(n):
    # 获取最新价格（CoinGecko）
    try:
        url = 'https://api.coingecko.com/api/v3/simple/price?ids=dash&vs_currencies=usd'
        price = requests.get(url).json()['dash']['usd']
    except:
        price = 25.0  # 模拟数据

    now = datetime.now().strftime('%H:%M:%S')
    
    # 更新数据
    fig.data[0].x = fig.data[0].x + (now,)
    fig.data[0].y = fig.data[0].y + (price,)

    # 限制显示最近 20 个点
    if len(fig.data[0].x) > 20:
        fig.data[0].x = fig.data[0].x[-20:]
        fig.data[0].y = fig.data[0].y[-20:]

    fig.update_layout(xaxis_range=[fig.data[0].x[0], fig.data[0].x[-1]])
    return fig

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

效果：每 5 秒添加一个点，平滑动画，自动缩放。

---

示例 2：WebSocket 实时交易推送（Insight API）

使用 Dash Insight WebSocket 监听新交易。

python

import dash
from dash import dcc, html, Input, Output
import plotly.graph_objects as go
import json
import websocket
import threading
import pandas as pd
from datetime import datetime

# 全局数据
tx_data = pd.DataFrame(columns=['time', 'value', 'txid'])

app = dash.Dash(__name__)

app.layout = html.Div([
    html.H2("DASH 实时交易流"),
    dcc.Graph(id='tx-chart'),
    dcc.Interval(id='interval', interval=1000, n_intervals=0)
])

def on_message(ws, message):
    global tx_data
    data = json.loads(message)
    if data.get('op') == 'utx':
        for tx in data.get('x', {}).get('txs', []):
            value = tx.get('value', 0) / 1e8
            txid = tx.get('txid', '')
            time = datetime.now().strftime('%H:%M:%S')
            new_row = pd.DataFrame([{'time': time, 'value': value, 'txid': txid}])
            tx_data = pd.concat([tx_data, new_row]).tail(50)  # 保留最近50笔

def start_ws():
    ws = websocket.WebSocketApp(
        "wss://testnet-insight.dashevo.org/insight-api",
        on_message=on_message
    )
    ws.run_forever()

# 启动 WebSocket 线程
threading.Thread(target=start_ws, daemon=True).start()

@app.callback(
    Output('tx-chart', 'figure'),
    Input('interval', 'n_intervals')
)
def update_tx_chart(n):
    if tx_data.empty:
        return go.Figure().add_scatter(x=[], y=[])

    fig = go.Figure()
    fig.add_trace(go.Bar(
        x=tx_data['time'],
        y=tx_data['value'],
        text=tx_data['txid'].apply(lambda x: x[:8] + '...'),
        hovertemplate='TX: %{text}<br>价值: %{y} DASH'
    ))
    fig.update_layout(title='最近 50 笔交易 (DASH)', xaxis_title='时间', yaxis_title='金额')
    return fig

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

效果：新交易立即出现在柱状图，悬停显示 TXID。

---

示例 3：K 线图动态更新（每分钟）

python

import dash
from dash import dcc, html, Input, Output
import plotly.graph_objects as go
import pandas as pd
import requests
from datetime import datetime, timedelta

app = dash.Dash(__name__)

# 初始 K 线数据
df = pd.DataFrame(columns=['date', 'open', 'high', 'low', 'close'])

app.layout = html.Div([
    html.H2("DASH 实时 K 线图"),
    dcc.Graph(id='candlestick'),
    dcc.Interval(id='kline-interval', interval=60*1000, n_intervals=0)  # 每分钟
])

@app.callback(
    Output('candlestick', 'figure'),
    Input('kline-interval', 'n_intervals')
)
def update_kline(n):
    global df
    try:
        url = 'https://api.coingecko.com/api/v3/coins/dash/ohlc?vs_currency=usd&days=1'
        data = requests.get(url).json()
        new_df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close'])
        new_df['date'] = pd.to_datetime(new_df['timestamp'], unit='ms')
        df = pd.concat([df, new_df]).drop_duplicates('timestamp').tail(50)
    except:
        pass

    fig = go.Figure(data=[go.Candlestick(
        x=df['date'],
        open=df['open'], high=df['high'], low=df['low'], close=df['close']
    )])
    fig.update_layout(title='DASH 1分钟 K 线', xaxis_rangeslider_visible=False)
    return fig

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

---

示例 4：主节点数量实时监控

python

@app.callback(
    Output('masternode-gauge', 'figure'),
    Input('interval-mn', 'n_intervals')
)
def update_mn_count(n):
    try:
        count = requests.get('https://dashnexus.org/api/masternodes').json()['total']
    except:
        count = 3800

    fig = go.Figure(go.Indicator(
        mode = "gauge+number",
        value = count,
        title = {'text': "主节点数量"},
        gauge = {'axis': {'range': [None, 5000]}}
    ))
    return fig

---

完整实时仪表板（整合所有）

python

import dash
from dash import dcc, html, Input, Output
import plotly.graph_objects as go
import plotly.express as px
import pandas as pd
import requests
import threading
import websocket
import json
from datetime import datetime

app = dash.Dash(__name__)

# 全局数据
price_history = []
tx_stream = pd.DataFrame(columns=['time', 'value'])

app.layout = html.Div([
    html.H1("DASH 实时监控中心", className="text-center mb-4"),
    
    # 价格 + 交易量
    html.Div([
        html.Div([dcc.Graph(id='price-chart')], className="col-md-8"),
        html.Div([dcc.Graph(id='volume-gauge')], className="col-md-4")
    ], className="row"),
    
    # 交易流 + 主节点
    html.Div([
        html.Div([dcc.Graph(id='tx-stream')], className="col-md-6"),
        html.Div([dcc.Graph(id='mn-gauge')], className="col-md-6")
    ], className="row mt-4"),
    
    dcc.Interval(id='fast-interval', interval=5*1000, n_intervals=0),
    dcc.Interval(id='slow-interval', interval=60*1000, n_intervals=0)
])

# WebSocket 监听交易
def ws_thread():
    def on_message(ws, msg):
        global tx_stream
        data = json.loads(msg)
        if data.get('op') == 'utx':
            for tx in data.get('x', {}).get('txs', []):
                new_row = pd.DataFrame([{
                    'time': datetime.now().strftime('%H:%M:%S'),
                    'value': tx.get('value', 0) / 1e8
                }])
                tx_stream = pd.concat([tx_stream, new_row]).tail(30)
    ws = websocket.WebSocketApp("wss://testnet-insight.dashevo.org/insight-api", on_message=on_message)
    ws.run_forever()

threading.Thread(target=ws_thread, daemon=True).start()

# 价格更新
@app.callback(
    Output('price-chart', 'figure'),
    Input('fast-interval', 'n_intervals')
)
def update_price(n):
    try:
        p = requests.get('https://api.coingecko.com/api/v3/simple/price?ids=dash&vs_currencies=usd').json()['dash']['usd']
        price_history.append({'time': datetime.now().strftime('%H:%M:%S'), 'price': p})
        df = pd.DataFrame(price_history[-50:])
    except:
        df = pd.DataFrame(price_history[-50:]) if price_history else pd.DataFrame()

    fig = px.line(df, x='time', y='price', title='实时价格')
    return fig

# 其他图表类似...

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

---

性能优化建议

技巧	说明
限制数据点	df.tail(100) 避免内存爆炸
WebSocket > 轮询	实时性更强
使用 FigureWidget	更平滑动画
部署	gunicorn app:server + Nginx

---

部署为 Web 应用

bash

# requirements.txt
dash==2.17.0
plotly==5.18.0
pandas
requests
websocket-client

# 启动
gunicorn app:server -b 0.0.0.0:8050

---

资源

• 实时 API：https://insight.dash.org
• Dash Nexus：https://dashnexus.org/api
• Plotly Dash 文档：https://dash.plotly.com/live-updates
• 示例项目：GitHub – dash-live-dashboard

---

一句话总结：使用 dcc.Interval + WebSocket + Plotly，轻松实现 DASH 价格、交易、主节点等数据的毫秒级动态更新，一键部署为专业监控面板。

需要我打包成 Docker 镜像、React 前端嵌入，或 Telegram 通知集成 吗？随时告诉我！