Entraîner des modèles de trading neuronaux avec validation fondée sur le ratio de Sharpe et arrêt anticipé
Code Machine Learning for Trading
Résumé
Cet utilitaire d’entraînement décrit une boucle d’optimisation pour un modèle de type DeePM. Il utilise AdamW pour mettre à jour les paramètres du modèle à partir d’une fonction de perte robuste orientée vers le ratio de Sharpe, en transmettant les rendements futurs, l’ajustement de volatilité, les masques, les coûts de transaction et les paramètres définis de période de chauffe et de soft-min. Les pertes non finies sont ignorées et les gradients peuvent être écrêtés avant chaque étape de l’optimiseur.
À intervalles prévus, le modèle est évalué sur les rendements de validation regroupés. L’utilitaire consigne l’objectif d’entraînement et les mesures du ratio de Sharpe avec le ratio de Sharpe de validation, conserve l’état associé au meilleur score de validation et applique un arrêt anticipé à partir d’une mesure de validation lissée exponentiellement, après une période de chauffe définie. La mesure regroupée concatène les rendements valides de plusieurs lots et annualise la moyenne par rapport à la dispersion des rendements. Elle ne renvoie aucun score s’il y a trop peu d’observations valides. Il s’agit de conseils d’implémentation et non de preuves empiriques : aucun jeu de données, résultat ou performance de trading n’est spécifié, et le score de validation dépend toujours de la division des données et de la configuration choisies.
Idées clés
- L’objectif d’entraînement intègre une fonction de perte robuste fondée sur le ratio de Sharpe, l’ajustement de volatilité, des masques et des ajustements facultatifs des coûts.
- Les valeurs de perte non finies entraînent l’omission d’une étape d’optimisation.
- La troncature des gradients est appliquée lorsque la norme maximale configurée est positive.
- Le ratio de Sharpe de validation est calculé à partir des rendements regroupés sur l’ensemble du chargeur ; les échantillons courts ou vides sont considérés comme indisponibles.
- L’arrêt anticipé lisse les scores de validation au moyen d’une moyenne exponentielle et autorise une période de chauffe définie.
- L’état du modèle renvoyé est celui qui a obtenu le meilleur ratio de Sharpe de validation observé.
Étiquettes
Texte intégral
# 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
```Reproduit dans son intégralité avec attribution, conformément à la licence de la source. Licence: MIT
Ce résumé a été rédigé par l’agent de recherche de Stratmill à partir de la source originale ; il n’en est pas une copie.