Creating Multi-Asset Sequences and Training Deep Learning Models
Code Machine Learning for Trading
Summary
This utility module supports deep learning workflows for financial time series across multiple assets. It resolves dataset aliases and loads canonical case study data, then creates sliding-window sequences independently for each symbol. The sequence functions return features, targets, timestamps, and symbol identifiers; a patched variant groups consecutive observations into tokens for transformer inputs. Sorting timestamps within each asset prevents windows from crossing symbol boundaries, while deterministic symbol ordering helps make pooled sequence order reproducible.
The module also connects sequence samples to existing walk-forward splits, offers training with AdamW, gradient clipping, and validation-based early stopping, and standardizes prediction output. These are implementation methods rather than evidence that a model can forecast returns profitably. Results still depend on the dataset, target, feature construction, fold design, and model architecture. The excerpt is a shared code utility, so it explains reusable mechanics but does not report a trading strategy, predictive performance, or safeguards beyond the described time-based splitting and early stopping.
Key ideas
- Sequences are built separately for each asset so that a lookback window never crosses symbols.
- Patched sequences reshape a lookback period into grouped tokens for transformer-style models.
- Stable ordering of assets supports reproducible pooled training data.
- Sequence timestamps can be mapped to the case study's shared walk-forward fold boundaries.
- Training uses validation loss for early stopping and restores the best observed model state.
Tags
Full text
# dl_sequences.py
```py
"""Multi-asset DL utilities for Chapter 13 notebooks.
This module provides canonical functions for:
- Loading case study data via load_modeling_dataset() (the shared Ch11+ API)
- Creating sequences for RNN/Transformer/CNN models
- Training models with early stopping (shared across notebooks)
- Creating time-based cross-validation folds
- Standardized prediction output
Usage:
from dl_sequences import (
# Data loading (thin wrapper around utils/modeling.py)
load_dl_dataset,
# Sequence creation
create_sequences_multi_asset,
create_patched_sequences_multi_asset,
# Training
train_model,
# Cross-validation
create_sequence_folds,
create_expanding_folds,
create_train_val_split,
# Predictions
make_predictions_df,
save_predictions,
)
Dataset IDs:
Use canonical case study IDs (etfs, crypto_perps_funding) or short aliases (crypto, etf).
See DATASET_ALIASES for the mapping.
"""
from pathlib import Path
from typing import Any
import numpy as np
import polars as pl
from utils.modeling import ModelingDataset, load_modeling_dataset
from utils.paths import get_case_study_dir
# =============================================================================
# Dataset Configuration
# =============================================================================
# Canonical case study IDs (match directory names in case_studies/)
CANONICAL_DATASET_IDS = {
"etfs",
"crypto_perps_funding",
"nasdaq100_microstructure",
"cme_futures",
"us_equities_panel",
"us_firm_characteristics",
"fx_pairs",
"sp500_options",
"sp500_equity_option_analytics",
}
# Short aliases and backward-compatible old IDs → canonical IDs
DATASET_ALIASES = {
# Short aliases
"crypto": "crypto_perps_funding",
"etf": "etfs",
"algoseek": "nasdaq100_microstructure",
"futures": "cme_futures",
"wiki": "us_equities_panel",
"fx": "fx_pairs",
# Backward-compatible old IDs
"crypto_premium": "crypto_perps_funding",
"etf_momentum": "etfs",
"nasdaq100_reversal": "nasdaq100_microstructure",
"futures_carry": "cme_futures",
"us_factors": "us_equities_panel",
"fx_momentum": "fx_pairs",
}
# Default primary labels per dataset
DEFAULT_LABELS = {
"etfs": "fwd_ret_21d",
"crypto_perps_funding": "fwd_ret_8h",
"nasdaq100_microstructure": "fwd_ret_15m",
"cme_futures": "fwd_ret_5d",
"us_equities_panel": "fwd_ret_1d",
"us_firm_characteristics": "fwd_ret_1m",
"fx_pairs": "fwd_ret_1d",
"sp500_options": "fwd_ret_dh_10d",
"sp500_equity_option_analytics": "fwd_ret_5d",
}
def resolve_dataset_id(dataset: str) -> str:
"""Resolve a dataset name to its canonical case study ID.
Args:
dataset: Canonical ID (e.g., 'crypto_perps_funding'),
short alias (e.g., 'crypto'), or
old ID (e.g., 'crypto_premium') for backward compatibility.
Returns:
Canonical case study ID
"""
if dataset in CANONICAL_DATASET_IDS:
return dataset
if dataset in DATASET_ALIASES:
return DATASET_ALIASES[dataset]
raise ValueError(
f"Unknown dataset: {dataset!r}. "
f"Valid IDs: {sorted(CANONICAL_DATASET_IDS)}. "
f"Valid aliases: {sorted(DATASET_ALIASES.keys())}"
)
# =============================================================================
# Data Loading (delegates to utils/modeling.py)
# =============================================================================
def load_dl_dataset(
dataset: str,
label: str | None = None,
max_symbols: int = 0,
) -> ModelingDataset:
"""Load a modeling dataset for DL notebooks.
Thin wrapper around load_modeling_dataset() that:
- Resolves short aliases (e.g., 'crypto' → 'crypto_perps_funding')
- Defaults to the primary label if none specified
Args:
dataset: Dataset name (canonical ID or alias)
label: Label file stem (e.g., 'fwd_ret_8h'). None = primary label.
max_symbols: Universe reduction for fast development. 0 = all.
Returns:
ModelingDataset with .dataset, .feature_names, .label_col,
.date_col, .entity_cols, .splits, etc.
"""
dataset_id = resolve_dataset_id(dataset)
if label is None:
label = DEFAULT_LABELS[dataset_id]
mds = load_modeling_dataset(dataset_id, label, max_symbols=max_symbols)
n_entities = mds.dataset[mds.entity_cols[0]].n_unique() if mds.entity_cols else 0
print(
f"Loaded {dataset_id}: {len(mds.dataset):,} rows, "
f"{len(mds.feature_names)} features, "
f"{n_entities} entities, label={mds.label_col}"
)
return mds
# =============================================================================
# Sequence Creation (for RNNs, Transformers, etc.)
# =============================================================================
def create_sequences_multi_asset(
df: pl.DataFrame,
feature_cols: list[str],
target_col: str,
lookback: int,
timestamp_col: str = "timestamp",
symbol_col: str = "symbol",
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
"""Create sequences WITH symbol tracking for multi-asset learning.
Creates sliding window sequences from each symbol independently,
then pools them together while preserving symbol identity.
Args:
df: DataFrame with features, target, canonical time column, and asset
feature_cols: List of feature column names
target_col: Name of target column
lookback: Number of timesteps in each sequence
timestamp_col: Name of date/timestamp column
symbol_col: Name of asset column
Returns:
Tuple of (X, y, timestamps, symbols):
- X: np.ndarray of shape (n_samples, lookback, n_features)
- y: np.ndarray of shape (n_samples,)
- timestamps: np.ndarray of timestamps for each sample
- symbols: np.ndarray of symbol names for each sample
"""
X_list: list[np.ndarray] = []
y_list: list[float] = []
dates_list: list[Any] = []
symbols_list: list[str] = []
# sorted(), not unique() alone: polars does not order the result of unique(),
# and it returns a different order on each run. The pooled row order would then
# differ between runs, which changes mini-batch composition and makes training
# irreproducible even with every seed fixed.
symbols = sorted(df.select(symbol_col).unique().to_series().to_list())
for symbol in symbols:
sym_df = df.filter(pl.col(symbol_col) == symbol).sort(timestamp_col)
if len(sym_df) < lookback + 1:
continue
features = sym_df.select(feature_cols).to_numpy()
targets = sym_df[target_col].to_numpy()
timestamps = sym_df[timestamp_col].to_numpy()
for i in range(lookback, len(features)):
X_list.append(features[i - lookback : i])
y_list.append(float(targets[i]))
dates_list.append(timestamps[i])
symbols_list.append(symbol)
if not X_list:
raise ValueError(f"No sequences created. Check lookback={lookback} vs data size.")
X = np.array(X_list, dtype=np.float32)
y = np.array(y_list, dtype=np.float32)
timestamps_arr = np.array(dates_list)
symbols_arr = np.array(symbols_list)
return X, y, timestamps_arr, symbols_arr
def create_patched_sequences_multi_asset(
df: pl.DataFrame,
feature_cols: list[str],
target_col: str,
lookback: int,
patch_size: int,
timestamp_col: str = "timestamp",
symbol_col: str = "symbol",
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
"""Create patched sequences for Transformer models.
Patching groups consecutive timesteps into tokens for Transformer input.
Args:
df: DataFrame with features, target, canonical time column, and asset
feature_cols: List of feature column names
target_col: Name of target column
lookback: Number of timesteps in each sequence
patch_size: Size of each patch (must divide lookback evenly)
timestamp_col: Name of date/timestamp column
symbol_col: Name of asset column
Returns:
Tuple of (X, y, timestamps, symbols):
- X: np.ndarray of shape (n_samples, n_patches, patch_size * n_features)
- y, timestamps, symbols: as in create_sequences_multi_asset
"""
if lookback % patch_size != 0:
raise ValueError(f"lookback ({lookback}) must be divisible by patch_size ({patch_size})")
X_list: list[np.ndarray] = []
y_list: list[float] = []
dates_list: list[Any] = []
symbols_list: list[str] = []
symbols = df.select(symbol_col).unique().to_series().to_list()
n_patches = lookback // patch_size
for symbol in symbols:
sym_df = df.filter(pl.col(symbol_col) == symbol).sort(timestamp_col)
if len(sym_df) < lookback + 1:
continue
features = sym_df.select(feature_cols).to_numpy()
targets = sym_df[target_col].to_numpy()
timestamps = sym_df[timestamp_col].to_numpy()
for i in range(lookback, len(features)):
seq = features[i - lookback : i]
patched = seq.reshape(n_patches, patch_size * len(feature_cols))
X_list.append(patched)
y_list.append(float(targets[i]))
dates_list.append(timestamps[i])
symbols_list.append(symbol)
if not X_list:
raise ValueError(f"No sequences created. Check lookback={lookback} vs data size.")
X = np.array(X_list, dtype=np.float32)
y = np.array(y_list, dtype=np.float32)
timestamps_arr = np.array(dates_list)
symbols_arr = np.array(symbols_list)
return X, y, timestamps_arr, symbols_arr
# =============================================================================
# Cross-Validation
# =============================================================================
def create_sequence_folds(
timestamps: np.ndarray,
mds: ModelingDataset,
) -> list[dict[str, Any]]:
"""Create CV folds for sequence data using setup.yaml splits.
Maps walk-forward fold boundaries (from mds.splits) to sequence data
using the timestamps from create_sequences_multi_asset(). Ensures
Ch13 DL models use the SAME fold boundaries as Ch11/Ch12.
Args:
timestamps: Array of timestamps from create_sequences_multi_asset()
mds: ModelingDataset with .splits containing fold date boundaries
Returns:
List of fold dicts with 'fold_id', 'train_indices', 'test_indices'
"""
import pandas as pd
seq_timestamps = pd.to_datetime(timestamps)
if seq_timestamps.tz is not None:
seq_timestamps = seq_timestamps.tz_localize(None)
folds = []
for split in mds.splits:
fold_id = split["fold"]
train_end = pd.Timestamp(split["train_end"])
val_start = pd.Timestamp(split["val_start"])
val_end = pd.Timestamp(split["val_end"])
# Normalize timezone awareness to match sequence timestamps
if train_end.tz is not None:
train_end = train_end.tz_localize(None)
val_start = val_start.tz_localize(None)
val_end = val_end.tz_localize(None)
train_mask = seq_timestamps <= train_end
test_mask = (seq_timestamps >= val_start) & (seq_timestamps <= val_end)
train_indices = np.where(train_mask)[0].tolist()
test_indices = np.where(test_mask)[0].tolist()
if len(train_indices) < 100 or len(test_indices) < 50:
continue
folds.append(
{
"fold_id": fold_id,
"train_indices": train_indices,
"test_indices": test_indices,
"train_end": train_end,
"test_start": val_start,
"test_end": val_end,
}
)
return folds
def create_expanding_folds(
n_samples: int,
n_folds: int = 5,
min_train_size: int = 100,
) -> list[dict[str, Any]]:
"""Create simple time-based expanding window folds.
For quick experiments where exact fold matching is not required.
Use create_sequence_folds() for Ch16-compatible results.
Args:
n_samples: Total number of samples
n_folds: Number of folds to create
min_train_size: Minimum training set size
Returns:
List of fold dicts with 'fold_id', 'train_indices', 'test_indices'
"""
fold_size = n_samples // (n_folds + 1)
folds = []
for i in range(n_folds):
train_end = fold_size * (i + 1)
test_start = train_end
test_end = min(train_end + fold_size, n_samples)
if train_end < min_train_size or test_end <= test_start:
continue
folds.append(
{
"fold_id": i,
"train_indices": list(range(train_end)),
"test_indices": list(range(test_start, test_end)),
}
)
return folds
def create_train_val_split(
train_indices: list[int],
val_ratio: float = 0.2,
) -> tuple[list[int], list[int]]:
"""Split training indices into train/validation (temporal split)."""
n = len(train_indices)
val_size = int(n * val_ratio)
train_end = n - val_size
return train_indices[:train_end], train_indices[train_end:]
# =============================================================================
# Prediction Output
# =============================================================================
def make_predictions_df(
timestamps: np.ndarray,
symbols: np.ndarray,
y_true: np.ndarray,
y_score: np.ndarray,
fold_id: int,
model_id: str,
horizon: str,
dataset: str,
time_col: str = "timestamp",
asset_col: str = "symbol",
) -> pl.DataFrame:
"""Create standardized predictions DataFrame."""
dataset_id = resolve_dataset_id(dataset)
n = len(timestamps)
return pl.DataFrame(
{
time_col: timestamps,
asset_col: symbols,
"y_true": y_true.astype(np.float64),
"y_score": y_score.astype(np.float64),
"fold_id": [fold_id] * n,
"model_id": [model_id] * n,
"horizon": [horizon] * n,
"dataset": [dataset_id] * n,
}
)
def save_predictions(preds: pl.DataFrame, dataset: str, model_id: str) -> Path:
"""Save predictions to case study models/deep_learning directory.
Path: case_studies/{dataset_id}/models/deep_learning/{model_id}_predictions.parquet
Args:
preds: Predictions DataFrame
dataset: Dataset name (canonical ID or alias)
model_id: Model identifier
Returns:
Path to saved file
"""
dataset_id = resolve_dataset_id(dataset)
case_dir = get_case_study_dir(dataset_id)
output_dir = case_dir / "models" / "deep_learning"
output_dir.mkdir(parents=True, exist_ok=True)
output_path = output_dir / f"{model_id}_predictions.parquet"
preds.write_parquet(output_path)
print(f"Saved {len(preds):,} predictions to {output_path}")
return output_path
def validate_predictions(
preds: pl.DataFrame,
require_multi_symbol: bool = True,
time_col: str = "timestamp",
asset_col: str = "symbol",
) -> None:
"""Validate prediction DataFrame schema and content."""
required = {
time_col,
asset_col,
"y_true",
"y_score",
"fold_id",
"model_id",
"horizon",
"dataset",
}
missing = required - set(preds.columns)
if missing:
raise AssertionError(f"Missing columns: {missing}")
n_symbols = preds[asset_col].n_unique()
n_folds = preds["fold_id"].n_unique()
if require_multi_symbol and n_symbols <= 1:
raise AssertionError(f"Must have multiple symbols, got {n_symbols}")
if n_folds < 1:
raise AssertionError("Must have at least one fold")
null_counts = preds.select(list(required)).null_count()
total_nulls = null_counts.sum_horizontal().item()
if total_nulls > 0:
raise AssertionError(f"Found {total_nulls} null values in required columns")
datasets = preds["dataset"].unique().to_list()
for ds in datasets:
if ds not in CANONICAL_DATASET_IDS:
print(f"Warning: Non-canonical dataset ID: {ds}")
print(f"Validated: {len(preds):,} rows, {n_symbols} symbols, {n_folds} folds")
def get_output_path(dataset: str, model_id: str) -> Path:
"""Get canonical output path for predictions."""
dataset_id = resolve_dataset_id(dataset)
case_dir = get_case_study_dir(dataset_id, create=False)
return case_dir / "models" / "deep_learning" / f"{model_id}_predictions.parquet"
def load_predictions(dataset: str, model_id: str) -> pl.DataFrame:
"""Load predictions from canonical location."""
path = get_output_path(dataset, model_id)
if not path.exists():
raise FileNotFoundError(f"Predictions not found: {path}")
return pl.read_parquet(path)
# =============================================================================
# Training
# =============================================================================
def train_model(
model,
X_train: np.ndarray,
y_train: np.ndarray,
X_val: np.ndarray,
y_val: np.ndarray,
epochs: int,
lr: float,
batch_size: int,
device,
weight_decay: float = 0.0,
patience: int = 5,
log_interval: int = 5,
) -> dict[str, list[float]]:
"""Train a PyTorch model with early stopping.
Uses AdamW optimizer (equivalent to Adam when weight_decay=0) with
gradient clipping. The model is modified in-place: best weights are
loaded via load_state_dict before returning.
Args:
model: PyTorch nn.Module to train
X_train, y_train: Training arrays (numpy)
X_val, y_val: Validation arrays (numpy)
epochs: Maximum training epochs
lr: Learning rate
batch_size: Mini-batch size
device: torch.device for computation
weight_decay: AdamW weight decay (default 0 = plain Adam behavior)
patience: Early stopping patience (epochs without improvement)
log_interval: Print progress every N epochs
Returns:
Dict with 'train_loss' and 'val_loss' lists (per-epoch averages)
"""
import torch
import torch.nn as nn
criterion = nn.MSELoss()
optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=weight_decay)
X_train_t = torch.FloatTensor(X_train).to(device)
y_train_t = torch.FloatTensor(y_train).to(device)
X_val_t = torch.FloatTensor(X_val).to(device)
y_val_t = torch.FloatTensor(y_val).to(device)
best_val_loss = float("inf")
best_state = None
patience_counter = 0
history = {"train_loss": [], "val_loss": []}
for epoch in range(epochs):
model.train()
indices = torch.randperm(len(X_train_t))
epoch_loss = 0.0
n_batches = 0
for i in range(0, len(indices), batch_size):
batch_idx = indices[i : i + batch_size]
optimizer.zero_grad()
preds = model(X_train_t[batch_idx])
loss = criterion(preds, y_train_t[batch_idx])
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
epoch_loss += loss.item()
n_batches += 1
avg_train = epoch_loss / max(n_batches, 1)
history["train_loss"].append(avg_train)
model.eval()
with torch.no_grad():
val_sum = 0.0
val_count = 0
for i in range(0, len(X_val_t), batch_size):
xb = X_val_t[i : i + batch_size]
yb = y_val_t[i : i + batch_size]
val_preds = model(xb)
batch_loss = criterion(val_preds, yb).item()
n_batch = len(xb)
val_sum += batch_loss * n_batch
val_count += n_batch
val_loss = val_sum / max(val_count, 1)
history["val_loss"].append(val_loss)
if val_loss < best_val_loss:
best_val_loss = val_loss
best_state = {k: v.cpu().clone() for k, v in model.state_dict().items()}
patience_counter = 0
else:
patience_counter += 1
if (epoch + 1) % log_interval == 0 or epoch == 0:
print(f" Epoch {epoch + 1}/{epochs}: val_loss={val_loss:.6f}")
if patience_counter >= patience:
print(f" Early stopping at epoch {epoch + 1}")
break
if best_state is not None:
model.load_state_dict(best_state)
return history
```Shown in full with attribution under the source's licence. Licence: MIT
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.