背景

株価データのリアルタイム可視化を目的に、TornadoとWebSocketを使用する方法。
Polygon.ioのAPIを使用して株価データを取得し、Chart.jsを使ったリアルタイムチャートに反映します。

サーバー側の実装

server.py

import json, os, polygon, asyncio
import tornado.websocket
import tornado.web
import tornado.ioloop
from polygon.enums import StreamCluster 

class SendWebSocket(tornado.websocket.WebSocketHandler):
    async def open(self):
        print ('Session Opened. IP:' + self.request.remote_ip)
        await self.send_websocket()

    def on_close(self):
        print("Session closed")

    def check_origin(self, origin):
        return True

    async def send_websocket(self):
        api_key = os.environ.get('POLYGON_API')
        stream_client = polygon.AsyncStreamClient(api_key, StreamCluster.STOCKS)
        await stream_client.subscribe_stock_minute_aggregates(['NVDA'], self.stock_trades_handler) # 複数の場合は['AMD', 'NVDA']のようにする

        while 1:
            await stream_client.handle_messages()  # the lib provides auto reconnect functionality. See docs for info

    async def stock_trades_handler(self, msg):
        message = json.dumps({
            'timestamp':msg['t'],
            'price':msg['p'],
            })
        print(message)
        await self.write_message(message)

def make_app():
    return tornado.web.Application([
        (r"/ws/display", SendWebSocket)
        ])

async def main():
    app = make_app()
    app.listen(8080)
    shutdown_event = asyncio.Event()
    await shutdown_event.wait()

if __name__ == "__main__":
    asyncio.run(main())

クライアント側の実装

clinet.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Display Demo</title>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/moment.js/2.10.6/moment.js"></script>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/moment.js/2.10.6/locale/ja.js"></script>
    <script src="https://cdn.jsdelivr.net/npm/chart.js@3.3.2"></script>
    <script src="https://cdn.jsdelivr.net/npm/luxon@1.27.0"></script>
    <script src="https://cdn.jsdelivr.net/npm/chartjs-adapter-luxon@1.0.0"></script>
    <script src="https://cdn.jsdelivr.net/npm/chartjs-plugin-streaming@2.0.0"></script>
</head>

<body>
    <h1> Display Demo </h1>
    <canvas id="line-chart"></canvas>

    <script>
    const ctx = document.getElementById('line-chart').getContext('2d');
    const chart = new Chart(ctx, {
        type: 'line',
        data: {
            labels: [], // ラベルを空の配列で初期化
            datasets: [{
                label: 'Value',
                data: [],   // データを空の配列で初期化
                borderColor: 'rgba(75, 192, 192, 1)',   // グラフの線の色
                backgroundColor: 'rgba(75, 192, 192, 0.2)',
                fill: false  // グラフの塗りつぶしを無効にする
            }]
        },
        options: {
            scales: {
                x: {
                    type: 'time',
                    time: {
                        unit: 'second', // x軸の単位を秒にする
                        displayFormats: {
                            second: 'HH:mm:ss' // x軸の表示フォーマットを指定
                        }
                    },
                    //min: '2023-04-19T15:50:00', // x軸の最小値を指定
                    max: '2023-04-21T16:00:00', // x軸の最大値を指定
                },
                y: {
                    // y軸の設定を追加する場合はここに記述
                }
            }
        } 
    });

    // WebSocketの接続
    const connection = new WebSocket('ws://127.0.0.1:8080/ws/display');

    // WebSocketのメッセージ受信時の処理
    connection.onmessage = function (e) {
        chart.data.datasets[0].data.push({
            x: JSON.parse(e.data)['timestamp'],
            y: JSON.parse(e.data)['price']
        });
        chart.update(); // チャートの更新
    };
    </script>
</body>
</html>

実行結果

server.pyを実行した後、client.htmlを開くとchartが表示されました。

まとめ

他のデータベンダーのAPIもこのような形でリアルタイムチャートを表示できると思います。

参考文献・引用

async_streming_example

import asyncio
import polygon
from polygon.enums import StreamCluster

async def stock_trades_handler(msg):   # it is possible to create one common message handler for different services.
    print(f'msg received: {msg}')
    
async def main():
    api_key = 'YOUR_KEY'
    
    stream_client = polygon.AsyncStreamClient(api_key, StreamCluster.STOCKS)
    
    await stream_client.subscribe_stock_trades(['AMD', 'NVDA'], stock_trades_handler)
    
    while 1:
        await stream_client.handle_messages()  # the lib provides auto reconnect functionality. See docs for info
if __name__ == '__main__':
    asyncio.run(main())

おわりに

荒削りなので、改善点などございましたらコメントしていただけると有り難いです。