MetaTrader 5 / Эксперты
English
preview
Анализ CSV-данных (Часть 5): Потоковая передача CSV-данных в реальном времени из активных сессий MetaTrader 5

Анализ CSV-данных (Часть 5): Потоковая передача CSV-данных в реальном времени из активных сессий MetaTrader 5

MetaTrader 5 — Эксперты |
123 0
Ushana Kevin Iorkumbul
Ushana Kevin Iorkumbul

От артефактов бэктеста к потокам данных в реальном времени

В первых четырёх частях этой серии рассматривалась конкретная задача: экспорт структурированных метрик из Тестера стратегий MetaTrader 5 в аналитический пайплайн Python для последующего анализа. Данные в этих реализациях по определению были историческими: каждая строка CSV представляла собой завершённый проход тестирования. Такой пайплайн был рассчитан на задержки выполнения, измеряемые минутами или часами, поэтому ни одному компоненту не требовалось обрабатывать данные в реальном времени.

Работа терминала в реальном времени предъявляет принципиально иной набор требований. Советник или индикатор, запущенный на активном графике, непрерывно обрабатывает тики в течение дней или недель. Формируемые им данные — значения индикаторов в реальном времени, торговые метрики на уровне сессии, счётчики качества сигнала и снимки эквити — по мере накопления приобретают значительную аналитическую ценность. Ожидание окончания торговой сессии перед экспортом данных приводит к потере оперативной видимости происходящего во время работы рынка.

Чтобы устранить этот пробел, в статье предлагается архитектура потокового экспорта. Система состоит из двух компонентов: экспортера MQL5, записывающего структурированные строки метрик в CSV-файл с ежедневной ротацией, и демона Python, который непрерывно читает дописываемый файл в режиме tail и формирует актуальную аналитическую картину в реальном времени. Вместе они превращают непрозрачную торговую сессию в прозрачный для наблюдения и аудита поток данных в реальном времени.

В статье полностью разбирается реализация потокового фреймворка. Рассматриваются: (1) ограничения экспорта в реальном времени по сравнению с бэктестами, (2) стратегии буферизации записи и ротации файлов и (3) архитектура демона Python для обновляемой панели мониторинга.

Примечание. Если вы пропустили часть 4 этой серии, прочитать её можно здесь.



Обзор архитектуры: трёхуровневый пайплайн

Полный конвейер потокового экспорта в реальном времени состоит из трёх независимо развёртываемых уровней:

Уровень 1 — потоковый экспортер MQL5 (LiveCSVStreamer.mqh)

Подключаемый файл, который можно добавить к любому индикатору или советнику, работающему на активном графике. В течение каждого бара он накапливает записи метрик в буфере памяти, сбрасывает буфер на диск при достижении настраиваемого порога и в полночь по UTC выполняет ротацию выходного файла, предотвращая его неограниченный рост. Этот уровень отвечает за все операции файлового ввода-вывода и сводит к минимуму влияние задержек на сигнальную логику индикатора или советника, использующего модуль.

Уровень 2 — выходной CSV-файл

Ротируемый CSV-файл, записываемый в общую папку файлов MetaTrader 5. Каждый файл охватывает ровно один календарный день данных и получает суффикс с датой, например LiveStream_ETHUSD_M1_20260613.csv. Демон Python определяет текущий активный файл по дате и автоматически переходит к следующему файлу на границе суточной ротации. На протяжении сессии файл остаётся открытым для дозаписи, а демон Python одновременно читает его, используя дескриптор только для чтения без эксклюзивной блокировки.

Уровень 3 — демон Python для tail-чтения файла (live_stream_daemon.py)

Длительно работающий процесс Python, который отслеживает активный CSV-файл с помощью tail-чтения. Он разбирает каждую новую строку по мере её появления, хранит в памяти скользящие окна ключевых метрик и через настраиваемые интервалы выводит аналитические сводки в консоль. Аномалии сигналов — например, резкие всплески частоты ложных срабатываний или резкое ухудшение эквити в скользящем окне — записываются в отдельный журнал предупреждений.

Три уровня намеренно слабо связаны. Экспортер MQL5 не знает, запущен ли демон Python. Демон Python не управляет экспортером MQL5. Любой компонент можно перезапустить независимо, не нарушая согласованный контракт данных между ними.



Механизм экспорта в реальном времени: проектные ограничения и поведенческие контракты

Почему шаблоны экспорта из бэктеста не работают в реальных сессиях

Шаблоны экспорта, использованные в частях 1–4, основаны на общем структурном предположении: операция записи выполняется один раз, в самом конце тестового прохода, внутри OnDeinit(). В Тестере стратегий это безопасно, поскольку весь жизненный цикл теста — инициализация, обработка данных и деинициализация — ограничен и синхронен. Экспортер записывает одну строку, файл закрывается, процесс завершается.

При работе терминала в реальном времени это предположение нарушают три условия:

  • Неограниченная длительность сессии. Советник или индикатор на активном графике может непрерывно работать днями, неделями или месяцами. Предсказуемой точки завершения нет. Внезапный перезапуск терминала или разрыв соединения приведёт к полной потере всех накопленных данных, если они ещё не были записаны на диск.
  • Необходимо контролировать частоту записи по тикам. Терминал в реальном времени обрабатывает тики с той частотой, с которой их передаёт брокер. Запись в файл на каждом тике создаёт значительную нагрузку на ввод-вывод и увеличивает задержки расчётов. Буфер записи, объединяющий записи в пакет перед сбросом, устраняет эту проблему.
  • Файловые дескрипторы нельзя держать открытыми бесконечно. Открытый в течение нескольких часов дескриптор повышает риск некорректного восстановления после сбоя терминала и может вызывать конфликты блокировок с антивирусным ПО. Правильный шаблон: открыть файл, записать пакет из буфера и сразу закрыть файл.

Модель буфера записи для высокочастотных данных

Буфер записи хранит в памяти массив строк фиксированного размера. Каждая новая запись метрик добавляется в этот массив вместо немедленной записи на диск. Когда массив достигает настраиваемого порога сброса (по умолчанию 20 строк), экспортер открывает файл, записывает все строки буфера одной операцией, закрывает файл и сбрасывает индекс буфера.

Таким образом, 20 отдельных циклов «открыть–записать–закрыть» объединяются в один, что сокращает общее число операций ввода-вывода примерно на 95% по сравнению с записью каждой строки отдельно. Компромисс состоит в том, что в любой момент до flush_threshold – 1 записей существуют только в памяти и будут потеряны при аварийном завершении. Для мониторинга поведения в реальном времени это приемлемая цена за существенный выигрыш в эффективности ввода-вывода. В сценариях, где требуется полная сохранность данных, установите порог сброса равным 1, фактически отключив буферизацию.

Ротация файлов: предотвращение неограниченного роста

Без ротации размер файла быстро увеличивается. На H1 формируется около 24 строк в день, на M15 — около 96, а потоковая передача на уровне тиков может достигать десятков тысяч строк в день. Один неротируемый файл быстро становится громоздким и неудобным для обработки.

Ротация решает эту проблему: в полночь по UTC создаётся новый выходной файл, а к его имени добавляется суффикс с датой. В начале каждого цикла сброса логика ротации сравнивает текущую дату UTC с датой, встроенной в имя активного файла. Если даты различаются, текущий файл закрывается и открывается новый файл с сегодняшним суффиксом. В результате формируется аккуратный архив: каждый файл содержит ровно один день потоковых данных, а его естественный размер ограничен частотой тиков инструмента. Позднее демон Python может восстановить любую историческую сессию, последовательно загрузив файлы из каталога архива по датам.

Потокобезопасность в терминале реального времени

MetaTrader 5 выполняет советник или индикатор каждого графика независимо; один экземпляр индикатора обычно не сталкивается с параллельной конкуренцией за запись, поэтому механизм спин-блокировки, необходимый в многоагентных сценариях Тестера из части 1 здесь не нужен. Достаточно одноагентного шаблона: открыть, перейти в конец, записать, закрыть.

Однако одновременный запуск потокового компонента на нескольких графиках может привести к конфликтам файлов, если имена слишком общие. Реализация ниже устраняет эту проблему, включая символ и таймфрейм в имя каждого файла. Благодаря этому каждый экземпляр на графике записывает в уникальный файл независимо от количества параллельно работающих экземпляров.


Реализация потокового экспорта на MQL5: LiveCSVStreamer.mqh

Потоковый модуль реализован в виде одного подключаемого файла, который можно добавить к любому индикатору или советнику. Далее последовательно рассмотрим его компоненты в том порядке, в котором они расположены в исходном коде.

Входные параметры конфигурации

Блок входных параметров основной программы расширяется пятью новыми параметрами:

  • InpStreamingEnabled — главный переключатель всей потоковой системы.
  • InpFlushThreshold — количество строк, накапливаемых в буфере перед принудительным сбросом (по умолчанию 20).
  • InpStreamTicks — если true, запись на уровне тика формируется при каждом обновлении цены.
  • InpStreamBars — если true, запись на уровне бара формируется при каждом закрытии бара.
  • InpUseCommonFolder — если true, выходные файлы записываются в общую папку файлов MetaTrader 5; иначе — в локальную папку терминала MQL5\Files.

Эти параметры объявлены непосредственно в подключаемом файле, поэтому автоматически появляются на вкладке «Входные параметры» основной программы без дублирования. Флаг потоковой передачи тиков по умолчанию равен false, поскольку поток тиков создаёт значительно больше данных, чем поток по барам, и его следует включать только тогда, когда действительно требуется разрешение внутри бара.

Структуры SLiveBarRecord и SLiveTickRecord

Две структуры определяют формат экспортируемых данных.

SLiveBarRecord содержит поля, записываемые при каждом закрытии бара:

Поле Описание
bar_time Время открытия закрытого бара (UTC)
symbol Символ графика
timeframe Обозначение таймфрейма, например «M1» или «H1»
open, high, low, close Значения OHLC закрытого бара
volume Тиковый объём закрытого бара
filter_value Значение индикатора при закрытии бара
filter_slope 1 = рост, –1 = снижение, 0 = без изменения
false_flips Накопленное число ложных срабатываний с начала сессии
avg_lag_bars Скользящее среднее значение задержки в барах с начала сессии
session_equity Средства счёта в терминале при закрытии бара 

SLiveTickRecord — облегчённая структура для экспорта тиковых данных:

Поле Описание
tick_time Метка времени тика (UTC)
symbol Символ графика
bid, ask Текущие цены bid и ask
spread_points Текущий спред в пунктах
filter_value Значение индикатора на данном тике

Буфер записи и механизм сброса

Управление буфером полностью инкапсулировано в классе CStreamBuffer. Его закрытые члены включают динамический массив строк m_rows[] (размер изменяется до STREAMER_MAX_BUFFER, то есть 500 элементов), счётчик m_count, порог сброса и идентификатор активного файла.

  • Конструктор (CStreamBuffer::CStreamBuffer) инициализирует все переменные состояния и задаёт размер буфера строк.
  • Деструктор (~CStreamBuffer): при завершении все оставшиеся в памяти строки принудительно записываются на диск через FlushToFile().
  • Initialize(): вызывается один раз перед любыми вызовами Push(). Задаёт символ, строку таймфрейма, порог сброса, флаг общей папки, а также вычисляет исходную активную дату и имя файла.
  • Push(const string row): добавляет отформатированную CSV-строку в буфер. Перед добавлением вызывает CheckRotation(). Если число строк достигает порога сброса или жёсткого предела STREAMER_MAX_BUFFER - 1, автоматически вызывается FlushToFile().
  • FlushToFile(): открывает активный файл для дозаписи, при необходимости создавая его с заголовком, переходит в конец, за один проход записывает все строки буфера, закрывает дескриптор и сбрасывает m_count в ноль.
  • ForceFlush(): принудительно выполняет немедленный сброс независимо от порога; вызывается из OnDeinit() основной программы.
  • ActiveFile(): возвращает имя текущего целевого выходного файла.

Такая архитектура гарантирует, что файловый ввод-вывод выполняется только при заполнении буфера или завершении работы компонента.

Логика ротации файлов

Ротацией файлов управляет закрытый метод CheckRotation(). При каждом вызове Push() он вычисляет текущую дату UTC, приведённую к полуночи UTC, и сравнивает её с m_active_date. Если today_utc > m_active_date, все оставшиеся строки буфера сбрасываются в файл предыдущего дня, после чего m_active_date и m_active_file обновляются для нового дня.

Имя файла формируется функцией BuildFileName(datetime utc_date) по шаблону:

Примечание: комментарии, строковые литералы и сообщения внутри блоков кода сохранены в оригинале, чтобы не изменять представленные рабочие примеры.

LiveStream_<symbol>_<timeframe>_YYYYMMDD.csv

Это гарантирует, что у каждого экземпляра на графике будет собственный файл с датой в имени и обеспечивает бесшовную ротацию без ручного вмешательства.

Сбор метрик на уровне бара в OnCalculate()

После определения класса CStreamBuffer и его глобального экземпляра g_stream_buffer файл предоставляет две глобальные вспомогательные функции, которые форматируют и передают записи:

//+------------------------------------------------------------------+
//| Transform and pass structured bar records to operational frames  |
//+------------------------------------------------------------------+
void StreamBarRecord(const SLiveBarRecord &rec)
  {
   if(!InpStreamingEnabled || !InpStreamBars)
      return;

   string row = ::StringFormat("%s,%s,%s,%.5f,%.5f,%.5f,%.5f,%.0f,%.5f,%d,%d,%.2f,%.2f",
                               ::TimeToString(rec.bar_time, TIME_DATE | TIME_MINUTES),
                               rec.symbol,
                               rec.timeframe,
                               rec.open, rec.high, rec.low, rec.close,
                               rec.volume,
                               rec.filter_value,
                               rec.filter_slope,
                               rec.false_flips,
                               rec.avg_lag_bars,
                               rec.session_equity);

   g_stream_buffer.Push(row);
  }

Основной индикатор вызывает StreamBarRecord() после обновления всех накопителей качества сигналов для нового бара. Сначала выполняется собственная сигнальная логика индикатора — расчёт EMA, обнаружение изменений наклона и подсчёт ложных срабатываний, — поэтому поля структуры содержат уже рассчитанные значения.

Сбор метрик на уровне тиков

Вторая глобальная вспомогательная функция StreamTickRecord() формирует записи тикового уровня:

//+------------------------------------------------------------------+
//| Transform and pass structured tick records to operational frames |
//+------------------------------------------------------------------+
void StreamTickRecord(const SLiveTickRecord &rec)
  {
   if(!InpStreamingEnabled || !InpStreamTicks)
      return;

   string row = ::StringFormat("%s,%s,%.5f,%.5f,%.1f,%.5f",
                               ::TimeToString(rec.tick_time, TIME_DATE | TIME_SECONDS),
                               rec.symbol,
                               rec.bid,
                               rec.ask,
                               rec.spread_points,
                               rec.filter_value);

   g_stream_buffer.Push(row);
  }

Когда включён InpStreamTicks, индикатор или советник вызывает эту функцию при каждом обновлении цены, передавая актуальные bid, ask, спред и значение индикатора.

Наконец, для подключения к OnInit() и OnDeinit() основной программы предоставляются две функции жизненного цикла — InitStreamer() и ShutdownStreamer(). Они инициализируют глобальный экземпляр буфера и принудительно выполняют окончательный сброс при завершении работы.


Демонстрационный индикатор: LiveStream_Indicator.mq5

Демонстрационный индикатор подключается к графику и передаёт данные на уровне баров, используя EMA в качестве эталонного фильтра. Он служит минимальным примером интеграции и показывает, где именно InitStreamer(), ShutdownStreamer() и вызовы потоковой передачи записей располагаются в жизненном цикле индикатора.

Ниже приведён полный код индикатора, после которого следует пояснение логики реализации.

//+------------------------------------------------------------------+
//|                                         LiveStream_Indicator.mq5 |
//|        Live streaming indicator using LiveCSVStreamer.mqh        |
//+------------------------------------------------------------------+

#property indicator_chart_window
#property indicator_buffers 1
#property indicator_plots   1

//--- Plot attributes
#property indicator_label1  "Filter"
#property indicator_type1   DRAW_LINE
#property indicator_color1  clrDodgerBlue
#property indicator_width1  2

//--- Include dependencies
#include <CSV_Data_Analysis_Part_5/LiveCSVStreamer.mqh>

//--- Indicator input parameters
input int InpFilterPeriod = 14; // EMA lookback period

//--- Indicator buffer allocation
double g_filter_buf[];

//--- Signal quality metrics trackers
int      g_current_slope   = 0;
int      g_false_flips     = 0;
long     g_sum_lag_bars    = 0;
int      g_slope_changes   = 0;
int      g_bars_since_flip = 0;

//--- Native handles and tracking metrics
int      g_ema_handle        = INVALID_HANDLE;
double   g_last_filter_value = 0.0;
datetime g_last_bar_time     = 0;
bool     g_live_confirmed    = false;

Индикатор начинается с обычных объявлений свойств и подключения потокового механизма. Глобальные переменные отслеживают фильтр EMA, его наклон, накопленное число ложных срабатываний, скользящее среднее задержки и флаг g_live_confirmed, используемый для регистрации обнаружения первого бара реального времени.

//+------------------------------------------------------------------+
//| Custom indicator initialization function                         |
//+------------------------------------------------------------------+
int OnInit()
  {
//--- Bind linear output buffer mappings
   SetIndexBuffer(0, g_filter_buf, INDICATOR_DATA);
   ArraySetAsSeries(g_filter_buf, true);

//--- Instantiate exponential moving average system handler
   g_ema_handle = iMA(_Symbol, _Period, InpFilterPeriod, 0, MODE_EMA, PRICE_CLOSE);
   if(g_ema_handle == INVALID_HANDLE)
      return(INIT_FAILED);

//--- Parse cleanly formatted period label strings for path targets
   string tf_str = EnumToString(_Period);
   StringReplace(tf_str, "PERIOD_", "");
   InitStreamer(_Symbol, tf_str);

   return(INIT_SUCCEEDED);
  }

В OnInit() подготавливается буфер индикатора, создаётся дескриптор EMA, а потоковый механизм инициализируется текущим символом и строкой таймфрейма без префикса PERIOD, например «H1» вместо «PERIOD_H1».

//+------------------------------------------------------------------+
//| Custom indicator iteration function                              |
//+------------------------------------------------------------------+
int OnCalculate(const int      rates_total,
                const int      prev_calculated,
                const datetime &time[],
                const double   &open[],
                const double   &high[],
                const double   &low[],
                const double   &close[],
                const long     &tick_volume[],
                const long     &volume[],
                const int      &spread[])
  {
//--- Verify minimum operational lookback threshold bounds
   if(rates_total < InpFilterPeriod + 2)
      return(0);

//--- Explicitly set system array indexing as reverse chronological series
   ArraySetAsSeries(time,        true);
   ArraySetAsSeries(open,        true);
   ArraySetAsSeries(high,        true);
   ArraySetAsSeries(low,         true);
   ArraySetAsSeries(close,       true);
   ArraySetAsSeries(tick_volume, true);

//--- Pull indicator array segments out of internal buffer registers
   double ema_vals[3];
   if(CopyBuffer(g_ema_handle, 0, 0, 3, ema_vals) < 3)
      return(prev_calculated);

   g_filter_buf[0]     = ema_vals[2];
   g_last_filter_value = ema_vals[2];

Сначала OnCalculate() переводит входные массивы цен в режим временных рядов, где индекс 0 соответствует текущему бару. Это критически важно, поскольку при полном пересчёте истории MetaTrader 5 по умолчанию передаёт эти массивы в обычном порядке, из-за чего индикатор мог бы считывать неверные бары. После этого копируются три последних значения EMA: самое новое отображается на графике, а два предыдущих используются для определения наклона.

//--- Gate 1: skip the entire historical recalculation pass.
//--- Skip computational routines during historical parsing iterations
   if(prev_calculated == 0)
      return(rates_total);

При первом подключении индикатора или запуске терминала prev_calculated равен нулю. Во время этого начального прохода MetaTrader 5 рассчитывает индикатор по всей доступной истории. Потоковый механизм не должен формировать строки для такой исторической дозагрузки, поскольку демон Python ожидает данные реального времени, а заполнение буфера тысячами исторических баров исказило бы скользящие окна и напрасно заняло дисковое пространство. Условие if(prev_calculated == 0) return rates_total; пропускает потоковую обработку во время загрузки истории. Код потоковой передачи запускается только после первого полного прохода, когда prev_calculated > 0.

//--- Tick-level streaming (if enabled)
   if(InpStreamTicks)
     {
      SLiveTickRecord tick_rec;
      tick_rec.tick_time     = TimeCurrent();
      tick_rec.symbol        = _Symbol;
      tick_rec.bid           = SymbolInfoDouble(_Symbol, SYMBOL_BID);
      tick_rec.ask           = SymbolInfoDouble(_Symbol, SYMBOL_ASK);
      tick_rec.spread_points = (tick_rec.ask - tick_rec.bid) / SymbolInfoDouble(_Symbol, SYMBOL_POINT);
      tick_rec.filter_value  = g_last_filter_value;
      StreamTickRecord(tick_rec);
     }

Если потоковая передача на уровне тиков включена, запись помещается в буфер при каждом вызове OnCalculate(), то есть на каждом новом тике. Значения bid, ask и спреда считываются в реальном времени.

//--- Bar-level logic: execute only once per newly closed bar.
//--- time[0] is the current forming bar; a change in time[0]
//--- means a new bar has opened and time[1] just closed.
   datetime current_bar_time = time[0];
   if(current_bar_time <= g_last_bar_time)
      return(rates_total);

   g_last_bar_time = current_bar_time;

Обработка на уровне бара запускается только при изменении time[0], что означает открытие нового бара и полное закрытие предыдущего бара time[1].

//--- Confirm the first live bar close in the Experts log
   if(!g_live_confirmed)
     {
      PrintFormat("[LiveCSVStreamer] First live bar close detected: %s",
                  TimeToString(time[1], TIME_DATE | TIME_MINUTES));
      g_live_confirmed = true;
     }

Однократное сообщение в журнале отмечает переход от загрузки истории к сбору баров реального времени и помечает переход на вкладке «Эксперты» терминала.

//--- Slope detection and signal quality accumulation
   int new_slope = (ema_vals[2] > ema_vals[1]) ?  1
                   : (ema_vals[2] < ema_vals[1]) ? -1
                   : 0;

   g_bars_since_flip++;

//--- Accumulate metrics testing noise versus meaningful momentum swings
   if(new_slope != 0 && new_slope != g_current_slope && g_current_slope != 0)
     {
      g_slope_changes++;
      g_sum_lag_bars += g_bars_since_flip;

      if(g_bars_since_flip <= 3)
         g_false_flips++;

      g_bars_since_flip = 0;
     }

   g_current_slope = (new_slope != 0) ? new_slope : g_current_slope;

Логика определения наклона повторяет шаблон, представленный в части 1: ложное срабатывание учитывается, когда наклон EMA меняет знак, а предшествующий тренд длился не более трёх баров.

//--- Populate and stream the bar-close record (time[1] = last closed bar)
   SLiveBarRecord bar_rec;
   bar_rec.bar_time = time[1];
   bar_rec.symbol   = _Symbol;

   string tf_str = EnumToString(_Period);
   StringReplace(tf_str, "PERIOD_", "");
   bar_rec.timeframe = tf_str;

   bar_rec.open           = open[1];
   bar_rec.high           = high[1];
   bar_rec.low            = low[1];
   bar_rec.close          = close[1];
   bar_rec.volume         = (double)tick_volume[1];
   bar_rec.filter_value   = ema_vals[1];
   bar_rec.filter_slope   = g_current_slope;
   bar_rec.false_flips    = g_false_flips;
   bar_rec.avg_lag_bars   = (g_slope_changes > 0) ? (double)g_sum_lag_bars / g_slope_changes : 0.0;
   bar_rec.session_equity = AccountInfoDouble(ACCOUNT_EQUITY);

   StreamBarRecord(bar_rec);

   return(rates_total);
  }

Запись закрытия бара заполняется данными последнего закрытого бара time[1] и соответствующими значениями OHLC. Значение EMA на этом баре равно ema_vals[1], поскольку ema_vals[2] относится к текущему, ещё формирующемуся бару. Средства счёта считываются непосредственно из информации о счёте терминала. Затем запись передаётся через StreamBarRecord().

Функция деинициализации просто вызывает ShutdownStreamer() и освобождает дескриптор EMA.

LiveStream_Indicator.mq5 attached to an ETHUSD M1 Chart

Индикатор LiveStream_Indicator, подключённый к графику ETHUSD M1 в MetaTrader 5. Синяя линия — EMA с периодом 14; в фоновом режиме индикатор экспортирует метрики по каждому бару в CSV-файл с ежедневной ротацией.


Демон Python для чтения конца файла: live_stream_daemon.py

Демон Python — это автономный процесс, который считывает CSV-поток, не вмешиваясь в работу терминала. Он использует только стандартную библиотеку Python и может быть развёрнут на любом компьютере с Python 3.8 или новее.

Полный файл live_stream_daemon.py

"""
live_stream_daemon.py
Real-time CSV tail daemon for MetaTrader 5 LiveCSVStreamer output files.

Requires: Python 3.8+, standard library only.

Usage:
    python live_stream_daemon.py

Configure SYMBOL, TIMEFRAME, POLL_INTERVAL, and ROLLING_WINDOW
in the Configuration block below before starting.
"""
import os
import time
import csv
import io
import logging
from datetime import datetime, timezone
from collections import deque

# ── Configuration ─────────────────────────────────────────────────────────────
MT5_APPDATA    = os.path.join(os.environ.get("APPDATA", ""),
                               "MetaQuotes", "Terminal")
SYMBOL         = "ETHUSD"
TIMEFRAME      = "M1"
POLL_INTERVAL  = 5          # Seconds between tail reads
ROLLING_WINDOW = 50         # Bars retained in rolling metric windows
ALERT_LOG      = "stream_alerts.log"

# Anomaly thresholds
WHIPSAW_DENSITY_THRESHOLD = 4     # False flips in last 5 bars
EQUITY_DRAWDOWN_THRESHOLD = 5.0   # Percent decline over last 10 bars
# ─────────────────────────────────────────────────────────────────────────────

logging.basicConfig(
    filename = ALERT_LOG,
    level    = logging.WARNING,
    format   = "%(asctime)s  %(levelname)s  %(message)s",
    datefmt  = "%Y-%m-%d %H:%M:%S"
)

def resolve_active_file(symbol: str, timeframe: str) -> str:
    """
    Resolves the path to today's active streaming CSV file.
    Searches all MetaTrader 5 terminal instance directories and the common
    files folder. Returns the most recently modified match, or None.
    """
    today_str = datetime.now(timezone.utc).strftime("%Y%m%d")
    target    = f"LiveStream_{symbol}_{timeframe}_{today_str}.csv"

    try:
        instances = [
            d for d in os.listdir(MT5_APPDATA)
            if os.path.isdir(os.path.join(MT5_APPDATA, d))
        ]
    except FileNotFoundError:
        raise FileNotFoundError(
            f"MT5 AppData directory not found at:\n  {MT5_APPDATA}\n"
            "Verify the APPDATA environment variable and MT5 installation."
        )

    candidates = []

    for inst in instances:
        candidate = os.path.join(
            MT5_APPDATA, inst, "MQL5", "Files", target
        )
        if os.path.isfile(candidate):
            candidates.append(candidate)

    common_candidate = os.path.join(
         MT5_APPDATA, "Common", "Files", target
    )
    if os.path.isfile(common_candidate):
        candidates.append(common_candidate)

    if not candidates:
        return None

    return max(candidates, key=os.path.getmtime)

class LiveStreamDaemon:
    """
    Tails an active LiveCSVStreamer output file, maintains rolling
    metric windows, checks for signal anomalies, and renders a live
    console dashboard at each polling interval.
    """

    def __init__(self, symbol        : str,
                       timeframe     : str,
                       poll_interval : int = POLL_INTERVAL,
                       rolling_window: int = ROLLING_WINDOW):
        self.symbol         = symbol
        self.timeframe      = timeframe
        self.poll_interval  = poll_interval
        self.rolling_window = rolling_window

        # Tail state
        self.active_file    = None
        self.file_offset    = 0
        self.header_parsed  = False
        self.column_map     = {}

        # Rolling metric windows
        self.equity_window   = deque(maxlen=rolling_window)
        self.slope_window    = deque(maxlen=rolling_window)
        self.whipsaw_deltas  = deque(maxlen=rolling_window)
        self.spread_window   = deque(maxlen=rolling_window)

        # Session-level state
        self.total_bars_seen  = 0
        self.last_false_flips = 0
        self.session_start    = datetime.now(timezone.utc)

    def _resolve_file(self) -> bool:
        path = resolve_active_file(self.symbol, self.timeframe)
        if path is None:
            return False

        if path != self.active_file:
            print(f"\n[Daemon] Active file: {os.path.basename(path)}")
            self.active_file   = path
            self.file_offset   = 0
            self.header_parsed = False
            self.column_map    = {}

        return True

    def _tail_new_rows(self) -> list:
        if not self.active_file or not os.path.isfile(self.active_file):
            return []

        new_rows = []

        with open(self.active_file, "r",
                  encoding="ansi", errors="replace") as f:
            f.seek(self.file_offset)
            new_bytes        = f.read()
            self.file_offset = f.tell()

        if not new_bytes.strip():
            return []

        reader = csv.DictReader(
            io.StringIO(new_bytes),
            fieldnames = list(self.column_map.keys())
                         if self.column_map else None
        )

        for row in reader:
            if not self.header_parsed:
                if "Bar_Time" in row or "Tick_Time" in row:
                    self.column_map    = {k: i
                                          for i, k in enumerate(row.keys())}
                    self.header_parsed = True
                    continue

            if row:
                new_rows.append(row)

        return new_rows

    def _process_bar_row(self, row: dict):
        try:
            equity      = float(row.get("Session_Equity",        0) or 0)
            slope       = int(row.get("Filter_Slope",            0) or 0)
            false_flips = int(row.get("False_Flips_Cumulative",  0) or 0)

            self.equity_window.append(equity)
            self.slope_window.append(slope)

            whipsaw_delta = max(0, false_flips - self.last_false_flips)
            self.whipsaw_deltas.append(whipsaw_delta)
            self.last_false_flips  = false_flips
            self.total_bars_seen  += 1

        except (ValueError, TypeError):
            pass

    def _process_tick_row(self, row: dict):
        try:
            spread = float(row.get("Spread_Points", 0) or 0)
            self.spread_window.append(spread)
        except (ValueError, TypeError):
            pass

    def _check_anomalies(self):
        if len(self.whipsaw_deltas) >= 5:
            recent_whipsaws = sum(list(self.whipsaw_deltas)[-5:])
            if recent_whipsaws >= WHIPSAW_DENSITY_THRESHOLD:
                logging.warning(
                    f"[{self.symbol} {self.timeframe}] High whipsaw density: "
                    f"{recent_whipsaws} false flips in last 5 bars."
                )

        if len(self.equity_window) >= 10:
            recent_equity = list(self.equity_window)
            equity_drop   = recent_equity[-10] - recent_equity[-1]
            equity_pct    = (equity_drop
                             / (recent_equity[-10] + 1e-9)) * 100
            if equity_pct > EQUITY_DRAWDOWN_THRESHOLD:
                logging.warning(
                    f"[{self.symbol} {self.timeframe}] Equity drawdown alert: "
                    f"{equity_pct:.1f}% decline over last 10 bars."
                )

    def _render_dashboard(self):
        equity_now  = self.equity_window[-1]  if self.equity_window  else 0.0
        equity_peak = max(self.equity_window)  if self.equity_window  else 0.0
        equity_dd   = ((equity_peak - equity_now)
                       / (equity_peak + 1e-9)) * 100

        slope_vals   = list(self.slope_window)
        rising_pct   = (slope_vals.count(1)  / len(slope_vals) * 100
                        if slope_vals else 0)
        falling_pct  = (slope_vals.count(-1) / len(slope_vals) * 100
                        if slope_vals else 0)

        avg_spread   = (sum(self.spread_window) / len(self.spread_window)
                        if self.spread_window else 0.0)

        recent_whips  = (sum(list(self.whipsaw_deltas)[-10:])
                        if self.whipsaw_deltas else 0)

        elapsed      = datetime.now(timezone.utc) - self.session_start
        hours, rem   = divmod(int(elapsed.total_seconds()), 3600)
        minutes      = rem // 60

        print("\033[H\033[J", end="")
        print(f"{'=' * 58}")
        print(f"  Live Stream Dashboard  |  "
              f"{self.symbol} {self.timeframe}")
        print(f"  Session: {hours:02d}h {minutes:02d}m  |  "
              f"Bars Seen: {self.total_bars_seen:,}")
        print(f"{'=' * 58}")
        print(f"  Equity Now    : {equity_now:>12.2f}")
        print(f"  Equity Peak   : {equity_peak:>12.2f}")
        print(f"  Rolling DD    : {equity_dd:>11.2f}%")
        print(f"{'─' * 58}")
        print(f"  Filter Slope  :  Rising {rising_pct:.0f}%  |  "
              f"Falling {falling_pct:.0f}%")
        print(f"  Avg Spread    : {avg_spread:>11.1f} pts")
        print(f"  Whipsaws (10b): {recent_whips:>11d}")
        print(f"{'─' * 58}")
        print(f"  Alert Log     :  {ALERT_LOG}")
        print(f"{'=' * 58}")

    def run(self):
        print(f"[Daemon] Starting. Symbol={self.symbol} "
              f"TF={self.timeframe} "
              f"Poll={self.poll_interval}s  "
              f"Window={self.rolling_window} bars")
        print(f"[Daemon] Press Ctrl+C to stop.\n")

        while True:
            try:
                if not self._resolve_file():
                    print(
                        f"[Daemon] Waiting for "
                        f"LiveStream_{self.symbol}_{self.timeframe}_"
                        f"{datetime.now(timezone.utc).strftime('%Y%m%d')}"
                        f".csv ..."
                    )
                    time.sleep(self.poll_interval)
                    continue

                new_rows = self._tail_new_rows()

                for row in new_rows:
                    if "Bar_Time" in row:
                        self._process_bar_row(row)
                    elif "Tick_Time" in row:
                        self._process_tick_row(row)

                if new_rows:
                    self._check_anomalies()

                if self.total_bars_seen > 0:
                    self._render_dashboard()

                time.sleep(self.poll_interval)

            except KeyboardInterrupt:
                print("\n[Daemon] Shutdown requested. Exiting cleanly.")
                break
            except Exception as exc:
                print(f"[Daemon] Unhandled error: {exc}")
                time.sleep(self.poll_interval)


# ── Entry Point ───────────────────────────────────────────────────────────────
if __name__ == "__main__":
    daemon = LiveStreamDaemon(
        symbol         = SYMBOL,
        timeframe      = TIMEFRAME,
        poll_interval  = POLL_INTERVAL,
        rolling_window = ROLLING_WINDOW
    )
    daemon.run()

Чтение конца файла и периодическая полная перезагрузка

Существуют две стратегии чтения файла, который параллельно дописывается другим процессом:

  • Периодическая полная перезагрузка повторно считывает весь файл с первого байта на каждом цикле опроса. Несмотря на простоту, её стоимость линейно растёт вместе с размером файла; к концу продолжительной сессии процесс тратит ресурсы CPU на повторный разбор неизменившихся данных.
  • Чтение конца файла хранит смещение в байтах, соответствующее последней прочитанной позиции. На каждом цикле демон переходит к этому смещению, считывает только новые добавленные байты и затем сдвигает указатель. Стоимость цикла зависит только от числа новых строк, а не от общего размера файла.

Демон реализует чтение файла в режиме tail файла с настраиваемым интервалом опроса.

Пошаговый разбор входящих строк

Функция resolve_active_file() находит файл, соответствующий текущей дате UTC. Она просматривает все каталоги экземпляров терминала MetaTrader 5 в APPDATA\MetaQuotes\Terminal, а также общую папку файлов, и возвращает последнее изменённое совпадение.

Класс LiveStreamDaemon инкапсулирует всё состояние чтения конца файла: путь к активному файлу, смещение в байтах, флаг и карту столбцов для однократного пропуска строки заголовка, а также скользящие очереди для средств, наклона, приращений ложных срабатываний и спреда. Его основные методы:

  • _resolve_file() — вызывает resolve_active_file() и при изменении пути из-за ротации сбрасывает состояние чтения.
  • _tail_new_rows() — открывает файл только для чтения в кодировке ANSI, переходит к сохранённому смещению, считывает новые байты и разбирает их с помощью csv.DictReader. Первая обнаруженная строка заголовка используется для построения карты столбцов; последующие строки возвращаются в виде словарей.
  • _process_bar_row(row) — извлекает Session_Equity, Filter_Slope и False_Flips_Cumulative из строки на уровне бара, обновляет скользящие окна средств и наклона, вычисляет приращение ложных срабатываний с предыдущего бара и увеличивает счётчик баров сессии.
  • _process_tick_row(row) — извлекает Spread_Points и добавляет значение в окно спреда.

Цикл обновления скользящей панели мониторинга

Метод _render_dashboard() очищает консоль и выводит текущий снимок:

  • Длительность сессии и общее число просмотренных баров.
  • Текущее эквити, максимальное эквити сессии и процент скользящей просадки.
  • Доля последних N баров, на которых наклон фильтра был восходящим и нисходящим.
  • Средний спред по тиковым данным, если они доступны.
  • Количество ложных срабатываний за последние 10 баров.

Метод run() представляет собой бесконечный цикл опроса. На каждой итерации он определяет файл, считывает новые строки, обрабатывает их по типу записи, проверяет аномалии, обновляет панель мониторинга и приостанавливает выполнение на заданный интервал.

Обнаружение аномалий сигналов

Метод _check_anomalies() проверяет два условия:

  • Плотность ложных срабатываний: если суммарное число ложных срабатываний за пять последних баров достигает или превышает WHIPSAW_DENSITY_THRESHOLD (по умолчанию 4), в журнал записывается предупреждение.
  • Темп просадки эквити: если эквити сессии снижается более чем на EQUITY_DRAWDOWN_THRESHOLD (по умолчанию 5%) за последние десять баров, в журнал записывается предупреждение.

Оба порога задаются настраиваемыми константами в начале скрипта. Предупреждения записываются в stream_alerts.log с меткой времени UTC, формируя постоянный журнал аудита всех событий аномалий, обнаруженных в ходе сессии.

Live Dashboard in Windows PowerShell

Панель мониторинга в реальном времени, формируемая демоном Python в терминале Windows PowerShell. Она обновляется на каждом цикле опроса и показывает текущее эквити, распределение наклона фильтра, средний спред и число ложных срабатываний.


Практический сценарий развёртывания

  1. Запустите MetaTrader 5 и подключите LiveStream_Indicator.mq5 к нужному графику. Убедитесь, что в журнале «Эксперты» появилось сообщение [LiveCSVStreamer] Initialized. Active file: LiveStream_<symbol>_<tf>_<date>.csv.
  2. Проверьте, что выходной файл появился в общей папке файлов MetaTrader 5 или в локальной папке терминала MQL5\Files — в зависимости от InpUseCommonFolder.
  3. Запустите демон Python из терминала командой python live_stream_daemon.py. Он найдёт активный файл в течение первого цикла опроса и начнёт отображать панель мониторинга.
  4. Ротация файлов в полночь по UTC автоматически выполняется на стороне MQL5; демон обнаруживает новый файл при следующем опросе и бесшовно переключается на него, сохраняя все накопители уровня сессии.


Эксплуатационные аспекты сбора данных по нескольким сессиям

  • Управление архивом. Каждый день создаётся один файл. За месяц работы на H1 формируется около 30 файлов примерно по 480 строк каждый. Эти файлы сразу совместимы с LoadOptResults() из части 2 и функциями визуализации Python из части 3. Чтобы создать набор данных для анализа нескольких сессий, загрузите их командой pd.concat([pd.read_csv(f) for f in sorted(glob("LiveStream_ETHUSD_M1_*.csv"))]).
  • Несколько инструментов. Одновременный запуск индикатора на нескольких графиках создаёт независимые файловые потоки благодаря включению символа и таймфрейма в имя файла. Для одновременного мониторинга нескольких потоков создайте по одному экземпляру LiveStreamDaemon для каждой пары в отдельных потоках или подпроцессах.
  • Обработка переподключения. Если терминал MetaTrader 5 теряет соединение с брокером и затем подключается снова, OnCalculate() может получить историческую дозагрузку баров. Условие prev_calculated == 0 предотвращает потоковую передачу при первоначальной загрузке истории, однако последующие разрывы, вызывающие частичную перезагрузку, пройдут через проверку, разрешающую потоковую обработку, поскольку prev_calculated уже не равен нулю. Если это различие важно для анализа, добавьте в запись бара логическое поле Is_Backfill и устанавливайте его в зависимости от того, находится ли время бара более чем на два периода от текущего времени сервера.
  • Оценка дискового пространства. Запись на уровне бара занимает в среднем около 120 байт. H1: 24 строки в день — примерно 3 КБ в день, что пренебрежимо мало. M1: 1440 строк в день — примерно 170 КБ в день. Потоковая передача тиков по активной паре в часы пик может создавать 50–200 тыс. строк в день и занимать 6–24 МБ в день. Перед включением тикового потока на высокочастотных инструментах заранее предусмотрите достаточный объём архива.


Заключение

Бэктест создаёт статический артефакт, а торговая сессия в реальном времени формирует непрерывный поток поведенческих данных. Чтобы извлечь из него аналитическую ценность, данные необходимо непрерывно собирать и обрабатывать по мере их появления.

Разработанная в статье архитектура создаёт такой уровень мониторинга в реальном времени с помощью двух основных компонентов:

  • LiveCSVStreamer.mqh — подключаемый файл, который записывает структурированные строки метрик в ежедневно ротируемый файл, учитывает ограничения параллельной работы в терминале реального времени и гарантирует запись буферизованных данных на диск при минимальных накладных расходах ввода-вывода. Механизм ротации сохраняет управляемый объём хранилища при длительной работе, а условие пропуска загрузки истории предотвращает заполнение демона данными дозагрузки.
  • Демон Python — фоновый процесс, который считывает экспортированные данные по мере дозаписи файла и поддерживает скользящие расчёты для средств, распределения наклона, спреда и частоты ложных срабатываний. Уровень обнаружения аномалий в фоновом режиме отслеживает превышение порогов и записывает предупреждения в постоянный файл.

Долгосрочная полезность пайплайна основана на двух структурных решениях:

  • Единообразие схемы. Формат выходного CSV совпадает со схемой, используемой в предыдущих статьях, поэтому файлы напрямую интегрируются с ранее созданными модулями аналитики, нормализации и проверки устойчивости без преобразования данных.
  • Отсутствие внешних зависимостей. Демон Python использует только стандартную библиотеку, поэтому его можно развернуть на любом компьютере без установки дополнительных пакетов.

Мониторинг в реальном времени — это страховочный механизм наблюдения, а не замена бэктесту или walk-forward-валидации. Он в реальном времени проверяет, сохраняются ли в условиях живого рынка поведенческие свойства, выявленные при историческом тестировании. Без мониторинга структурные сбои стратегии становятся заметны только после просадки; автоматизированный поток реального времени выявляет ранние отклонения до того, как возникнут существенные потери капитала.


Программы, использованные в статье:

# Название Тип Описание
1 LiveCSVStreamer.mqh Подключаемый файл Буферизует записи метрик по барам и тикам в памяти и сбрасывает их в ежедневно ротируемые CSV-файлы с минимальными накладными расходами ввода-вывода.
2 LiveStream_Indicator.mq5 Пользовательский индикатор Демонстрационный индикатор, который рассчитывает EMA и передаёт в поток данные реального времени по барам и, при необходимости, по тикам, используя подключаемый файл.
3 live_stream_daemon.py Скрипт Python Отслеживает активный CSV-файл, поддерживает скользящие окна метрик, отображает консольную панель мониторинга в реальном времени и регистрирует аномалии сигналов.
4 CSV_Data_Analysis_Part_5.zip  ZIP-архив  ZIP-архив, содержащий все приложенные файлы и пути к ним относительно корневой папки терминала.

Перевод с английского произведен MetaQuotes Ltd.
Оригинальная статья: https://www.mql5.com/en/articles/23065

Прикрепленные файлы |
Загрузить ZIP
LiveCSVStreamer.mqh (13.07 KB)
LiveStream_Indicator.mq5 (6.51 KB)
live_stream_daemon.py (11.04 KB)
CSV_Data_Analysis_Part_5.zip (10.22 KB)

Предупреждение: все права на данные материалы принадлежат MetaQuotes Ltd. Полная или частичная перепечатка запрещена.

Данная статья написана пользователем сайта и отражает его личную точку зрения. Компания MetaQuotes Ltd не несет ответственности за достоверность представленной информации, а также за возможные последствия использования описанных решений, стратегий или рекомендаций.

Ushana Kevin Iorkumbul
Ushana Kevin Iorkumbul
Writer and developer

Другие статьи автора

Перейти к обсуждению на форуме трейдеров
Моделирование рынка: Position View (IX) Моделирование рынка: Position View (IX)
В данной статье, которая станет поворотным моментом, мы начнем более глубоко изучать взаимодействие между разрабатываемыми нами приложениями для обеспечения полной поддержки системы повторов и симуляции. Здесь мы проанализируем проблему, которая, с одной стороны, довольно неприятна, но с другой — очень интересна для объяснения и решения. Проблема заключается в следующем: как вернуть линии тейк-профита и стоп-лосса после того, как они были удалены, и сделать это без использования терминала путем выполнения операции прямо на графике. На первый взгляд, это кажется простым. Однако, существуют несколько препятствий, которые необходимо преодолеть.
Автоматизация классических рыночных методов в MQL5 (Часть 2): Закон причины и следствия Вайкоффа — ценовые цели по графику "крестики-нолики" Автоматизация классических рыночных методов в MQL5 (Часть 2): Закон причины и следствия Вайкоффа — ценовые цели по графику "крестики-нолики"
В статье создаётся автономный советник на MQL5, реализующий полный цикл Вайкоффа: он распознаёт накопление и распределение с помощью конечного автомата, входит в рынок в точке LPS/LPSY — последней точке поддержки или последней точке предложения — и рассчитывает цели выхода по графику «крестики-нолики» в соответствии с законом причины и следствия Вайкоффа. Подробно рассматриваются размер клетки на основе ATR диапазона, разворот на одну клетку, проверка корректности целевой цены и резервная цель 2R. Читатель получает рабочий код, который можно запускать без внешних зависимостей.
Модульная архитектура индикаторов в MQL5 (Часть 1): Хватит копировать и вставлять — начните писать масштабируемый, переиспользуемый код Модульная архитектура индикаторов в MQL5 (Часть 1): Хватит копировать и вставлять — начните писать масштабируемый, переиспользуемый код
В статье разработан объектно-ориентированный каркас для индикаторов MQL5 путем преобразования примитивного примера в пригодные для постоянного использования модули. В ней формализуется частичный пересчет буфера в OnCalculate, логика переносится в классы, вынесенные в заголовочные файлы (CAppliedPrice, CSma), а также вводятся CSubIndiBase, CIndicatorBase и реестр для централизации требований. В результате получаются переносимые компоненты, изолированные входные параметры и аккуратно организованные буферы при минимуме шаблонного кода, что ускоряет сборку новых индикаторов и упрощает их обслуживание.
Встраивание торговой дисциплины в код (Часть 7): Автоматизация защиты эквити с помощью логики управления Встраивание торговой дисциплины в код (Часть 7): Автоматизация защиты эквити с помощью логики управления
Автоматизированные торговые системы часто уделяют большое внимание генерации сигналов, пренебрегая механизмами, необходимыми для защиты капитала в периоды стресса. В данной статье представлен фреймворк управления эквити в MQL5, который отслеживает условия просадки, оценивает нагрузку на эквити и динамически управляет торговой активностью с помощью модели управления рисками, основанной на состояниях. Объединяя анализ просадки, логику охлаждения, авторизацию сделок и ограничения на исполнение, эта структура демонстрирует, как торговая дисциплина может быть непосредственно встроена в код с использованием модульной и расширяемой архитектуры.