16. Advanced WebSocket Streaming

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>"   # leave empty if you only need market data
LISTEN_MS = 30_000                  # listen for 30 seconds, then clean up

counts: Counter = Counter()

# Callbacks run on the SDK's own thread: keep them SHORT, just count/record
def on_connection(event):
    # state: connected | reconnected | disconnected | failed
    err = f" | error: {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:  # print only the first one to see the structure
        print(f"  [INDEX-SUMMARY] {msg.index} | fields: {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)} | filled {msg.filled_quantity}/{msg.quantity}"
    )

def on_order_match(msg: OrderMatchMessage):
    print(f"  [MATCH] {msg.symbol} {enum_value(msg.side)} | match price {msg.match_price} | qty {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:
    # Market data needs no OTP; the trading channel (orders/portfolio) does
    ensure_auth(auth, otp="<your_otp>") if has_account else ensure_auth(auth)

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

        # Step 1: Register callbacks BEFORE connect so the "connected" event is not missed
        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

        # Step 2: Connect and check the state
        ws.connect()
        print(f"is_connected={ws.is_connected} reconnect_count={ws.reconnect_count} last_error={ws.last_error}")

        # Step 3: Subscribe
        ws.subscribe_trades(["SSI", "HPG"])
        ws.subscribe_quotes(["SSI"])
        ws.subscribe_board([Board.HNX])           # all trade/quote/room of HNX (a lot!)
        ws.subscribe_index(["VN30"])              # data of the VN30 CONSTITUENTS
        ws.subscribe_index_ticks(["VNINDEX"])     # ticks of the INDEX ITSELF (IndexTickMessage)
        ws.subscribe_index_summary(["VN30"])      # IndexSummaryMessage
        ws.subscribe_session_flag()               # ATO/LO/ATC session flag
        ws.subscribe_candles(["SSI"], "1m")       # 1-minute candles; tick/1m/5m only

        if has_account:
            print(f"Subscribing to orders/portfolio of account {mask_account(ACCOUNT_NO)}")
            ws.subscribe_orders(ACCOUNT_NO)       # order status + matches
            ws.subscribe_positions(ACCOUNT_NO)    # portfolio (events for derivatives only)

        # Step 4: Listen
        print(f"Listening for {LISTEN_MS // 1000} seconds...")
        try:
            ws.wait(timeout_ms=LISTEN_MS)
        except KeyboardInterrupt:
            print("Interrupted...")

        print("Messages received by type:")
        for name, n in counts.most_common():
            print(f"  {name:<22} {n}")

        # Step 5: See which subscriptions the server holds
        subs = ws.list_subscription()  # {"trading": [...], "data": [...]}
        print(f"list_subscription: data={len(subs['data'])} topics, trading={len(subs['trading'])} topics")

        # Step 6: Cancel piece by piece, then cancel everything
        ws.unsubscribe_board([Board.HNX])          # heaviest board: cancel first
        ws.unsubscribe_index_ticks(["VNINDEX"])
        removed = ws.unsubscribe_all()             # asks the server, then cancels whatever is left
        print(f"unsubscribe_all removed: data={len(removed['data'])}, trading={len(removed['trading'])}")

        print(f"Done: 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)  # market data only, no OTP needed

        async with AsyncStream(auth) as stream:
            ws = stream.streaming
            ws.on_connection = on_connection   # register BEFORE 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'])} topics, trading={len(subs['trading'])} topics")

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

asyncio.run(main())