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

//+------------------------------------------------------------------+
//|                   MarketDataSender.mq5 (Final Fixed Version)     |
//+------------------------------------------------------------------+
#property strict
#property description "Fetches all symbols' candles and live prices, sends to API."

#include <Trade\SymbolInfo.mqh>

input string ApiBaseUrl = "http://market-price.insightbull.io";
input int HistoricalCandleCount = 1000;
input ENUM_TIMEFRAMES HistoricalTimeframe = PERIOD_H1;
input int LivePriceIntervalSeconds = 5;

// Globals
string symbols[];
int symbolIds[];
datetime lastSend = 0;

//+------------------------------------------------------------------+
int OnInit()
{
   Print("Initializing MarketDataSender EA...");
   if(!InitializeSymbols())
   {
      Print("❌ Failed to initialize symbols.");
      return(INIT_FAILED);
   }

   Print("✅ Symbols initialized: ", ArraySize(symbols));
   SendAllHistoricalCandles();
   return(INIT_SUCCEEDED);
}

//+------------------------------------------------------------------+
void OnTick()
{
   if(TimeCurrent() - lastSend >= LivePriceIntervalSeconds)
   {
      SendLivePrices();
      lastSend = TimeCurrent();
   }
}

//+------------------------------------------------------------------+
bool InitializeSymbols()
{
   int total = SymbolsTotal(true);
   if(total <= 0)
   {
      Print("❌ No symbols found!");
      return false;
   }

   ArrayResize(symbols, total);
   ArrayResize(symbolIds, total);

   for(int i = 0; i < total; i++)
   {
      symbols[i] = SymbolName(i, true);
      symbolIds[i] = -1;
   }

   if(!SyncSymbolsWithDatabase())
   {
      Print("❌ Failed to sync symbols with database");
      return false;
   }

   return true;
}

//+------------------------------------------------------------------+
bool SyncSymbolsWithDatabase()
{
   Print("Syncing symbols with database...");

     string url = ApiBaseUrl + "/api/symbols";
   string headers = "Content-Type: application/json\r\n";
   string resultHeaders = "";
   char result[];
   char emptyData[];   // ✅ required placeholder for GET request

   ResetLastError();

   // ✅ Correct GET request signature: includes empty data[]
   int res = WebRequest("GET", url, headers, 5000, emptyData, result, resultHeaders);

   if(res == -1)
   {
      int err = GetLastError();
      Print("❌ WebRequest connection error: ", err, " URL=", url);
      return false;
   }

   if(res != 200)
   {
      Print("❌ Failed to fetch symbols from API: HTTP ", res, " Response: ", CharArrayToString(result));
      return false;
   }

   string symbolsResponse = CharArrayToString(result);

   if(StringFind(symbolsResponse, "\"data\"") < 0)
   {
      Print("⚠️ Unexpected response format from symbols API: ", symbolsResponse);
   }

   for(int i = 0; i < ArraySize(symbols); i++)
   {
      string symbolName = symbols[i];
      int symbolId = FindSymbolId(symbolsResponse, symbolName);

      if(symbolId > 0)
      {
         symbolIds[i] = symbolId;
         Print("✅ Found existing symbol: ", symbolName, " (ID: ", symbolId, ")");
      }
      else
      {
         Sleep(300); // prevent overload (0.3 second delay)
         symbolId = CreateSymbolInDatabase(symbolName);
         if(symbolId > 0)
         {
            symbolIds[i] = symbolId;
            Print("✅ Created new symbol: ", symbolName, " (ID: ", symbolId, ")");
         }
         else
         {
            Print("❌ Failed to create symbol: ", symbolName," (ID: ", symbolId, ")");
            symbolIds[i] = -1;
         }
      }
   }

   return true;
}

//+------------------------------------------------------------------+
int FindSymbolId(string response, string symbolName)
{
   string searchPattern = StringFormat("\"symbol\":\"%s\"", symbolName);
   int symbolPos = StringFind(response, searchPattern);
   if(symbolPos < 0) return -1;

   int idPos = StringFind(response, "\"id\":", symbolPos);
   if(idPos < 0) return -1;

   int startPos = idPos + 5;
   int endPos = StringFind(response, ",", startPos);
   if(endPos < 0) endPos = StringFind(response, "}", startPos);

   string idStr = StringSubstr(response, startPos, endPos - startPos);
   return (int)StringToInteger(idStr);
}

//+------------------------------------------------------------------+
int CreateSymbolInDatabase(string symbolName)
{
   string baseAsset = "";
   string quoteAsset = "";
   string exchange = "MT5";
   string instrumentType = "forex";

   if(StringLen(symbolName) >= 6)
   {
      baseAsset = StringSubstr(symbolName, 0, 3);
      quoteAsset = StringSubstr(symbolName, 3, 3);
   }

   string json = StringFormat(
      "{\"symbol\":\"%s\",\"baseAsset\":\"%s\",\"quoteAsset\":\"%s\",\"exchange\":\"%s\",\"instrumentType\":\"%s\",\"isActive\":true}",
      symbolName, baseAsset, quoteAsset, exchange, instrumentType
   );

   string url = ApiBaseUrl + "/api/symbols";
   string headers = "Content-Type: application/json\r\n";
   string resultHeaders = "";

   char postData[];
   StringToCharArray(json, postData, 0, CP_UTF8);

   // ✅ FIX: Remove trailing null terminator from JSON
   if(ArraySize(postData) > 0 && postData[ArraySize(postData) - 1] == 0)
      ArrayResize(postData, ArraySize(postData) - 1);

   char result[];

   int res = WebRequest("POST", url, headers, 5000, postData, result, resultHeaders);

   if(res != 201 && res != 200)
   {
      Print("❌ Failed to create symbol: ", res, " Response: ", CharArrayToString(result));
      return -1;
   }

   string createResponse = CharArrayToString(result);
   int idPos = StringFind(createResponse, "\"id\":");
   if(idPos < 0) return -1;

   int startPos = idPos + 5;
   int endPos = StringFind(createResponse, ",", startPos);
   if(endPos < 0) endPos = StringFind(createResponse, "}", startPos);

   string idStr = StringSubstr(createResponse, startPos, endPos - startPos);
   return (int)StringToInteger(idStr);
}

//+------------------------------------------------------------------+
//+------------------------------------------------------------------+
//| Send all historical candles to the API (Fixed Version)           |
//+------------------------------------------------------------------+
void SendAllHistoricalCandles()
{
   Print("Starting historical upload for ", ArraySize(symbols), " symbols...");

   for(int i = 0; i < ArraySize(symbols); i++)
   {
      string sym = symbols[i];

      // --- Ensure data is ready ---
      Sleep(300);
      int tries = 0;
      while(!SeriesInfoInteger(sym, HistoricalTimeframe, SERIES_SYNCHRONIZED) && tries < 10)
      {
         Print("⏳ Waiting for ", sym, " history to load...");
         Sleep(500);
         tries++;
      }

      // --- Now copy candles ---
      MqlRates rates[];
      ResetLastError();
      int copied = CopyRates(sym, HistoricalTimeframe, 0, HistoricalCandleCount, rates);
      int err = GetLastError();

      if(copied <= 0)
      {
         Print("⚠️ Failed to copy candles for ", sym, " (copied=", copied, ", err=", err, ")");
         continue;
      }

      Print("✅ Copied ", copied, " candles for ", sym);

      // --- Print a few sample candles ---
      int sampleCount = MathMin(5, copied); // show up to 5 examples
      for(int j = 0; j < sampleCount; j++)
      {
         MqlRates r = rates[j];
         string openTime = TimeToString(r.time, TIME_DATE|TIME_SECONDS);
         string closeTime = TimeToString(r.time + PeriodSeconds(HistoricalTimeframe), TIME_DATE|TIME_SECONDS);
         PrintFormat(
            "🕒 [%s] %s → %s | O=%.5f H=%.5f L=%.5f C=%.5f | Vol=%.2f",
            sym, openTime, closeTime, r.open, r.high, r.low, r.close, r.tick_volume
         );
      }

      // --- Send candles in batches to API ---
      int sentTotal = 0;
      int batchSize = 200;
      for(int start = 0; start < copied; start += batchSize)
      {
         int size = MathMin(batchSize, copied - start);
         int symbolId = symbolIds[i];
         if(symbolId <= 0) continue;

         string json = BuildCandleJSONFromRates(symbolId, rates, start, size);
         string url = ApiBaseUrl + "/api/candles/bulk";
         string response;
         bool ok = SendJSON(url, json, response);
         if(!ok)
         {
            Print("❌ Failed to send candle batch for ", sym, " start=", start);
            break;
         }
         sentTotal += size;
         Print("📤 Sent candles for ", sym, ": ", sentTotal, "/", copied);
      }
   }
   Print("✅ Historical upload finished.");
}


//+------------------------------------------------------------------+
void SendLivePrices()
{
   bool firstItem = true;
   string json = "{\"prices\":[";

   int sentCount = 0;
   for(int i = 0; i < ArraySize(symbols); i++)
   {
      string sym = symbols[i];
      double bid = SymbolInfoDouble(sym, SYMBOL_BID);
      double ask = SymbolInfoDouble(sym, SYMBOL_ASK);
      double last = SymbolInfoDouble(sym, SYMBOL_LAST);
      if(bid <= 0 || ask <= 0 || last <= 0) continue;

      int symId = symbolIds[i];
      if(symId <= 0) continue;

      string item = StringFormat("{\"symbolId\":%d,\"price\":%.8f,\"bid\":%.8f,\"ask\":%.8f,\"bidSize\":%.8f,\"askSize\":%.8f}",
                                 symId, last, bid, ask, 0.0, 0.0);
      if(!firstItem) json += ",";
      json += item;
      firstItem = false;
      sentCount++;
   }

   json += "]}";

   if(sentCount == 0)
   {
      Print("No valid live prices to send right now.");
      return;
   }

   string url = ApiBaseUrl + "/api/live-prices/bulk";
   string response;
   bool ok = SendJSON(url, json, response);
   if(ok)
      Print("✅ Sent ", sentCount, " live prices.");
   else
      Print("❌ Failed to send live prices (sent ", sentCount, " items). Response: ", response);
}

//+------------------------------------------------------------------+
string ToISO8601(datetime t)
{
   MqlDateTime st;
   TimeToStruct(t, st);
   return StringFormat("%04d-%02d-%02dT%02d:%02d:%02d.000Z", st.year, st.mon, st.day, st.hour, st.min, st.sec);
}

string BuildCandleJSONFromRates(int symbolId, MqlRates &rates[], int startIndex, int count)
{
   string json = "{\"candles\":[";
   bool first = true;
   int ratesSize = ArraySize(rates);

   for(int i = startIndex; i < startIndex + count && i < ratesSize; i++)
   {
      MqlRates r = rates[i];
      if(r.time <= 0) continue;

      datetime open_dt  = (datetime)r.time;
      datetime close_dt = (datetime)(r.time + (datetime)PeriodSeconds(HistoricalTimeframe));

      string openTime  = ToISO8601(open_dt);
      string closeTime = ToISO8601(close_dt);

      double volume       = (r.tick_volume > 0 ? r.tick_volume : 1);
      double quoteVolume  = (r.real_volume > 0 ? r.real_volume : volume);

      string one = StringFormat(
         "{\"symbolId\":%d,\"openTime\":\"%s\",\"closeTime\":\"%s\",\"open\":%.5f,\"high\":%.5f,\"low\":%.5f,\"close\":%.5f,\"volume\":%.5f,\"tradesCount\":%d,\"quoteVolume\":%.5f}",
         symbolId, openTime, closeTime,
         r.open, r.high, r.low, r.close,
         volume, (int)volume, quoteVolume
      );

      if(!first) json += ",";
      json += one;
      first = false;
   }

   json += "]}";
   return json;
}

//+------------------------------------------------------------------+
bool SendJSON(string url, string json, string &response)
{
   ResetLastError();

   char postData[];
   StringToCharArray(json, postData, 0, CP_UTF8);

   // ✅ Remove trailing null terminator
   if(ArraySize(postData) > 0 && postData[ArraySize(postData) - 1] == 0)
      ArrayResize(postData, ArraySize(postData) - 1);

   if(ArraySize(postData) <= 0)
   {
      Print("❌ Empty postData for URL: ", url);
      return false;
   }

   char result[];
   string headers = "Content-Type: application/json\r\n";
   string resultHeaders = "";
   int timeout = 15000;

   int res = WebRequest("POST", url, headers, timeout, postData, result, resultHeaders);

   if(res == -1)
   {
      int err = GetLastError();
      Print("WebRequest error: ", err, " url=", url);
      return false;
   }

   response = CharArrayToString(result);
   if(res == 200 || res == 201)
      return true;

   Print("HTTP status ", res, " response: ", response);
   return false;
}

//+------------------------------------------------------------------+