リターン予測のためのscikit-learnパイプライン

リターンに対する機械学習

教師あり学習は、特徴量 X から目的変数 y への関数を当てはめます。金融では目的変数は通常将来のリターンで、特徴量は事前に分かっているもの（過去のリターン、ボラティリティ、出来高）です。scikit-learn は統一されたインターフェースを持ち、どのモデルにも fit(X, y) と predict(X) があります。教科書の問題との大きな違いは信号対雑音比で、リターンはほぼ純粋なノイズなので、良いモデルでもごくわずかしか説明できず、雑な設定は存在しないパターンを見つけてしまいます。

import numpy as np

returns = np.array([0.01, -0.02, 0.015, 0.005, -0.01, 0.02])
n_lags = 2

# Row i holds the two returns before the target; y[i] is the return that follows them
windows = np.lib.stride_tricks.sliding_window_view(returns, n_lags)
X, y = windows[:-1], returns[n_lags:]
print(X)
print(y)                                             # the targets are one step ahead of the features

ラグから作る特徴量

リターンの最も単純な特徴量は、その前のリターンです。ラグ行列を作ります。i 行目は returns[i : i + n_lags] で、目的変数は次の値 returns[i + n_lags] です。スライディングウィンドウのビューを使えばループなしで作れます。1行の中身はすべて、予測する時点で既知でなければなりません。これは先読みのルールをモデルに適用したもので、1つ逆方向にずれた特徴量は、モデルに答えを見せてしまいます。

import numpy as np
from sklearn.linear_model import Ridge
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

# Pipeline: scale each feature, then a ridge regression (a linear model with a penalty on large coefficients)
pipe = Pipeline([("scale", StandardScaler()), ("model", Ridge(alpha=10.0))])
print([name for name, _ in pipe.steps])

rng = np.random.default_rng(1)
returns = np.zeros(400)
noise = rng.normal(0, 0.01, size=400)
for t in range(1, 400):
    returns[t] = 0.3 * returns[t - 1] + noise[t]                    # a series with a little memory

windows = np.lib.stride_tricks.sliding_window_view(returns, 3)
X, y = windows[:-1], returns[3:]
pipe.fit(X[:300], y[:300])                                          # fit on the first 300 rows
print(np.round(pipe.named_steps["model"].coef_, 4))                 # the last lag weighs most

パイプラインとリーク

Pipeline はスケーラーとモデルをひとまとめにし、fit は学習行だけからスケーリングの統計（各特徴量の平均と標準偏差）を学び、predict はそれを新しい行にそのまま適用します。データ全体をスケーリングしてから分割すると、テスト期間の平均とばらつきが学習に漏れます。これは先読みバイアスが別の場所に現れたものです。時系列では分割も時間順にし、前半で学習し後半でテストします。ランダムにシャッフルしてはいけません。

import numpy as np
from sklearn.linear_model import Ridge
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

def build_pipeline(alpha):
    return Pipeline([("scale", StandardScaler()), ("model", Ridge(alpha=alpha))])

rng = np.random.default_rng(1)
returns = np.zeros(600)
noise = rng.normal(0, 0.01, size=600)
for t in range(1, 600):
    returns[t] = 0.3 * returns[t - 1] + noise[t]

windows = np.lib.stride_tricks.sliding_window_view(returns, 3)
X, y = windows[:-1], returns[3:]
model = build_pipeline(10.0).fit(X[:400], y[:400])                  # chronological split: early rows to train
pred = model.predict(X[400:])                                       # later rows to test
print(round(float(np.mean(np.sign(y[400:]) == np.sign(pred))), 3))  # direction accuracy on unseen data

結果を誠実に評価する

売買シグナルでは方向が重要なことが多いので、簡単な指標は方向の的中率、つまり予測リターンと実際のリターンの符号が一致した日の割合です。50%はコイン投げと同じです。実際のリターン予測器は51〜53%に達する程度で、テストデータで60%が出たなら、発見よりもリークのほうがありそうです。何かを信じる前に、自明な基準（常に上昇と予測する、ゼロと予測する）と比べてください。

import numpy as np

y_true = np.array([0.01, -0.02, 0.015, -0.005, 0.02, -0.01])
y_pred = np.array([0.004, 0.001, 0.002, -0.001, 0.003, 0.002])        # the model is right on 4 of 6 signs

accuracy = float(np.mean(np.sign(y_true) == np.sign(y_pred)))
always_up = float(np.mean(np.sign(y_true) == 1))                      # the trivial baseline: always predict a rise
print(round(accuracy, 3), round(always_up, 3))                        # compare the model with the baseline

• ラグ行列（sliding_window_view）が各目的変数の前のリターンから特徴量を作る。1行の中身は予測時点で既知でなければならない
• Pipeline([("scale", StandardScaler()), ("model", Ridge(alpha=alpha))]) は学習行だけでスケーリングを学ぶ。分割は時間順で、ランダムにはしない
• 方向（符号一致の割合）を自明な基準と比べる。51〜53%が現実的で、60%はリークを疑う。ノイズの多いデータには正則化付き線形モデルが向く

次は金融データの特徴量エンジニアリングです。生のラグ付きリターンよりも良い入力を作ります。

ラボ：採点される内容

• lag_matrix で系列を特徴量に変える25点
• build_pipeline で処理をつなぐ25点
• fit_predict で過去で学習し残りを予測する25点
• direction_accuracy で方向の的中率を測る25点