時系列クロスバリデーションと訓練/検証分割

1
import numpy as np
2
import pandas as pd
3

4
rng = np.random.default_rng(seed=43)
5
n_days = 1500
6

7
ret = rng.normal(loc=0.0003, scale=0.012, size=n_days)
8
price = 1500 * np.exp(np.cumsum(ret))
9

10
df = pd.DataFrame({
11
    "C": price,
12
    "ret": np.r_[np.nan, np.diff(np.log(price))],
13
}, index=pd.date_range("2020-01-02", periods=n_days, freq="B"))
14

15
split_idx = int(len(df) * 0.7)
16
train = df.iloc[:split_idx]
17
test = df.iloc[split_idx:]
18
print(f"train: {train.index.min().date()} 〜 {train.index.max().date()}  ({len(train)} 行)")
19
print(f"test:  {test.index.min().date()} 〜 {test.index.max().date()}  ({len(test)} 行)")

1
from sklearn.model_selection import TimeSeriesSplit
2

3
tscv = TimeSeriesSplit(n_splits=5)
4

5
for fold, (tr_idx, va_idx) in enumerate(tscv.split(df)):
6
    tr_period = (df.index[tr_idx[0]].date(), df.index[tr_idx[-1]].date())
7
    va_period = (df.index[va_idx[0]].date(), df.index[va_idx[-1]].date())
8
    print(f"fold {fold}: train {tr_period} -> valid {va_period}")

1
tscv_gap = TimeSeriesSplit(n_splits=5, gap=20, max_train_size=500)
2

3
for fold, (tr_idx, va_idx) in enumerate(tscv_gap.split(df)):
4
    print(f"fold {fold}: train [{df.index[tr_idx[0]].date()} .. {df.index[tr_idx[-1]].date()}]  "
5
          f"valid [{df.index[va_idx[0]].date()} .. {df.index[va_idx[-1]].date()}]")

1
def walk_forward_splits(
2
    n: int,
3
    train_size: int,
4
    test_size: int,
5
    step: int | None = None,
6
):
7
    """ローリング窓で (train_idx, test_idx) を生成する。"""
8
    step = step or test_size
9
    start = 0
10
    while start + train_size + test_size <= n:
11
        tr = np.arange(start, start + train_size)
12
        te = np.arange(start + train_size, start + train_size + test_size)
13
        yield tr, te
14
        start += step
15

16

17
for fold, (tr_idx, va_idx) in enumerate(walk_forward_splits(len(df), train_size=400, test_size=80)):
18
    print(f"fold {fold}: train [{df.index[tr_idx[0]].date()} .. {df.index[tr_idx[-1]].date()}]  "
19
          f"valid [{df.index[va_idx[0]].date()} .. {df.index[va_idx[-1]].date()}]")

1
from itertools import product
2

3
TRADING_DAYS = 252
4
COST = 0.001
5

6
def evaluate_sma_pair(
7
    df_local: pd.DataFrame,
8
    short_w: int,
9
    long_w: int,
10
    cost: float = COST,
11
) -> float:
12
    """SMA クロス戦略の年率シャープを返す。"""
13
    sma_s = df_local["C"].rolling(short_w).mean()
14
    sma_l = df_local["C"].rolling(long_w).mean()
15
    signal = (sma_s > sma_l).astype(int)
16
    position = signal.shift(1)
17
    daily_ret = df_local["C"].pct_change()
18

19
    pos_change = position.diff().abs().fillna(position)
20
    strat_ret = (position * daily_ret - pos_change * cost).dropna()
21
    if strat_ret.std(ddof=0) == 0:
22
        return float("nan")
23
    return (strat_ret.mean() / strat_ret.std(ddof=0)) * np.sqrt(TRADING_DAYS)
24

25

26
candidates = [(5, 30), (25, 75), (50, 200)]
27
results: list[dict] = []
28

29
for fold, (tr_idx, va_idx) in enumerate(walk_forward_splits(len(df), train_size=500, test_size=120)):
30
    train_df = df.iloc[tr_idx]
31
    valid_df = df.iloc[va_idx]
32
    # 訓練区間で最良パラメータを選ぶ
33
    best = max(candidates, key=lambda p: evaluate_sma_pair(train_df, *p) or -np.inf)
34
    # 検証区間で評価
35
    va_score = evaluate_sma_pair(valid_df, *best)
36
    results.append({"fold": fold, "best_pair": best, "valid_sharpe": va_score})
37

38
print(pd.DataFrame(results))

1
from sklearn.linear_model import LogisticRegression
2
from sklearn.preprocessing import StandardScaler
3
from sklearn.pipeline import Pipeline
4
from sklearn.model_selection import cross_val_score
5

6
# 例: 過去 5 日のリターン平均で「翌日の上昇」を当てる
7
df["feat"] = df["ret"].rolling(5).mean().shift(1)
8
df["label"] = (df["ret"].shift(-1) > 0).astype(int)
9

10
X = df[["feat"]].dropna()
11
y = df.loc[X.index, "label"]
12

13
pipe = Pipeline([
14
    ("scaler", StandardScaler()),
15
    ("clf", LogisticRegression()),
16
])
17

18
scores = cross_val_score(pipe, X.values, y.values, cv=TimeSeriesSplit(n_splits=5), scoring="accuracy")
19
print("scores:", scores)
20
print("mean:  ", scores.mean())

[ ] 検証区間が常に訓練区間より「未来」になっている
[ ] gap を入れる必要があるラベルか確認した(N 日先のリターン予測など)
[ ] 前処理(標準化・補完)は Pipeline でカプセル化されている
[ ] パラメータ選定は訓練区間内で完結している
[ ] 検証区間が 1 つではなく、複数 fold の平均で評価している
[ ] fold 間のばらつき(標準偏差)も併記している

入力:
- X: pd.DataFrame, 説明変数(全て float)
- y: pd.Series,    目的変数(0/1)
- models: dict[str, sklearn estimator]

次の処理を行う関数 compare_time_series_cv(X, y, models, n_splits=5, gap=10) を
書いてください。

仕様:
- TimeSeriesSplit(n_splits=n_splits, gap=gap) で時系列順 CV
- 各モデルを Pipeline([("scaler", StandardScaler()), ("model", model)]) でラップ
- accuracy / roc_auc を fold ごとに計算し、平均と標準偏差を集計
- 戻り値: pd.DataFrame(列: model, acc_mean, acc_std, auc_mean, auc_std)

要件:
- pandas 2.2 / sklearn 1.4 系
- docstring を日本語で書く
- 標準偏差は ddof=0