開啟 Redis

 docker run --rm -p 6379:6379 redis:7

建立聊天室

#chat/fugle.py
import time

from fugle_marketdata import WebSocketClient
import asyncio
from asgiref.sync import async_to_sync, sync_to_async
import channels.layers
import django, sys
from os.path import join, dirname, abspath
from os import environ

# ---------------------- django setting --------------------------
PROJECT_DIR = dirname(dirname(abspath(__file__)))
sys.path.insert(0, PROJECT_DIR)

#  Set the correct path to you settings module
environ.setdefault("DJANGO_SETTINGS_MODULE", "django_channels.settings")

# All django stuff has to come after the setup:
django.setup()

# ---------------------- django setting --------------------------
channel_layer = channels.layers.get_channel_layer()

YOUR_API_KEY = 'xxx'

def handle_message(message):
    print(message)
    if message['event'] in ['snapshot', 'data']:
        data = message['data']
        symbol = data['symbol']

        async_to_sync(channel_layer.group_send)(f"chat_{symbol}",
                                                {'type': 'candles.message', "message": message})


async def main():
    client = WebSocketClient(api_key=YOUR_API_KEY)
    stock = client.stock
    stock.on('message', handle_message)
    await stock.connect()
    stock.subscribe({
        'channel': 'trades',
        'symbol': '2330'
    })


if __name__ == '__main__':

    asyncio.run(main())

透過 xxx 的方式傳送訊息至 Redis 上，然後讓 Redis 傳送訊息到 Consumer 上，由 Consumer 接收訊息轉發至 client 端上

# chat/consumers.py
import json
from datetime import datetime
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer

from fugle_marketdata import RestClient

api_key = 'xxx'
#

class ChatConsumer(WebsocketConsumer):
    def connect(self):
        self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
        self.room_group_name = f"chat_{self.room_name}"
        client = RestClient(api_key=api_key)
        self.stock = client.stock

        # Join room group
        async_to_sync(self.channel_layer.group_add)(
            self.room_group_name, self.channel_name
        )

        self.accept()

        message = f"""
歡迎來到 {self.room_name} 股票即時推播聊天室,
            """
        async_to_sync(self.channel_layer.group_send)(
            self.room_group_name, {"type": "chat.message", "message": message}
        )

    def disconnect(self, close_code):
        # Leave room group
        async_to_sync(self.channel_layer.group_discard)(
            self.room_group_name, self.channel_name
        )

    # Receive message from WebSocket
    def receive(self, text_data):
        text_data_json = json.loads(text_data)


    # Receive message from room group
    def chat_message(self, event):
        message = event["message"]

        # Send message to WebSocket
        self.send(text_data=json.dumps({"message": message}))

    def candles_message(self, event):
        message = event["message"]

        # Send message to WebSocket
        msg = f'現在時間為: {message["date"]}, ' + f','.join([f' {i}: {j}' for i,j in message.items() if i != 'date'])
        self.send(text_data=json.dumps({"message": msg}))

python manage.py runserver 8000

開啟另一個視窗

python3 manage.py chat/fugle.py

預期進入聊天室後可以看到成交明細資料

恭喜達到即時推播的功能

牛刀小試

• 訂閱五個商品的股票最新成交資訊

注意

• 若開啟時間非盤中時間，會收到最後一筆的資料，如想看到類似盤中情況，可使用歷史行情的 candles 來替代.

參考

fugle-metadata
fugle 官方文件

使用規範與聲明

透過本服務取得之行情資料僅供參考，成交值及成交量不含零股及鉅額交易，使用者依本資料交易發生交易損失需自行負責。
時報資訊與群馥科技對資料內容錯誤、更新延誤或傳輸中斷不負任何責任。您應對您所為之任何金融或投資決策自行負責。
使用者應遵守臺灣證券交易所股份有限公司交易資訊使用管理辦法、臺灣期貨交易所股份有限公司交易資訊使用管理辦法、財團法人中華民國證券櫃檯買賣中心有價證券交易資訊使用管理辦法、各資訊來源提供者所定之資訊使用相關規範及智慧財產權相關法令，如有盜接、轉接交易資訊，或以其他方式出售、出租、轉讓、再授權交易資訊，或將交易資訊另行取樣並編製指數、其他衍生性商品或將之傳送予第三人，應負違約及侵權之相關民、刑事責任。
使用者須遵守臺灣證券交易所「交易資訊使用管理辦法」等交易資訊管理相關規定，所有資訊以臺灣證券交易所公告資料為準。