Все блоги / Разное

вместо КодоБейз - пишем тики протоколом ILP

3 сентября 2025, 18:23

Maxim Kuznetsov

Собственно SUBJ - пример записи тиков в TimeSeries database

почти все TSDB и брокеры данных поддерживают протокол и формат записей ILP

детальнее : https://docs.influxdata.com/influxdb3/cloud-serverless/reference/syntax/line-protocol/

сейчас использую совместно с QuestDB (поэтому порт 9009). Чё-то 3-й Influx не понравился

Всё довольно просто : на локальном хосте ставится база TSDB на выбор или брокер данных типа telegraf

в него по Tcp льются строчки в формате: measurement,tag=value,tag=value field1=value,field2=value timestamp

measurement - что измеряем, в данном случай ticks, то есть тики. Аналог имени таблицы в SQL
tag - теги определяющие расположение. В данном примере это server="mt",symbol="EURUSD"
field - отдельные измеряемые поля. bid=1.223,ask=1.224
timestamp - штамп времени в наносекундах. Можно не указывать тогда база или брокер должны сами проставить. Проставляет-ли QuestDB не проверял. Influx проставляет

#property service
#property copyright "Maxim Kuznetsov"
#property link      "https://www.mql5.com"
#property version   "1.00"

/*** транслируем тики в телеграф
***/
class SymbolInfo {
public:
   SymbolInfo(string symbol);
   ~SymbolInfo();
   bool CheckTick();
   string LastMetric(ulong stamp);
public:
   string symbol;
   MqlTick ticks[2];   
};
SymbolInfo::SymbolInfo(string _symbol)
{
   symbol=_symbol;
   ZeroMemory(ticks);
}
SymbolInfo::~SymbolInfo()
{
}
bool SymbolInfo::CheckTick()
{
   MqlTick tmp[1];
   if (CopyTicks(symbol,tmp,COPY_TICKS_ALL,0,1)!=1) return false;
   if (tmp[0].ask==ticks[0].ask && tmp[0].bid==ticks[0].bid) return false;
   ticks[1]=ticks[0];
   ticks[0]=tmp[0];
   return true;
}
string SymbolInfo::LastMetric(ulong stamp)
{
   if (ticks[0].time_msc!=0) stamp=ticks[0].time_msc*1000000;
   return StringFormat("ticks,server=\"mt\",symbol=\"%s\" bid=%f,ask=%f %I64u\n",symbol,ticks[0].bid,ticks[0].ask,stamp);
}
class TickSender {
public:
   TickSender(ushort port=9009);
   ~TickSender();
   int Add(string symbol);
   bool CheckTicks();
public:
   int sock;
   ushort port;
   bool connected;
   datetime reconnectTimer;
   SymbolInfo *SYMBOLS[];      
};
TickSender::TickSender(ushort _port)
{
   sock=SocketCreate();
   port=_port;
   connected=false;
   reconnectTimer=0;
   ArrayResize(SYMBOLS,0,16);
}
TickSender::~TickSender()
{
   for(int id=ArraySize(SYMBOLS)-1;id>=0;id--) {
      if (SYMBOLS[id]==NULL) continue;
      delete SYMBOLS[id];
      SYMBOLS[id]=NULL;
   }
   ArrayResize(SYMBOLS,0);
   if (connected) SocketClose(sock);
}
int TickSender::Add(string symbol)
{
   int id=ArraySize(SYMBOLS);
   if (ArrayResize(SYMBOLS,id+1)!=id+1) return -1;
   SYMBOLS[id]=new SymbolInfo(symbol);
   return id;
}
bool TickSender::CheckTicks()
{
   bool haveTicks=false;
   if (!connected && TimeLocal()>reconnectTimer) {
      if (SocketConnect(sock,"127.0.0.1",port,100)) {
         connected=true;   
         Print("Connected!!");
      } else {
         reconnectTimer=TimeLocal()+2;
      }
   }
   string allmetrics=NULL;
   for(int i=ArraySize(SYMBOLS)-1;i>=0;i--) {
      if (SYMBOLS[i]==NULL) continue;
      if (SYMBOLS[i].CheckTick()) {
         string metric=SYMBOLS[i].LastMetric(0);
         if (connected) {
            if (allmetrics==NULL) allmetrics=metric;
            else allmetrics=allmetrics+metric;
         }
         haveTicks=true;
      }
   }
   if (connected && allmetrics!=NULL) {
      uchar data[];
      int len=StringToCharArray(allmetrics,data,0,WHOLE_ARRAY,CP_UTF8);
      if (len>0) {
         if (!SocketSend(sock,data,len-1)) {
            Print("Disconnected :-(");
            connected=false;
            reconnectTimer=TimeLocal()+1;
         } else {
            //Print(allmetrics);
         }
      }
   }
   return haveTicks;
}
void OnStart()
{
   TickSender ts;
   ts.Add("XAUUSD");
   ts.Add("EURUSD");
   while(!IsStopped() && !_StopFlag) {
      if (ts.CheckTicks()) {
         Sleep(1);
      } else {
         Sleep(20);
      }
   }
}