【Python】pubnub を使って bitflyer の Ticker 情報をリアルタイム表示する

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

おはようございます。

今回は、pubnubというサービスを用いてリアルタイムメッセージング機能というのを試してみたくて色々調べてみました。

（ただ、結果的にあまりうまくできてませんが）

pubnubについて調べていると Bitflyer（仮想通貨取引所）が板等の更新情報を配信しているということなので、それを利用してWEBサービスにリアルタイムでデータを表示してみます。

1. 新規プロジェクトの作成
2. パッケージ追加
3. 画面の作成
4. プログラムの作成
5. 起動してみる
6. まとめ

新規プロジェクトの作成

PyCharmを起動し、新しくプロジェクトを作成します。
今回は「BfTool」としました。

プロジェクトの作成については、大した手順ではないですが次の記事を参考にしてください。

【Python】PyCharmのインストールからHello Worldまで

また、構成は次の通りです。
BfTool
│ Sample.py
├─static
│ ├─css
│ │ style.css
│ └─js
│ script.js
├─templates
│ index.html
└─venv

パッケージ追加

今回は Tornado の他に「pubnub」というパッケージを追加します。

パッケージ追加の方法についても前回の記事を参考にしていただけると幸いです。

【Python】PyCharm でパッケージ追加と別ファイルのクラス読込

画面の作成

index.html

<!DOCTYPE html>
<html>
   <head>
      <title>{{ title }}</title>
      <link rel="stylesheet" href="{{ static_url('css/style.css') }}"/>
      <script type="text/javascript" src="{{ static_url('js/script.js') }}"></script>
      <script type="text/javascript" src="http://ajax.googleapis.com/ajax/libs/jquery/1.7.2/jquery.min.js"></script>
   </head>
   <body onload="initialize();">
      <div id="container">
         <div style="clear:both; padding-top:10px;">
            <table id="dataTable">
               <tr id="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>
   </body>
</html>

<!DOCTYPE html>

<html>

<head>

<title>{{ title }}</title>

</head>

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

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

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

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

</tr>

</table>

</div>

</body>

</html>

style.css

body {
   font-family:"ＭＳ Ｐゴシック","MS PGothic",sans-serif;
   width: 100%;
   margin: 0 auto;
}
div {
   margin:20px;
}
div#container{
   width: 100%;
}
table {
   width:95%;
   border: 1px solid #ccc;
   border-collapse:collapse;
}
th {
   text-align:center;
   background-color:#404040;
   color:#ffffff;
   width: 100px;
   height: 25px;
   border: 1px solid #ccc;
}
td {
   padding-left:5px;
   width: 200px;
   height: 20px;
   border: 1px solid #ccc;
}

body {

font-family:"ＭＳＰゴシック","MSPGothic",sans-serif;

width:100%;

margin:0auto;

}

div {

margin:20px;

}

div#container{

width:100%;

}

table {

width:95%;

border:1pxsolid#ccc;

border-collapse:collapse;

}

th {

text-align:center;

background-color:#404040;

color:#ffffff;

width:100px;

height:25px;

border:1pxsolid#ccc;

}

td {

padding-left:5px;

width:200px;

height:20px;

border:1pxsolid#ccc;

}

プログラムの作成

Sample.py

import json
import logging
import os
import signal
import time
import tornado.ioloop
import tornado.web
import tornado.websocket

from tornado.options import options
from pubnub.pnconfiguration import PNConfiguration
from pubnub.pubnub import PubNub, SubscribeListener

is_closing = False
pb = None
my_listener = None


def signal_handler(signum, stack):
    u""" サーバー停止信号を受信した際にフラグをオンにします."""

    global is_closing
    logging.info("exiting...")
    is_closing = True


def try_exit():
    u""" サーバー停止フラグがオンの場合にインスタンスを停止します. """

    global is_closing
    if is_closing:
        # clean up here
        tornado.ioloop.IOLoop.instance().stop()
        logging.info("exit success")


class MainHandler(tornado.web.RequestHandler):
    u""" メイン処理 """

    def initialize(self):
        logging.info("initialize")

    def get(self):

        self.render("index.html", title="Pythonサンプル")


class SendWebSocket(tornado.websocket.WebSocketHandler):
    def open(self):
        logging.info('SendWebSocket IP:' + self.request.remote_ip)
        self.ioloop = tornado.ioloop.IOLoop.instance()
        self.send_websocket()

    def on_close(self):
        print("Session closed")
        global pb
        global my_listener

        pb.unsubscribe().channels(['lightning_ticker_BTC_JPY', 'lightning_ticker_FX_BTC_JPY']).execute()
        my_listener.wait_for_disconnect()

    def check_origin(self, origin):
        return True

    def send_websocket(self):
        self.ioloop.add_timeout(time.time() + 1, self.send_websocket)
        if self.ws_connection:
            global pb
            global my_listener

            # Bitflyer の キーを設定してインスタンスを生成、
            # リスナーを追加
            pnc = PNConfiguration()
            pnc.subscribe_key = "sub-c-52a9ab50-291b-11e5-baaa-0619f8945a4f"
            pnc.ssl = False
            pb = PubNub(pnc)
            my_listener = SubscribeListener()
            pb.add_listener(my_listener)

            # チャンネルの設定
            pb.subscribe().channels(['lightning_ticker_BTC_JPY', 'lightning_ticker_FX_BTC_JPY']).execute()
            my_listener.wait_for_connect()

            # BTC-FX/JPY の結果を取得
            fx_result = my_listener.wait_for_message_on('lightning_ticker_FX_BTC_JPY')
            fx_data = fx_result.message
            message = json.dumps(fx_data)
            logging.info(message)
            self.write_message(message)


app = tornado.web.Application([
    (r"/", MainHandler),
    (r"/ticker", SendWebSocket)
    ],
    template_path=os.path.join(os.getcwd(), "templates"),
    static_path=os.path.join(os.getcwd(), "static"),
    js_path=os.path.join(os.getcwd(), "js"),
)

if __name__ == "__main__":
    tornado.options.parse_command_line()
    signal.signal(signal.SIGINT, signal_handler)
    app.listen(8888)
    logging.info("server started")
    tornado.ioloop.PeriodicCallback(try_exit, 100).start()
    tornado.ioloop.IOLoop.instance().start()

100

101

102

103

104

105

106

107

importjson

importlogging

importos

importsignal

importtime

importtornado.ioloop

importtornado.web

importtornado.websocket

fromtornado.options importoptions

frompubnub.pnconfiguration importPNConfiguration

frompubnub.pubnub importPubNub,SubscribeListener

is_closing=False

pb=None

my_listener=None

defsignal_handler(signum,stack):

u""" サーバー停止信号を受信した際にフラグをオンにします."""

globalis_closing

logging.info("exiting...")

is_closing=True

deftry_exit():

u""" サーバー停止フラグがオンの場合にインスタンスを停止します. """

globalis_closing

ifis_closing:

# clean up here

tornado.ioloop.IOLoop.instance().stop()

logging.info("exit success")

classMainHandler(tornado.web.RequestHandler):

u""" メイン処理 """

definitialize(self):

logging.info("initialize")

defget(self):

self.render("index.html",title="Pythonサンプル")

classSendWebSocket(tornado.websocket.WebSocketHandler):

defopen(self):

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

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

self.send_websocket()

defon_close(self):

print("Session closed")

globalpb

globalmy_listener

pb.unsubscribe().channels(['lightning_ticker_BTC_JPY','lightning_ticker_FX_BTC_JPY']).execute()

my_listener.wait_for_disconnect()

defcheck_origin(self,origin):

returnTrue

defsend_websocket(self):

self.ioloop.add_timeout(time.time()+1,self.send_websocket)

ifself.ws_connection:

globalpb

globalmy_listener

# Bitflyer のキーを設定してインスタンスを生成、

# リスナーを追加

pnc=PNConfiguration()

pnc.subscribe_key="sub-c-52a9ab50-291b-11e5-baaa-0619f8945a4f"

pnc.ssl=False

pb=PubNub(pnc)

my_listener=SubscribeListener()

pb.add_listener(my_listener)

# チャンネルの設定

pb.subscribe().channels(['lightning_ticker_BTC_JPY','lightning_ticker_FX_BTC_JPY']).execute()

my_listener.wait_for_connect()

# BTC-FX/JPY の結果を取得

fx_result=my_listener.wait_for_message_on('lightning_ticker_FX_BTC_JPY')

fx_data=fx_result.message

message=json.dumps(fx_data)

logging.info(message)

self.write_message(message)

app=tornado.web.Application([

(r"/",MainHandler),

(r"/ticker",SendWebSocket)

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

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

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

)

if__name__=="__main__":

tornado.options.parse_command_line()

signal.signal(signal.SIGINT,signal_handler)

app.listen(8888)

logging.info("server started")

tornado.ioloop.PeriodicCallback(try_exit,100).start()

tornado.ioloop.IOLoop.instance().start()

script.js

// スクリプト読み込み時の処理
function initialize() {
    var connection = new WebSocket('ws://127.0.0.1:8888/ticker');
    connection.onmessage = function (e) {
        var data = JSON.parse(e.data.replace( /\\/g , "" ));

        var table = $("#dataTable");

        // 日付け変換
        var date = new Date(data.timestamp);
        data.timestamp = date.toLocaleString();

        // テーブルに追加
        var tr = document.createElement("tr");
        $.each(data, function(i, cell){
            var td = document.createElement("td");
            td.innerHTML = cell;
            tr.appendChild(td);
        });
        $(tr).insertAfter("#header");

    };
}