16. WebSocket nâng cao

Client ↔ Streaming API ↔ Market Data + Trading Event

from collections import Counter

from ssi_sdk import Auth, Stream, Config
from ssi_sdk.enums import Board
from ssi_sdk.models.streaming import (
    IndexSummaryMessage, MarketFlagMessage,
    OrderMatchMessage, OrderStatusMessage, PortfolioMessage,
)
from auth_helper import ensure_auth
from helpers import enum_value, mask_account

config = Config(
    client_id="<your_client_id>",
    api_key="<your_api_key>",
    api_secret="<your_api_secret>",
    private_key="<your_private_key>",
)
ACCOUNT_NO = "<your_account_no>"   # để trống nếu chỉ cần dữ liệu thị trường
LISTEN_MS = 30_000                  # nghe 30 giây rồi dọn dẹp

counts: Counter = Counter()

# Callback chạy trên thread riêng của SDK: giữ NGẮN, chỉ đếm/ghi nhận
def on_connection(event):
    # state: connected | reconnected | disconnected | failed
    err = f" | lỗi: {type(event.error).__name__}" if event.error else ""
    print(f"  [CONN] {event.state} (reconnect_count={event.reconnect_count}){err}")

def on_data(msg):
    counts[type(msg).__name__] += 1

def on_market_flag(msg: MarketFlagMessage):
    print(f"  [FLAG] {msg.board} | {msg.flag} | {msg.trading_time}")

def on_index_summary(msg: IndexSummaryMessage):
    counts["IndexSummary"] += 1
    if counts["IndexSummary"] == 1:  # chỉ in lần đầu để xem cấu trúc
        print(f"  [INDEX-SUMMARY] {msg.index} | số trường: {len(msg.data)}")

def on_order(msg: OrderStatusMessage):
    print(
        f"  [ORDER] {msg.symbol} {enum_value(msg.side)} {enum_value(msg.order_type)} "
        f"| {enum_value(msg.status)} | khớp {msg.filled_quantity}/{msg.quantity}"
    )

def on_order_match(msg: OrderMatchMessage):
    print(f"  [MATCH] {msg.symbol} {enum_value(msg.side)} | giá khớp {msg.match_price} | KL {msg.match_qty}")

def on_portfolio(msg: PortfolioMessage):
    print(f"  [PORTFOLIO] {msg.symbol} | long={msg.long_qty} short={msg.short_qty} net={msg.net}")

has_account = bool(ACCOUNT_NO) and not ACCOUNT_NO.startswith("<")

with Auth(config) as auth:
    # Dữ liệu thị trường không cần OTP; kênh trading (lệnh/danh mục) cần OTP
    ensure_auth(auth, otp="<your_otp>") if has_account else ensure_auth(auth)

    with Stream(auth) as stream:
        ws = stream.streaming

        # Bước 1: Đăng ký callback TRƯỚC khi connect để không lỡ sự kiện "connected"
        ws.on_connection = on_connection
        ws.on_data = on_data
        ws.on_market_flag = on_market_flag
        ws.on_index_summary = on_index_summary
        ws.on_order = on_order
        ws.on_order_match = on_order_match
        ws.on_portfolio = on_portfolio

        # Bước 2: Kết nối và xem trạng thái
        ws.connect()
        print(f"is_connected={ws.is_connected} reconnect_count={ws.reconnect_count} last_error={ws.last_error}")

        # Bước 3: Subscribe
        ws.subscribe_trades(["SSI", "HPG"])
        ws.subscribe_quotes(["SSI"])
        ws.subscribe_board([Board.HNX])           # toàn bộ trade/quote/room của sàn HNX (nhiều!)
        ws.subscribe_index(["VN30"])              # dữ liệu các MÃ THÀNH PHẦN của VN30
        ws.subscribe_index_ticks(["VNINDEX"])     # tick của CHÍNH chỉ số (IndexTickMessage)
        ws.subscribe_index_summary(["VN30"])      # IndexSummaryMessage
        ws.subscribe_session_flag()               # cờ phiên ATO/LO/ATC
        ws.subscribe_candles(["SSI"], "1m")       # nến 1 phút; chỉ tick/1m/5m

        if has_account:
            print(f"Subscribe lệnh/danh mục của tài khoản {mask_account(ACCOUNT_NO)}")
            ws.subscribe_orders(ACCOUNT_NO)       # trạng thái lệnh + khớp lệnh
            ws.subscribe_positions(ACCOUNT_NO)    # danh mục (chỉ phái sinh có event)

        # Bước 4: Lắng nghe
        print(f"Đang lắng nghe {LISTEN_MS // 1000} giây...")
        try:
            ws.wait(timeout_ms=LISTEN_MS)
        except KeyboardInterrupt:
            print("Ngắt theo yêu cầu...")

        print("Số message nhận được theo loại:")
        for name, n in counts.most_common():
            print(f"  {name:<22} {n}")

        # Bước 5: Xem server đang giữ những subscription nào
        subs = ws.list_subscription()  # {"trading": [...], "data": [...]}
        print(f"list_subscription: data={len(subs['data'])} topic, trading={len(subs['trading'])} topic")

        # Bước 6: Hủy từng phần rồi hủy tất cả
        ws.unsubscribe_board([Board.HNX])          # board nặng nhất: hủy trước
        ws.unsubscribe_index_ticks(["VNINDEX"])
        removed = ws.unsubscribe_all()             # hỏi server rồi hủy đúng những gì còn lại
        print(f"unsubscribe_all đã gỡ: data={len(removed['data'])}, trading={len(removed['trading'])}")

        print(f"Kết thúc: is_connected={ws.is_connected} reconnect_count={ws.reconnect_count}")

import asyncio
from collections import Counter

from ssi_sdk import AsyncAuth, AsyncStream, Config
from ssi_sdk.enums import Board
from auth_helper import ensure_auth_async

config = Config(...)
LISTEN_MS = 30_000
counts: Counter = Counter()

def on_connection(event):
    print(f"  [CONN] {event.state} (reconnect_count={event.reconnect_count})")

def on_data(msg):
    counts[type(msg).__name__] += 1

async def main():
    async with AsyncAuth(config) as auth:
        await ensure_auth_async(auth)  # chỉ dữ liệu thị trường, không cần OTP

        async with AsyncStream(auth) as stream:
            ws = stream.streaming
            ws.on_connection = on_connection   # đăng ký TRƯỚC connect
            ws.on_data = on_data

            await ws.connect()
            print(f"is_connected={ws.is_connected} reconnect_count={ws.reconnect_count}")

            await ws.subscribe_trades(["SSI", "HPG"])
            await ws.subscribe_board([Board.HNX])
            await ws.subscribe_index(["VN30"])
            await ws.subscribe_index_ticks(["VNINDEX"])
            await ws.subscribe_index_summary(["VN30"])

            try:
                await ws.wait(timeout_ms=LISTEN_MS)
            except KeyboardInterrupt:
                pass

            subs = await ws.list_subscription()
            print(f"data={len(subs['data'])} topic, trading={len(subs['trading'])} topic")

            await ws.unsubscribe_board([Board.HNX])
            await ws.unsubscribe_index_ticks(["VNINDEX"])
            await ws.unsubscribe_all()

asyncio.run(main())