ウォークフォワード検証

実際に取引したであろうやり方で検証する

通常の機械学習のように学習行とテスト行をランダムに分けると、時系列では破綻します。学習データにテスト日より後の日が含まれ、隣り合う日は相関するため、テストは学習の一部の写しになってしまいます。ウォークフォワード検証は時間を守ります。過去のブロックで学習し、その直後のブロックでテストし、窓をずらして繰り返します。どの予測も未来を一度も見たことのないモデルが出したもので、これがまさにライブ取引の状況です。

import numpy as np

n, train_size, test_size = 10, 4, 2
start = 0
while start + train_size + test_size <= n:
    train = np.arange(start, start + train_size)
    test = np.arange(start + train_size, start + train_size + test_size)
    print("train", train.tolist(), "test", test.tolist())
    start += test_size                                   # slide forward by one test block

ローリングか拡大か、そしてエンバーゴ

2種類あります。ローリング窓は学習の長さを固定し、最も古い行を捨てます。局面変化に適応でき、コストは一定です。拡大窓は常に0行目から始まるので学習データが増え、すべての観測を使えますが、古いデータは陳腐化しているかもしれません。3つ目の要素がギャップ（エンバーゴ）です。特徴量や目的変数が h 日前や先を見る場合、テスト期間の直前の行はそれと情報を共有するので、学習とテストの間の数行を飛ばします。

def walk_forward_splits(n, train_size, test_size, gap=0):
    splits, start = [], 0
    while start + train_size + gap + test_size <= n:
        train = list(range(start, start + train_size))
        test = list(range(start + train_size + gap, start + train_size + gap + test_size))
        splits.append((train, test))
        start += test_size
    return splits

for train, test in walk_forward_splits(12, 5, 2, gap=1):
    print(train, "->", test)                         # the row between them is skipped

フォールドごとに採点する

各フォールドで学習行に当てはめ、テスト行で採点するので、1つの数ではなくフォールドごとの誤差が得られます。そのばらつきが、モデルが時間を通じてどれだけ安定しているかを示します。有用な基準は自明な予測で、ここでは各テスト日に学習期間の平均を使います。誤差をフォールドごとに表示してください。平均では良いがある局面でひどいモデルは、常にそこそこ良いモデルとは違います。

import numpy as np

def walk_forward_splits(n, train_size, test_size):
    out, start = [], 0
    while start + train_size + test_size <= n:
        out.append((np.arange(start, start + train_size), np.arange(start + train_size, start + train_size + test_size)))
        start += test_size
    return out

rng = np.random.default_rng(2)
returns = rng.normal(0.0004, 0.01, size=300)
errors = []
for train, test in walk_forward_splits(len(returns), 100, 50):
    errors.append(np.mean((returns[test] - returns[train].mean()) ** 2))
print(np.round(np.array(errors) * 1e4, 3))                      # squared error in basis points squared, fold by fold

リークを見つける

分割がクリーンなのは、すべての学習行がすべてのテスト行より厳密に前にあるときだけです。train.max() < test.min() の確認は1行で済み、最もよくあるバグを捕まえます。並べ替えられた・シャッフルされたインデックス、1つずれたスライス、デフォルトでシャッフルするライブラリ関数の分割などです。この確認をフォールドを作るコードの中に入れ、リークが起きたら見栄えの良い結果を出す代わりに実行を止めるようにしましょう。

import numpy as np

def leaks_future(train_idx, test_idx):
    return bool(np.max(train_idx) >= np.min(test_idx))

train = np.arange(0, 100)
print(leaks_future(train, np.arange(100, 120)))                     # False: clean
shuffled = np.random.default_rng(1).permutation(120)                # what a shuffling splitter does
print(leaks_future(shuffled[:100], shuffled[100:]))                 # True: training rows after test rows

• ウォークフォワード：過去のブロックで学習し次のブロックでテストして窓をずらす。どのテスト行もその学習行すべての後にある
• ローリング窓は学習サイズ固定、拡大窓は0行目から。特徴量が複数日にまたがるなら、学習とテストの間にギャップ（エンバーゴ）を置く
• フォールドを個別に採点してばらつきを見る。train.max() < test.min() を確認する。報告するフォールドでは決して調整しない

次は指値注文板（リミット・オーダーブック）です。価格が実際に形成されるデータ構造です。

ラボ：採点される内容

• walk_forward_splits で窓を時間方向に動かす25点
• expanding_splits で履歴をすべて残す25点
• forecast_mse で予測をフォールドごとに検証する25点
• leaks_future でリークを見つける25点