quantlab.model.torch_data.SymbolSequenceDataset#

class quantlab.model.torch_data.SymbolSequenceDataset(panel, bars, window_bars, training)[source]#

Bases: Dataset

One item per (bar, symbol) cell: the symbol’s last window_bars bars.

Qlib’s sequence models (LSTM, GRU, ALSTM, Transformer) draw random (timestamp, symbol) samples; this is that sample shape. Item i is a Batch for one cell (t, s) with x shaped [N, F], y and y_raw shaped [L], a scalar mask and scalar where indices, so PyTorch’s default collation batches B items to x [B, N, F] with mask and where shaped [B]. Rows before the panel’s first bar are NaN. In training the dataset holds the cells of bars with a valid training target; in evaluation it holds every present cell of bars, so prediction covers the whole cross-section. Cells are ordered by bar, then by symbol.

The training target was computed per bar over the whole cross-section before the dataset was built, so a batch mixing bars still sees each bar’s cross-sectional target. __getitems__ gathers a whole batch of windows with one indexing call on the panel. window_bars=1 gives row samples, x shaped [B, 1, F].

Batch it with a batch_size (Qlib uses 800); batch_size=None would feed single cells. batch_loader over IndexBatches is faster: it gathers each batch whole, where the default collation splits the gathered block into items and stacks them back.

Parameters:

panelTrainingPanel: The panel to read.
barsarray-like of int: Positions of the bars on the panel’s time axis.
window_barsint: N, bars per window; at least 1.
trainingbool: Whether the dataset feeds training steps.

Raises:

ValueError: If window_bars is below 1.

Examples

>>> panel = TrainingPanel.from_arrays(
...     np.arange(6.0).reshape(3, 2, 1), timestamps=np.arange(3),
...     symbols=np.array(["A", "B"]),
... )
>>> dataset = SymbolSequenceDataset(panel, bars=[0, 2], window_bars=2, training=False)
>>> len(dataset), dataset[1].x[:, 0], [int(i) for i in dataset[1].where]
(4, tensor([nan, 1.]), [0, 1])
>>> from torch.utils.data import DataLoader
>>> batch = next(iter(DataLoader(dataset, batch_size=4)))
>>> tuple(batch.x.shape), batch.where[0].tolist()
((4, 2, 1), [0, 0, 2, 2])

times#: The timestamp index and symbol index of every item, [len(self)] each.