Как настроить валидацию временных рядов?

Как настроить валидацию временных рядов

Валидация временных рядов существенно отличается от стандартной кросс-валидации для табличных данных. Главное отличие: нельзя случайно перемешивать данные, потому что это нарушает временную последовательность, на которой строится прогноз. Я расскажу о правильных подходах на примере реальных задач.

Почему обычная кросс-валидация не работает

Проблема: K-fold кросс-валидация выбирает данные случайно из всего датасета, что означает, что модель может видеть будущее:

from sklearn.model_selection import KFold
import pandas as pd

# Создаём временной ряд
timeseries = pd.DataFrame({
    time: pd.date_range(2023-01-01, periods=1000, freq=D),
    value: np.random.randn(1000).cumsum()
})

# ПЛОХО: K-fold перемешивает данные
kf = KFold(n_splits=5, shuffle=True)
for train_idx, test_idx in kf.split(timeseries):
    # train_idx и test_idx случайные!
    # Модель видит данные из будущего и прошлого вперемешку
    # Результаты будут оптимистичными и неправдивыми
    pass

Правильный подход 1: Time Series Split

Растущее окно (expanding window) используется, когда нужно обучать на всё растущем наборе исторических данных:

from sklearn.model_selection import TimeSeriesSplit

# TimeSeriesSplit гарантирует, что тестовый набор всегда после обучающего
tscv = TimeSeriesSplit(n_splits=5)

for train_idx, test_idx in tscv.split(timeseries):
    X_train, X_test = X[train_idx], X[test_idx]
    y_train, y_test = y[train_idx], y[test_idx]
    
    # Обучение всегда на прошлых данных, тестирование на будущих
    model.fit(X_train, y_train)
    score = model.score(X_test, y_test)

# Пример разбиения для 5 фолдов:
# Fold 1: train=[0:200],    test=[200:250]
# Fold 2: train=[0:400],    test=[400:450]
# Fold 3: train=[0:600],    test=[600:650]
# Fold 4: train=[0:800],    test=[800:850]
# Fold 5: train=[0:1000],   test=[1000:1050]

Правильный подход 2: Walk-Forward Validation

Ролирующее окно фиксированного размера часто реалистичнее, потому что имитирует действительный сценарий, когда модель переучивается на фиксированном историческом окне:

def walk_forward_validation(X, y, train_size=400, test_size=50, step=50):
    """
    Walk-Forward validation с фиксированным размером обучающего окна
    """
    n_samples = len(X)
    scores = []
    
    for i in range(0, n_samples - train_size - test_size, step):
        # Обучающий набор: фиксированное окно
        train_start = i
        train_end = i + train_size
        
        # Тестовый набор: сразу после обучающего
        test_start = train_end
        test_end = test_start + test_size
        
        X_train, X_test = X[train_start:train_end], X[test_start:test_end]
        y_train, y_test = y[train_start:train_end], y[test_start:test_end]
        
        model = RandomForestRegressor(random_state=42)
        model.fit(X_train, y_train)
        score = model.score(X_test, y_test)
        scores.append(score)
    
    return np.mean(scores), np.std(scores)

# Использование
mean_score, std_score = walk_forward_validation(
    X, y,
    train_size=400,  # 400 дней истории
    test_size=50,    # 50 дней прогноза
    step=50          # Скользящий на 50 дней каждой итерации
)
print(f"Walk-forward R2: {mean_score:.4f} +/- {std_score:.4f}")

Правильный подход 3: Nested Cross-Validation

Для гиперпараметр-тюнинга на временных рядах используй вложенную валидацию:

from sklearn.model_selection import TimeSeriesSplit, GridSearchCV
from sklearn.ensemble import RandomForestRegressor

# Внешний loop: оценка итогового качества
outer_cv = TimeSeriesSplit(n_splits=5)

scores = []

for train_idx, test_idx in outer_cv.split(timeseries):
    X_train_outer, X_test_outer = X[train_idx], X[test_idx]
    y_train_outer, y_test_outer = y[train_idx], y[test_idx]
    
    # Внутренний loop: гиперпараметр-тюнинг на обучающих данных
    inner_cv = TimeSeriesSplit(n_splits=3)
    
    param_grid = {
        n_estimators: [50, 100, 200],
        max_depth: [5, 10, 15]
    }
    
    grid_search = GridSearchCV(
        RandomForestRegressor(random_state=42),
        param_grid,
        cv=inner_cv,
        scoring=neg_mean_squared_error
    )
    
    # Гиперпараметр-тюнинг только на обучающих данных
    grid_search.fit(X_train_outer, y_train_outer)
    
    # Тестирование на новых данных (внешний fold)
    score = grid_search.score(X_test_outer, y_test_outer)
    scores.append(score)

print(f"Nested CV R2: {np.mean(scores):.4f} +/- {np.std(scores):.4f}")

Чувствительность к началу: Gap вставка

Добавь gap между обучением и тестированием, чтобы избежать утечки информации (leakage):

def walk_forward_with_gap(
    X, y,
    train_size=400,
    gap=10,           # 10 дней не используем
    test_size=50,
    step=50
):
    """
    Walk-Forward validation с разделением между train и test
    """
    n_samples = len(X)
    scores = []
    
    for i in range(0, n_samples - train_size - gap - test_size, step):
        train_start = i
        train_end = i + train_size
        
        # Gap для предотвращения утечки
        gap_start = train_end
        gap_end = gap_start + gap
        
        # Тестовый набор после gap
        test_start = gap_end
        test_end = test_start + test_size
        
        X_train = X[train_start:train_end]
        y_train = y[train_start:train_end]
        X_test = X[test_start:test_end]
        y_test = y[test_start:test_end]
        
        model = RandomForestRegressor(random_state=42)
        model.fit(X_train, y_train)
        score = model.score(X_test, y_test)
        scores.append(score)
    
    return np.mean(scores)

mean_score = walk_forward_with_gap(X, y)
print(f"Walk-forward with gap R2: {mean_score:.4f}")

Реальный пример: прогнозирование цены акций

import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.preprocessing import StandardScaler

# Загрузим исторические данные
prices = pd.read_csv(stock_prices.csv, parse_dates=[date])
prices = prices.sort_values(by=date).reset_index(drop=True)

# Создаём признаки на основе лагов
def create_features(data, n_lags=10):
    X = []
    y = []
    
    for i in range(n_lags, len(data) - 1):
        # Используем последние n_lags дней как признаки
        X.append(data[i-n_lags:i])
        # Прогнозируем следующий день
        y.append(data[i+1])
    
    return np.array(X), np.array(y)

X, y = create_features(prices[close].values, n_lags=30)

# Walk-forward validation
train_size = 200  # 200 торговых дней (примерно 1 год)
test_size = 20    # 20 торговых дней прогноза
gap = 5           # 5 дней gap

mean_rmse = []

for i in range(0, len(X) - train_size - gap - test_size, test_size):
    X_train = X[i:i+train_size]
    y_train = y[i:i+train_size]
    
    X_test = X[i+train_size+gap:i+train_size+gap+test_size]
    y_test = y[i+train_size+gap:i+train_size+gap+test_size]
    
    # Масштабируем признаки
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    
    # Обучаем модель
    model = GradientBoostingRegressor(
        n_estimators=100,
        max_depth=5,
        learning_rate=0.1
    )
    model.fit(X_train_scaled, y_train)
    
    # Оцениваем
    y_pred = model.predict(X_test_scaled)
    rmse = np.sqrt(np.mean((y_pred - y_test) ** 2))
    mean_rmse.append(rmse)

print(f"Mean RMSE: {np.mean(mean_rmse):.4f}")
print(f"Std RMSE: {np.std(mean_rmse):.4f}")

Сравнение подходов

Для большинства практических задач рекомендую Walk-Forward Validation с gap = 5-10 дней — это лучший баланс между реалистичностью и производительностью.