今回の記事は、前職消防士でゼロからプログラミングを始めた超未熟者の私が、異常検知ライブラリを作った話です。
なぜ作ったか
マインド的背景
消防士を辞めてエンジニアに転職してから1年、いろんな技術に触れました。TensorFlow、scikit-learn、Dockerなどなど、必死になって使い方を覚えました。しかしだんだん、「これ、コマンド覚えてるだけで自分に何も技術身についてなくない？」という疑問や焦りが湧いてきて、自分は便利なツールを使いこなせるようになりたいんじゃなくて、いつの日かみんなが使って便利なツールを作る側になりたいんだ、ということに気づきました。そのような思いから今回初めてライブラリを作り、公開してみました。
データサイエンス的背景
世の中は時系列データで溢れています。ビジネスの場において、データの何かしらの変化の兆候を捉えて、いち早く意思決定をしたいという場面がよくあります。例えば、観測データが商品の受注データであれば、急に売れなくなった（もしくは売れ出した）変化をいち早く検知し、次の一手を打てると機会損失を防げるかもしれません。私も実際に業務で時系列データの変化を捉えたい場面がありましたが、変化の兆候というのはデータによって様々であり、明らかなものもあれば、そうでないものもありました。様々なタイプの変化に対して柔軟に検知するためには、閾値やルール設定といった力技では難しく、コンピュータに「いつもと違う状態」を自動に検知させることが求められています。これがいわゆる異常検知（ここでは特に変化点検知）の技術です。
作ったもの
ソースコード
Banpei（番兵）
github.com
使い方
インストールは下記のコマンドで行います。（Pythonパッケージ化はしていますが、PyPIには登録していないのでgit cloneしたのちにpip installする必要があります。）
git clone https://github.com/tsurubee/banpei.git
cd banpei
pip install .
一番簡単な使い方は、異常を検知したいデータをPythonのリスト型やNumPy arrayなどの配列にして、detect関数に渡してやると、データの変化度を計算して配列を返します。
下記のコードのように3行のみで異常検知できます。まず、Banpeiをimportし、SSTという特異スペクトル変換法（異常検知アルゴリズムの一つ）を使うためのインスタンスを作成し、detect関数にインプットデータを渡すだけです。インスタンス作成時の引数wはスライド窓の幅であり、チューニングが必要なパラメータです。（詳しくは以前の記事を参照ください）
import banpei model = banpei.SST(w=50) results = model.detect(data)
resultsには変化度が入っており、インプットであるdataと、アウトプットであるresultsを可視化すると、
（data: 緑、results: 赤）※見やすいように縦軸は調整しています。
このようにデータの兆候が変化する点で、変化度である赤線が立ち上がっているのが見て取れます。この赤の立ち上がりをみることで変化点検知ができそうです。
リアルタイム異常監視への挑戦
もともと私がBanpeiをつくったときは、リアルタイム監視の環境を作りたいというモチベーションがありました。（売り上げのリアルタイム監視やサーバリソースのリアルタイム監視などなど）そこで今回、ブラウザを通してリアルタイム異常監視ができる簡単な仕組みを構築しました。
可視化のツールとしてBokehというPython製の可視化ライブラリを採用しました。簡単なデモ動画を載せておきます。動画の10秒あたりで変化点検知をしています。
このようなリアルタイム監視環境を構築したければ、pip install bokehでBokehをインストールして、下のようなスクリプトで可視化できます。（私の環境ではbokeh==0.12.10）
import banpei import numpy as np import pandas as pd from bokeh.io import curdoc from bokeh.models import ColumnDataSource, DatetimeTickFormatter from bokeh.plotting import figure from datetime import datetime from math import radians from pytz import timezone data = [] results = [] def get_new_data(): global data #ここでデータ取得（例：new_data = psutil.cpu_percent()） data.append(new_data) def update_data(): global results get_new_data() ret = model.stream_detect(data) results.append(ret) now = datetime.now(tz=timezone("Asia/Tokyo")) new_data = dict(x=[now], y=[data[-1]], ret=[results[-1]]) source.stream(new_data, rollover=500) # Create Data Source source = ColumnDataSource(dict(x=[], y=[], ret=[])) # Create Banpei instance model = banpei.SST(w=30) # Draw a graph fig = figure(x_axis_type="datetime", x_axis_label="Datetime", plot_width=950, plot_height=650) fig.title.text = "Realtime monitoring with Banpei" fig.line(source=source, x='x', y='y', line_width=2, alpha=.85, color='blue', legend='observed data') fig.line(source=source, x='x', y='ret', line_width=2, alpha=.85, color='red', legend='change-point score') fig.circle(source=source, x='x', y='y', line_width=2, line_color='blue', color='blue') fig.legend.location = "top_left" # Configuration of the axis 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) # Configuration of the callback curdoc().add_root(fig) curdoc().add_periodic_callback(update_data, 1000) #ms
get_new_data関数の部分をカスタマイズして、独自のデータ取得の処理を書けば、bokeh serve file名.pyでlocalhostが立ち上がりブラウザ上で上記の動画のようなものが見れます。Banpei側ではstream_detectという関数を使っていますが、これは渡されたインプットデータのうち、最新の1点の変化度を計算して、数値を返す関数です。
Banpeiの異常検知アルゴリズム
Banpeiには異常検知アルゴリズムとして特異スペクトル変換を採用しています。概要は以前記事に書きました。
特異スペクトル変換の特徴としては、特定の確率分布を仮定していないため、①多様な変化に頑強に対応できる、②パラメータチューニングが容易、などが挙げられます。パラメータに関しては、スライド窓のサイズwの実質1個のみのチューニングで済みます。（他の必要なパラメータはwから経験的に算出）特異スペクトル変換は愚直に線形計算しているだけ、というところがすごく素敵です。
一方、内部の計算過程で特異分解と呼ばれる行列計算を用いており、これがかなり計算コストが高いです。したがって、検知を行いたい時間間隔や、パラメータwの最適値によっては、リアルタイム計算が厳しくなるかもしれません。
今後、特異スペクトル変換の高速化のために、クリロフ部分空間を用いた下の論文を実装したいと思ってます。
http://epubs.siam.org/doi/pdf/10.1137/1.9781611972771.54
さいごに
作った感想としては、やっぱりモノ作り楽しいな、です。一旦、画面上で動くものまで作れたのですごく達成感はありました。ここからどのように実際の現場で使えるように持っていくかは真剣に考えています。例えば、前述のアルゴリズムの高速化でよりリアルタイム性を高めたり、メールやSlackに異常通知を送れる機能を実装したりと・・まだまだやりたいことが山積みなので、着実に開発を進めていきたいと思っています。
