K_DM Book
用語集
対象者 標準
対象者別の入口 →

対象者の設定はこのブラウザの中にだけ保存されます。

言語 JA

ruptures

中級

04.17.1

ruptures

最終更新 2020-01-29 読了時間 7 分
まとめ
変化点検出は、時系列を「平均やばらつきがほぼ一定の区間」に切り分け、その境目(変化点)を見つける問題です。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 はペナルティの代わりに変化点の個数を指定して厳密に解きます。

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()

公式ドキュメント:ruptures documentationの図

何が起きているか(ステップで見る) #

コードを読む前に、この手法が図の上で何をしているかを追ってみましょう。「再生」を押すか、スライダーを動かすと 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
)

動的計画法を使う場合、変化点の数(n_bkps)をあらかじめ指定する必要があるの図

人工データでの実験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)

(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)

(0.0, 1.0)の図

(0.0, 1.0)の図

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)

(0.0, 1.0)の図

(0.0, 1.0)の図

変化点が 2 つ(区間の始まりと終わり)と分かっているなら、Dynp で n_bkps=2 を指定するのが最も確実です。

1
vizualize_change_point_dynp(data2.values, min_size=90, n_bkps=2)

(0.0, 1.0)の図

実データでの実験 #

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()

Yahoo financeのDDOGのデータを使って変化点を見てみようと思いますの図

1
vizualize_change_point(ddog.values, method="rbf", pen=0.5, min_size=40)

Yahoo financeのDDOGのデータを使って変化点を見てみようと思いますの図

使いどころと落とし穴 #

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 のように正規化してから渡さないと、桁の大きい列だけで変化点が決まります。