> ## Documentation Index
> Fetch the complete documentation index at: https://docs.oms.verolabs.co/llms.txt
> Use this file to discover all available pages before exploring further.

# Nhận cập nhật dữ liệu thị trường tổng quát

> Nhận cập nhật dữ liệu thị trường tổng quát

## Channel

```text theme={null}
mkt:{dataType}:{symbol}
```

## Mô tả

Dùng kênh này khi cần stream dữ liệu thị trường theo `dataType` và `symbol`. Tạo channel theo `mkt:${dataType}:${symbol}` rồi subscribe sau khi WebSocket kết nối thành công. Với depth, OHLCV và product info, ưu tiên các trang channel riêng vì các stream đó dùng định dạng channel cụ thể hơn.

## WebSocket URL

```text theme={null}
wss://streaming.verolabs.co/connection/websocket
```

## Ví dụ subscribe

Các ví dụ bên dưới subscribe vào:

```text theme={null}
mkt:tradeLog:VN30F1M
```

Thay `tradeLog` hoặc `VN30F1M` bằng data type và mã giao dịch cần subscribe.

<CodeGroup dropdown>
  ```javascript JavaScript theme={null}
  const streamUrl = "wss://streaming.verolabs.co/connection/websocket";
  const channel = "mkt:tradeLog:VN30F1M";

  const socket = new WebSocket(streamUrl);
  let requestId = 1;

  socket.addEventListener("open", () => {
    send({ id: requestId++, connect: {} });
  });

  socket.addEventListener("message", (event) => {
    if (typeof event.data !== "string") return;

    const replies = event.data
      .split("\n")
      .filter(Boolean)
      .map((item) => JSON.parse(item));

    for (const reply of replies) {
      if (Object.keys(reply).length === 0) {
        send({});
        continue;
      }

      if (reply.connect && !reply.error) {
        send({ id: requestId++, subscribe: { channel } });
        continue;
      }

      const payload = reply.push?.pub?.data;

      if (reply.push?.channel === channel && payload !== undefined) {
        console.log(payload);
      }
    }
  });

  function send(command) {
    socket.send(JSON.stringify(command) + "\n");
  }
  ```

  ```python Python theme={null}
  import asyncio
  import json

  import websockets

  STREAM_URL = "wss://streaming.verolabs.co/connection/websocket"
  channel = "mkt:tradeLog:VN30F1M"


  async def main():
      global channel

      async with websockets.connect(STREAM_URL) as ws:
          request_id = 1

          async def send(command):
              await ws.send(json.dumps(command) + "\n")

          await send({"id": request_id, "connect": {}})
          request_id += 1

          async for message in ws:
              for raw in filter(None, message.split("\n")):
                  reply = json.loads(raw)

                  if reply == {}:
                      await send({})
                      continue

                  if "connect" in reply and "error" not in reply:
                      await send({"id": request_id, "subscribe": {"channel": channel}})
                      request_id += 1
                      continue

                  push = reply.get("push", {})
                  pub = push.get("pub", {})

                  if push.get("channel") == channel and "data" in pub:
                      print(pub["data"])


  asyncio.run(main())
  ```

  ```go Go theme={null}
  package main

  import (
  	"encoding/json"
  	"fmt"
  	"strings"

  	"github.com/gorilla/websocket"
  )

  func main() {
  	streamURL := "wss://streaming.verolabs.co/connection/websocket"
  	channel := "mkt:tradeLog:VN30F1M"

  	conn, _, err := websocket.DefaultDialer.Dial(streamURL, nil)
  	if err != nil {
  		panic(err)
  	}
  	defer conn.Close()

  	requestID := 1

  	send := func(command map[string]any) {
  		payload, err := json.Marshal(command)
  		if err != nil {
  			panic(err)
  		}

  		if err := conn.WriteMessage(websocket.TextMessage, append(payload, '\n')); err != nil {
  			panic(err)
  		}
  	}

  	send(map[string]any{"id": requestID, "connect": map[string]any{}})
  	requestID++

  	for {
  		_, data, err := conn.ReadMessage()
  		if err != nil {
  			panic(err)
  		}

  		for _, raw := range strings.Split(string(data), "\n") {
  			if raw == "" {
  				continue
  			}

  			var reply map[string]any
  			if err := json.Unmarshal([]byte(raw), &reply); err != nil {
  				panic(err)
  			}

  			if len(reply) == 0 {
  				send(map[string]any{})
  				continue
  			}

  			if _, ok := reply["connect"]; ok && reply["error"] == nil {
  				send(map[string]any{"id": requestID, "subscribe": map[string]any{"channel": channel}})
  				requestID++
  				continue
  			}

  			push, _ := reply["push"].(map[string]any)
  			pub, _ := push["pub"].(map[string]any)

  			if push["channel"] == channel {
  				if payload, ok := pub["data"]; ok {
  					fmt.Printf("%#v\n", payload)
  				}
  			}
  		}
  	}
  }
  ```

  ```csharp C# theme={null}
  using System;
  using System.IO;
  using System.Linq;
  using System.Net.WebSockets;
  using System.Text;
  using System.Text.Json;
  using System.Threading;

  var streamUrl = new Uri("wss://streaming.verolabs.co/connection/websocket");
  var channel = "mkt:tradeLog:VN30F1M";

  using var socket = new ClientWebSocket();
  await socket.ConnectAsync(streamUrl, CancellationToken.None);

  var requestId = 1;
  await SendAsync(new { id = requestId++, connect = new { } });

  while (socket.State == WebSocketState.Open)
  {
      var message = await ReceiveStringAsync(socket);

      foreach (var raw in message.Split('\n', StringSplitOptions.RemoveEmptyEntries))
      {
          using var document = JsonDocument.Parse(raw);
          var root = document.RootElement;

          if (root.ValueKind == JsonValueKind.Object && !root.EnumerateObject().Any())
          {
              await SendAsync(new { });
              continue;
          }

          if (root.TryGetProperty("connect", out _) && !root.TryGetProperty("error", out _))
          {
              await SendAsync(new { id = requestId++, subscribe = new { channel } });
              continue;
          }

          if (root.TryGetProperty("push", out var push) &&
              push.TryGetProperty("channel", out var pushChannel) &&
              pushChannel.GetString() == channel &&
              push.TryGetProperty("pub", out var pub) &&
              pub.TryGetProperty("data", out var data))
          {
              Console.WriteLine(data.GetRawText());
          }
      }
  }

  async Task SendAsync(object command)
  {
      var json = JsonSerializer.Serialize(command) + "\n";
      var bytes = Encoding.UTF8.GetBytes(json);
      await socket.SendAsync(bytes, WebSocketMessageType.Text, true, CancellationToken.None);
  }

  async Task<string> ReceiveStringAsync(ClientWebSocket client)
  {
      var buffer = new byte[64 * 1024];
      using var message = new MemoryStream();
      WebSocketReceiveResult result;

      do
      {
          result = await client.ReceiveAsync(buffer, CancellationToken.None);
          message.Write(buffer, 0, result.Count);
      } while (!result.EndOfMessage);

      return Encoding.UTF8.GetString(message.ToArray());
  }
  ```

  ```rust Rust theme={null}
  use futures_util::{SinkExt, StreamExt};
  use serde_json::{json, Value};
  use tokio_tungstenite::{connect_async, tungstenite::Message};

  #[tokio::main]
  async fn main() -> anyhow::Result<()> {
      let stream_url = "wss://streaming.verolabs.co/connection/websocket";
      let channel = "mkt:tradeLog:VN30F1M";

      let (mut ws, _) = connect_async(stream_url).await?;
      let mut request_id = 1;

      ws.send(Message::Text(format!("{}\n", json!({
          "id": request_id,
          "connect": {}
      })).into())).await?;
      request_id += 1;

      while let Some(message) = ws.next().await {
          let Message::Text(text) = message? else { continue };

          for raw in text.split('\n').filter(|value| !value.is_empty()) {
              let reply: Value = serde_json::from_str(raw)?;

              if reply.as_object().is_some_and(|object| object.is_empty()) {
                  ws.send(Message::Text("{}\n".into())).await?;
                  continue;
              }

              if reply.get("connect").is_some() && reply.get("error").is_none() {
              ws.send(Message::Text(format!("{}\n", json!({
                  "id": request_id,
                  "subscribe": { "channel": channel }
              })).into())).await?;
                  request_id += 1;
                  continue;
              }

              if reply.pointer("/push/channel").and_then(Value::as_str) == Some(channel) {
                  if let Some(payload) = reply.pointer("/push/pub/data") {
                      println!("{payload}");
                  }
              }
          }
      }

      Ok(())
  }
  ```

  ```cpp C++ theme={null}
  #include <ixwebsocket/IXWebSocket.h>
  #include <ixwebsocket/IXNetSystem.h>
  #include <nlohmann/json.hpp>
  #include <iostream>
  #include <sstream>
  #include <string>

  using json = nlohmann::json;

  int main() {
      ix::initNetSystem();

      std::string stream_url = "wss://streaming.verolabs.co/connection/websocket";
      std::string channel = "mkt:tradeLog:VN30F1M";

      ix::WebSocket webSocket;
      webSocket.setUrl(stream_url);

      int request_id = 1;

      auto send_command = [&webSocket](const json& command) {
          if (command.empty()) {
              webSocket.send("\n");
          } else {
              webSocket.send(command.dump() + "\n");
          }
      };

      webSocket.setOnMessageCallback([&](const ix::WebSocketMessagePtr& msg) {
          if (msg->type == ix::WebSocketMessageType::Open) {
              send_command({
                  {"id", request_id++},
                  {"connect", json::object()}
              });
          }
          else if (msg->type == ix::WebSocketMessageType::Message) {
              std::stringstream ss(msg->str);
              std::string raw;
              while (std::getline(ss, raw, '\n')) {
                  if (raw.empty()) continue;

                  try {
                      json reply = json::parse(raw);

                      if (reply.empty()) {
                          send_command(json::object());
                          continue;
                      }

                      if (reply.contains("connect") && !reply.contains("error")) {
                          send_command({
                              {"id", request_id++},
                              {"subscribe", {{"channel", channel}}}
                          });
                          continue;
                      }

                      if (reply.contains("push")) {
                          auto push = reply["push"];
                          if (push.value("channel", "") == channel && push.contains("pub")) {
                              auto pub = push["pub"];
                              if (pub.contains("data")) {
                                  std::cout << pub["data"].dump() << std::endl;
                              }
                          }
                      }
                  } catch (const std::exception& e) {
                      std::cerr << "Parse error: " << e.what() << std::endl;
                  }
              }
          }
      });

      webSocket.start();

      std::string line;
      std::getline(std::cin, line);

      webSocket.stop();
      return 0;
  }
  ```
</CodeGroup>

## Tham số channel

| Tham số    | Kiểu   | Mô tả                                   | Giá trị                                         |
| ---------- | ------ | --------------------------------------- | ----------------------------------------------- |
| `dataType` | string | Loại dữ liệu thị trường của stream này. | enum: depth, productInfo, productStat, tradeLog |
| `symbol`   | string | Mã giao dịch, ví dụ VN30F1M.            | example: VN30F1M                                |

## Message

| Mục       | Giá trị                    |
| --------- | -------------------------- |
| Operation | `receiveGenericMarketData` |
| Tên       | `GenericMarketData`        |
| Payload   | `GenericObject`            |

## Payload schema

Payload là object JSON linh hoạt và có thể chứa thêm field theo publication thực tế.