まとめ 変化点検出は、時系列を「平均やばらつきがほぼ一定の区間」に切り分け、その境目(変化点)を見つける問題です。ruptures は「各区間のコストの合計 + 区間数 × ペナルティ」を最小にする区切り方を探すライブラリで、Pelt(model="rbf").fit(signal).predict(pen=10) の 2 行で変化点のリストが返ります。このページでは平均が 100 日ごとに段階的に変わる人工データ、分散だけが 500〜600 日目で変わるデータ、Datadog の株価の 3 種類に Pelt と Dynp を当て、pen を 0.1 から 1.0 に動かすと変化点が減っていく様子、model を rbf と l2 で替えると分散の変化を拾えるかどうかが変わる様子を rpt.display の図で確かめます。
Truong, Charles, Laurent Oudre, and Nicolas Vayatis. “ruptures: change point detection in Python.” arXiv preprint arXiv:1801.00826 (2018). (arXiv )
直感
# 1 年分の日次売上を眺めて「ここから客層が変わった」と感じる場所を探すとします。区間を 1 つに分けなければ平均からのずれ(コスト)が大きく、1 日ごとに区切ればコストは 0 になりますが意味がありません。そこで「区切りを 1 つ増やすごとにペナルティを払う」というルールを決め、コストとペナルティの合計が最小になる区切り方を選びます。pen が大きいと区切りが減って大きな変化だけが残り、小さいと細かい揺れまで変化点になります。ruptures の Pelt はこの最小化を動的計画法と枝刈りで高速に解き、Dynp はペナルティの代わりに変化点の個数を指定して厳密に解きます。
信号 →
コスト関数(rbf / l2)を選ぶ →
区切りの候補ごとに「区間コストの合計 + 区間数 × pen」を計算 →
最小になる区切りを Pelt で探す →
変化点のインデックスを返す →
rpt.display で正解と重ねる
ruptures
# 公式ドキュメント: ruptures documentation
まず ruptures 付属の生成関数で「平均が 4 回ジャンプする 3 次元の信号」を作り、Pelt で変化点を検出して正解と重ねます。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# seaborn内部のpandas使用箇所でFutureWarningが出ていたので表示しないようにする
import warnings
from datetime import datetime , timedelta
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import ruptures as rpt
import seaborn as sns
warnings . simplefilter ( action = "ignore" , category = FutureWarning )
# generate signal
n_samples , dim , sigma = 1000 , 3 , 4
n_bkps = 4 # number of breakpoints
signal , bkps = rpt . pw_constant ( n_samples , dim , n_bkps , noise_std = sigma )
# detection
algo = rpt . Pelt ( model = "rbf" ) . fit ( signal )
result = algo . predict ( pen = 10 )
# display
rpt . display ( signal , bkps , result )
plt . show ()
何が起きているか(ステップで見る)
# コードを読む前に、この手法が図の上で何をしているかを追ってみましょう。「再生」を押すか、スライダーを動かすと 1 手順ずつ進みます。
見るポイント :どこで区切ると左右のばらつきの合計が最小になるかを総当たりで探す。
さまざまなデータでの比較
# 変化点検出+可視化のコード
# Pelt による検出と rpt.display による描画を 1 つの関数にまとめ、method・pen・min_size を変えて何度も呼べるようにします。
1
2
3
4
5
6
7
8
9
def vizualize_change_point (
signals , true_change_points = [], method = "rbf" , pen = 0.5 , min_size = 3
):
algo = rpt . Pelt ( model = method , min_size = min_size ) . fit ( signals )
result = algo . predict ( pen = pen )
rpt . display ( signals , true_change_points , result )
plt . suptitle ( f "method= { method } , penalty= { pen } , " , fontsize = 20 )
plt . tight_layout ()
plt . show ()
サンプルデータの作成
# 100 日ごとに value = i // 100 が 1 ずつ増え、その値を cos(i) に掛けて振れ幅が段階的に大きくなる 2 列の信号を 1000 日分作ります。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
X = []
dates = [ datetime ( 2021 , 1 , 1 ) + timedelta ( days = i ) for i in range ( 1000 )]
for i , di in enumerate ( dates ):
value = i // 100
X . append (
[
np . sin ( di . weekday () * 0.001 ) * 2
+ np . random . rand () * 0.1
+ value * np . cos ( i ),
np . sin ( di . day * 0.01 ) * 3 + np . random . rand () * 0.1 + value * np . cos ( i ),
]
)
data = pd . DataFrame ( X )
data . columns = [ f "f { c } " for c in data . columns ]
data . index = dates
plt . figure ( figsize = ( 12 , 3 ))
sns . lineplot ( data = data )
<Axes: >
ペナルティの違いによる結果の違い
# pen を 0.1 から 1.0 まで 5 段階に変えて、検出される変化点の数がどう変わるかを見ます。
1
2
for pen in np . linspace ( 0.1 , 1 , 5 ):
vizualize_change_point ( data . values , method = "rbf" , pen = pen )
動的計画法を使った変化点の検出
# Dynp は厳密な動的計画法で、ペナルティの代わりに変化点の数 n_bkps をあらかじめ指定する必要があります。「変化点はちょうど 8 個」と分かっている場面向きです。
1
2
3
4
5
6
7
8
9
10
11
12
13
def vizualize_change_point_dynp (
signals , true_change_points = [], method = "rbf" , min_size = 3 , n_bkps = 3
):
algo = rpt . Dynp ( model = method , min_size = min_size ) . fit ( signals )
result = algo . predict ( n_bkps = n_bkps )
rpt . display ( signals , true_change_points , result )
plt . suptitle ( f "method= { method } , penalty= { pen } , " , fontsize = 20 )
plt . tight_layout ()
plt . show ()
vizualize_change_point_dynp (
data . values , true_change_points = [], method = "rbf" , min_size = 3 , n_bkps = 8
)
人工データでの実験2
# 平均はほぼ同じまま、500〜600 日目の 100 日間だけ分散が大きくなるデータです。平均の変化しか見ないコスト関数では見つけにくい変化点になります。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
X2 = []
dates2 = [ datetime ( 2021 , 1 , 1 ) + timedelta ( days = i ) for i in range ( 1000 )]
for i , di in enumerate ( dates2 ):
if i == 550 :
X2 . append (
[
np . sin ( di . weekday () * 0.001 ) * 2 + np . random . rand () * 0.1 ,
np . sin ( di . day * 0.01 ) * 3 + np . random . rand () * 0.1 ,
]
)
elif 500 < i < 600 :
X2 . append (
[
np . sin ( di . weekday () * 0.001 ) * 2
+ np . random . rand () * 0.1
+ np . random . randint ( - 2 , 2 ) * 0.1 ,
np . sin ( di . day * 0.01 ) * 3 + np . random . rand () * 0.1 ,
]
)
else :
X2 . append (
[
np . sin ( di . weekday () * 0.001 ) * 2 + np . random . rand () * 0.1 ,
np . sin ( di . day * 0.01 ) * 3 + np . random . rand () * 0.1 ,
]
)
data2 = pd . DataFrame ( X2 )
data2 . columns = [ f "f { c } " for c in data2 . columns ]
data2 . index = dates2
plt . figure ( figsize = ( 12 , 3 ))
sns . lineplot ( data = data2 )
plt . ylim ( 0 , 1 )
(0.0, 1.0)
同じデータに rbf と l2 のコスト関数を当て、分散の変化を拾えるかを比べます。
1
2
3
# 注意点:値の絶対値や損失関数によって指定すべきpenaltyの値は異なる
vizualize_change_point ( data2 . values , method = "rbf" , pen = 1.5 , min_size = 50 )
vizualize_change_point ( data2 . values , method = "l2" , pen = 1.5 , min_size = 50 )
l2 で分散の変化を拾わせるには、分散そのものを列にして「平均の変化」に変えてしまう手があります。f0 の 10 日移動分散を 3 列目に足して再実行します。
1
2
3
data2 [ "var" ] = data2 [ "f0" ] . rolling ( window = 10 ) . var () . fillna ( 0 )
vizualize_change_point ( data2 . values , method = "rbf" , pen = 1.5 , min_size = 50 )
vizualize_change_point ( data2 . values , method = "l2" , pen = 0.1 , min_size = 100 )
変化点が 2 つ(区間の始まりと終わり)と分かっているなら、Dynp で n_bkps=2 を指定するのが最も確実です。
1
vizualize_change_point_dynp ( data2 . values , min_size = 90 , n_bkps = 2 )
実データでの実験
# Yahoo finance の DDOG のデータ (Datadog 株の 1 年分の日足)を使って変化点を見てみます。正解のない実データなので、破線の位置を決算発表などの出来事と照らし合わせて解釈します。
1
2
3
4
5
6
7
8
9
10
ddog = pd . read_csv ( "./DDOG.csv" )
ddog . index = ddog [ "Date" ]
ddog . drop ([ "Date" ], axis = 1 , inplace = True )
ddog [ "Volume" ] = 100 * ddog [ "Volume" ] / ddog [ "Volume" ] . median ()
ddog = ddog [[ "Open" , "Close" , "Volume" ]]
plt . figure ( figsize = ( 12 , 4 ))
sns . lineplot ( data = ddog )
plt . xticks ( rotation = 90 )
plt . show ()
1
vizualize_change_point ( ddog . values , method = "rbf" , pen = 0.5 , min_size = 40 )
使いどころと落とし穴
# ruptures を実データに使うときの調整点は model・pen・min_size の 3 つです。平均のジャンプを探すなら l2、分散や分布の形の変化も拾いたいなら rbf を選び、l2 の pen は信号の分散に比例した桁、rbf の pen は 0.1〜数の桁から始めます。min_size は「変化が意味を持つ最短の長さ」を業務知識から決め、pen と対で動かします。変化点の数が分かっているなら Dynp(n_bkps=k) が最も確実ですが、計算量が \(O(n^2)\) 以上なので 1 万点を超える系列では Pelt か、ウィンドウ法の Window を使います。列ごとのスケールが違うデータでは、DDOG の Volume のように正規化してから渡さないと、桁の大きい列だけで変化点が決まります。
学習ナビ
PHASE 4 · 応用・展開
読んだページに印をつけると、トップページとセクション一覧に進み具合が表示されます(ブラウザ内にのみ保存)。