Merge branch 'feature/cherry-picked-to-master' of MQL-Development/market-data-service into master

MT5/Experts/MarketDataSender.mq5

@@ -15,7 +15,7 @@ input int LivePriceIntervalSeconds = 5;
 string symbols[];
 int symbolIds[];
 datetime lastSend = 0;
-
+datetime lastCandleSync = 0;
 //+------------------------------------------------------------------+
 int OnInit()
 {
@@ -28,6 +28,8 @@ int OnInit()
 
    Print("✅ Symbols initialized: ", ArraySize(symbols));
    SendAllHistoricalCandles();
+   EventSetTimer(60);  // ⏱️ Trigger OnTimer() every 30 minutes
+   Print("✅ Timer set: SendAllHistoricalCandles() will run every 30 minutes.");
    return(INIT_SUCCEEDED);
 }
 
@@ -135,22 +137,63 @@ bool SyncSymbolsWithDatabase()
    return true;
 }
 
+//+------------------------------------------------------------------+
+//+------------------------------------------------------------------+
+//| Find exact symbolId from JSON response                           |
+//+------------------------------------------------------------------+
+//+------------------------------------------------------------------+
+//| Robust JSON search: matches exact symbol only                    |
 //+------------------------------------------------------------------+
 int FindSymbolId(string response, string symbolName)
 {
-   string searchPattern = StringFormat("\"symbol\":\"%s\"", symbolName);
-   int symbolPos = StringFind(response, searchPattern);
-   if(symbolPos < 0) return -1;
+   int pos = 0;
+   string patternSymbol = "\"symbol\":\"";
+   string patternId = "\"id\":";
 
-   int idPos = StringFind(response, "\"id\":", symbolPos);
-   if(idPos < 0) return -1;
+   while(true)
+   {
+      // find each symbol occurrence
+      int symPos = StringFind(response, patternSymbol, pos);
+      if(symPos < 0)
+         break;
 
-   int startPos = idPos + 5;
-   int endPos = StringFind(response, ",", startPos);
-   if(endPos < 0) endPos = StringFind(response, "}", startPos);
+      // extract actual symbol value
+      int symStart = symPos + StringLen(patternSymbol);
+      int symEnd = StringFind(response, "\"", symStart);
+      if(symEnd < 0) break;
 
-   string idStr = StringSubstr(response, startPos, endPos - startPos);
-   return (int)StringToInteger(idStr);
+      string foundSymbol = StringSubstr(response, symStart, symEnd - symStart);
+
+      // 🟩 DEBUG LOG: show all symbols found
+      // ✅ exact match check (case-sensitive)
+      if(foundSymbol == symbolName)
+      {
+         // find id that comes *before* this symbol entry
+         int blockStart = StringFind(response, patternId, symPos - 100);
+         if(blockStart < 0)
+            blockStart = StringFind(response, patternId, symPos);
+
+         if(blockStart >= 0)
+         {
+            int idStart = blockStart + StringLen(patternId);
+            int idEnd = StringFind(response, ",", idStart);
+            if(idEnd < 0) idEnd = StringFind(response, "}", idStart);
+
+            string idStr = StringSubstr(response, idStart, idEnd - idStart);
+            int id = (int)StringToInteger(idStr);
+
+            // 🟩 Success log
+            Print("✅ Exact match found → symbol='", symbolName, "' | ID=", id);
+            return id;
+         }
+      }
+
+      // move to next
+      pos = symEnd + 1;
+   }
+
+   Print("⚠️ No exact match for symbol ", symbolName);
+   return -1;
 }
 
 //+------------------------------------------------------------------+
@@ -161,11 +204,53 @@ int CreateSymbolInDatabase(string symbolName)
    string exchange = "MT5";
    string instrumentType = "forex";
 
+   // --- Clean suffixes like ".pro", ".m", ".r", "_i" ---
+   int dotPos = StringFind(symbolName, ".");
+   if(dotPos > 0)
+      symbolName = StringSubstr(symbolName, 0, dotPos);
+
+   // --- Try basic split for 6-char pairs like EURUSD, GBPJPY, BTCUSD ---
    if(StringLen(symbolName) >= 6)
    {
       baseAsset = StringSubstr(symbolName, 0, 3);
       quoteAsset = StringSubstr(symbolName, 3, 3);
    }
+   else
+   {
+      // --- Try alternate detection ---
+      if(StringFind(symbolName, "USD") >= 0)
+      {
+         int pos = StringFind(symbolName, "USD");
+         baseAsset = StringSubstr(symbolName, 0, pos);
+         quoteAsset = "USD";
+      }
+      else if(StringFind(symbolName, "EUR") >= 0)
+      {
+         int pos = StringFind(symbolName, "EUR");
+         baseAsset = StringSubstr(symbolName, 0, pos);
+         quoteAsset = "EUR";
+      }
+      else
+      {
+         // Fallback safe defaults
+         baseAsset = symbolName;
+         quoteAsset = "USD";
+      }
+   }
+
+   // --- Final safety: ensure no empty fields ---
+   if(StringLen(baseAsset) == 0) baseAsset = "UNKNOWN";
+   if(StringLen(quoteAsset) == 0) quoteAsset = "USD";
+
+   // --- Decide instrument type ---
+   if(StringFind(symbolName, "BTC") == 0 || StringFind(symbolName, "ETH") == 0)
+      instrumentType = "crypto";
+   else if(StringFind(symbolName, "XAU") == 0 || StringFind(symbolName, "XAG") == 0)
+      instrumentType = "metal";
+   else if(StringFind(symbolName, "US30") == 0 || StringFind(symbolName, "NAS") == 0)
+      instrumentType = "index";
+   else
+      instrumentType = "forex";
 
    string json = StringFormat(
       "{\"symbol\":\"%s\",\"baseAsset\":\"%s\",\"quoteAsset\":\"%s\",\"exchange\":\"%s\",\"instrumentType\":\"%s\",\"isActive\":true}",
@@ -178,18 +263,15 @@ int CreateSymbolInDatabase(string symbolName)
 
    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));
+      PrintFormat("❌ Failed to create symbol %s | HTTP %d | Response: %s", symbolName, res, CharArrayToString(result));
       return -1;
    }
 
@@ -205,10 +287,62 @@ int CreateSymbolInDatabase(string symbolName)
    return (int)StringToInteger(idStr);
 }
 
+//+------------------------------------------------------------------+
+//| Fetch latest stored candle openTime from API                     |
+//+------------------------------------------------------------------+
+datetime GetLatestCandleTime(int symbolId)
+{
+   string url = ApiBaseUrl + "/api/candles/" + IntegerToString(symbolId) + "/latest";
+   string headers = "Content-Type: application/json\r\n";
+   string resultHeaders = "";
+   char result[];
+   char emptyData[];
+
+   ResetLastError();
+   int res = WebRequest("GET", url, headers, 10000, emptyData, result, resultHeaders);
+
+   if(res != 200)
+   {
+      Print("⚠️ Could not fetch latest candle for symbolId=", symbolId, " (HTTP ", res, ")");
+      return 0;
+   }
+
+   string response = CharArrayToString(result);
+   int pos = StringFind(response, "\"openTime\":\"");
+   if(pos < 0)
+   {
+      Print("⚠️ No openTime found in response for symbolId=", symbolId);
+      return 0;
+   }
+
+   pos += StringLen("\"openTime\":\"");
+   int end = StringFind(response, "\"", pos);
+   string openTimeStr = StringSubstr(response, pos, end - pos);
+
+   // --- Parse ISO8601 to datetime ---
+   int year  = (int)StringToInteger(StringSubstr(openTimeStr, 0, 4));
+   int month = (int)StringToInteger(StringSubstr(openTimeStr, 5, 2));
+   int day   = (int)StringToInteger(StringSubstr(openTimeStr, 8, 2));
+   int hour  = (int)StringToInteger(StringSubstr(openTimeStr, 11, 2));
+   int min   = (int)StringToInteger(StringSubstr(openTimeStr, 14, 2));
+   int sec   = (int)StringToInteger(StringSubstr(openTimeStr, 17, 2));
+
+   MqlDateTime t;
+   t.year = year; t.mon = month; t.day = day;
+   t.hour = hour; t.min = min; t.sec = sec;
+
+   datetime dt = StructToTime(t);
+   Print("🕓 Latest stored candle openTime for symbolId=", symbolId, " → ", TimeToString(dt, TIME_DATE|TIME_SECONDS));
+   return dt;
+}
+
 //+------------------------------------------------------------------+
 //+------------------------------------------------------------------+
 //| Send all historical candles to the API (Fixed Version)           |
 //+------------------------------------------------------------------+
+//+------------------------------------------------------------------+
+//| Send all historical candles to the API (Fixed + Timeout Safe)    |
+//+------------------------------------------------------------------+
 void SendAllHistoricalCandles()
 {
    Print("Starting historical upload for ", ArraySize(symbols), " symbols...");
@@ -216,90 +350,200 @@ void SendAllHistoricalCandles()
    for(int i = 0; i < ArraySize(symbols); i++)
    {
       string sym = symbols[i];
+      int symbolId = symbolIds[i];
+      if(symbolId <= 0) continue;
+
+      // --- Get last stored candle time ---
+      datetime latestApiTime = GetLatestCandleTime(symbolId);
 
-      // --- Ensure data is ready ---
+      // --- Ensure history data is available ---
       Sleep(300);
       int tries = 0;
-      while(!SeriesInfoInteger(sym, HistoricalTimeframe, SERIES_SYNCHRONIZED) && tries < 10)
+      bool historyReady = false;
+
+      while(tries < 10)
       {
-         Print("⏳ Waiting for ", sym, " history to load...");
+         if(SeriesInfoInteger(sym, HistoricalTimeframe, SERIES_SYNCHRONIZED))
+         {
+            historyReady = true;
+            break;
+         }
+         PrintFormat("⏳ Waiting for %s history to load... (try %d/10)", sym, tries + 1);
          Sleep(500);
          tries++;
       }
 
-      // --- Now copy candles ---
+      if(!historyReady)
+      {
+         PrintFormat("⚠️ Skipping %s — history not loaded after 10 tries (~5s timeout).", sym);
+         continue;
+      }
+
+      // --- Copy rates ---
       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, ")");
+         int err = GetLastError();
+         PrintFormat("⚠️ Failed to copy candles for %s (error %d)", sym, err);
          continue;
       }
 
-      Print("✅ Copied ", copied, " candles for ", sym);
+      PrintFormat("✅ Copied %d candles for %s", copied, sym);
 
-      // --- Print a few sample candles ---
-      int sampleCount = MathMin(5, copied); // show up to 5 examples
-      for(int j = 0; j < sampleCount; j++)
+      // --- Filter new candles ---
+      int startIndex = 0;
+      for(int j = 0; j < copied; 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
-         );
+         if(rates[j].time > latestApiTime)
+         {
+            startIndex = j;
+            break;
+         }
       }
 
-      // --- Send candles in batches to API ---
-      int sentTotal = 0;
+      int newCount = copied - startIndex;
+      if(newCount <= 0)
+      {
+         PrintFormat("ℹ️ No new candles to send for %s", sym);
+         continue;
+      }
+
+      PrintFormat("🆕 Sending %d new candles for %s after %s", newCount, sym, TimeToString(latestApiTime, TIME_DATE|TIME_SECONDS));
+
+      // --- Send new candles in batches ---
       int batchSize = 200;
-      for(int start = 0; start < copied; start += batchSize)
+      int sentTotal = 0;
+
+      for(int start = startIndex; 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);
+            PrintFormat("❌ Failed to send candle batch for %s (start=%d)", sym, start);
             break;
          }
+
          sentTotal += size;
-         Print("📤 Sent candles for ", sym, ": ", sentTotal, "/", copied);
+         PrintFormat("📤 Sent %d/%d new candles for %s", sentTotal, newCount, sym);
       }
    }
-   Print("✅ Historical upload finished.");
+
+   Print("✅ Incremental candle upload finished.");
 }
 
 
+//+------------------------------------------------------------------+
+//| Send live prices of all active symbols                           |
 //+------------------------------------------------------------------+
 void SendLivePrices()
 {
    bool firstItem = true;
    string json = "{\"prices\":[";
-
    int sentCount = 0;
+
    for(int i = 0; i < ArraySize(symbols); i++)
    {
       string sym = symbols[i];
+      int symId = symbolIds[i];
+      if(symId <= 0) continue;
+
+      // Ensure symbol is visible in Market Watch
+      if(!SymbolSelect(sym, true))
+      {
+         Print("⚠️ Failed to select symbol: ", sym);
+         continue;
+      }
+
+      // Read primary prices
       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;
+      // If last = 0 (some providers), use midprice as fallback
+      if(last <= 0 && bid > 0 && ask > 0)
+         last = (bid + ask) / 2.0;
 
-      string item = StringFormat("{\"symbolId\":%d,\"price\":%.8f,\"bid\":%.8f,\"ask\":%.8f,\"bidSize\":%.8f,\"askSize\":%.8f}",
-                                 symId, last, bid, ask, 0.0, 0.0);
+      // Skip if prices are still invalid
+      if(bid <= 0 || ask <= 0 || last <= 0)
+      {
+         Print("⚠️ Skipping symbol ", sym, " — invalid bid/ask/last (", DoubleToString(bid,8), "/", DoubleToString(ask,8), "/", DoubleToString(last,8), ")");
+         continue;
+      }
+
+      // Initialize sizes
+      double bidSize = 0.0;
+      double askSize = 0.0;
+
+      // Try to fetch market depth (book) and classify volumes by price vs bid/ask
+      MqlBookInfo book[];
+      if(MarketBookGet(sym, book) && ArraySize(book) > 0)
+      {
+         for(int j = 0; j < ArraySize(book); j++)
+         {
+            double p = book[j].price;
+            double v = book[j].volume;
+
+            // If price is >= ask => ask side
+            if(p >= ask) askSize += v;
+            // If price is <= bid => bid side
+            else if(p <= bid) bidSize += v;
+            else
+            {
+               // price in-between -> assign to nearer side
+               double distToBid = MathAbs(p - bid);
+               double distToAsk = MathAbs(ask - p);
+               if(distToBid <= distToAsk) bidSize += v; else askSize += v;
+            }
+         }
+         PrintFormat("ℹ️ MarketBook for %s → bid=%.8f ask=%.8f bidSize=%.2f askSize=%.2f (book entries=%d)", sym, bid, ask, bidSize, askSize, ArraySize(book));
+      }
+      else
+      {
+         // MarketBook not available or empty
+         // Try SymbolInfoTick as fallback
+         MqlTick tick;
+         if(SymbolInfoTick(sym, tick))
+         {
+            // tick.volume is aggregated tick volume — not exact bid/ask sizes but better than zero
+            double tickVol = (double)tick.volume;
+            if(tickVol > 0.0)
+            {
+               // assign tick volume to both sides conservatively
+               if(bidSize <= 0) bidSize = tickVol;
+               if(askSize <= 0) askSize = tickVol;
+               PrintFormat("ℹ️ tick fallback for %s → tick.volume=%.2f", sym, tickVol);
+            }
+            else
+            {
+               Print("ℹ️ tick available but volume zero for ", sym);
+            }
+         }
+         else
+         {
+            Print("ℹ️ MarketBook and tick not available for ", sym);
+         }
+      }
+
+      // Final safety: ensure API-required positive numbers
+      // If a side is zero or negative, set to minimal positive 1.0
+      if(bidSize <= 0.0) bidSize = 1.0;
+      if(askSize <= 0.0) askSize = 1.0;
+
+      // Build JSON item for this symbol
+      string item = StringFormat(
+         "{\"symbolId\":%d,\"price\":%.8f,\"bid\":%.8f,\"ask\":%.8f,\"bidSize\":%.8f,\"askSize\":%.8f}",
+         symId, last, bid, ask, bidSize, askSize
+      );
+
+      // Add to aggregate payload
       if(!firstItem) json += ",";
       json += item;
       firstItem = false;
@@ -310,17 +554,25 @@ void SendLivePrices()
 
    if(sentCount == 0)
    {
-      Print("No valid live prices to send right now.");
+      Print("⚠️ No valid live prices to send right now. Check if market is open and symbols have tick data.");
       return;
    }
 
+   // Log URL and truncated payload for debugging
    string url = ApiBaseUrl + "/api/live-prices/bulk";
+   int maxShow = 1000;
+   string payloadLog = (StringLen(json) > maxShow) ? StringSubstr(json, 0, maxShow) + "...(truncated)" : json;
+   Print("📤 Calling API: ", url);
+   Print("📦 Payload (truncated): ", payloadLog);
+
+   // Send and report
    string response;
    bool ok = SendJSON(url, json, response);
+
    if(ok)
-      Print("✅ Sent ", sentCount, " live prices.");
+      Print("✅ Successfully sent ", sentCount, " live prices to API.");
    else
-      Print("❌ Failed to send live prices (sent ", sentCount, " items). Response: ", response);
+      Print("❌ Failed to send live prices (", sentCount, " items). API response: ", response);
 }
 
 //+------------------------------------------------------------------+
@@ -408,3 +660,57 @@ bool SendJSON(string url, string json, string &response)
 }
 
 //+------------------------------------------------------------------+
+//+------------------------------------------------------------------+
+//| Cleanup old candles (keep only last 1000)                        |
+//+------------------------------------------------------------------+
+void CleanupOldCandles(int symbolId)
+{
+   string url = ApiBaseUrl + "/api/candles/cleanup/" + IntegerToString(symbolId) + "?keep=1000";
+   string headers = "Content-Type: application/json\r\n";
+   string resultHeaders = "";
+   char result[];
+   char emptyData[];
+
+   ResetLastError();
+   int res = WebRequest("DELETE", url, headers, 10000, emptyData, result, resultHeaders);
+
+   string response = CharArrayToString(result);
+   if(res == 200 || res == 204)
+      Print("🧹 Cleanup successful for symbolId=", symbolId, " → kept last 1000 candles.");
+   else
+      Print("⚠️ Cleanup failed for symbolId=", symbolId, " HTTP=", res, " Response=", response);
+}
+
+//+------------------------------------------------------------------+
+//| Timer event: runs every 60 seconds                               |
+//+------------------------------------------------------------------+
+void OnTimer()
+{
+   datetime now = TimeCurrent();
+
+   // ✅ Run full candle sync only once every minute
+   if(now - lastCandleSync >= 600)
+   {
+      Print("⏰ Running scheduled candle sync and cleanup...");
+      SendAllHistoricalCandles();
+
+      // ✅ After uploading candles, clean up old ones
+      for(int i = 0; i < ArraySize(symbols); i++)
+      {
+         int symId = symbolIds[i];
+         if(symId <= 0) continue;
+         CleanupOldCandles(symId);
+         Sleep(500); // small delay to avoid API overload
+      }
+
+      lastCandleSync = now;
+      Print("✅ Candle sync + cleanup cycle completed.");
+   }
+}
+
+
+
+void OnDeinit(const int reason)
+{
+   EventKillTimer();
+}

README.md

@@ -658,7 +658,7 @@ DB_USER=your_username
 DB_PASSWORD=your_password
 
 # Server Configuration
-PORT=3000
+PORT=3001
 NODE_ENV=development
 
 # JWT Configuration (if needed for authentication)

package-lock.json

@@ -17,6 +17,8 @@
         "morgan": "^1.10.1",
         "pg": "^8.16.3",
         "sequelize": "^6.37.7",
+        "socket.io": "^4.8.1",
+        "socket.io-client": "^4.8.1",
         "winston": "^3.18.3"
       },
       "devDependencies": {
@@ -1187,6 +1189,12 @@
         "text-hex": "1.0.x"
       }
     },
+    "node_modules/@socket.io/component-emitter": {
+      "version": "3.1.2",
+      "resolved": "https://registry.npmjs.org/@socket.io/component-emitter/-/component-emitter-3.1.2.tgz",
+      "integrity": "sha512-9BCxFwvbGg/RsZK9tjXd8s4UcwR0MWeFQ1XEKIQVVvAGJyINdrqKMcTRyLoK8Rse1GjzLV9cwjWV1olXRWEXVA==",
+      "license": "MIT"
+    },
     "node_modules/@standard-schema/spec": {
       "version": "1.0.0",
       "resolved": "https://registry.npmjs.org/@standard-schema/spec/-/spec-1.0.0.tgz",
@@ -1249,6 +1257,15 @@
         "@babel/types": "^7.28.2"
       }
     },
+    "node_modules/@types/cors": {
+      "version": "2.8.19",
+      "resolved": "https://registry.npmjs.org/@types/cors/-/cors-2.8.19.tgz",
+      "integrity": "sha512-mFNylyeyqN93lfe/9CSxOGREz8cpzAhH+E93xJ4xWQf62V8sQ/24reV2nyzUWM6H6Xji+GGHpkbLe7pVoUEskg==",
+      "license": "MIT",
+      "dependencies": {
+        "@types/node": "*"
+      }
+    },
     "node_modules/@types/debug": {
       "version": "4.1.12",
       "resolved": "https://registry.npmjs.org/@types/debug/-/debug-4.1.12.tgz",
@@ -1887,6 +1904,15 @@
       "dev": true,
       "license": "MIT"
     },
+    "node_modules/base64id": {
+      "version": "2.0.0",
+      "resolved": "https://registry.npmjs.org/base64id/-/base64id-2.0.0.tgz",
+      "integrity": "sha512-lGe34o6EHj9y3Kts9R4ZYs/Gr+6N7MCaMlIFA3F1R2O5/m7K06AxfSeO5530PEERE6/WyEg3lsuyw4GHlPZHog==",
+      "license": "MIT",
+      "engines": {
+        "node": "^4.5.0 || >= 5.9"
+      }
+    },
     "node_modules/baseline-browser-mapping": {
       "version": "2.8.17",
       "resolved": "https://registry.npmjs.org/baseline-browser-mapping/-/baseline-browser-mapping-2.8.17.tgz",
@@ -2661,6 +2687,125 @@
         "node": ">= 0.8"
       }
     },
+    "node_modules/engine.io": {
+      "version": "6.6.4",
+      "resolved": "https://registry.npmjs.org/engine.io/-/engine.io-6.6.4.tgz",
+      "integrity": "sha512-ZCkIjSYNDyGn0R6ewHDtXgns/Zre/NT6Agvq1/WobF7JXgFff4SeDroKiCO3fNJreU9YG429Sc81o4w5ok/W5g==",
+      "license": "MIT",
+      "dependencies": {
+        "@types/cors": "^2.8.12",
+        "@types/node": ">=10.0.0",
+        "accepts": "~1.3.4",
+        "base64id": "2.0.0",
+        "cookie": "~0.7.2",
+        "cors": "~2.8.5",
+        "debug": "~4.3.1",
+        "engine.io-parser": "~5.2.1",
+        "ws": "~8.17.1"
+      },
+      "engines": {
+        "node": ">=10.2.0"
+      }
+    },
+    "node_modules/engine.io-client": {
+      "version": "6.6.3",
+      "resolved": "https://registry.npmjs.org/engine.io-client/-/engine.io-client-6.6.3.tgz",
+      "integrity": "sha512-T0iLjnyNWahNyv/lcjS2y4oE358tVS/SYQNxYXGAJ9/GLgH4VCvOQ/mhTjqU88mLZCQgiG8RIegFHYCdVC+j5w==",
+      "license": "MIT",
+      "dependencies": {
+        "@socket.io/component-emitter": "~3.1.0",
+        "debug": "~4.3.1",
+        "engine.io-parser": "~5.2.1",
+        "ws": "~8.17.1",
+        "xmlhttprequest-ssl": "~2.1.1"
+      }
+    },
+    "node_modules/engine.io-client/node_modules/debug": {
+      "version": "4.3.7",
+      "resolved": "https://registry.npmjs.org/debug/-/debug-4.3.7.tgz",
+      "integrity": "sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ==",
+      "license": "MIT",
+      "dependencies": {
+        "ms": "^2.1.3"
+      },
+      "engines": {
+        "node": ">=6.0"
+      },
+      "peerDependenciesMeta": {
+        "supports-color": {
+          "optional": true
+        }
+      }
+    },
+    "node_modules/engine.io-parser": {
+      "version": "5.2.3",
+      "resolved": "https://registry.npmjs.org/engine.io-parser/-/engine.io-parser-5.2.3.tgz",
+      "integrity": "sha512-HqD3yTBfnBxIrbnM1DoD6Pcq8NECnh8d4As1Qgh0z5Gg3jRRIqijury0CL3ghu/edArpUYiYqQiDUQBIs4np3Q==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=10.0.0"
+      }
+    },
+    "node_modules/engine.io/node_modules/accepts": {
+      "version": "1.3.8",
+      "resolved": "https://registry.npmjs.org/accepts/-/accepts-1.3.8.tgz",
+      "integrity": "sha512-PYAthTa2m2VKxuvSD3DPC/Gy+U+sOA1LAuT8mkmRuvw+NACSaeXEQ+NHcVF7rONl6qcaxV3Uuemwawk+7+SJLw==",
+      "license": "MIT",
+      "dependencies": {
+        "mime-types": "~2.1.34",
+        "negotiator": "0.6.3"
+      },
+      "engines": {
+        "node": ">= 0.6"
+      }
+    },
+    "node_modules/engine.io/node_modules/debug": {
+      "version": "4.3.7",
+      "resolved": "https://registry.npmjs.org/debug/-/debug-4.3.7.tgz",
+      "integrity": "sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ==",
+      "license": "MIT",
+      "dependencies": {
+        "ms": "^2.1.3"
+      },
+      "engines": {
+        "node": ">=6.0"
+      },
+      "peerDependenciesMeta": {
+        "supports-color": {
+          "optional": true
+        }
+      }
+    },
+    "node_modules/engine.io/node_modules/mime-db": {
+      "version": "1.52.0",
+      "resolved": "https://registry.npmjs.org/mime-db/-/mime-db-1.52.0.tgz",
+      "integrity": "sha512-sPU4uV7dYlvtWJxwwxHD0PuihVNiE7TyAbQ5SWxDCB9mUYvOgroQOwYQQOKPJ8CIbE+1ETVlOoK1UC2nU3gYvg==",
+      "license": "MIT",
+      "engines": {
+        "node": ">= 0.6"
+      }
+    },
+    "node_modules/engine.io/node_modules/mime-types": {
+      "version": "2.1.35",
+      "resolved": "https://registry.npmjs.org/mime-types/-/mime-types-2.1.35.tgz",
+      "integrity": "sha512-ZDY+bPm5zTTF+YpCrAU9nK0UgICYPT0QtT1NZWFv4s++TNkcgVaT0g6+4R2uI4MjQjzysHB1zxuWL50hzaeXiw==",
+      "license": "MIT",
+      "dependencies": {
+        "mime-db": "1.52.0"
+      },
+      "engines": {
+        "node": ">= 0.6"
+      }
+    },
+    "node_modules/engine.io/node_modules/negotiator": {
+      "version": "0.6.3",
+      "resolved": "https://registry.npmjs.org/negotiator/-/negotiator-0.6.3.tgz",
+      "integrity": "sha512-+EUsqGPLsM+j/zdChZjsnX51g4XrHFOIXwfnCVPGlQk/k5giakcKsuxCObBRu6DSm9opw/O6slWbJdghQM4bBg==",
+      "license": "MIT",
+      "engines": {
+        "node": ">= 0.6"
+      }
+    },
     "node_modules/error-ex": {
       "version": "1.3.4",
       "resolved": "https://registry.npmjs.org/error-ex/-/error-ex-1.3.4.tgz",
@@ -5906,6 +6051,173 @@
         "node": ">=8"
       }
     },
+    "node_modules/socket.io": {
+      "version": "4.8.1",
+      "resolved": "https://registry.npmjs.org/socket.io/-/socket.io-4.8.1.tgz",
+      "integrity": "sha512-oZ7iUCxph8WYRHHcjBEc9unw3adt5CmSNlppj/5Q4k2RIrhl8Z5yY2Xr4j9zj0+wzVZ0bxmYoGSzKJnRl6A4yg==",
+      "license": "MIT",
+      "dependencies": {
+        "accepts": "~1.3.4",
+        "base64id": "~2.0.0",
+        "cors": "~2.8.5",
+        "debug": "~4.3.2",
+        "engine.io": "~6.6.0",
+        "socket.io-adapter": "~2.5.2",
+        "socket.io-parser": "~4.2.4"
+      },
+      "engines": {
+        "node": ">=10.2.0"
+      }
+    },
+    "node_modules/socket.io-adapter": {
+      "version": "2.5.5",
+      "resolved": "https://registry.npmjs.org/socket.io-adapter/-/socket.io-adapter-2.5.5.tgz",
+      "integrity": "sha512-eLDQas5dzPgOWCk9GuuJC2lBqItuhKI4uxGgo9aIV7MYbk2h9Q6uULEh8WBzThoI7l+qU9Ast9fVUmkqPP9wYg==",
+      "license": "MIT",
+      "dependencies": {
+        "debug": "~4.3.4",
+        "ws": "~8.17.1"
+      }
+    },
+    "node_modules/socket.io-adapter/node_modules/debug": {
+      "version": "4.3.7",
+      "resolved": "https://registry.npmjs.org/debug/-/debug-4.3.7.tgz",
+      "integrity": "sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ==",
+      "license": "MIT",
+      "dependencies": {
+        "ms": "^2.1.3"
+      },
+      "engines": {
+        "node": ">=6.0"
+      },
+      "peerDependenciesMeta": {
+        "supports-color": {
+          "optional": true
+        }
+      }
+    },
+    "node_modules/socket.io-client": {
+      "version": "4.8.1",
+      "resolved": "https://registry.npmjs.org/socket.io-client/-/socket.io-client-4.8.1.tgz",
+      "integrity": "sha512-hJVXfu3E28NmzGk8o1sHhN3om52tRvwYeidbj7xKy2eIIse5IoKX3USlS6Tqt3BHAtflLIkCQBkzVrEEfWUyYQ==",
+      "license": "MIT",
+      "dependencies": {
+        "@socket.io/component-emitter": "~3.1.0",
+        "debug": "~4.3.2",
+        "engine.io-client": "~6.6.1",
+        "socket.io-parser": "~4.2.4"
+      },
+      "engines": {
+        "node": ">=10.0.0"
+      }
+    },
+    "node_modules/socket.io-client/node_modules/debug": {
+      "version": "4.3.7",
+      "resolved": "https://registry.npmjs.org/debug/-/debug-4.3.7.tgz",
+      "integrity": "sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ==",
+      "license": "MIT",
+      "dependencies": {
+        "ms": "^2.1.3"
+      },
+      "engines": {
+        "node": ">=6.0"
+      },
+      "peerDependenciesMeta": {
+        "supports-color": {
+          "optional": true
+        }
+      }
+    },
+    "node_modules/socket.io-parser": {
+      "version": "4.2.4",
+      "resolved": "https://registry.npmjs.org/socket.io-parser/-/socket.io-parser-4.2.4.tgz",
+      "integrity": "sha512-/GbIKmo8ioc+NIWIhwdecY0ge+qVBSMdgxGygevmdHj24bsfgtCmcUUcQ5ZzcylGFHsN3k4HB4Cgkl96KVnuew==",
+      "license": "MIT",
+      "dependencies": {
+        "@socket.io/component-emitter": "~3.1.0",
+        "debug": "~4.3.1"
+      },
+      "engines": {
+        "node": ">=10.0.0"
+      }
+    },
+    "node_modules/socket.io-parser/node_modules/debug": {
+      "version": "4.3.7",
+      "resolved": "https://registry.npmjs.org/debug/-/debug-4.3.7.tgz",
+      "integrity": "sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ==",
+      "license": "MIT",
+      "dependencies": {
+        "ms": "^2.1.3"
+      },
+      "engines": {
+        "node": ">=6.0"
+      },
+      "peerDependenciesMeta": {
+        "supports-color": {
+          "optional": true
+        }
+      }
+    },
+    "node_modules/socket.io/node_modules/accepts": {
+      "version": "1.3.8",
+      "resolved": "https://registry.npmjs.org/accepts/-/accepts-1.3.8.tgz",
+      "integrity": "sha512-PYAthTa2m2VKxuvSD3DPC/Gy+U+sOA1LAuT8mkmRuvw+NACSaeXEQ+NHcVF7rONl6qcaxV3Uuemwawk+7+SJLw==",
+      "license": "MIT",
+      "dependencies": {
+        "mime-types": "~2.1.34",
+        "negotiator": "0.6.3"
+      },
+      "engines": {
+        "node": ">= 0.6"
+      }
+    },
+    "node_modules/socket.io/node_modules/debug": {
+      "version": "4.3.7",
+      "resolved": "https://registry.npmjs.org/debug/-/debug-4.3.7.tgz",
+      "integrity": "sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ==",
+      "license": "MIT",
+      "dependencies": {
+        "ms": "^2.1.3"
+      },
+      "engines": {
+        "node": ">=6.0"
+      },
+      "peerDependenciesMeta": {
+        "supports-color": {
+          "optional": true
+        }
+      }
+    },
+    "node_modules/socket.io/node_modules/mime-db": {
+      "version": "1.52.0",
+      "resolved": "https://registry.npmjs.org/mime-db/-/mime-db-1.52.0.tgz",
+      "integrity": "sha512-sPU4uV7dYlvtWJxwwxHD0PuihVNiE7TyAbQ5SWxDCB9mUYvOgroQOwYQQOKPJ8CIbE+1ETVlOoK1UC2nU3gYvg==",
+      "license": "MIT",
+      "engines": {
+        "node": ">= 0.6"
+      }
+    },
+    "node_modules/socket.io/node_modules/mime-types": {
+      "version": "2.1.35",
+      "resolved": "https://registry.npmjs.org/mime-types/-/mime-types-2.1.35.tgz",
+      "integrity": "sha512-ZDY+bPm5zTTF+YpCrAU9nK0UgICYPT0QtT1NZWFv4s++TNkcgVaT0g6+4R2uI4MjQjzysHB1zxuWL50hzaeXiw==",
+      "license": "MIT",
+      "dependencies": {
+        "mime-db": "1.52.0"
+      },
+      "engines": {
+        "node": ">= 0.6"
+      }
+    },
+    "node_modules/socket.io/node_modules/negotiator": {
+      "version": "0.6.3",
+      "resolved": "https://registry.npmjs.org/negotiator/-/negotiator-0.6.3.tgz",
+      "integrity": "sha512-+EUsqGPLsM+j/zdChZjsnX51g4XrHFOIXwfnCVPGlQk/k5giakcKsuxCObBRu6DSm9opw/O6slWbJdghQM4bBg==",
+      "license": "MIT",
+      "engines": {
+        "node": ">= 0.6"
+      }
+    },
     "node_modules/source-map": {
       "version": "0.6.1",
       "resolved": "https://registry.npmjs.org/source-map/-/source-map-0.6.1.tgz",
@@ -6745,6 +7057,35 @@
         "node": "^14.17.0 || ^16.13.0 || >=18.0.0"
       }
     },
+    "node_modules/ws": {
+      "version": "8.17.1",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-8.17.1.tgz",
+      "integrity": "sha512-6XQFvXTkbfUOZOKKILFG1PDK2NDQs4azKQl26T0YS5CxqWLgXajbPZ+h4gZekJyRqFU8pvnbAbbs/3TgRPy+GQ==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=10.0.0"
+      },
+      "peerDependencies": {
+        "bufferutil": "^4.0.1",
+        "utf-8-validate": ">=5.0.2"
+      },
+      "peerDependenciesMeta": {
+        "bufferutil": {
+          "optional": true
+        },
+        "utf-8-validate": {
+          "optional": true
+        }
+      }
+    },
+    "node_modules/xmlhttprequest-ssl": {
+      "version": "2.1.2",
+      "resolved": "https://registry.npmjs.org/xmlhttprequest-ssl/-/xmlhttprequest-ssl-2.1.2.tgz",
+      "integrity": "sha512-TEU+nJVUUnA4CYJFLvK5X9AOeH4KvDvhIfm0vV1GaQRtchnG0hgK5p8hw/xjv8cunWYCsiPCSDzObPyhEwq3KQ==",
+      "engines": {
+        "node": ">=0.4.0"
+      }
+    },
     "node_modules/xtend": {
       "version": "4.0.2",
       "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz",

package.json

@@ -30,6 +30,8 @@
     "morgan": "^1.10.1",
     "pg": "^8.16.3",
     "sequelize": "^6.37.7",
+    "socket.io": "^4.8.1",
+    "socket.io-client": "^4.8.1",
     "winston": "^3.18.3"
   },
   "devDependencies": {

src/controllers/candleController.js

@@ -120,6 +120,29 @@ class CandleController {
 
       const candle = await Candle1h.create(candleData);
 
+      // Emit WebSocket event for real-time updates
+      const io = req.app.get('io');
+      if (io) {
+        const eventData = {
+          symbol: symbol.symbol,
+          symbolId: symbol.id,
+          openTime: candle.openTime,
+          open: candle.open,
+          high: candle.high,
+          low: candle.low,
+          close: candle.close,
+          volume: candle.volume,
+          exchange: symbol.exchange,
+          instrumentType: symbol.instrumentType
+        };
+
+        // Emit to all clients subscribed to this symbol
+        io.to(`symbol:${symbol.symbol}`).emit('candleUpdate', eventData);
+
+        // Also emit to general candle updates
+        io.emit('candleUpdate', eventData);
+      }
+
       res.status(201).json({
         success: true,
         data: candle,
@@ -159,6 +182,47 @@ class CandleController {
 
       const createdCandles = await Candle1h.bulkCreate(candles);
 
+      // Emit WebSocket events for real-time updates
+      const io = req.app.get('io');
+      if (io) {
+        // Group candles by symbol for efficient emission
+        const symbolGroups = {};
+        for (const candle of createdCandles) {
+          if (!symbolGroups[candle.symbolId]) {
+            symbolGroups[candle.symbolId] = [];
+          }
+          symbolGroups[candle.symbolId].push(candle);
+        }
+
+        // Emit events for each symbol
+        for (const [symbolId, symbolCandles] of Object.entries(symbolGroups)) {
+          const symbol = await Symbol.findByPk(symbolId);
+          if (symbol) {
+            // Emit the latest candle for this symbol (most recent)
+            const latestCandle = symbolCandles.sort((a, b) => new Date(b.openTime) - new Date(a.openTime))[0];
+
+            const eventData = {
+              symbol: symbol.symbol,
+              symbolId: symbol.id,
+              openTime: latestCandle.openTime,
+              open: latestCandle.open,
+              high: latestCandle.high,
+              low: latestCandle.low,
+              close: latestCandle.close,
+              volume: latestCandle.volume,
+              exchange: symbol.exchange,
+              instrumentType: symbol.instrumentType
+            };
+
+            // Emit to all clients subscribed to this symbol
+            io.to(`symbol:${symbol.symbol}`).emit('candleUpdate', eventData);
+
+            // Also emit to general candle updates
+            io.emit('candleUpdate', eventData);
+          }
+        }
+      }
+
       res.status(201).json({
         success: true,
         data: createdCandles,

src/controllers/livePriceController.js

@@ -90,6 +90,29 @@ class LivePriceController {
         lastUpdated: new Date()
       });
 
+      // Emit WebSocket event for real-time updates
+      const io = req.app.get('io');
+      if (io) {
+        const eventData = {
+          symbol: symbol.symbol,
+          symbolId: symbol.id,
+          price,
+          bid,
+          ask,
+          bidSize,
+          askSize,
+          lastUpdated: livePrice.lastUpdated,
+          exchange: symbol.exchange,
+          instrumentType: symbol.instrumentType
+        };
+
+        // Emit to all clients subscribed to this symbol
+        io.to(`symbol:${symbol.symbol}`).emit('livePriceUpdate', eventData);
+
+        // Also emit to general live price updates
+        io.emit('livePriceUpdate', eventData);
+      }
+
       res.status(created ? 201 : 200).json({
         success: true,
         data: livePrice,
@@ -142,9 +165,36 @@ class LivePriceController {
       }));
 
       const updatedPrices = [];
+      const io = req.app.get('io');
+
       for (const data of upsertData) {
         const [livePrice] = await LivePrice.upsert(data);
         updatedPrices.push(livePrice);
+
+        // Emit WebSocket event for each updated price
+        if (io) {
+          const symbol = await Symbol.findByPk(data.symbolId);
+          if (symbol) {
+            const eventData = {
+              symbol: symbol.symbol,
+              symbolId: symbol.id,
+              price: data.price,
+              bid: data.bid,
+              ask: data.ask,
+              bidSize: data.bidSize,
+              askSize: data.askSize,
+              lastUpdated: data.lastUpdated,
+              exchange: symbol.exchange,
+              instrumentType: symbol.instrumentType
+            };
+
+            // Emit to all clients subscribed to this symbol
+            io.to(`symbol:${symbol.symbol}`).emit('livePriceUpdate', eventData);
+
+            // Also emit to general live price updates
+            io.emit('livePriceUpdate', eventData);
+          }
+        }
       }
 
       res.json({

src/server.js

@@ -1,6 +1,8 @@
 const app = require('./app');
 const { testConnection } = require('./config/database');
 const logger = require('./utils/logger');
+const { createServer } = require('http');
+const { Server } = require('socket.io');
 
 const PORT = process.env.PORT || 3000;
 
@@ -10,11 +12,63 @@ const startServer = async () => {
     // Test database connection
     await testConnection();
 
+    // Create HTTP server
+    const server = createServer(app);
+
+    // Initialize Socket.io
+    const io = new Server(server, {
+      cors: {
+        origin: process.env.CORS_ORIGIN || "*",
+        methods: ["GET", "POST"],
+        credentials: true
+      }
+    });
+
+    // Socket.io connection handling
+    io.on('connection', (socket) => {
+      logger.info(`Client connected: ${socket.id}`);
+
+      // Handle subscription to symbols
+      socket.on('subscribe', (symbols) => {
+        if (Array.isArray(symbols)) {
+          symbols.forEach(symbol => {
+            socket.join(`symbol:${symbol}`);
+            logger.info(`Client ${socket.id} subscribed to ${symbol}`);
+          });
+        } else {
+          socket.join(`symbol:${symbols}`);
+          logger.info(`Client ${socket.id} subscribed to ${symbols}`);
+        }
+      });
+
+      // Handle unsubscription
+      socket.on('unsubscribe', (symbols) => {
+        if (Array.isArray(symbols)) {
+          symbols.forEach(symbol => {
+            socket.leave(`symbol:${symbol}`);
+            logger.info(`Client ${socket.id} unsubscribed from ${symbol}`);
+          });
+        } else {
+          socket.leave(`symbol:${symbols}`);
+          logger.info(`Client ${socket.id} unsubscribed from ${symbols}`);
+        }
+      });
+
+      // Handle disconnect
+      socket.on('disconnect', () => {
+        logger.info(`Client disconnected: ${socket.id}`);
+      });
+    });
+
+    // Make io accessible to routes/controllers
+    app.set('io', io);
+
     // Start the server
-    const server = app.listen(PORT, () => {
+    server.listen(PORT, () => {
       logger.info(`Market Data Service is running on port ${PORT}`);
       logger.info(`Environment: ${process.env.NODE_ENV || 'development'}`);
       logger.info(`Health check available at: http://localhost:${PORT}/health`);
+      logger.info(`WebSocket server ready for connections`);
     });
 
     // Graceful shutdown