Машинное обучение и Data Science (Часть 48): Насколько важны трансформеры для трейдинга?
44
1
Содержание
- Что такое модель-трансформер?
- Общая информация
- Архитектура модели-трансформера
- Модель Temporal Fusion Transformer (TFT)
- Подготовка данных для модели TFT
- Обучение модели Temporal Fusion Transformer
- Поиск оптимальных параметров для модели Temporal Fusion Transformer
- Создание торгового робота на основе модели TFT
- Заключение
Что такое модель-трансформер?
В области глубокого обучения трансформер представляет собой архитектуру искусственной нейронной сети, основанную на механизме многоголового внимания. Эта архитектура была впервые представлена в статье 2017 года "Attention Is All You Need", авторами которой выступили восемь исследователей из Google. В статье была представлена новая модель, построенная на основе механизма внимания, изначально предложенного Бахданау и соавт. в 2014 году; эта работа широко признается одним из основополагающих вкладов в современный искусственный интеллект.
Трансформеры добились выдающихся успехов в самых разных областях. В области обработки естественного языка (NLP) они продемонстрировали свои возможности в машинном переводе, анализе тональности и суммаризации текста. При расширении области применения на обработку изображений трансформеры были успешно адаптированы для задач компьютерного зрения, продемонстрировав высокую эффективность в классификации изображений и обнаружении объектов. Кроме того, их эффективность распространяется на анализ временных рядов: уникальная способность учитывать дальние зависимости делает их пригодными для прогнозирования последовательных данных, что наглядно демонстрируется в таких задачах, как прогнозирование цен акций или погодных явлений.
Термин "трансформер", используемый в данной статье, относится к семейству архитектур, основанных на механизме внимания, а не к какой-либо одной конкретной модели.
Трансформеры – это семейство архитектур нейронных сетей, построенных на основе механизмов внимания, а не рекуррентности. В отличие от традиционных последовательных моделей, они позволяют каждому элементу последовательности напрямую учитывать все остальные элементы с помощью механизма внимания, что обеспечивает эффективное моделирование дальних зависимостей и параллельные вычисления во время обучения.
Хотя изначально варианты трансформеров разрабатывались для обработки естественного языка, впоследствии их адаптировали для задач компьютерного зрения и анализа временных рядов, часто с архитектурными изменениями для учета предметно-специфических ограничений, таких как известные будущие входные данные, многогоризонтное прогнозирование и ограниченная доступность данных.
Общая информация
В области обработки естественного языка (NLP) широкое распространение получили последовательные модели, такие как RNN и LSTM. Эти модели продемонстрировали высокую эффективность при решении таких задач, как машинный перевод и языковое моделирование. Однако при обработке последовательных входных данных им присущи внутренние структурные ограничения.
Основные ограничения заключаются в следующем:
Структура существующей последовательной модели
Длительное время вычислений
В традиционных последовательных моделях информация из предыдущих временных шагов, как правило, используется в качестве входных данных для интерпретации текущего временного шага. Как показано на примере на рисунке 1, последовательная зависимость формирует структуру, в которой скрытые признаки обрабатываются по одному шагу за раз. Следовательно, время вычислений линейно зависит от длины входных данных, что приводит к снижению эффективности по мере увеличения длины последовательностей.
Сложность учета дальних зависимостей
В силу своей последовательной природы модели рекуррентных нейронных сетей (RNN) с трудом улавливают дальние зависимости между элементами последовательности. Хотя для решения этой проблемы были предложены такие архитектуры, как LSTM и GRU, они по-прежнему используют представления скрытого состояния фиксированного размера, что ограничивает их способность эффективно обрабатывать чрезвычайно длинные последовательности.
Разработчики Google предложили модель-трансформер, в которой полностью исключена рекурсия, а вместо этого используется механизм внимания для установления глобальных зависимостей между элементами входа и выхода. Эта архитектура была разработана для эффективного устранения вышеупомянутых ограничений и повышения масштабируемости и эффективности модели, хотя в современных вариантах для конкретных областей, таких как временные ряды, рекуррентность может вводиться снова.
Архитектура модели-трансформера
Механизм самовнимания
В отличие от традиционных рекуррентных нейронных сетей (RNN) и нейронных сетей с долговременной краткосрочной памятью (LSTM), трансформеры используют механизм самовнимания, который позволяет модели оценивать значимость различных частей входной последовательности при построении прогнозов.
Параллелизация
Трансформеры допускают параллелизацию на этапе обучения, хотя некоторые их варианты остаются последовательными на этапе инференса, что делает их более эффективными, чем последовательные модели, такие как RNN. Благодаря этому время обучения сокращается.
Структура энкодера-декодера
Модель состоит из энкодера и декодера. Энкодер обрабатывает входную последовательность, извлекая контекстную информацию, а декодер генерирует выходную последовательность.
Многоголовое внимание
Механизм самовнимания дополняется несколькими головами внимания, что позволяет модели одновременно сосредоточиваться на различных аспектах входной последовательности. Это повышает способность модели выявлять сложные взаимосвязи.
В отличие от скрытых состояний фиксированного размера, механизм внимания позволяет модели динамически выбирать релевантную информацию из любой точки последовательности.
Позиционное кодирование
Трансформеры по своей сути не понимают порядок входной последовательности. Для решения этой проблемы к входным эмбеддингам добавляются позиционные кодировки, которые содержат информацию о положениях токенов в последовательности.
Повторим: термин "трансформеры", который несколько раз используется в этой статье, не относится к какой-то одной конкретной модели; архитектура разных типов моделей-трансформеров может значительно различаться. Например:
| Модель | Архитектура | Основное назначение / область применения |
|---|---|---|
| Исходный трансформер | Энкодер-декодер. | Машинный перевод, задачи sequence-to-sequence. |
| BERT (двунаправленные энкодерные представления на основе трансформеров) | Только энкодер. | Понимание текста (классификация, распознавание именованных сущностей, эмбеддинги) |
| GPT (генеративный предобученный трансформер) | Только декодер. | Генерация текста, языковое моделирование, генерация кода. |
| ViT (визуальный трансформер) | Только энкодер. | Классификация изображений, обучение представлениям для компьютерного зрения. |
| TFT (Temporal Fusion Transformer) | Гибридная модель (LSTM + механизм внимания) | Прогнозирование временных рядов, финансовые и бизнес-данные. |
Более подробную информацию о том, как устроены трансформеры, и другие сведения можно найти в разделе "Ссылки" в конце этой статьи.
Среди всех моделей-трансформеров именно Temporal Fusion Transformer (TFT) подходит для прогнозирования временных рядов. В этой статье мы расскажем, что это такое, как создать такую модель и как с ее помощью делать прогнозы по рынку.
Модель Temporal Fusion Transformer (TFT)
По определению:
Temporal Fusion Transformer (TFT) – это современная модель глубокого обучения на основе механизма внимания, разработанная для многогоризонтного прогнозирования временных рядов.
Эта модель, представленная Google и исследователями Оксфордского университета еще в 2021 году, предназначена для получения высокоточных и интерпретируемых прогнозов за счет объединения компонентов энкодера-декодера на основе LSTM-сетей (сетей с долговременной краткосрочной памятью) с механизмами самовнимания.
Более подробную информацию о его теории и архитектуре см. в этой статье: https://medium.com/dataness-ai/understanding-temporal-fusion-transformer-9a7a4fcde74b
Реализуем эту модель с использованием фреймворка PyTorch Forecasting.
Для начала установим в вашей виртуальной среде Python все зависимости, используемые в этом проекте (файл requirements.txt находится в приложенном ZIP-файле, размещенном в конце этой статьи).
pip install -r requirements.txt
Подготовка данных для модели TFT
Для начала мы импортируем данные из терминала MetaTrader 5.
import MetaTrader5 as mt5 # Get rates from the MetaTrader5 app if not mt5.initialize(): print("initialize() failed, error code =", mt5.last_error()) quit() symbol = "EURUSD" rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M15, 0, 10000) rates_df = pd.DataFrame(rates) rates_df["time"] = pd.to_datetime(rates_df["time"], unit="s") data = pd.concat([rates_df, get_features(rates_df)], axis=1)
Конструирование признаков
Мы не можем использовать все полученные признаки; удалим те, которые считаем ненужными.
rates_df.drop(columns=[ "spread", "real_volume" ], inplace=True)
Оставшихся после этой операции признаков тоже недостаточно; добавим в наш DataFrame новые признаки.
Ниже приведен простой класс, который поможет нам добавлять различные признаки (конструировать новые переменные):
features.py
- Признаки на основе даты и времени
features.pyimport pandas as pd from ta.trend import sma_indicator, ema_indicator, macd_diff, macd_signal from ta.momentum import stochrsi_k, stochrsi_d, rsi from ta.volatility import bollinger_hband, bollinger_lband class FeatureEngineer: # date/time features @staticmethod def hour(date_series: pd.Series) -> pd.Series: return date_series.dt.hour @staticmethod def dayofweek(date_series: pd.Series) -> pd.Series: return date_series.dt.dayofweek @staticmethod def dayofmonth(date_series: pd.Series) -> pd.Series: return date_series.dt.day @staticmethod def month(date_series: pd.Series) -> pd.Series: return date_series.dt.month
- Индикаторы следования за трендом
# trend following indicators @staticmethod def sma(price: pd.Series, window: int=20) -> pd.Series: return sma_indicator(price, window) @staticmethod def ema(price: pd.Series, window: int=20) -> pd.Series: return ema_indicator(price, window) @staticmethod def macd_diff(price: pd.Series, window_slow: int=26, window_fast: int=12, window_signal: int=9) -> pd.Series: return macd_diff(price, window_slow=window_slow, window_fast=window_fast, window_sign=window_signal) @staticmethod def macd_signal(price: pd.Series, window_slow: int=26, window_fast: int=12, window_signal: int=9) -> pd.Series: return macd_signal(price, window_slow=window_slow, window_fast=window_fast, window_sign=window_signal
- Индикаторы моментума
# momentum indicators @staticmethod def rsi(price: pd.Series, window: int=14) -> pd.Series: return rsi(price, window) @staticmethod def stochrsi_k(price: pd.Series, window: int=14, smooth1: int=3, smooth2: int=3) -> pd.Series: return stochrsi_k(price, window=window, smooth1=smooth1, smooth2=smooth2) @staticmethod def stochrsi_d(price: pd.Series, window: int=14, smooth1: int=3, smooth2: int=3) -> pd.Series: return stochrsi_d(price, window=window, smooth1=smooth1, smooth2=
- Индикаторы волатильности
# volatility indicators @staticmethod def bollinger_hband(price: pd.Series, window: int=20, window_dev: int=2) -> pd.Series: return bollinger_hband(price, window=window, window_dev=window_dev) @staticmethod def bollinger_lband(price: pd.Series, window: int=20, window_dev: int=2) -> pd.Series: return bollinger_lband(price, window=window, window_dev=window_dev)
Мы вызываем методы этого класса в единственной статической функции того же класса с именем get_all.
@staticmethod def get_all(data: pd.DataFrame) -> pd.DataFrame: return pd.DataFrame({ "hour": FeatureEngineer.hour(data["time"]), "dayofweek": FeatureEngineer.dayofweek(data["time"]), "dayofmonth": FeatureEngineer.dayofmonth(data["time"]), "month": FeatureEngineer.month(data["time"]), "sma_20": FeatureEngineer.sma(data["close"]), "ema_20": FeatureEngineer.ema(data["close"]), "macd_diff": FeatureEngineer.macd_diff(data["close"]), "macd_signal": FeatureEngineer.macd_signal(data["close"]), "rsi": FeatureEngineer.rsi(data["close"]), "stochrsi_k": FeatureEngineer.stochrsi_k(data["close"]), "stochrsi_d": FeatureEngineer.stochrsi_d(data["close"]), "bollinger_hband": FeatureEngineer.bollinger_hband(data["close"]), "bollinger_lband": FeatureEngineer.bollinger_lband(data["close"]), })
Мы используем этот класс для получения нового Pandas DataFrame, заполненного признаками.
new_features = features.FeatureEngineer.get_all(rates_df)
Результаты.
(.env) C:\Users\Omega Joctan\OneDrive\mql5 articles\Data Science and ML\Part 48\TFT>python train.py hour dayofweek dayofmonth month sma_20 ema_20 macd_diff macd_signal rsi stochrsi_k stochrsi_d bollinger_hband bollinger_lband 0 20 3 21 8 NaN NaN NaN NaN NaN NaN NaN NaN NaN 1 20 3 21 8 NaN NaN NaN NaN NaN NaN NaN NaN NaN 2 20 3 21 8 NaN NaN NaN NaN NaN NaN NaN NaN NaN 3 20 3 21 8 NaN NaN NaN NaN NaN NaN NaN NaN NaN 4 21 3 21 8 NaN NaN NaN NaN NaN NaN NaN NaN NaN ... ... ... ... ... ... ... ... ... ... ... ... ... ... 9985 20 4 16 1 1.160491 1.160265 -0.000107 -0.000404 41.591063 0.303937 0.316652 1.162598 1.158383 9986 20 4 16 1 1.160384 1.160213 -0.000068 -0.000421 43.685146 0.499779 0.367431 1.162418 1.158349 9987 20 4 16 1 1.160269 1.160151 -0.000047 -0.000433 42.436779 0.698607 0.500775 1.162214 1.158323 9988 21 4 16 1 1.160154 1.160067 -0.000046 -0.000444 40.194753 0.774743 0.657710 1.162051 1.158257 9989 21 4 16 1 1.160052 1.160010 -0.000026 -0.000450 42.452830 0.687333 0.720228 1.161864 1.158240 [9990 rows x 13 columns]
Затем мы объединяем оба DataFrame, чтобы сформировать один более крупный, который будем использовать для финального обучения модели.
data = pd.concat([rates_df, new_features], axis=1) # concatenate dataframes
Результаты.
time open high low close tick_volume ... macd_signal rsi stochrsi_k stochrsi_d bollinger_hband bollinger_lband 0 2025-08-21 20:00:00 1.16112 1.16171 1.16112 1.16170 642 ... NaN NaN NaN NaN NaN NaN 1 2025-08-21 20:15:00 1.16170 1.16173 1.16126 1.16136 557 ... NaN NaN NaN NaN NaN NaN 2 2025-08-21 20:30:00 1.16136 1.16167 1.16129 1.16158 414 ... NaN NaN NaN NaN NaN NaN 3 2025-08-21 20:45:00 1.16158 1.16187 1.16149 1.16151 513 ... NaN NaN NaN NaN NaN NaN 4 2025-08-21 21:00:00 1.16151 1.16152 1.16103 1.16106 473 ... NaN NaN NaN NaN NaN NaN ... ... ... ... ... ... ... ... ... ... ... ... ... ... 9985 2026-01-16 20:15:00 1.15911 1.15954 1.15905 1.15951 407 ... -0.000404 41.591063 0.303937 0.316652 1.162598 1.158383 9986 2026-01-16 20:30:00 1.15951 1.15973 1.15936 1.15972 289 ... -0.000421 43.685146 0.499779 0.367431 1.162418 1.158349 9987 2026-01-16 20:45:00 1.15972 1.15994 1.15955 1.15956 447 ... -0.000433 42.436779 0.698607 0.500775 1.162214 1.158323 9988 2026-01-16 21:00:00 1.15956 1.15967 1.15921 1.15927 457 ... -0.000444 40.194753 0.774743 0.657710 1.162051 1.158257 9989 2026-01-16 21:15:00 1.15927 1.15958 1.15915 1.15947 290 ... -0.000450 42.452830 0.687333 0.720228 1.161864 1.158240 [9990 rows x 19 columns]
Создание целевой переменной
Для типичного процесса машинного обучения с учителем требуется целевая переменная – переменная, значение которой модель должна предсказать на основе других признаков (предикторов).
Поскольку речь идет о задаче с временными рядами, возьмем доходность (процентное изменение цены) в качестве целевой переменной.
data["returns"] = data["close"].pct_change()
Создание объекта набора данных временных рядов
PyTorch Forecasting требует, чтобы данные для модели хранились в объекте под названием TimeSeriesDataset.
Перед передачей DataFrame в объект TimeSeriesDataset в нем должен присутствовать столбец с именем time_idx.
data["time_idx"] = data.index data.drop(columns=["time"], inplace=True)
time_idx (str) – столбец целочисленного типа, обозначающий временной индекс в данных. Этот столбец используется для определения последовательности наблюдений. Если пропущенных наблюдений нет, временной индекс должен увеличиваться на +1 для каждого последующего наблюдения. Первое значение time_idx для каждого ряда не обязательно должно равняться 0; допускается любое значение.
После добавления столбца time_idx в DataFrame исходный столбец, содержащий время (datetime), необходимо удалить (TFT напрямую не использует переменные типа datetime).
data.drop(columns=["time"], inplace=True)
max_prediction_length = 6 max_encoder_length = 24 training_cutoff = data["time_idx"].max() - max_prediction_length training = TimeSeriesDataSet( data[lambda x: x.time_idx <= training_cutoff], time_idx="time_idx", target="returns", group_ids=["symbol"], min_encoder_length=max_encoder_length // 2, # keep encoder length long (as it is in the validation set) max_encoder_length=max_encoder_length, min_prediction_length=1, max_prediction_length=max_prediction_length, static_categoricals=["symbol"], # time_varying_known_categoricals=[], time_varying_known_reals=[ "hour", "dayofweek", "dayofmonth", "month", "time_idx", "stochrsi_k", "stochrsi_d", "rsi", "macd_diff", ], time_varying_unknown_categoricals=[], time_varying_unknown_reals=[ "open", "high", "low", "close", "tick_volume", "ema_20", "sma_20", "bollinger_hband", "bollinger_lband" ], target_normalizer=GroupNormalizer( groups=["symbol"], transformation="softplus" ), # use softplus and normalize by group add_target_scales=True, add_encoder_length=True, )
Этот объект принимает множество переменных; в таблице ниже описаны некоторые из них. Подробнее.
| Переменная | Описание |
|---|---|
| max_encoder_length | Насколько далеко в прошлое модели разрешено заглядывать. |
| max_prediction_length | На какой горизонт в будущем модель должна давать прогноз. |
| time_varying_known_reals | Это список непрерывных переменных, которые изменяются во времени и известны в будущем. Мы выбрали стационарные индикаторы, имеющиеся в нашем DataFrame, а также временные признаки. |
| time_varying_unknown_reals | Это должен быть список непрерывных переменных, значения которых в будущем неизвестны и которые изменяются во времени. Целевые переменные следует включать сюда, если они являются вещественными. Для этого мы выбрали такие признаки, как цена открытия, максимум, минимум, цена закрытия и т. д. Признаки, значения которых в будущем нам неизвестны. |
| group_ids | Список имен столбцов, определяющих экземпляр временного ряда в данных; это означает, что group_ids вместе с time_idx идентифицируют отдельную выборку. Если у вас только один временной ряд, укажите здесь название столбца, значение которого остается постоянным. Поскольку в нашем DataFrame Pandas собраны данные только по одному символу, единственная назначаемая нами группа – это текущий символ. data["symbol"] = "EURUSD"Группы могут представлять данные временных рядов по различным инструментам (символам) и таймфреймам. |
Мы формируем данные для валидации по образцу обучающего объекта.
# (predict=True) which means to predict the last max_prediction_length points in time for each series validation = TimeSeriesDataSet.from_dataset( training, data, predict=True, stop_randomization=True )
В заключение мы создаем загрузчики данных PyTorch как для обучающего, так и для валидационного наборов данных (загрузчики данных подходят для подачи данных в модели).
batch_size = 128 # set this between 32 to 128 train_dataloader = training.to_dataloader( train=True, batch_size=batch_size, num_workers=0 ) val_dataloader = validation.to_dataloader( train=False, batch_size=batch_size * 10, num_workers=0 )
Обучение Temporal Fusion Transformer (TFT)
Мы обучаем нашу модель с помощью PyTorch Lightning; ниже приведен объект Lightning Trainer для задачи обучения.
pl.seed_everything(42) # random seed for reproducibility lr_logger = LearningRateMonitor() # log the learning rate logger = TensorBoardLogger("lightning_logs") # logging results to a tensorboard # configure network and trainer early_stop_callback = EarlyStopping( monitor="val_loss", min_delta=1e-4, patience=10, verbose=False, mode="min" ) trainer = pl.Trainer( max_epochs=50, accelerator="cpu", enable_model_summary=True, gradient_clip_val=0.1, limit_train_batches=50, # comment in for training, running validation every 30 batches # fast_dev_run=True, # comment in to check that networkor dataset has no serious bugs callbacks=[lr_logger, early_stop_callback], logger=logger, )
Поскольку у нас нет возможности определить правильную скорость обучения для нашей модели, подберем ее.
Мы создаем экземпляр модели TFT, который будет использоваться для определения оптимальной скорости обучения.
tft = TemporalFusionTransformer.from_dataset( training, # not meaningful for finding the learning rate but otherwise very important learning_rate=0.03, hidden_size=8, # most important hyperparameter apart from learning rate # number of attention heads. Set to up to 4 for large datasets attention_head_size=2, dropout=0.1, # between 0.1 and 0.3 are good values hidden_continuous_size=8, # set to <= hidden_size loss=metrics.QuantileLoss(), optimizer="ranger", # reduce learning rate if no improvement in validation loss after x epochs # reduce_on_plateau_patience=1000, )
Определение оптимальной скорости обучения.
res = Tuner(trainer).lr_find( tft, train_dataloaders=train_dataloader, val_dataloaders=val_dataloader, max_lr=10.0, min_lr=1e-6, ) optimal_lr = res.suggestion() print(f"suggested learning rate: {optimal_lr}") fig = res.plot(show=False, suggest=True) plots_path = os.path.join(outputs_dir, "Plots") os.makedirs(plots_path, exist_ok=True) fig.savefig(os.path.join(plots_path, "lr_finder.png"))
Результаты.
Finding best initial lr: 91%|█████████████████████████████████████████████████████████████████████████████████████████████▋ | 91/100 [00:41<00:04, 2.20it/s] LR finder stopped early after 91 steps due to diverging loss. Restoring states from the checkpoint path at C:\Users\Omega Joctan\OneDrive\mql5 articles\Data Science and ML\Part 48\TFT\.lr_find_6df0be87-4347-4325-98f1-3a2b5a244c46.ckpt Restored all states from the checkpoint at C:\Users\Omega Joctan\OneDrive\mql5 articles\Data Science and ML\Part 48\TFT\.lr_find_6df0be87-4347-4325-98f1-3a2b5a244c46.ckpt Learning rate set to 1.3182567385564071e-05 suggested learning rate: 1.3182567385564071e-05
После определения оптимальной скорости обучения мы создаем новый экземпляр модели и обучаем его с использованием данного значения скорости обучения.
tft = TemporalFusionTransformer.from_dataset( training, learning_rate=optimal_lr, hidden_size=16, attention_head_size=2, dropout=0.1, hidden_continuous_size=8, loss=metrics.QuantileLoss(), log_interval=10, # uncomment for learning rate finder and otherwise, e.g., to 10 for logging every 10 batches optimizer="ranger", reduce_on_plateau_patience=4, ) print(f"Number of parameters in network: {tft.size() / 1e3:.1f}k") trainer.fit( tft, train_dataloaders=train_dataloader, val_dataloaders=val_dataloader, ) tft_predictions = tft.predict(val_dataloader, return_y=True) print("TFT MAE: ", metrics.MAE()(tft_predictions.output, tft_predictions.y))
Результаты.
Epoch 4: 100%|██████████████████████████████████████████| 50/50 [01:04<00:00, 0.77it/s, v_num=1, train_loss_step=0.000846, val_loss=0.0013, train_loss_epoch=0.00092]`Trainer.fit` stopped: `max_epochs=5` reached. Epoch 4: 100%|██████████████████████████████████████████| 50/50 [01:05<00:00, 0.77it/s, v_num=1, train_loss_step=0.000846, val_loss=0.0013, train_loss_epoch=0.00092] 💡 Tip: For seamless cloud uploads and versioning, try installing [litmodels](https://pypi.org/project/litmodels/) to enable LitModelCheckpoint, which syncs automatically with the Lightning model registry. GPU available: False, used: False TPU available: False, using: 0 TPU cores TFT MAPE: tensor(1.2644)+
Визуализируем прогнозы модели вместе с фактическими значениями, чтобы понять, как прогнозы модели соотносятся с истинными (исходными) значениями.
best_model_path = trainer.checkpoint_callback.best_model_path best_tft = TemporalFusionTransformer.load_from_checkpoint(best_model_path) # raw predictions are a dictionary from which all kinds of information, including quantiles, can be extracted raw_predictions = best_tft.predict( val_dataloader, mode="raw", return_x=True, trainer_kwargs=dict(accelerator="cpu") ) n = raw_predictions.output.prediction.shape[0] print(f"Plotting {n} predictions...") for idx in range(n): fig = best_tft.plot_prediction( raw_predictions.x, raw_predictions.output, idx=idx, add_loss_to_title=True ) fig.savefig(os.path.join(plots_path, f"tft_prediction_{idx}.png")) plt.close(fig=fig)
Результаты.
Прогнозные значения, похоже, близки к исходным.
Показатель MAE не имеет смысла без сравнения; нам необходимо знать, как наша модель выглядит на фоне других. Выясним это с помощью базовой модели.
Базовая модель
Это модель, которая использует последнее известное значение целевой переменной для построения прогноза. Это дает нам простой ориентир, который мы хотим превзойти.
baseline_predictions = Baseline().predict(val_dataloader, return_y=True) print("Baseline model MAE: ",metrics.MAE()(baseline_predictions.output, baseline_predictions.y))
После повторного запуска скрипта точность модели TFT оказалась в два раза выше, чем у базовой модели.
TFT MAE: tensor(0.0002) Baseline model MAE: tensor(0.0004)
Поиск оптимальных параметров модели Temporal Fusion Transformer
Поскольку модель-трансформер представляет собой модель на основе нейронной сети (в ее основе лежит Long Short-Term Memory (LSTM)), она, как и любые другие нейросетевые модели, чувствительна к гиперпараметрам.
Чтобы получить от таких моделей максимальную отдачу, нам нужен правильный набор гиперпараметров для конкретной задачи, которую пытается решить модель.
Согласно документации:
Настройка гиперпараметров с помощью optuna встроена непосредственно в PyTorch Forecasting. Для оптимизации гиперпараметров TFT можно использовать функцию optimize_hyperparameters().
Например:
import pickle from pytorch_forecasting.models.temporal_fusion_transformer.tuning import optimize_hyperparameters # create study study = optimize_hyperparameters( train_dataloader, val_dataloader, model_path="optuna_test", n_trials=200, max_epochs=50, gradient_clip_val_range=(0.01, 1.0), hidden_size_range=(8, 128), hidden_continuous_size_range=(8, 128), attention_head_size_range=(1, 4), learning_rate_range=(0.001, 0.1), dropout_range=(0.1, 0.3), trainer_kwargs=dict(limit_train_batches=30), reduce_on_plateau_patience=4, use_learning_rate_finder=False, # use Optuna to find ideal learning rate or use in-built learning rate finder ) # save study results - also we can resume tuning at a later point in time with open("test_study.pkl", "wb") as fout: pickle.dump(study, fout) # show best hyperparameters print(study.best_trial.params)
В отличие от других Python-фреймворков для машинного обучения, таких как scikit-learn и Keras, модули PyTorch требуют немного ручного написания кода, что часто приводит к необходимости писать дополнительный код практически для всего. Из-за этого процесс написания кода может стать утомительным и чреватым ошибками.
Чтобы значительно упростить себе жизнь, объединим весь необходимый код в один класс.
model.py
class TFTModel: def __init__(self, training: TimeSeriesDataSet, train_dataloader: DataLoader, val_dataloader: DataLoader, parameters: dict, loss: metrics=metrics.QuantileLoss(), trainer_max_epochs = 10): """ Initialize the Temporal Fusion Transformer model with training and validation data. Args: training (TimeSeriesDataSet): The training dataset loader containing time series data for model training. parameters (dict): A dictionary containing hyperparameters for the model configuration: - learning_rate (float, optional): Learning rate for the optimizer. Default is 0.03. - hidden_size (int, optional): Size of hidden layers. Most important hyperparameter apart from learning rate. Default is 8. - attention_head_size (int, optional): Number of attention heads. Set to up to 4 for large datasets. Default is 2. - dropout (float, optional): Dropout rate for regularization. Values between 0.1 and 0.3 are recommended. Default is 0.1. - hidden_continuous_size (int, optional): Size of continuous hidden layers. Should be set to <= hidden_size. Default is 8. loss (metrics): Loss function to be used for model training, e.g., QuantileLoss. Attributes: model (TemporalFusionTransformer): The initialized Temporal Fusion Transformer model with a given loss function and Ranger optimizer. trainer: PyTorch Lightning trainer instance configured for model training. """ # configure network and trainer pl.seed_everything(42) self.train_dataloader = train_dataloader self.val_dataloader = val_dataloader self.training = training self.loss = loss self.model = self._create_model(parameters=parameters) self.trainer = self._create_trainer(max_epochs=trainer_max_epochs) def _create_model(self, parameters: dict) -> TemporalFusionTransformer: return TemporalFusionTransformer.from_dataset( self.training, # not meaningful for finding the learning rate but otherwise very important learning_rate=parameters.get("learning_rate", 0.03), hidden_size=parameters.get("hidden_size", 8), # most important hyperparameter apart from learning rate # number of attention heads. Set to up to 4 for large datasets attention_head_size=parameters.get("attention_head_size", 2), dropout=parameters.get("dropout", 0.1), # between 0.1 and 0.3 are good values hidden_continuous_size=parameters.get("hidden_continuous_size", 8), # set to <= hidden_size loss=self.loss, optimizer="ranger", # reduce learning rate if no improvement in validation loss after x epochs # reduce_on_plateau_patience=1000, ) def _create_trainer(self, max_epochs: int=50, grad_clip_val=0.1, limit_train_batches: int=50) -> pl.Trainer: lr_logger = LearningRateMonitor() # log the learning rate logger = TensorBoardLogger("lightning_logs") # logging results to a tensorboard # configure network and trainer early_stop_callback = EarlyStopping( monitor="val_loss", min_delta=1e-4, patience=10, verbose=False, mode="min" ) return pl.Trainer( max_epochs=max_epochs, accelerator="cpu", enable_model_summary=True, gradient_clip_val=grad_clip_val, limit_train_batches=limit_train_batches, # comment in for training, running validation every 30 batches # fast_dev_run=True, # comment in to check that networkor dataset has no serious bugs callbacks=[lr_logger, early_stop_callback], logger=logger, ) def find_optimal_lr(self, plot_output_dir: str, max_lr: float=10.0, min_lr: float=1e-6, show_plot: bool=False, save_plot: bool=True) -> float: """find an optimal learning rate""" res = Tuner(self.trainer).lr_find( self.model, train_dataloaders=self.train_dataloader, val_dataloaders=self.val_dataloader, max_lr=max_lr, min_lr=min_lr, ) optimal_lr = res.suggestion() # ---- optional, saving the plot ---- fig = res.plot(show=show_plot, suggest=True) if save_plot: try: fig.savefig(os.path.join(plot_output_dir, "lr_finder.png")) except Exception as e: print("Error saving learning rate finder plot: ", e) return optimal_lr def load_best_model(self) -> bool: """Load the best model checkpoint after training.""" model = None try: best_model_path = self.trainer.checkpoint_callback.best_model_path model = TemporalFusionTransformer.load_from_checkpoint(best_model_path) except Exception as e: print("Error loading best model checkpoint: ", e) return False self.model = model return True def fit(self): self.trainer.fit( self.model, train_dataloaders=self.train_dataloader, val_dataloaders=self.val_dataloader, ) def predict(self, x: TimeSeriesDataSet, return_x: Optional[bool]=False, mode: Optional[str]="prediction", return_y: bool=True): try: tft_predictions = self.model.predict(x, mode=mode, return_x=return_x, return_y=return_y) except Exception as e: print(f"Failed to predict: {e}") return None return tft_predictions @staticmethod def find_optimal_parameters(train_dataloader: TimeSeriesDataSet, val_dataloader: TimeSeriesDataSet, max_epochs: int=50, n_trials: int=100, use_learning_rate_finder: bool=False, model_path: str="optuna_test", best_params_path: str="best_params.pkl", timeout: int=300) -> dict: """ Find optimal hyperparameters for a Temporal Fusion Transformer model using Optuna. Best parameters are saved for potential later usage Args: train_dataloader (TimeSeriesDataSet): Training dataset loader containing time series data. val_dataloader (TimeSeriesDataSet): Validation dataset loader for evaluating model performance. max_epochs (int, optional): Maximum number of training epochs per trial. Defaults to 50. n_trials (int, optional): Number of optimization trials to run. Defaults to 100. use_learning_rate_finder (bool, optional): Whether to use built-in learning rate finder instead of Optuna-based learning rate optimization. Defaults to False. model_path (str, optional): Directory path to save model checkpoints during optimization. Defaults to "optuna_test". best_params_path (str, optional): File path to save the best hyperparameters. Defaults to "best_params.pkl". timeout (int, optional): Maximum time in seconds to run the optimization study. Defaults to 300. Returns: dict: Dictionary containing the best hyperparameters found during optimization. """ # create study study = optimize_hyperparameters( train_dataloader, val_dataloader, model_path=model_path, n_trials=n_trials, max_epochs=max_epochs, gradient_clip_val_range=(0.01, 1.0), hidden_size_range=(8, 128), hidden_continuous_size_range=(8, 128), attention_head_size_range=(1, 4), learning_rate_range=(0.001, 0.1), dropout_range=(0.1, 0.3), trainer_kwargs=dict(limit_train_batches=30), reduce_on_plateau_patience=4, use_learning_rate_finder=use_learning_rate_finder, # use Optuna to find ideal learning rate or use in-built learning rate finder timeout=timeout, # stop study after given seconds ) # save study results - also we can resume tuning at a later point in time best_params = study.best_trial.params try: with open(best_params_path, "wb") as fout: pickle.dump(best_params, fout) print("Best parameters saved to: ", best_params_path) except Exception as e: print("Error saving best parameters: ", e) # return best hyperparameters return best_params def plot_raw_predictions(self, raw_predictions, plots_path: str, show=False): n = raw_predictions.output.prediction.shape[0] print(f"Plotting {n} predictions...") for idx in range(n): fig = self.model.plot_prediction( raw_predictions.x, raw_predictions.output, idx=idx, add_loss_to_title=True ) if show: plt.show(fig=fig) fig.savefig(os.path.join(plots_path, f"tft_prediction_{idx}.png")) plt.close(fig=fig)
Класс обеспечивает более чистый подход к работе с повторно используемым обучающим компонентом, обучению модели, поиску оптимальных параметров и т. д.
Если функция find_optimal_parameters задана как статический метод класса TFTModel, мы можем вызвать ее до создания экземпляра класса, получить оптимальные параметры, а затем присвоить их этому экземпляру.
Например:
best_params = model.TFTModel.find_optimal_parameters(train_dataloader=train_dataloader, val_dataloader=val_dataloader, timeout=optuna_timeout, best_params_path=best_params_path ) print("Best hyperparameters found: ", best_params) tft_model = model.TFTModel( training=training, train_dataloader=train_dataloader, val_dataloader=val_dataloader, parameters=best_params, trainer_max_epochs=max_training_epochs )
Создание торгового робота на основе модели TFT
Чтобы создать работоспособного торгового робота, нам необходимо получать прогнозы, сгенерированные моделью с использованием самых свежих рыночных данных на этапе инференса модели.
Поскольку TFT ожидает те же признаки при инференсе (включая целевую переменную), нам потребуется глобальная функция для сбора данных и конструирования признаков.
bot.py
def prepare_data(rates_df: pd.DataFrame) -> pd.DataFrame: rates_df["time"] = pd.to_datetime(rates_df["time"], unit="s") # convert time in seconds to datetime features_df = features.FeatureEngineer.get_all(rates_df) data = pd.concat([rates_df, features_df], axis=1) # concatenate dataframes # making the target variable data["returns"] = data["close"].pct_change() data["symbol"] = "EURUSD" # assigning symbol name as a group # drop NANs if any data.dropna(inplace=True) # assigning a time index data = data.reset_index(drop=True) data["time_idx"] = data.index # let's keep track of unused features unused_features = ["time", "spread", "real_volume"] return data.drop(columns=unused_features)
Приведенная выше функция принимает необработанный DataFrame из MetaTrader 5, создает новые признаки, включая целевую переменную, группу, к которой относится данный DataFrame, и временной индекс, а затем возвращает все необходимые признаки.
Нам также понадобится функция для обучения модели; она должна сохранять обученную TFTModel в глобальной переменной.
bot.py
def train_model(start_bar: int=100, num_bars: int=1000, symbol: int = "EURUSD", timeframe: int=mt5.TIMEFRAME_M15, max_prediction_length: int = 6, max_encoder_length: int = 24, load_best_parameters = False): # we extract training data from MetaTrader 5 try: rates = mt5.copy_rates_from_pos(symbol, timeframe, start_bar, num_bars) except Exception as e: print("Error retrieving data from MetaTrader 5: ", e) return data = prepare_data(rates_df=pd.DataFrame(rates)) # ------------ preparing training data and data loaders ------------ training_cutoff = data["time_idx"].max() - max_prediction_length training = TimeSeriesDataSet( data[lambda x: x.time_idx <= training_cutoff], time_idx="time_idx", target="returns", group_ids=["symbol"], min_encoder_length=max_encoder_length // 2, # keep encoder length long (as it is in the validation set) max_encoder_length=max_encoder_length, min_prediction_length=1, max_prediction_length=max_prediction_length, static_categoricals=["symbol"], # time_varying_known_categoricals=[], time_varying_known_reals=[ "hour", "dayofweek", "dayofmonth", "month", "time_idx", "stochrsi_k", "stochrsi_d", "rsi", "macd_diff", ], time_varying_unknown_categoricals=[], time_varying_unknown_reals=[ "open", "high", "low", "close", "tick_volume", "ema_20", "sma_20", "bollinger_hband", "bollinger_lband" ], target_normalizer=GroupNormalizer( groups=["symbol"], transformation="softplus" ), # use softplus and normalize by group add_relative_time_idx=True, add_target_scales=True, add_encoder_length=True, ) # create validation set (predict=True) which means to predict the last max_prediction_length points in time # for each series validation = TimeSeriesDataSet.from_dataset( training, data, predict=True, stop_randomization=True ) # create dataloaders for model batch_size = 128 # set this between 32 to 128 train_dataloader = training.to_dataloader( train=True, batch_size=batch_size, num_workers=4, persistent_workers=True ) val_dataloader = validation.to_dataloader( train=False, batch_size=batch_size * 10, num_workers=4, persistent_workers=True ) best_params_path = os.path.join(outputs_dir, "best_params.pkl") if load_best_parameters: try: with open(best_params_path, "rb") as fin: best_params = pickle.load(fin) except Exception as e: print("Error loading best parameters: ", e) print("Finding optimal parameters instead...") best_params = model.TFTModel.find_optimal_parameters(train_dataloader=train_dataloader, val_dataloader=val_dataloader, timeout=optuna_timeout, best_params_path=best_params_path, ) else: best_params = model.TFTModel.find_optimal_parameters(train_dataloader=train_dataloader, val_dataloader=val_dataloader, timeout=optuna_timeout, best_params_path=best_params_path ) print("Best hyperparameters found: ", best_params) global trained_model trained_model = model.TFTModel( training=training, train_dataloader=train_dataloader, val_dataloader=val_dataloader, parameters=best_params, trainer_max_epochs=max_training_epochs ) trained_model.load_best_model() trained_model.fit()
Все торговые операции выполняются внутри функции с именем trading_function.
def trading_function(): global trained_model if trained_model is None: train_model(symbol=symbol, timeframe=timeframe, max_encoder_length=lookback_window, max_prediction_length=lookahead_window, load_best_parameters=True) # get a trained model instance return # ---------- get data for model's inference ------- rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, 100) rates_df = pd.DataFrame(rates) if rates_df.empty: return data = prepare_data(rates_df=rates_df) predicted_returns = trained_model.predict(x=data, return_x=False, return_y=False) print(f"predicted returns: {np.array(predicted_returns)}") next_return = np.array(predicted_returns).ravel()[-1] print(f"next_return: {next_return:.2f}") # ------------- some trading strategy ---------------- tick_info = mt5.symbol_info_tick(symbol) if tick_info is None: print("Failed to get tick information. Error = ",mt5.last_error()) return symbol_info = mt5.symbol_info(symbol) if symbol_info is None: print(f"Failed to get information for {symbol}") return lotsize = symbol_info.volume_min if next_return > 0: if not pos_exists(symbol=symbol, magic=magic_number, type=mt5.POSITION_TYPE_BUY): m_trade.buy(volume=lotsize, symbol=symbol, price=tick_info.ask) close_by_type(symbol=symbol, magic=magic_number, type=mt5.POSITION_TYPE_SELL) # close a different type else: if not pos_exists(symbol=symbol, magic=magic_number, type=mt5.POSITION_TYPE_SELL): m_trade.sell(volume=lotsize, symbol=symbol, price=tick_info.bid) close_by_type(symbol=symbol, magic=magic_number, type=mt5.POSITION_TYPE_BUY) # close a different type
Первым делом внутри указанной выше функции мы проверяем, есть ли корректная модель в глобальной переменной trained_model; если ее нет, мы впервые обучаем модель с помощью функции train_model.
Используемая торговая стратегия элементарна: если последнее прогнозируемое значение доходности в ряду положительно, мы воспринимаем это как сигнал на покупку (лонг) и открываем сделку на покупку; в противном случае это сигнал на продажу (шорт), и мы открываем сделку на продажу. Все сделки, направленные в сторону, противоположную сигналу, закрываются с помощью функции close_by_type.
Наконец, мы задаем расписание: как часто нужно проверять торговые сигналы и выполнять соответствующие торговые действия, а также когда и с какой периодичностью модель переобучается, чтобы оставаться актуальной с учетом новой рыночной информации.
bot.py
timeframe = mt5.TIMEFRAME_M15 symbol = "EURUSD" magic_number = 20012026 slippage = 100 lookback_window = 24 lookahead_window = 6 if __name__ == "__main__": mt5_exe_path = r"C:\Program Files\MetaTrader 5 IC Markets Global\terminal64.exe" if not mt5.initialize(mt5_exe_path): print("initialize() failed, error code =", mt5.last_error()) quit() m_trade = CTrade(magic_number=magic_number, filling_type_symbol=symbol, deviation_points=slippage, mt5_instance=mt5) def trading_function(): global trained_model if trained_model is None: train_model(symbol=symbol, timeframe=timeframe, max_encoder_length=lookback_window, max_prediction_length=lookahead_window, load_best_parameters=True) # get a trained model instance return # ---------- get data for model's inference ------- rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, 100) rates_df = pd.DataFrame(rates) if rates_df.empty: return data = prepare_data(rates_df=rates_df) predicted_returns = trained_model.predict(x=data, return_x=False, return_y=False) print(f"predicted returns: {np.array(predicted_returns)}") next_return = np.array(predicted_returns).ravel()[-1] print(f"next_return: {next_return:.2f}") # ------------- some trading strategy ---------------- tick_info = mt5.symbol_info_tick(symbol) if tick_info is None: print("Failed to get tick information. Error = ",mt5.last_error()) return symbol_info = mt5.symbol_info(symbol) if symbol_info is None: print(f"Failed to get information for {symbol}") return lotsize = symbol_info.volume_min if next_return > 0: if not pos_exists(symbol=symbol, magic=magic_number, type=mt5.POSITION_TYPE_BUY): m_trade.buy(volume=lotsize, symbol=symbol, price=tick_info.ask) close_by_type(symbol=symbol, magic=magic_number, type=mt5.POSITION_TYPE_SELL) # close a different type else: if not pos_exists(symbol=symbol, magic=magic_number, type=mt5.POSITION_TYPE_SELL): m_trade.sell(volume=lotsize, symbol=symbol, price=tick_info.bid) close_by_type(symbol=symbol, magic=magic_number, type=mt5.POSITION_TYPE_BUY) # close a different type schedule.every(15).minutes.do(trading_function) # check for signals after 15 minutes (according to the timeframe) schedule.every(lookback_window*15).minutes.do(train_model, max_encoder_length=lookback_window, max_prediction_length=lookahead_window) while True: schedule.run_pending() time.sleep(1)
Заключение
Temporal Fusion Transformer (TFT) обеспечивает многогоризонтное прогнозирование, что может быть полезно для подтверждения по нескольким окнам. Модель поддерживает несколько групп данных, например данные по разным инструментам и таймфреймам, что позволяет ей выявлять полезные закономерности в разных областях. Кроме того, в этой модели есть различные методы, облегчающие интерпретацию, например методы построения графиков, где прогнозы модели сопоставляются с фактическими значениями, а также важности признаков, что помогает понять, как модель принимает решения.
Справедливо сказать, что TFT – достойная модель для построения прогнозов по данным временных рядов.
Однако это одна из сложных моделей: в ее основе лежит архитектура LSTM, поэтому она требует значительных вычислительных ресурсов (если вы хотите поработать с ней на большом наборе данных, вам точно понадобится GPU), а обучение на CPU занимает много времени. Кроме того, для хорошей обобщающей способности ей требуется много наблюдений в наборе данных, но даже при этом она может улавливать шум вместо сигналов; чтобы решить эту проблему, разработчики используют квантили, предоставляемые моделью.
Несмотря на то, что трансформеры отлично зарекомендовали себя в других областях, об их эффективности в финансовой сфере, которая гораздо сложнее других и в которой прогнозирование представляет собой серьезную проблему, известно очень мало; эта уникальная статья демонстрирует возможность применения TFT в данной сфере и служит отправной точкой для дальнейших исследований.
Не стесняйтесь делиться своими мыслями и мнениями в разделе комментариев к этой статье.
Таблица вложений
| Имя файла | Описание и использование |
|---|---|
| train.py | Это своего рода "полигон" для большей части кода, используемого в данной статье; он демонстрирует процесс обучения и оценки модели TFT. |
| features.py | Модуль, содержащий класс, отвечающий за конструирование признаков (создание дополнительных признаков, например индикаторов на основе значений OHLC). |
| model.py | Содержит класс TFTModel, в котором собраны все полезные методы для развертывания модели Temporal Fusion Transformer. |
| bot.py | Итоговый торговый робот, использующий модель TFT для принятия торговых решений. |
| error_description.py | Содержит функции, которые интерпретируют коды ошибок MetaTrader 5 в понятные человеку сообщения (ошибки). |
| Trade/Trade.py | Каталог, аналогичный каталогу MQL5/Include/Trade. В этом каталоге находятся модули Python, аналогичные стандартным библиотекам торговых классов. |
| requirements.txt | Содержит все зависимости Python и их версии, используемые в данном проекте. В данном проекте используется версия Python 3.11.1 |
Список литературы
- https://pytorch-forecasting.readthedocs.io/en/stable/tutorials/stallion.html
- https://medium.com/@ejcacciatore/the-transformer-revolution-in-financial-markets-technical-insights-applications-and-caveats-34806db92e8e
- https://dl.acm.org/doi/fullHtml/10.1145/3674029.3674037
- https://arxiv.org/pdf/1912.09363
- https://medium.com/dataness-ai/understanding-temporal-fusion-transformer-9a7a4fcde74b
- https://en.wikipedia.org/wiki/Transformer_(deep_learning)
- https://medium.com/@kdk199604/kdks-review-attention-is-all-you-need-what-makes-the-transformer-so-revolutionary-c91f135583b0
Перевод с английского произведен MetaQuotes Ltd.
Оригинальная статья: https://www.mql5.com/en/articles/18885
Предупреждение: все права на данные материалы принадлежат MetaQuotes Ltd. Полная или частичная перепечатка запрещена.
Данная статья написана пользователем сайта и отражает его личную точку зрения. Компания MetaQuotes Ltd не несет ответственности за достоверность представленной информации, а также за возможные последствия использования описанных решений, стратегий или рекомендаций.
Особенности написания Пользовательских Индикаторов
База данных — это просто (Часть 1): Легковесный ORM-фреймворк для MQL5 на базе SQLite
Машинное обучение и Data Science (Часть 47): Прогнозирование рынка с помощью модели DeepAR на Python
- Бесплатные приложения для трейдинга
- 8 000+ сигналов для копирования
- Экономические новости для анализа финансовых рынков
Вы принимаете политику сайта и условия использования
Опубликована статья Машинное обучение и Data Science (Часть 48): Насколько важны трансформеры для трейдинга?:
Автор: Omega J Msigwa
"Базовая модель
Это модель, которая использует последнее известное значение целевой переменной для построения прогноза. Это дает нам простой ориентир, который мы хотим превзойти."
Корректный базовый прогноз для стационарного ряда с нулевыми автокорреляциями это среднее этого ряда.
Для нестационарного ряда(типа случайного блуждания) корректный базовый прогноз это последнее значение.
У нас целевая переменная это приращения цен, ряд приблизительно стационарный, следовательно, модель нужно было сравнивать с прогнозом по среднему, а не по последнему значению.