10. WebSocket Market Data

Nhận giá khớp, bảng giá, room nước ngoài tức thời qua WebSocket

Mục tiêu

Nhận tick data real-time (giá khớp, bảng giá, room nước ngoài) qua WebSocket. Không cần OTP cho market data.

Luồng xử lý

Client ↔ Streaming API ↔ Market Data Event

Mở kết nối WebSocket bằng token hợp lệ (không cần OTP)
Subscribe stream theo symbol (subscribe_symbol), theo các mã thành phần của chỉ số (subscribe_index) và theo tick của chính chỉ số (subscribe_index_ticks)
Server push event khi có giao dịch mới hoặc bảng giá thay đổi
Callback on_data phân loại message: IndexTickMessage, TradeMessage, QuoteMessage, ForeignRoomMessage
Khi mất kết nối, SDK có cơ chế reconnect

Các loại Message

Type	Mô tả
`TradeMessage`	Giá khớp, khối lượng, chiều mua/bán
`QuoteMessage`	Bảng giá bid/ask
`ForeignRoomMessage`	Room nước ngoài còn lại
`IndexTickMessage`	Tick của chính chỉ số (điểm, thay đổi), nhận qua `subscribe_index_ticks`
`HeartbeatMessage`	Keepalive từ server

Từ SDK 3.3.0, subscribe_index(["VN30"]) gửi trade/quote/room.<code>@* nên nhận dữ liệu của các mã thành phần (TradeMessage, QuoteMessage...), không phải tick của chỉ số. Muốn tick của chính chỉ số (ví dụ VNINDEX) thì dùng subscribe_index_ticks. Xem thêm WebSocket nâng cao.

Sample Code — Sync

python/sample_10_websocket_data.py

from ssi_sdk import Auth, Stream, Config
from auth_helper import ensure_auth
from ssi_sdk.models.streaming import (
    TradeMessage, QuoteMessage, ForeignRoomMessage,
    IndexTickMessage, HeartbeatMessage,
)

config = Config(
    client_id="<your_client_id>",
    api_key="<your_api_key>",
    api_secret="<your_api_secret>",
    private_key="<your_private_key>",
)

# Callback xử lý dữ liệu
def on_market_data(msg):
    if isinstance(msg, IndexTickMessage):
        # Tick của chính chỉ số (subscribe_index_ticks), KHÔNG phải tick của mã thành phần
        print(f"  [INDEX] {msg.symbol} | Điểm: {msg.value} | Thay đổi: {msg.change} ({msg.change_percent}%)")
    elif isinstance(msg, TradeMessage):
        print(f"  [TRADE] {msg.symbol} | Giá: {msg.price} | KL: {msg.quantity} | Side: {msg.side}")
    elif isinstance(msg, QuoteMessage):
        print(f"  [QUOTE] {msg.symbol} | Bid: {msg.bid_prices[:3]} | Ask: {msg.ask_prices[:3]}")
    elif isinstance(msg, ForeignRoomMessage):
        print(f"  [ROOM]  {msg.symbol} | Room còn: {msg.current_room}/{msg.total_room}")
    else:
        print(f"  [DATA]  {msg}")

def on_heartbeat(msg: HeartbeatMessage):
    print(f"  [HEARTBEAT] {msg.status} - {msg.message}")

with Auth(config) as auth:
    ensure_auth(auth)  # Không cần OTP

    with Stream(auth) as stream:
        stream.streaming.connect()
        stream.streaming.on_data = on_market_data
        stream.streaming.on_heartbeat = on_heartbeat

        # Subscribe theo symbol
        stream.streaming.subscribe_symbol(["SSI", "HPG", "VNM"])

        # Từ SDK 3.3.0, subscribe_index gửi trade/quote/room.<code>@* nên server
        # phát dữ liệu của CÁC MÃ THÀNH PHẦN (vd VN30), không phải tick của chỉ số.
        stream.streaming.subscribe_index(["VN30"])
        # Muốn nhận tick của chính chỉ số (IndexTickMessage) thì dùng subscribe_index_ticks.
        stream.streaming.subscribe_index_ticks(["VNINDEX"])

        # Lắng nghe 5 phút (đơn vị mili-giây)
        try:
            stream.streaming.wait(timeout_ms=300_000)
        except KeyboardInterrupt:
            print("Ngắt kết nối...")

Sample Code — Async

python/sample_10_websocket_data_async.py

import asyncio
from ssi_sdk import AsyncAuth, AsyncStream, Config
from auth_helper import ensure_auth_async
from ssi_sdk.models.streaming import (
    TradeMessage, QuoteMessage, ForeignRoomMessage, IndexTickMessage,
)

config = Config(...)

def on_market_data(msg):
    if isinstance(msg, IndexTickMessage):
        print(f"  [INDEX] {msg.symbol} | Điểm: {msg.value}")
    elif isinstance(msg, TradeMessage):
        print(f"  [TRADE] {msg.symbol} | {msg.price} x {msg.quantity}")
    elif isinstance(msg, QuoteMessage):
        print(f"  [QUOTE] {msg.symbol}")

async def main():
    async with AsyncAuth(config) as auth:
        await ensure_auth_async(auth)
        async with AsyncStream(auth) as stream:
            await stream.streaming.connect()
            stream.streaming.on_data = on_market_data

            await stream.streaming.subscribe_symbol(["SSI", "HPG", "VNM"])
            await stream.streaming.subscribe_index(["VN30"])         # các mã thành phần của VN30
            await stream.streaming.subscribe_index_ticks(["VNINDEX"])  # tick của chính chỉ số

            try:
                await stream.streaming.wait(timeout_ms=300_000)
            except KeyboardInterrupt:
                print("Disconnecting...")

asyncio.run(main())

Lưu ý

Không cần OTP cho market data streaming.
SDK tự xử lý reconnect khi mất kết nối.
Sử dụng wait(timeout_ms=...) để giới hạn thời gian lắng nghe. Từ SDK 3.3.0 đơn vị là mili-giây; tên cũ timeout= (giây) đã bị xóa và báo TypeError.
Có thể subscribe nhiều symbol và index cùng lúc.
subscribe_index nhận dữ liệu các mã thành phần; subscribe_index_ticks nhận tick của chính chỉ số. Các tính năng khác (board, index summary, cờ phiên, nến, sự kiện kết nối) xem WebSocket nâng cao.