シャープレシオに基づく検証と早期終了を用いたトレードモデルの学習

コード Machine Learning for Trading

サマリー

この学習用ユーティリティでは、DeePM型モデルの最適化ループを説明します。AdamWでモデルパラメータを更新する際、ロバストなシャープレシオ重視の損失関数に、先行リターン、ボラティリティ調整、マスク、取引コスト、設定可能なウォームアップ期間とソフトミン制御を渡します。有限でない損失はスキップし、各最適化ステップの前に勾配をクリップできます。

指定した間隔で、複数バッチをまとめた検証リターンを使ってモデルを評価します。このユーティリティは、学習目的関数とシャープレシオ指標を検証シャープレシオとともに記録し、検証スコアが最良の状態を保持します。また、設定したウォームアップ期間の後、指数平滑化した検証指標を使って早期終了を適用します。統合指標はバッチ間の有効なリターンを連結し、リターンのばらつきに対する平均を年率換算します。有効な観測数が少なすぎる場合、スコアは返しません。これは実装の指針であり、実証的な証拠ではありません。データセット、結果、取引実績は示されておらず、検証スコアは選択したデータ分割と設定に左右されます。

主なアイデア

学習目的関数には、ボラティリティ調整、マスク、任意のコスト調整を含むロバストなシャープレシオ損失が組み込まれます。
損失値が有限でない場合、最適化ステップはスキップされます。
設定された最大ノルムが正の場合、勾配クリッピングが適用されます。
検証シャープレシオはローダー全体のリターンをまとめて計算し、短すぎるサンプルや空のサンプルは利用不可として扱います。
早期終了では検証スコアを指数移動平均で平滑化し、設定可能なウォームアップ期間を設けます。
返されるモデル状態は、観測された検証シャープレシオが最も高いものです。

# train.py


```py
"""Training utilities for DeePM-style models."""

from __future__ import annotations

from dataclasses import dataclass

import torch
from torch import nn
from torch.optim import AdamW
from torch.utils.data import DataLoader

from .configs import TrainingConfig, validate_training_config
from .dataset import StaticAssetMetadata
from .losses import robust_sharpe_loss
from .utils import set_seed


@dataclass(frozen=True, slots=True)
class TrainHistory:
    steps: list[int]
    train_objective: list[float]
    train_sharpe_pool: list[float]
    val_sharpe_pool: list[float]


class EarlyStopping:
    """Validation-metric early stopping with EMA smoothing."""

    def __init__(
        self,
        *,
        alpha: float,
        min_delta: float,
        patience: int,
        burn_in_iters: int,
    ) -> None:
        self.alpha = float(alpha)
        self.min_delta = float(min_delta)
        self.patience = int(patience)
        self.burn_in_iters = int(burn_in_iters)

        self._ema: float | None = None
        self._best_ema: float = float("-inf")
        self._bad_count: int = 0

    def update(self, value: float, step: int) -> tuple[bool, float, float]:
        """Update state. Returns (stop, ema, best_ema)."""
        if self._ema is None:
            self._ema = float(value)
        else:
            self._ema = self.alpha * float(value) + (1.0 - self.alpha) * self._ema

        if step < self.burn_in_iters:
            return False, self._ema, self._best_ema

        if self._ema >= self._best_ema + self.min_delta:
            self._best_ema = self._ema
            self._bad_count = 0
        else:
            self._bad_count += 1

        return self._bad_count >= self.patience, self._ema, self._best_ema


@torch.no_grad()
def evaluate_pooled_sharpe(
    model: nn.Module,
    loader: DataLoader,
    *,
    static_meta: StaticAssetMetadata,
    cfg: TrainingConfig,
    device: torch.device,
) -> float:
    """Compute pooled Sharpe over the entire loader."""
    model.eval()
    returns: list[torch.Tensor] = []

    for x, y, v, m in loader:
        x = x.to(device=device, dtype=torch.float32)
        y = y.to(device=device, dtype=torch.float32)
        v = v.to(device=device, dtype=torch.float32)
        m = m.to(device=device, dtype=torch.float32)

        p = model(
            x,
            mask=m,
            asset_ids=static_meta.asset_ids.to(device),
            group_ids=None if static_meta.group_ids is None else static_meta.group_ids.to(device),
            costs=None if static_meta.costs is None else static_meta.costs.to(device),
        )

        out = robust_sharpe_loss(
            p=p,
            y_fwd1=y,
            vol_scale=v,
            mask=m,
            costs=None if static_meta.costs is None else static_meta.costs.to(device),
            burn_in=cfg.burn_in,
            gamma_cost=cfg.gamma_cost,
            annualization_factor=cfg.annualization_factor,
            eps=cfg.sharpe_eps,
            tau=cfg.softmin_tau,
            lambda_soft=cfg.softmin_lambda,
        )
        returns.append(out.net_returns[:, cfg.burn_in :].reshape(-1).cpu())

    if not returns:
        return float("nan")

    r = torch.cat(returns)
    r = r[~torch.isnan(r)]
    if r.numel() < 10:
        return float("nan")

    mu = r.mean()
    var = r.var(unbiased=False)
    sharpe = (cfg.annualization_factor**0.5) * mu / torch.sqrt(var + cfg.sharpe_eps)
    return float(sharpe.item())


def train_model(
    model: nn.Module,
    *,
    train_loader: DataLoader,
    val_loader: DataLoader,
    static_meta: StaticAssetMetadata,
    cfg: TrainingConfig,
) -> tuple[dict[str, torch.Tensor], TrainHistory]:
    """Train a DeePM-style model and return the best state dict."""
    validate_training_config(cfg)
    set_seed(cfg.seed)

    device = torch.device(cfg.device)
    model.to(device)

    optimizer = AdamW(model.parameters(), lr=cfg.learning_rate, weight_decay=cfg.weight_decay)

    stopper = EarlyStopping(
        alpha=cfg.metric_ema_alpha,
        min_delta=cfg.metric_min_delta,
        patience=cfg.early_stopping_patience,
        burn_in_iters=cfg.early_stopping_burn_in_iters,
    )

    history = TrainHistory(steps=[], train_objective=[], train_sharpe_pool=[], val_sharpe_pool=[])
    best_state: dict[str, torch.Tensor] = {}
    best_val_sharpe: float = float("-inf")
    train_iter = iter(train_loader)

    for step in range(1, cfg.max_iters + 1):
        model.train()

        try:
            x, y, v, m = next(train_iter)
        except StopIteration:
            train_iter = iter(train_loader)
            x, y, v, m = next(train_iter)

        x = x.to(device=device, dtype=torch.float32)
        y = y.to(device=device, dtype=torch.float32)
        v = v.to(device=device, dtype=torch.float32)
        m = m.to(device=device, dtype=torch.float32)

        optimizer.zero_grad(set_to_none=True)

        p = model(
            x,
            mask=m,
            asset_ids=static_meta.asset_ids.to(device),
            group_ids=None if static_meta.group_ids is None else static_meta.group_ids.to(device),
            costs=None if static_meta.costs is None else static_meta.costs.to(device),
        )

        out = robust_sharpe_loss(
            p=p,
            y_fwd1=y,
            vol_scale=v,
            mask=m,
            costs=None if static_meta.costs is None else static_meta.costs.to(device),
            burn_in=cfg.burn_in,
            gamma_cost=cfg.gamma_cost,
            annualization_factor=cfg.annualization_factor,
            eps=cfg.sharpe_eps,
            tau=cfg.softmin_tau,
            lambda_soft=cfg.softmin_lambda,
        )

        if torch.isnan(out.loss) or torch.isinf(out.loss):
            optimizer.zero_grad(set_to_none=True)
            continue

        out.loss.backward()
        if cfg.max_grad_norm > 0:
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=cfg.max_grad_norm)
        optimizer.step()

        if step % cfg.eval_every != 0:
            continue

        val_sharpe = evaluate_pooled_sharpe(
            model, val_loader, static_meta=static_meta, cfg=cfg, device=device
        )

        stop, _ema, _best_ema = stopper.update(val_sharpe, step)

        history.steps.append(step)
        history.train_objective.append(float(out.objective.item()))
        history.train_sharpe_pool.append(float(out.sharpe_pool.item()))
        history.val_sharpe_pool.append(val_sharpe)

        if val_sharpe > best_val_sharpe:
            best_val_sharpe = val_sharpe
            best_state = {k: v.detach().cpu().clone() for k, v in model.state_dict().items()}

        if stop:
            break

    return best_state, history

```

出典を明記したうえで、ライセンスに従って全文を掲載しています。ライセンス: MIT

この要約は原文をもとにStratmillのリサーチエージェントが作成したもので、出典の複製ではありません。

シャープレシオに基づく検証と早期終了を用いたトレードモデルの学習

サマリー

主なアイデア

タグ