Python 與 Plotly 實作即時股票資料視覺化

隨著金融市場的快速變化，即時資料分析和視覺化變得越來越重要。本文將示範如何使用 Python 結合 ZeroMQ 與 Plotly 建立一個可以接收、處理和視覺化即時股票資料的應用程式。此應用程式將會展示股票價格、動量和簡單移動平均線（SMA）等指標，並透過 Plotly 繪製成互動式圖表。這些圖表可以幫助交易者即時監控市場變化，並根據最新的資料做出更明智的決策。此外，我們還將探討線上演算法在即時交易策略中的應用，並示範如何使用 Python 實作一個根據動量的交易策略。

股票價格模擬器實作

import math
import random
import time

class InstrumentPrice:
    def __init__(self, symbol, value, r, sigma):
        '''
        初始化股票價格模擬器。

        :param symbol: 股票程式碼
        :param value: 初始股票價格
        :param r: 無風險利率
        :param sigma: 波動率
        '''
        self.symbol = symbol
        self.value = value
        self.r = r
        self.sigma = sigma
        self.t = time.time()

    def simulate_value(self):
        '''
        生成新的隨機股票價格。

        :return: 新的股票價格
        '''
        t = time.time()
        dt = (t - self.t) / (252 * 8 * 60 * 60)  # 轉換為交易年分數
        dt *= 500  # 放大時間間隔以增加價格波動
        self.t = t
        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('SYMBOL', 100.0, 0.05, 0.2)

while True:
    # 生成新的股票價格並建立訊息
    msg = '{} {:.2f}'.format(ip.symbol, ip.simulate_value())
    print(msg)
    # 模擬隨機時間間隔
    time.sleep(random.random() * 2)

解釋

InstrumentPrice 類別用於模擬股票價格的變動。它的初始引數包括股票程式碼、初始價格、無風險利率和波動率。
simulate_value 方法根據幾何布朗運動模型生成新的隨機股票價格。時間間隔 dt 被轉換為交易年分數，並放大以增加價格波動。
在主迴圈中，建立一個 InstrumentPrice 例項，並不斷生成新的股票價格，建立訊息，並模擬隨機時間間隔。

應用

此股票價格模擬器可以用於金融市場的模擬、風險分析和投資策略的評估。透過調整模型引數和時間間隔，可以模擬不同的市場情況和投資策略。

連線簡單的Tick資料客戶端

在之前的Tick資料伺服器中，我們已經實作了隨機到達時間的模擬。現在，我們需要一個客戶端來連線這個伺服器，接收並處理Tick資料。

客戶端程式碼

以下是客戶端的程式碼：

import zmq

# 建立Context物件
context = zmq.Context()

# 建立訂閱socket
socket = context.socket(zmq.SUB)

# 連線到釋出socket
socket.connect("tcp://localhost:5555")

# 訂閱SYMBOL頻道
socket.setsockopt_string(zmq.SUBSCRIBE, "SYMBOL")

# 進入迴圈接收訊息
while True:
    # 接收字串訊息
    message = socket.recv_string()
    print(message)

這個客戶端程式碼非常簡潔。它建立了一個Context物件，然後建立了一個訂閱socket，連線到釋出socket，並訂閱SYMBOL頻道。最後，它進入一個迴圈，接收並列印預出收到的訊息。

Tick資料客戶端工作原理

當客戶端連線到釋出socket並訂閱SYMBOL頻道後，它就可以接收到釋出socket傳送的訊息。釋出socket會隨機傳送SYMBOL的價格資料，客戶端就可以接收並處理這些資料。

實際應用

在實際應用中，客戶端可以根據接收到的Tick資料進行各種分析和處理，例如計算移動平均線、判斷趨勢等。客戶端也可以將接收到的資料儲存到資料函式庫中，或者將其用於其他應用程式中。

內容解密：

在上面的程式碼中，zmq.SUB是訂閱socket的型別，socket.setsockopt_string(zmq.SUBSCRIBE, "SYMBOL")是用於訂閱SYMBOL頻道的。socket.recv_string()是用於接收字串訊息的。這些函式都是ZeroMQ函式庫的一部分，提供了高效的訊息傳遞機制。

圖表翻譯：

以下是客戶端程式碼的流程圖：

  flowchart TD
    A[建立Context物件] --> B[建立訂閱socket]
    B --> C[連線到釋出socket]
    C --> D[訂閱SYMBOL頻道]
    D --> E[進入迴圈接收訊息]
    E --> F[接收字串訊息]
    F --> G[列印訊息]

這個流程圖顯示了客戶端程式碼的執行流程。它從建立Context物件開始，然後建立訂閱socket，連線到釋出socket，訂閱SYMBOL頻道，進入迴圈接收訊息，接收字串訊息，最後列印預出訊息。

使用 ZeroMQ 進行實時資料傳輸

在實時資料傳輸中，ZeroMQ 是一個非常重要的工具。以下是使用 Python wrapper for ZeroMQ 進行實時資料傳輸的範例。

設定 Socket

首先，需要設定 socket 型別為 SUB，這代表著這個 socket 會從指定的 IP 地址和埠接收資料。

import zmq

# 設定 socket 型別為 SUB
socket = zmq.Context().socket(zmq.SUB)

# 設定 socket 的 IP 地址和埠
socket.connect("tcp://localhost:5555")

# 設定 socket 的 channel
socket.setsockopt_string(zmq.SUBSCRIBE, 'SYMBOL')

接收資料

然後，使用 while loop 不斷地接收資料。

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

資料傳輸過程

在這個範例中，socket 會不斷地接收資料，並且印出接收到的資料。資料的格式為 SYMBOL 加上一個價格值。

範例輸出

以下是這個範例的輸出結果：

SYMBOL 100.00
SYMBOL 99.65
SYMBOL 99.28
SYMBOL 99.09
SYMBOL 98.76
SYMBOL 98.83
SYMBOL 98.82
SYMBOL 98.92
SYMBOL 98.57
SYMBOL 98.81

即時訊號生成

在金融交易中，能夠即時生成訊號的能力至關重要。這需要一個根據增量接收資料的線上演算法。這類演算法只知道當前和之前的狀態，但不知道未來的狀態。這是一個現實的設定，尤其是在金融交易中，任何形式的預知都是不允許的。

線上演算法

線上演算法是一種根據增量接收資料的演算法。這種演算法只知道當前和之前的狀態，但不知道未來的狀態。這與離線演算法不同，離線演算法知道完整的資料集。許多電腦科學中的演算法都屬於離線演算法，例如排序演算法。

即時訊號生成實作

要在實時基礎上生成訊號，需要收集和處理資料。例如，考慮一個根據過去三個五秒間隔的時間序列動量的交易策略。需要收集和重取樣tick資料，然後計算動量。隨著時間的推移，會不斷更新資料。

以下是一個Python指令碼，實作了動量策略作為一個線上演算法：

import pandas as pd
import numpy as np
from datetime import datetime

# 初始化一個空的DataFrame來收集tick資料
df = pd.DataFrame()

# 定義動量計算的時間間隔數
mom = 3

# 指定訊號生成的最小長度
min_length = mom + 1

while True:
    # 從socket接收字串資料
    data = socket.recv_string()
    
    # 生成時間戳
    t = datetime.now()
    
    # 將字串資料分割為符號和數值
    sym, value = data.split()
    
    # 將新資料新增到DataFrame中
    df = df.append(pd.DataFrame({sym: float(value)}, index=[t]))
    
    # 重取樣資料
    dr = df.resample('5s', label='right').last()
    
    # 計算對數收益率
    dr['returns'] = np.log(dr / dr.shift(1))
    
    # 如果資料長度超過最小長度，則計算動量
    if len(dr) > min_length:
        min_length += 1
        dr['momentum'] = np.sign(dr['returns'].rolling(mom).mean())

這個指令碼首先初始化一個空的DataFrame來收集tick資料。然後，它定義了動量計算的時間間隔數和訊號生成的最小長度。接著，它進入一個無窮迴圈中，不斷從socket接收字串資料，將其新增到DataFrame中，重取樣資料，計算對數收益率和動量。

實時交易演算法訊號生成

在實時交易中，訊號生成是非常重要的一個環節。以下是使用 Python 實作的一個簡單的實時交易演算法訊號生成示例。

資料預處理

首先，我們需要對原始資料進行預處理。這包括對資料進行重取樣，以確保資料的時間間隔是一致的。在這個例子中，我們使用 5 秒的時間間隔。

import pandas as pd
import numpy as np
import datetime

# 對資料進行重取樣
dr = pd.DataFrame({
    'SYMBOL': ['AAPL'] * 100,
    'momentum': np.random.rand(100)
}, index=pd.date_range('2020-05-23 11:33:00', periods=100, freq='s'))

dr_resampled = dr.resample('5S').last()

計算動量

接下來，我們需要計算動量。動量是指資產價格的變化率。在這個例子中，我們使用 3 個重取樣時間間隔的對數回報來計算動量。

# 計算對數回報
log_returns = np.log(dr_resampled['SYMBOL'] / dr_resampled['SYMBOL'].shift(1))

# 計算動量
momentum = log_returns.rolling(window=3).mean()
dr_resampled['momentum'] = momentum

訊號生成

現在，我們可以根據動量的符號來生成訊號。如果動量為 1.0，則表示長市場位置；如果動量為 -1.0，則表示短市場位置。

# 生成訊號
if dr_resampled['momentum'].iloc[-2] == 1.0:
    print('\nLong market position.')
elif dr_resampled['momentum'].iloc[-2] == -1.0:
    print('\nShort market position.')

輸出結果

最後，我們可以輸出結果。

print('\n' + '=' * 51)
print('NEW SIGNAL | {}'.format(datetime.datetime.now()))
print('=' * 51)
print(dr_resampled.tail())

執行結果

以下是執行結果的示例。

===================================================
NEW SIGNAL | 2020-05-23 11:33:31.233606
===================================================
                     SYMBOL  momentum
2020-05-23 11:33:15  98.65      NaN
2020-05-23 11:33:20  98.53      NaN
2020-05-23 11:33:25  98.83      NaN
2020-05-23 11:33:30  99.33      1.0
[4 rows x 2 columns]
Long market position.

圖表翻譯：

  flowchart TD
    A[資料預處理] --> B[計算動量]
    B --> C[訊號生成]
    C --> D[輸出結果]
    D --> E[執行結果]

圖表解釋：

此圖表描述了實時交易演算法訊號生成的流程。首先，對資料進行預處理；然後，計算動量；接下來，根據動量的符號生成訊號；最後，輸出結果。

實時資料視覺化與策略實施

在實時資料分析中，能夠快速且有效地視覺化資料對於做出正確的決策至關重要。Plotly是一個強大的工具，能夠幫助我們生成互動式的圖表，以便更好地理解資料的趨勢和模式。

安裝Plotly和Jupyter Lab擴充功能

要使用Plotly，首先需要安裝Plotly套件和相關的Jupyter Lab擴充功能。可以透過以下命令完成安裝：

conda install plotly ipywidgets
jupyter labextension install jupyterlab-plotly
jupyter labextension install @jupyter-widgets/jupyterlab-manager
jupyter labextension install plotlywidget

實時資料視覺化

Plotly提供了多種方式來視覺化實時資料，包括線圖、柱狀圖和散點圖等。以下是一個簡單的例子，展示如何使用Plotly來視覺化實時資料：

import plotly.graph_objs as go
import pandas as pd

# 載入資料
df = pd.read_csv('data.csv')

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

# 更新圖表
fig.update_layout(title='實時資料視覺化', xaxis_title='時間', yaxis_title='值')

# 顯示圖表
fig.show()

策略實施

除了視覺化資料外，Plotly還可以用來實施不同的策略，例如移動平均線（SMA）策略和均值迴歸策略。以下是一個簡單的例子，展示如何使用Plotly來實施SMA策略：

import plotly.graph_objs as go
import pandas as pd

# 載入資料
df = pd.read_csv('data.csv')

# 計算移動平均線
df['sma'] = df['value'].rolling(window=10).mean()

# 建立圖表
fig = go.Figure(data=[go.Scatter(x=df['time'], y=df['value']), go.Scatter(x=df['time'], y=df['sma'])])

# 更新圖表
fig.update_layout(title='SMA策略', xaxis_title='時間', yaxis_title='值')

# 顯示圖表
fig.show()

即時資料視覺化基礎

首先，需要安裝必要的套件和擴充功能，才能夠高效地生成即時資料視覺化。第一步是建立一個 Plotly 圖表小工具：

import zmq
from datetime import datetime
import plotly.graph_objects as go

# 定義符號
symbol = 'SYMBOL'

# 建立 Plotly 圖表小工具
fig = go.FigureWidget()

# 新增散點圖
fig.add_scatter()

# 顯示圖表
fig

這些程式碼匯入了 Plotly 的圖形物件，並在 Jupyter Notebook 中例項化了 Plotly 圖表小工具。

接下來，需要設定通訊端通訊以連線到樣本 tick 資料伺服器，該伺服器需要在相同的機器上以獨立的 Python 程式執行。由玄貓提供的資料被豐富化，並用於更新圖表小工具的資料物件（見圖 7-1）：

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

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

# 初始化時間和價格列表
times = list()
prices = list()

# 接收和處理 50 次資料
for _ in range(50):
    # 接收字串訊息
    msg = socket.recv_string()
    
    # 將訊息轉換為資料並更新圖表
    # ... (處理和更新程式碼)

這些程式碼建立了通訊端通訊，設定了訂閱主題，並初始化了時間和價格列表。然後，它們接收和處理 50 次資料，更新圖表小工具的資料物件。

圖表翻譯：

  flowchart TD
    A[建立 Plotly 圖表小工具] --> B[新增散點圖]
    B --> C[設定通訊端通訊]
    C --> D[接收和處理資料]
    D --> E[更新圖表小工具]

內容解密：

這個程式碼示例展示瞭如何使用 Plotly 和 ZeroMQ 建立一個即時資料視覺化。首先，建立一個 Plotly 圖表小工具，並新增散點圖。然後，設定通訊端通訊以連線到樣本 tick 資料伺服器。最後，接收和處理資料，更新圖表小工具的資料物件。這個過程可以實作即時資料視覺化，讓使用者可以即時看到資料的變化。

即時資料流與Plotly視覺化

引言

在金融市場中，能夠即時取得和視覺化資料對於投資決策至關重要。Plotly是一個強大的視覺化工具，能夠幫助我們實作這一目標。本文將介紹如何使用Plotly建立一個即時資料流視覺化應用。

單一資料流

首先，我們需要建立一個列表來儲存時間戳和價格資料。然後，我們可以使用Plotly的FigureWidget來建立一個圖表，並更新圖表的資料。

import datetime
import plotly.graph_objects as go

# 建立列表來儲存時間戳和價格資料
times = []
prices = []

# 建立圖表
fig = go.FigureWidget()

# 更新圖表的資料
fig.add_scatter(x=times, y=prices)

# 更新圖表
def update_fig(t, price):
    times.append(t)
    prices.append(float(price))
    fig.data[0].x = times
    fig.data[0].y = prices

# 測試更新圖表
t = datetime.datetime.now()
price = 100.0
update_fig(t, price)

多個資料流

如果我們想要視覺化多個資料流，例如價格和兩個簡單移動平均線（SMA），我們可以建立多個散點圖物件，並更新圖表的資料。

import pandas as pd
import plotly.graph_objects as go

# 建立圖表
fig = go.FigureWidget()

# 建立散點圖物件
fig.add_scatter(name='SYMBOL')
fig.add_scatter(name='SMA1', line=dict(width=1, dash='dot'), mode='lines+markers')
fig.add_scatter(name='SMA2', line=dict(width=1, dash='dash'), mode='lines+markers')

# 更新圖表的資料
def update_fig(t, price):
    # 更新價格資料
    times.append(t)
    prices.append(float(price))
    fig.data[0].x = times
    fig.data[0].y = prices
    
    # 更新SMA資料
    df = pd.DataFrame({'price': prices})
    df['SMA1'] = df['price'].rolling(window=10).mean()
    df['SMA2'] = df['price'].rolling(window=20).mean()
    fig.data[1].x = times
    fig.data[1].y = df['SMA1']
    fig.data[2].x = times
    fig.data[2].y = df['SMA2']

# 測試更新圖表
t = datetime.datetime.now()
price = 100.0
update_fig(t, price)

即時串流資料視覺化

即時串流資料視覺化是一種動態展示資料的方法，能夠實時更新和反映最新的資料變化。這種技術在金融、物流、交通等領域中被廣泛應用。

即時串流資料收集

即時串流資料收集是指從各種資料源中實時收集和處理資料的過程。這種收集可以透過網路、感測器、物聯網裝置等方式進行。以下是使用Python收集即時串流資料的範例：

import pandas as pd
import socket
import datetime

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

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

即時串流資料視覺化

即時串流資料視覺化是指使用圖表、圖形等方式展示即時串流資料的過程。這種視覺化可以幫助使用者快速理解和分析資料的變化。以下是使用Plotly進行即時串流資料視覺化的範例：

import plotly.graph_objects as go
from plotly.subplots import make_subplots

# 建立一個子圖
f = make_subplots(rows=3, cols=1, shared_xaxes=True)

# 新增第一個子圖
f.append_trace(go.Scatter(name='SYMBOL'), row=1, col=1)

# 新增第二個子圖
f.append_trace(go.Scatter(name='RETURN', line=dict(width=1, dash='dot'),
                           mode='lines+markers', marker={'symbol': 'triangle-up'}), row=2, col=1)

# 新增第三個子圖
f.append_trace(go.Scatter(name='MOMENTUM'), row=3, col=1)

即時串流資料分析

即時串流資料分析是指對即時串流資料進行分析和處理的過程。這種分析可以包括資料清理、資料轉換、資料視覺化等步驟。以下是使用Python進行即時串流資料分析的範例：

import pandas as pd

# 計算移動平均值
df['SMA1'] = df[sym].rolling(5).mean()
df['SMA2'] = df[sym].rolling(10).mean()

# 更新圖表資料
fig.data[0].x = df.index
fig.data[1].x = df.index
fig.data[2].x = df.index
fig.data[0].y = df[sym]
fig.data[1].y = df['SMA1']
fig.data[2].y = df['SMA2']

實時資料處理與視覺化

在實時資料處理中，能夠即時接收和處理資料是非常重要的。以下是一個使用Python和Plotly進行實時資料處理和視覺化的例子。

從技術架構視角來看，本文展示瞭如何利用 ZeroMQ 建立即時資料管道，並結合 Plotly 進行動態視覺化，實作了從資料採集、處理到展示的完整流程。透過訂閱模式，客戶端能高效地接收伺服器推播的股票價格資料，並運用 Pandas 進行資料重取樣和動量計算，展現了線上演算法的實踐應用。然而，此範例程式碼的容錯機制和例外處理仍有待加強，例如在網路連線中斷或資料格式錯誤時，程式碼的穩定性可能受到影響。此外，僅使用簡單的動量策略進行訊號生成，在實際交易環境中可能不夠穩健。展望未來，整合機器學習模型進行更複雜的訊號預測，並結合更進階的 Plotly 互動式圖表功能，將能打造更強大的即時交易分析平臺。對於追求高效能和即時資料分析的金融科技應用，此架構值得參考，但需進一步完善錯誤處理和策略複雜度以應對真實市場的挑戰。玄貓認為，此技術組合在金融資料視覺化領域具有相當的應用潛力，值得深入研究並持續最佳化。

玄貓

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