【Python】pubnub と WebSocket で更にリアルタイムで情報を取得する

2018年3月22日2020年10月11日Python,開発

おはようございます。

えーと、SQLiteにデータ登録なんかしようと思ってたんですが、
どう実装すればいいか分からず別の方法で実装した、pubnub でティッカー情報が更新される度に画面の情報も更新する方法が分かりましたのでやってみました。

プログラムは前回のものを流用します。

【Python】bitflyer の API を使って注文を送信する

1. 画面の修正
2. プログラムの修正
3. 起動してみる

画面の修正

Main.html

ティッカー情報の配信を開始するボタンと停止するボタンを追加しました。

<div style="clear:both; padding-top:10px;">
   <div class="entry_title">
      <div class="pull_left">ティッカー情報</div>
      <div class="pull_right">
         <input type="button" value="更新開始" onclick="startTicker();" />
         <input type="button" value="更新停止" onclick="stopTicker();" />
      </div>
   </div>
   <table id="tickerTable">
      <tr class="header">
         <th style="width:5%">種別</th>
         <th style="width:10%">時刻</th>
         <th style="width:5%">ID</th>
         <th style="width:5%">売値</th>
         <th style="width:5%">買値</th>
         <th style="width:10%">売り数量</th>
         <th style="width:10%">買い数量</th>
         <th style="width:10%">売り注文総数</th>
         <th style="width:10%">買い注文総数</th>
         <th style="width:10%">最終取引価格</th>
         <th style="width:10%">出来高</th>
         <th style="width:10%">価格単位出来高</th>
      </tr>
   </table>
</div>

<div class="pull_left">ティッカー情報</div>

</div>

<th style="width:10%">売り注文総数</th>

<th style="width:10%">買い注文総数</th>

<th style="width:10%">最終取引価格</th>

<th style="width:10%">価格単位出来高</th>

</tr>

</table>

</div>

プログラムの修正

APIの修正

BfApi.py

インポートするクラス

import hashlib
import hmac
import json
import logging
import requests
import time
import urllib


from common.Constants import Constants

from pubnub.callbacks import SubscribeCallback
from pubnub.enums import PNStatusCategory
from pubnub.pnconfiguration import PNConfiguration
from pubnub.pubnub import SubscribeListener
from pubnub.pubnub_tornado import PubNubTornado

importhashlib

importhmac

importjson

importlogging

importrequests

importtime

importurllib

fromcommon.Constants importConstants

frompubnub.callbacks importSubscribeCallback

frompubnub.enums importPNStatusCategory

frompubnub.pnconfiguration importPNConfiguration

frompubnub.pubnub importSubscribeListener

frompubnub.pubnub_tornado importPubNubTornado

追加するメソッド、内部クラス

def start_pub_nub_ticker(self, cl, channels):
    """
    ティッカー情報の配信を開始
    :param cl: web_socket client
    :param channels: 配信するチャンネル
    :return:
    """

    self.listener = self.MySubscriberCallback(cl)
    self.pub_nub.add_listener(self.listener)
    self.pub_nub.subscribe().channels(channels).execute()

def stop_pub_nub_ticker(self, channels):
    """
    ティッカー情報の配信を停止
    :param channels: 停止するチャンネル
    :return:
    """
    self.pub_nub.unsubscribe().channels(channels).execute()
    self.pub_nub.remove_listener(self.listener)

class MySubscriberCallback(SubscribeCallback):
    """
    Pubnub登録のコールバッククラス
    """

    def __init__(self, clent):
        self.cl = clent

    def presence(self, pubnub, presence):
        pass  # handle incoming presence data

    def status(self, pubnub, status):
        if status.category == PNStatusCategory.PNUnexpectedDisconnectCategory:
            pass  # This event happens when radio / connectivity is lost

        elif status.category == PNStatusCategory.PNConnectedCategory:
            # Connect event. You can do stuff like publish, and know you'll get it.
            # Or just use the connected event to confirm you are subscribed for
            # UI / internal notifications, etc
            pass
        elif status.category == PNStatusCategory.PNReconnectedCategory:
            pass
            # Happens as part of our regular operation. This event happens when
            # radio / connectivity is lost, then regained.
        elif status.category == PNStatusCategory.PNDecryptionErrorCategory:
            pass
            # Handle message decryption error. Probably client configured to
            # encrypt messages and on live data feed it received plain text.

    def message(self, pubnub, message):
        """
        登録したチャンネルからメッセージを受信した際の処理
        :param pubnub:
        :param message:
        :return:
        """
        # WEBソケットを利用してクライアントに配信
        for c in self.cl:
            c.write_message(message.message)

defstart_pub_nub_ticker(self,cl,channels):

"""

ティッカー情報の配信を開始

:param cl: web_socket client

:param channels: 配信するチャンネル

:return:

"""

self.listener=self.MySubscriberCallback(cl)

self.pub_nub.add_listener(self.listener)

self.pub_nub.subscribe().channels(channels).execute()

defstop_pub_nub_ticker(self,channels):

"""

ティッカー情報の配信を停止

:param channels: 停止するチャンネル

:return:

"""

self.pub_nub.unsubscribe().channels(channels).execute()

self.pub_nub.remove_listener(self.listener)

classMySubscriberCallback(SubscribeCallback):

"""

Pubnub登録のコールバッククラス

"""

def__init__(self,clent):

self.cl=clent

defpresence(self,pubnub,presence):

pass # handle incoming presence data

defstatus(self,pubnub,status):

ifstatus.category==PNStatusCategory.PNUnexpectedDisconnectCategory:

pass # This event happens when radio / connectivity is lost

elifstatus.category==PNStatusCategory.PNConnectedCategory:

# Connect event. You can do stuff like publish, and know you'll get it.

# Or just use the connected event to confirm you are subscribed for

# UI / internal notifications, etc

pass

elifstatus.category==PNStatusCategory.PNReconnectedCategory:

pass

# Happens as part of our regular operation. This event happens when

# radio / connectivity is lost, then regained.

elifstatus.category==PNStatusCategory.PNDecryptionErrorCategory:

pass

# Handle message decryption error. Probably client configured to

# encrypt messages and on live data feed it received plain text.

defmessage(self,pubnub,message):

"""

登録したチャンネルからメッセージを受信した際の処理

:param pubnub:

:param message:

:return:

"""

# WEBソケットを利用してクライアントに配信

forcinself.cl:

c.write_message(message.message)

メインクラスのグローバル変数を追加

BfTool.py

    cl = []
    api_ticker = None

1 2	cl=[] api_ticker=None

WEBソケットの修正

BfTool.py

ひとまずは、既存のティッカー情報更新用メソッドはコメントアウトし、
新しい仕組みを追加しました。

class SendWebSocket(WebSocketHandler):
    """
    WEBソケット通信
    1秒毎にティッカー情報を画面に配信します
    """

    def open(self):
        logging.info('SendWebSocket [open] IP :' + self.request.remote_ip)
        self.ioloop = tornado.ioloop.IOLoop.instance()
        # self.send_ticker()
        if self not in cl:
            cl.append(self)

    def on_message(self, message):
        logging.info("WebSocketHandler [on_message]")
        for client in cl:
            client.write_message(message)

    def on_close(self):
        logging.info("SendWebSocket [on_closed]")
        if self in cl:
            cl.remove(self)

    def check_origin(self, origin):
        logging.info("SendWebSocket [check_origin]")

        return True

    def send_ticker(self):

        self.ioloop.add_timeout(time.time() + 2, self.send_ticker)
        if self.ws_connection:
            api = BfApi()
            data = api.call_pub_nub('lightning_ticker_FX_BTC_JPY')
            message = json.dumps(data)
            self.write_message(message)

classSendWebSocket(WebSocketHandler):

"""

WEBソケット通信

1秒毎にティッカー情報を画面に配信します

"""

defopen(self):

logging.info('SendWebSocket [open] IP :'+self.request.remote_ip)

self.ioloop=tornado.ioloop.IOLoop.instance()

# self.send_ticker()

ifselfnotincl:

cl.append(self)

defon_message(self,message):

logging.info("WebSocketHandler [on_message]")

forclient incl:

client.write_message(message)

defon_close(self):

logging.info("SendWebSocket [on_closed]")

ifselfincl:

cl.remove(self)

defcheck_origin(self,origin):

logging.info("SendWebSocket [check_origin]")

returnTrue

defsend_ticker(self):

self.ioloop.add_timeout(time.time()+2,self.send_ticker)

ifself.ws_connection:

api=BfApi()

data=api.call_pub_nub('lightning_ticker_FX_BTC_JPY')

message=json.dumps(data)

self.write_message(message)

ティッカー情報配信の開始と停止メソッド追加

BfTool.py

画面から開始、停止ができるようにURLメソッドを追加しました。

class StartTicker(RequestHandler):
    """
    ティッカー情報の更新を開始
    """

    def initialize(self):
        logging.info("StartTicker [initialize]")

    def post(self):
        logging.info("StartTicker [post]")
        global api_ticker
        api_ticker = BfApi()
        api_ticker.start_pub_nub_ticker(cl, ['lightning_ticker_FX_BTC_JPY'])


class StopTicker(RequestHandler):
    """
    ティッカー情報の更新を停止
    """

    def initialize(self):
        logging.info("StopTicker [initialize]")

    def post(self):
        logging.info("StopTicker [post]")
        global api_ticker
        if api_ticker:
            api_ticker.stop_pub_nub_ticker('lightning_ticker_FX_BTC_JPY')

classStartTicker(RequestHandler):

"""

ティッカー情報の更新を開始

"""

definitialize(self):

logging.info("StartTicker [initialize]")

defpost(self):

logging.info("StartTicker [post]")

globalapi_ticker

api_ticker=BfApi()

api_ticker.start_pub_nub_ticker(cl,['lightning_ticker_FX_BTC_JPY'])

classStopTicker(RequestHandler):

"""

ティッカー情報の更新を停止

"""

definitialize(self):

logging.info("StopTicker [initialize]")

defpost(self):

logging.info("StopTicker [post]")

globalapi_ticker

ifapi_ticker:

api_ticker.stop_pub_nub_ticker('lightning_ticker_FX_BTC_JPY')

URLマッピングの追加

BfTool.py

    app = tornado.web.Application([
        (r"/", MainHandler),
        (r"/ws", SendWebSocket),
        (r"/balance", GetBalanceHandler),
        (r"/execution", GetExecutionHandler),
        (r"/childOrder", GetChildOrderHandler),
        (r"/sendOrder", SendChildOrderHandler),
        (r"/cancelOrder", CancelChildOrderHandler),
        (r"/startTicker", StartTicker),
        (r"/stopTicker", StopTicker)
        ],
        template_path=os.path.join(os.getcwd(), "templates"),
        static_path=os.path.join(os.getcwd(), "static"),
        js_path=os.path.join(os.getcwd(), "js"),
    )

app=tornado.web.Application([

(r"/",MainHandler),

(r"/ws",SendWebSocket),

(r"/balance",GetBalanceHandler),

(r"/execution",GetExecutionHandler),

(r"/childOrder",GetChildOrderHandler),

(r"/sendOrder",SendChildOrderHandler),

(r"/cancelOrder",CancelChildOrderHandler),

(r"/startTicker",StartTicker),

(r"/stopTicker",StopTicker)

template_path=os.path.join(os.getcwd(),"templates"),

static_path=os.path.join(os.getcwd(),"static"),

js_path=os.path.join(os.getcwd(),"js"),

)

画面処理の追加

script.js

/**
 * ティッカーの更新を開始
 */
function startTicker() {

   $.ajax({
      url: "http://localhost:8080/startTicker",
      type: "POST",
      success: function(jsonResponse) {
          console.log(jsonResponse);
      },
      error: function() {
      }
   });
}

/**
 * ティッカーの更新を停止
 */
function stopTicker() {

   $.ajax({
      url: "http://localhost:8080/stopTicker",
      type: "POST",
      success: function(jsonResponse) {
          console.log(jsonResponse);
      },
      error: function() {
      }
   });
}