Fire Engine

消防士→ITエンジニア→研究者

Bokehを用いたビットコイン価格のリアルタイム可視化（Python）

Python データサイエンス

　今回はBokehというライブラリを使って、ビットコイン価格のリアルタイム可視化を行う方法について書いていきます。Bokehを使うと、Pythonオンリーで可視化までできるので、非常に便利です。

Bokehとは

　BokehとはPython製の対話的可視化ライブラリです。対話的とはどういうことかというと、作成したグラフのズームアップ・ズームアウト・軸の調整といった様々な操作を手元で簡単に行えるということです。

f:id:hirotsuru314:20171111151354p:plain

これは、Bokeh内部で、HTML・CSS・JavaScriptを自動で生成する仕組みがあることで達成されており、ユーザは基本的にはPythonコードしか書く必要がありません。
Bokehの公式のギャラリーを見るとすごくファンシーな可視化がたくさんあるので見るだけでも楽しいです！（実際に対話的な操作もできます）

Gallery — Bokeh 0.13.0 documentation

※以前の記事に対話的なグラフの作成方法を書きましたので、併せてどうぞ

Bokehの特徴

　Bokehの特徴をいくつか挙げると、

  • 対話的な可視化ができる
  • ストリーミングデータのリアルタイム可視化ができる
  • グラフを作るためにHTML・CSS・JavaScriptを書かなくてもよい

といった感じです。
今回は『ストリーミングデータのリアルタイム可視化ができる』に着目した記事になります。
対話的というだけでもすごいのですが、動的に流れてくるデータに対しても、Bokehにデータを流すことでリアルタイム可視化ができ、これが本当に便利です！

用いるデータについて

　ストリーミングデータの例として『ビットコイン価格』を取り上げて見ることにします。（私自身はシステムトレードなどは全くやっていません。笑）
　下記リンクの中のLast Tradeという部分を取得します。この部分は、数秒ごとにリロードすると値が変わるので、定期的にスクレイピングすることで、時系列データを得ることができます。

Bitcoincharts | bitFlyer - JPY Summary

f:id:hirotsuru314:20171111151600p:plain

ビットコイン価格のリアルタイム可視化

開発環境
Python 3.6.3
bokeh==0.12.10
beautifulsoup4==4.6.0
requests==2.13.0

作成プログラム

from bokeh.io import curdoc
from bokeh.models import ColumnDataSource, DatetimeTickFormatter
from bokeh.plotting import figure
import requests
from bs4 import BeautifulSoup
from datetime import datetime
from math import radians
from pytz import timezone

URL = "https://bitcoincharts.com/markets/"
MARKET="bitflyerJPY"

def get_data():
    res = requests.get(URL + MARKET + ".html", headers={"User-Agent": "Mozilla/5.0"})
    content = res.content
    soup = BeautifulSoup(content, "html.parser")
    last_trade = int(soup.find_all("p")[0].span.text)
    return last_trade

def update_data():
    now = datetime.now(tz=timezone("Asia/Tokyo"))
    new_data = dict(x=[now], y=[get_data()])
    source.stream(new_data, rollover=200)

#データソースの作成
source = ColumnDataSource(dict(x=[], y=[]))

#グラフの作成
fig = figure(x_axis_type="datetime",
             x_axis_label="Datetime",
             y_axis_label="Last Trade",
             plot_width=800,
             plot_height=600)
fig.title.text = "Bitcoin Charts"
fig.line(source=source, x="x", y="y", line_width=2, alpha=.85, color="blue")
fig.circle(source=source, x="x", y="y", line_width=2, color="blue")

#軸の設定
format = "%Y-%m-%d-%H-%M-%S"
fig.xaxis.formatter = DatetimeTickFormatter(
    seconds=[format],
    minsec =[format],
    minutes=[format],
    hourmin=[format],
    hours  =[format],
    days   =[format],
    months =[format],
    years  =[format]
)
fig.xaxis.major_label_orientation=radians(90)

#コールバックの設定
curdoc().add_root(fig)
curdoc().add_periodic_callback(update_data, 3000) #ms単位

上記プログラムにbitcoin_streaming.pyというファイル名をつけ、コマンドラインから下のように実行すると、私の場合

$ bokeh serve bitcoin_streaming.py
2017-11-11 12:24:32,513 Starting Bokeh server version 0.12.2
2017-11-11 12:24:32,520 Starting Bokeh server on port 5006 with applications at paths ['/bitcoin_streaming']
2017-11-11 12:24:32,520 Starting Bokeh server with process id: 1995

のように表示され、http://localhost:5006/bitcoin_streaming にアクセスすると、下のようにグラフが表示されます。

f:id:hirotsuru314:20171111151837p:plain

実際はリアルタイムで動くグラフなので、動画にしてみました。


Bokehを用いたビットコイン価格のリアルタイム可視化

ちょっとわかりづらいですが、3秒に1回スクレイピングでビットコイン価格を取得して、リアルタイムに可視化しています。

プログラムについての補足

Beautiful Soupドキュメント — BeautifulSoup Document 3.0 ドキュメント

  • ColumnarDataSource
    BokehにはColumnarDataSourceというデータ構造があります。ColumnarDataSourceは 列指向でデータを保持します。
    リアルタイム可視化の場合には、このデータ構造にstream関数を使って、データを流し込むとBokehの方でデータを保持から可視化までやってくれます。
    ちなみに
def update_data():
    now = datetime.now(tz=timezone("Asia/Tokyo"))
    new_data = dict(x=[now], y=[get_data()])
    source.stream(new_data, rollover=200)

ここのstream関数の引数であるrolloverは、保持するデータ数の上限を決めるもので、その上限がそのまま可視化にも反映されるので、重要です。

  • コールバック
    作成したプログラムの最後の行の
curdoc().add_periodic_callback(update_data, 3000)

部分で、update_data関数を3秒（3000ms）に1回呼び出す設定を書いていて、update_data関数が実行されると、ColumnarDataSourceのインスタンスに新しいデータが追加され、可視化される仕組みになっています。

参考

おわりに

　上記のソースコードはデータ取得のget_data関数の部分だけ変えるといろんなリアルタイム可視化ができます。例えば、CPU使用率を監視しようと思うと、こんな感じに入れ替えればできます

import psutil
def get_new_data():
    return psutil.cpu_percent()

　いろいろ楽しいことができそうですね！

ちなみに・・・

　私は最近個人で『Banpei』という異常検知ライブラリを作っており、Bokehと連携させ、リアルタイム異常監視などができるように開発を進めています。リアルタイム異常監視については近日中にまた記事にします！
　異常検知ライブラリBanpeiはGitHubで公開しているので見ていただけると嬉しいです！

github.com