Python Plotly 串流動態圖表與 ZeroMQ 整合應用

– 9 分鐘閱讀

在現今資料驅動的時代，實時資料處理與視覺化扮演著關鍵角色，特別是在金融市場、物聯網等需要即時監控和反應的領域。本文將示範如何結合 Python、Plotly 和 ZeroMQ，打造一個能動態更新圖表的應用程式，以滿足實時資料展示的需求。此應用程式能接收來自 ZeroMQ 伺服器的串流資料，並利用 Plotly 的互動式圖表功能，即時呈現資料變化，讓使用者能更直觀地掌握資料趨勢。

安裝必要的函式庫

首先，需要安裝必要的函式庫，包括plotly、pandas和numpy。

import plotly.graph_objects as go
import pandas as pd
import numpy as np

建立一個FigureWidget

接下來，建立一個FigureWidget物件，這將用於顯示實時資料。

f = go.FigureWidget()

新增資料到FigureWidget

新增一些初始資料到FigureWidget中。

f.append_trace(go.Scatter(name='SYMBOL', mode='lines+markers', marker={'symbol': 'x'}), row=1, col=1)
f.append_trace(go.Scatter(name='MOMENTUM', line=dict(width=1, dash='dash'), mode='lines+markers', marker={'symbol': 'x'}), row=3, col=1)

更新FigureWidget的佈局

更新FigureWidget的佈局，以適應實時資料的顯示。

# f.update_layout(height=600)

建立一個FigureWidget物件

建立一個FigureWidget物件，以顯示實時資料。

fig = go.FigureWidget(f)

接收和處理實時資料

使用socket接收實時資料，並將其處理為DataFrame格式。

import socket
import datetime

df = pd.DataFrame()

for _ in range(75):
    msg = socket.recv_string()
    t = datetime.datetime.now()
    sym, price = msg.split()
    df = df.append(pd.DataFrame({sym: float(price)}, index=[t]))

計算收益率和動量

計算收益率和動量，並將其新增到DataFrame中。

df['RET'] = np.log(df[sym] / df[sym].shift(1))
df['MOM'] = df['RET'].rolling(10).mean()

更新FigureWidget的資料

更新FigureWidget的資料，以顯示實時資料。

fig.data[0].x = df.index

內容解密：

以上程式碼示範瞭如何使用Python和Plotly進行實時資料處理和視覺化。首先，建立一個FigureWidget物件，並新增初始資料。然後，使用socket接收實時資料，並將其處理為DataFrame格式。接下來，計算收益率和動量，並將其新增到DataFrame中。最後，更新FigureWidget的資料，以顯示實時資料。

圖表翻譯：

以下是使用Mermaid語法建立的圖表，示範了實時資料處理和視覺化的流程。

  flowchart TD
    A[接收實時資料] --> B[處理資料]
    B --> C[計算收益率和動量]
    C --> D[更新FigureWidget的資料]
    D --> E[顯示實時資料]

此圖表顯示了實時資料處理和視覺化的流程，從接收實時資料到更新FigureWidget的資料，以顯示實時資料。

使用 Plotly 建立動態條形圖

Plotly 是一個強大的資料視覺化工具，允許使用者建立互動式和動態圖表。以下是使用 Plotly 建立動態條形圖的步驟：

安裝 Plotly

首先，需要安裝 Plotly。可以使用 pip 安裝：

pip install plotly

匯入 Plotly

匯入 Plotly：

import plotly.graph_objects as go

建立 FigureWidget

建立一個 FigureWidget 物件：

fig = go.FigureWidget()

新增條形圖

新增一個條形圖到 FigureWidget 中：

fig.add_bar()

設定資料

設定條形圖的資料。假設我們有以下資料：

data = [
    [60.361, 53.504, 67.782, 64.165, 35.046, 94.227, 20.221, 54.716],
    [79.508, 48.210, 84.163, 73.430, 53.288, 38.673, 4.962, 78.920],
    [53.316, 80.139, 73.733, 55.549, 21.015, 20.556, 49.090, 29.630],
    [86.664, 93.919, 33.762, 82.095, 3.108, 92.122, 84.194, 36.666],
    [37.192, 85.305, 48.397, 36.903, 81.835, 98.691, 61.818, 87.121]
]

更新條形圖

更新條形圖的資料：

for i, row in enumerate(data):
    fig.data[0].x = list(range(len(row)))
    fig.data[0].y = row
    fig.layout.title = f"條形圖 {i+1}"
    fig.show()

這會建立一個動態條形圖，資料會隨著時間更新。

使用 zmq 接收資料

如果需要從 zmq 接收資料，可以使用以下程式碼：

import zmq

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.setsockopt_string(zmq.SUBSCRIBE, '')

for _ in range(5):
    msg = socket.recv_string()
    print(msg)
    # 更新條形圖的資料
    data = [float(x) for x in msg.split()]
    fig.data[0].x = list(range(len(data)))
    fig.data[0].y = data
    fig.show()

這會從 zmq 接收資料，並更新條形圖的資料。

內容解密：

  • Plotly 是一個強大的資料視覺化工具，允許使用者建立互動式和動態圖表。
  • FigureWidget 是 Plotly 中的一個物件，允許使用者建立動態圖表。
  • 條形圖是一種常用的資料視覺化工具，允許使用者展示資料的分佈情況。
  • zmq 是一個強大的通訊工具，允許使用者在不同程式之間進行通訊。

圖表翻譯：

以下是使用 Mermaid 語法建立的圖表：

  graph LR
    A[Plotly] --> B[FigureWidget]
    B --> C[條形圖]
    C --> D[資料]
    D --> E[更新]
    E --> F[展示]

這個圖表展示了 Plotly、FigureWidget、條形圖、資料、更新和展示之間的關係。

處理實時資料和通訊端的重要性

在金融市場中，實時資料的處理和分析至關重要。許多市場，例如加密貨幣市場，24小時不間斷地運作，產生大量需要即時處理的資料。因此，瞭解如何使用通訊端（sockets）來處理實時資料是非常重要的。

通訊端的應用

通訊端是一種技術工具，允許不同應用程式之間進行通訊。它們在處理實時資料方面尤其有用，例如金融市場中的tick資料。ZeroMQ是一個強大且易於使用的通訊端型檔，常用於建立簡單的tick資料伺服器。

實時資料的視覺化

Plotly是一個強大的資料視覺化函式庫，允許在Jupyter Notebook中進行實時資料視覺化。它支援多個資料流，可以在單個圖表或多個子圖表中顯示。這使得即時跟蹤和分析資料變得更加容易。

樣本Tick資料伺服器

import zmq
import numpy as np

# 建立ZeroMQ上下文
context = zmq.Context()

# 建立通訊端
socket = context.socket(zmq.PUB)

# 繫結通訊端到埠
socket.bind("tcp://*:5555")

# 定義tick資料生成函式
def generate_tick_data():
    # 使用蒙特卡羅模擬生成隨機資料
    data = np.random.normal(0, 1, 100)
    return data

# 啟動伺服器
while True:
    # 生成tick資料
    data = generate_tick_data()
    
    # 將資料傳送到客戶端
    socket.send_string(" ".join(map(str, data)))

客戶端指令碼

import zmq
import plotly.graph_objs as go

# 建立ZeroMQ上下文
context = zmq.Context()

# 建立通訊端
socket = context.socket(zmq.SUB)

# 連線到伺服器
socket.connect("tcp://localhost:5555")

# 訂閱所有主題
socket.setsockopt_string(zmq.SUBSCRIBE, "")

# 建立圖表
fig = go.Figure(data=[go.Scatter(x=[], y=[])])

# 啟動客戶端
while True:
    # 接收資料
    data = socket.recv_string()
    
    # 將資料轉換為列表
    data = list(map(float, data.split()))
    
    # 更新圖表
    fig.data[0].y = data
    
    # 顯示圖表
    fig.show()

這些指令碼示範瞭如何使用ZeroMQ和Plotly建立一個簡單的tick資料伺服器和客戶端，以生成實時交易訊號和視覺化資料。

金融交易所資料伺服器

金融交易所資料伺服器是一個重要的基礎設施，負責提供即時的金融資料給交易員和投資者。以下是使用 Python 和 ZeroMQ 建立一個簡單的金融交易所資料伺服器的範例。

安裝必要的套件

pip install zmq

建立金融交易所資料伺服器

import zmq
import time
import random

# 建立 ZeroMQ 的 context 和 socket
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind('tcp://0.0.0.0:5555')

class InstrumentPrice(object):
    def __init__(self):
        self.symbol = 'SYMBOL'
        self.t = time.time()
        self.value = 100.
        self.sigma = 0.4
        self.r = 0.01

    def simulate_value(self):
        ''' 生成新的隨機股票價格。 '''
        t = time.time()
        dt = (t - self.t) / (252 * 8 * 60 * 60)
        dt *= 500
        # 使用隨機過程生成新的股票價格
        self.value = self.value * math.exp((self.r - 0.5 * self.sigma ** 2) * dt + self.sigma * math.sqrt(dt) * random.gauss(0, 1))
        self.t = t
        return self.value

# 建立 InstrumentPrice 物件
instrument_price = InstrumentPrice()

while True:
    # 生成新的隨機股票價格
    value = instrument_price.simulate_value()
    # 將股票價格傳送給訂閱者
    socket.send_string(f'{instrument_price.symbol} {value:.2f}')
    time.sleep(1)

訂閱金融交易所資料

import zmq

# 建立 ZeroMQ 的 context 和 socket
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect('tcp://localhost:5555')
socket.setsockopt_string(zmq.SUBSCRIBE, '')

while True:
    # 接收股票價格
    message = socket.recv_string()
    print(message)

圖表翻譯：

  flowchart TD
    A[金融交易所資料伺服器] --> B[生成隨機股票價格]
    B --> C[傳送股票價格給訂閱者]
    C --> D[訂閱者接收股票價格]
    D --> E[顯示股票價格]

圖表翻譯：

此圖表顯示金融交易所資料伺服器的工作流程。首先，伺服器生成新的隨機股票價格。然後，伺服器將股票價格傳送給訂閱者。訂閱者接收股票價格並顯示給使用者。

即時資料與 Socket 的應用

在金融市場中，實時資料的處理和傳輸對於即時交易和決策至關重要。Socket 是一種允許不同程式之間進行通訊的技術，特別是在網路上。以下，我們將探討如何使用 Python 和 ZeroMQ 這個函式庫來建立一個 tick 資料客戶端，連線到一個 tick 資料伺服器，並接收實時的金融市場資料。

Tick 資料模擬

首先，我們需要模擬 tick 資料的產生。這可以透過一個簡單的類別實作，該類別模擬金融工具的價格變化。

import math
import random
import time

class InstrumentPrice:
    def __init__(self, symbol, r, sigma):
        self.symbol = symbol
        self.r = r
        self.sigma = sigma
        self.value = 100.0  # 初始價格

    def simulate_value(self, dt):
        self.value *= math.exp((self.r - 0.5 * self.sigma ** 2) * dt + self.sigma * math.sqrt(dt) * random.gauss(0, 1))
        return self.value

# 建立一個模擬物件
ip = InstrumentPrice('AAPL', 0.05, 0.2)

while True:
    # 模擬價格變化
    msg = '{} {:.2f}'.format(ip.symbol, ip.simulate_value(1/252))  # 252 是一年中的交易日數
    print(msg)
    # 在實際應用中，這裡會將 msg 傳送給 socket
    time.sleep(random.random() * 2)  # 隨機延遲模擬實時資料的不均勻到來

使用 ZeroMQ 建立 Tick 資料客戶端

ZeroMQ 是一種高效能的通訊函式庫，支援多種通訊模式，包括發布/訂閱模式。以下是如何使用 ZeroMQ 建立一個 tick 資料客戶端的示例：

import zmq

# 建立一個 ZeroMQ 的 context
context = zmq.Context()

# 建立一個 socket 物件，使用 SUB 模式
socket = context.socket(zmq.SUB)

# 連線到 tick 資料伺服器
socket.connect("tcp://localhost:5555")  # 伺服器地址和埠

# 訂閱所有主題（在這個例子中，我們不使用特定的主題）
socket.setsockopt_string(zmq.SUBSCRIBE, "")

while True:
    # 接收來自伺服器的訊息
    msg = socket.recv_string()
    print("收到訊息:", msg)

# 當不再需要時，關閉 socket 和 context
socket.close()
context.term()

圖表翻譯：

  flowchart TD
    A[客戶端] -->|連線|> B[伺服器]
    B -->|傳送資料|> A
    A -->|接收資料|> C[處理資料]
    C -->|顯示結果|> D[使用者]

這個流程圖描述了客戶端如何連線到伺服器，接收實時資料，然後處理和顯示這些資料給使用者。

內容解密：

在這個例子中，我們使用 ZeroMQ 的發布/訂閱模式建立了一個 tick 資料客戶端。客戶端連線到伺服器，訂閱所有主題，然後不斷接收來自伺服器的實時資料。這些資料可以是金融工具的價格變化，或者其他型別的實時資料。客戶端可以根據需要處理這些資料，例如顯示給使用者，或者進行進一步的分析。

這個例子展示瞭如何使用 Python 和 ZeroMQ 來建立一個簡單的實時資料處理系統。這種系統可以應用在金融市場、物聯網、或其他需要實時資料處理的領域。

Momentum Online Algorithm 實作

連線 Tick 資料伺服器

為了實作 Momentum Online Algorithm，我們需要連線到 Tick 資料伺服器，以接收即時的市場資料。以下是使用 ZeroMQ 連線到 Tick 資料伺服器的程式碼：

import zmq
import datetime
import numpy as np
import pandas as pd

# 建立 ZeroMQ 的 context 和 socket
context = zmq.Context()
socket = context.socket(zmq.SUB)

# 訂閱 SYMBOL 主題
socket.setsockopt_string(zmq.SUBSCRIBE, 'SYMBOL')

# 接收資料
while True:
    data = socket.recv_string()
    print(data)

Momentum Online Algorithm

Momentum Online Algorithm 是一個根據時間序列動量的交易策略。以下是實作 Momentum Online Algorithm 的程式碼：

# 建立一個空的 DataFrame
df = pd.DataFrame()

# 設定動量的視窗大小
mom = 3

# 設定最小的資料長度
min_length = mom + 1

while True:
    # 接收資料
    data = socket.recv_string()
    
    # 將資料新增到 DataFrame 中
    df = pd.concat([df, pd.DataFrame([data])], ignore_index=True)
    
    # 檢查資料長度是否足夠
    if len(df) >= min_length:
        # 計算動量
        momentum = df.iloc[-mom:].mean()
        
        # 印出動量
        print(momentum)

圖表翻譯：

此圖示 Momentum Online Algorithm 的執行流程

  flowchart TD
    A[連線 Tick 資料伺服器] --> B[接收資料]
    B --> C[計算動量]
    C --> D[印出動量]
    D --> B

圖表翻譯：

此圖表示 Momentum Online Algorithm 的執行流程。首先，程式連線到 Tick 資料伺服器，然後接收資料。當資料長度足夠時，程式計算動量並印出結果。接著，程式繼續接收資料並計算動量，直到程式停止。

從技術架構視角來看，本文逐步展示瞭如何利用 Python 結合 Plotly 與 ZeroMQ 建構一個實時資料處理和視覺化系統，特別聚焦於金融領域的 tick data 應用。文章涵蓋了從安裝必要套件、建立 FigureWidget、處理 socket 資料流、計算金融指標（如收益率和動能），到最終實作一個簡化的線上動能交易演算法的完整流程。透過詳細的程式碼範例和流程圖，清晰地闡述了各個模組的功能和互動方式，降低了讀者理解和實作的門檻。然而，目前提供的程式碼範例仍停留在基礎功能演示階段，缺乏對錯誤處理、效能最佳化以及實際交易環境中可能遇到的複雜情況的考慮。例如，網路延遲、資料丟失、以及高頻交易下的系統穩定性等問題都需進一步探討和解決。對於追求生產級應用的開發者而言，需要在本文基礎上加入更健壯的錯誤處理機制、更精細的資料清洗和驗證流程，並針對特定應用場景進行效能調優。展望未來，整合機器學習模型進行更複雜的交易策略開發，以及探索雲原生架構下的佈署方案，將是此技術堆疊持續發展的重要方向。玄貓認為，此架構具備相當的實用價值，尤其適合金融科技領域的原型設計和教學示範，但要應用於實際交易系統，仍需更多工程化的努力。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。