market-data-service/MT5/Experts/MarketDataSender.mq5

//+------------------------------------------------------------------+
//|                                 MarketDataSender.mq5             |
//|                        Copyright 2025, MetaQuotes Software Corp. |
//|                                             https://www.mql5.com |
//+------------------------------------------------------------------+
#property copyright "Copyright 2025, MetaQuotes Software Corp."
#property link      "https://www.mql5.com"
#property version   "1.00"
#property description "Sends historical candles and live prices to REST API"
#property script_show_inputs

#include <Trade\SymbolInfo.mqh>
#include <JAson.mqh> // Include JSON library

//--- REST API Configuration
input string ApiBaseUrl = "http://localhost:3000"; // Base URL of your API
input string ApiKey = ""; // Optional API key if required

//--- Historical Data Configuration
input int    HistoricalCandleCount = 1000; // Number of historical candles to send
input ENUM_TIMEFRAMES HistoricalTimeframe = PERIOD_H1; // Timeframe for historical data

//--- Global variables
CJAson json;
CSymbolInfo symbolInfo;
string symbols[];
datetime lastSentTime = 0;

//+------------------------------------------------------------------+
//| Expert initialization function                                   |
//+------------------------------------------------------------------+
int OnInit()
{
   // Get all symbols
   int count = SymbolsTotal(true);
   ArrayResize(symbols, count);
   
   for(int i = 0; i < count; i++)
   {
      symbols[i] = SymbolName(i, true);
   }
   
   // Send initial historical data
   SendHistoricalData();
   
   return(INIT_SUCCEEDED);
}

//+------------------------------------------------------------------+
//| Expert tick function                                             |
//+------------------------------------------------------------------+
void OnTick()
{
   // Send live prices every second
   if(TimeCurrent() - lastSentTime >= 1)
   {
      SendLivePrices();
      lastSentTime = TimeCurrent();
   }
}

//+------------------------------------------------------------------+
//| Send historical candle data                                      |
//+------------------------------------------------------------------+
void SendHistoricalData()
{
   for(int s = 0; s < ArraySize(symbols); s++)
   {
      string symbol = symbols[s];
      
      // Get historical candles
      MqlRates rates[];
      int copied = CopyRates(symbol, HistoricalTimeframe, 0, HistoricalCandleCount, rates);
      
      if(copied <= 0) continue;
      
      // Prepare JSON payload
      CJAsonArray candlesArray;
      
      for(int i = 0; i < copied; i++)
      {
         CJAson candleObj;
         candleObj.Add("symbolId", GetSymbolId(symbol)); // You need to implement GetSymbolId()
         candleObj.Add("openTime", TimeToString(rates[i].time, TIME_DATE|TIME_MINUTES|TIME_SECONDS));
         candleObj.Add("closeTime", TimeToString(rates[i].time + PeriodSeconds(HistoricalTimeframe), TIME_DATE|TIME_MINUTES|TIME_SECONDS));
         candleObj.Add("open", rates[i].open);
         candleObj.Add("high", rates[i].high);
         candleObj.Add("low", rates[i].low);
         candleObj.Add("close", rates[i].close);
         candleObj.Add("volume", rates[i].tick_volume);
         
         candlesArray.Add(candleObj);
      }
      
      // Create final payload
      CJAson payload;
      payload.Add("candles", candlesArray);
      
      // Send to API
      string url = ApiBaseUrl + "/api/candles/bulk";
      string result;
      string headers = "Content-Type: application/json";
      if(StringLen(ApiKey) > 0) headers += "\r\nAuthorization: Bearer " + ApiKey;
      
      // Implement retry logic with exponential backoff
      int retries = 3;
      int delayMs = 1000;
      int res = -1;
      
      for(int attempt = 0; attempt < retries; attempt++) {
         res = WebRequest("POST", url, headers, 5000, payload.GetJson(), result);
         
         if(res == 200) break;
         
         Print("Attempt ", attempt+1, " failed (", res, "). Retrying in ", delayMs, "ms");
         Sleep(delayMs);
         delayMs *= 2; // Exponential backoff
      }
      
      if(res == 200) {
         Print("Successfully sent historical data for ", symbol);
      } else {
         Print("Permanent failure sending historical data for ", symbol, ": ", res, " - ", result);
         // TODO: Implement dead letter queue storage
      }
   }
}

//+------------------------------------------------------------------+
//| Send live prices                                                 |
//+------------------------------------------------------------------+
void SendLivePrices()
{
   CJAsonArray pricesArray;
   
   for(int s = 0; s < ArraySize(symbols); s++)
   {
      string symbol = symbols[s];
      
      if(!symbolInfo.Name(symbol)) continue;
      symbolInfo.RefreshRates();
      
      CJAson priceObj;
      priceObj.Add("symbolId", GetSymbolId(symbol));
      priceObj.Add("price", symbolInfo.Last());
      priceObj.Add("bid", symbolInfo.Bid());
      priceObj.Add("ask", symbolInfo.Ask());
      priceObj.Add("bidSize", symbolInfo.VolumeBid());
      priceObj.Add("askSize", symbolInfo.VolumeAsk());
      
      pricesArray.Add(priceObj);
   }
   
   // Create final payload
   CJAson payload;
   payload.Add("prices", pricesArray);
   
   // Send to API
   string url = ApiBaseUrl + "/api/live-prices/bulk";
   string result;
   string headers = "Content-Type: application/json";
   if(StringLen(ApiKey) > 0) headers += "\r\nAuthorization: Bearer " + ApiKey;
   
      // Implement retry logic with exponential backoff
      int retries = 3;
      int delayMs = 1000;
      int res = -1;
      
      for(int attempt = 0; attempt < retries; attempt++) {
         res = WebRequest("POST", url, headers, 5000, payload.GetJson(), result);
         
         if(res == 200) break;
         
         Print("Attempt ", attempt+1, " failed (", res, "). Retrying in ", delayMs, "ms");
         Sleep(delayMs);
         delayMs *= 2; // Exponential backoff
      }
      
      if(res == 200) {
         Print("Successfully sent live prices");
      } else {
         Print("Permanent failure sending live prices: ", res, " - ", result);
         // TODO: Implement dead letter queue storage
      }
}

//+------------------------------------------------------------------+
//| Initialize symbol map from database                              |
//+------------------------------------------------------------------+
bool InitializeSymbolMap()
{
   // Fetch existing symbols from API
   string url = ApiBaseUrl + "/api/symbols";
   string result;
   string headers = "Content-Type: application/json";
   if(StringLen(ApiKey) > 0) headers += "\r\nAuthorization: Bearer " + ApiKey;
   
   int res = WebRequest("GET", url, headers, 5000, "", result);
   
   if(res != 200)
   {
      Print("Failed to fetch symbols: ", res, " - ", result);
      return false;
   }
   
   // Parse response
   CJAson parser;
   if(!parser.Parse(result))
   {
      Print("Failed to parse symbols response");
      return false;
   }
   
   // Create lookup table
   CJAsonArray symbolsArray = parser.GetArray("data");
   int dbSymbolCount = symbolsArray.Size();
   
   for(int s = 0; s < ArraySize(symbols); s++)
   {
      string mt5Symbol = symbols[s];
      bool found = false;
      
      // Search for matching symbol in database
      for(int i = 0; i < dbSymbolCount; i++)
      {
         CJAson dbSymbol = symbolsArray.GetObject(i);
         string dbSymbolName = dbSymbol.GetString("symbol");
         
         if(dbSymbolName == mt5Symbol)
         {
            symbolIdMap[s] = (int)dbSymbol.GetInt("id");
            found = true;
            break;
         }
      }
      
      // Create symbol if not found
      if(!found)
      {
         int newId = CreateSymbol(mt5Symbol);
         if(newId > 0)
         {
            symbolIdMap[s] = newId;
            Print("Created new symbol: ", mt5Symbol, " (ID: ", newId, ")");
         }
         else
         {
            Print("Failed to create symbol: ", mt5Symbol);
            return false;
         }
      }
   }
   
   return true;
}

//+------------------------------------------------------------------+
//| Create new symbol in database                                    |
//+------------------------------------------------------------------+
int CreateSymbol(string symbol)
{
   string url = ApiBaseUrl + "/api/symbols";
   string result;
   string headers = "Content-Type: application/json";
   if(StringLen(ApiKey) > 0) headers += "\r\nAuthorization: Bearer " + ApiKey;
   
   // Extract exchange and instrument type from symbol name
   string parts[];
   StringSplit(symbol, '_', parts);
   string exchange = (ArraySize(parts) > 1) ? parts[0] : "MT5";
   string instrumentType = "forex"; // Default, can be improved
   
   // Prepare payload
   CJAson payload;
   payload.Add("symbol", symbol);
   payload.Add("exchange", exchange);
   payload.Add("instrumentType", instrumentType);
   payload.Add("isActive", true);
   
   int res = WebRequest("POST", url, headers, 5000, payload.GetJson(), result);
   
   if(res != 201)
   {
      Print("Error creating symbol: ", res, " - ", result);
      return -1;
   }
   
   // Parse response to get new ID
   CJAson parser;
   if(!parser.Parse(result))
   {
      Print("Failed to parse create symbol response");
      return -1;
   }
   
   return (int)parser.GetObject("data").GetInt("id");
}

//+------------------------------------------------------------------+
//| Get symbol ID from map                                           |
//+------------------------------------------------------------------+
int GetSymbolId(string symbol)
{
   for(int i = 0; i < ArraySize(symbols); i++)
   {
      if(symbols[i] == symbol)
      {
         return symbolIdMap[i];
      }
   }
   return -1;
}
//+------------------------------------------------------------------+