diff --git a/v2/futures/websocket_endpoints_test.go b/v2/futures/websocket_endpoints_test.go
new file mode 100644
index 00000000..1e364122
--- /dev/null
+++ b/v2/futures/websocket_endpoints_test.go
@@ -0,0 +1,1262 @@
+package futures
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+)
+
+// Integration tests for the /public, /market, /private WS endpoint routing.
+//
+// Verifies that each stream category connects and receives data on the correct
+// endpoint path for all three environments (main, testnet, demo).
+//
+// Run:
+//   BINANCE_APIKEY=... BINANCE_SECRET=... go test -v -run TestWsEndpoints -count=1 -timeout 120s ./futures/
+
+func requireKeys(t *testing.T) {
+	t.Helper()
+	if os.Getenv("BINANCE_APIKEY") == "" || os.Getenv("BINANCE_SECRET") == "" {
+		t.Skip("BINANCE_APIKEY and BINANCE_SECRET not set")
+	}
+}
+
+// --- Demo environment tests ---
+
+func TestWsEndpoints_Demo_Market_AggTrade(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsAggTradeServe("BTCUSDT", func(e *WsAggTradeEvent) {
+		fmt.Printf("  [demo/market] aggTrade symbol=%s price=%s qty=%s\n", e.Symbol, e.Price, e.Quantity)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_Kline(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsKlineServe("BTCUSDT", "1m", func(e *WsKlineEvent) {
+		fmt.Printf("  [demo/market] kline symbol=%s open=%s close=%s\n", e.Symbol, e.Kline.Open, e.Kline.Close)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_MarkPrice(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsMarkPriceServe("BTCUSDT", func(e *WsMarkPriceEvent) {
+		fmt.Printf("  [demo/market] markPrice symbol=%s price=%s funding=%s\n", e.Symbol, e.MarkPrice, e.FundingRate)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_Ticker(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsMarketTickerServe("BTCUSDT", func(e *WsMarketTickerEvent) {
+		fmt.Printf("  [demo/market] ticker symbol=%s last=%s\n", e.Symbol, e.ClosePrice)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_MiniTicker(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsMiniMarketTickerServe("BTCUSDT", func(e *WsMiniMarketTickerEvent) {
+		fmt.Printf("  [demo/market] miniTicker symbol=%s close=%s\n", e.Symbol, e.ClosePrice)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_CombinedAggTrade(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCombinedAggTradeServe([]string{"BTCUSDT", "ETHUSDT"}, func(e *WsAggTradeEvent) {
+		fmt.Printf("  [demo/market] combined aggTrade symbol=%s price=%s\n", e.Symbol, e.Price)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_CombinedKline(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCombinedKlineServe(map[string]string{"BTCUSDT": "1m", "ETHUSDT": "1m"}, func(e *WsKlineEvent) {
+		fmt.Printf("  [demo/market] combined kline symbol=%s\n", e.Symbol)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_BookTicker(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsBookTickerServe("BTCUSDT", func(e *WsBookTickerEvent) {
+		fmt.Printf("  [demo/public] bookTicker symbol=%s bid=%s ask=%s\n", e.Symbol, e.BestBidPrice, e.BestAskPrice)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_Depth(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsPartialDepthServe("BTCUSDT", 5, func(e *WsDepthEvent) {
+		fmt.Printf("  [demo/public] depth bids=%d asks=%d\n", len(e.Bids), len(e.Asks))
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_DiffDepth(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsDiffDepthServe("BTCUSDT", func(e *WsDepthEvent) {
+		fmt.Printf("  [demo/public] diffDepth bids=%d asks=%d\n", len(e.Bids), len(e.Asks))
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_CombinedBookTicker(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCombinedBookTickerServe([]string{"BTCUSDT", "ETHUSDT"}, func(e *WsBookTickerEvent) {
+		fmt.Printf("  [demo/public] combined bookTicker symbol=%s\n", e.Symbol)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_CombinedDepth(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCombinedDepthServe(map[string]string{"BTCUSDT": "5", "ETHUSDT": "5"}, func(e *WsDepthEvent) {
+		fmt.Printf("  [demo/public] combined depth bids=%d\n", len(e.Bids))
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_CombinedDiffDepth(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCombinedDiffDepthServe([]string{"BTCUSDT", "ETHUSDT"}, func(e *WsDepthEvent) {
+		fmt.Printf("  [demo/public] combined diffDepth bids=%d\n", len(e.Bids))
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Private_UserData(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	client := NewClient(os.Getenv("BINANCE_APIKEY"), os.Getenv("BINANCE_SECRET"))
+	listenKey, err := client.NewStartUserStreamService().Do(context.Background())
+	if err != nil {
+		t.Fatalf("listen key: %v", err)
+	}
+	t.Logf("listen key: %s...", listenKey[:8])
+
+	doneC, stopC, err := WsUserDataServe(listenKey, func(e *WsUserDataEvent) {
+		fmt.Printf("  [demo/private] userData event\n")
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+
+	select {
+	case <-doneC:
+		t.Fatal("connection closed unexpectedly")
+	case <-time.After(3 * time.Second):
+		t.Log("PASS: connection held for 3s")
+	}
+}
+
+// --- Parameterized variant tests (WithRate, MultiInterval, ContinuousKline, All*, etc.) ---
+
+func TestWsEndpoints_Demo_Market_MarkPriceWithRate(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsMarkPriceServeWithRate("BTCUSDT", 3*time.Second, func(e *WsMarkPriceEvent) {
+		fmt.Printf("  [demo/market] markPrice@1s symbol=%s price=%s\n", e.Symbol, e.MarkPrice)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_AllMarkPrice(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsAllMarkPriceServe(func(events WsAllMarkPriceEvent) {
+		if len(events) > 0 {
+			fmt.Printf("  [demo/market] !markPrice@arr count=%d first=%s\n", len(events), events[0].Symbol)
+		}
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_AllMarkPriceWithRate(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsAllMarkPriceServeWithRate(3*time.Second, func(events WsAllMarkPriceEvent) {
+		if len(events) > 0 {
+			fmt.Printf("  [demo/market] !markPrice@arr@1s count=%d\n", len(events))
+		}
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_CombinedMarkPriceWithRate(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCombinedMarkPriceServeWithRate(
+		map[string]time.Duration{"BTCUSDT": 3 * time.Second, "ETHUSDT": 3 * time.Second},
+		func(e *WsMarkPriceEvent) {
+			fmt.Printf("  [demo/market] combined markPrice@1s symbol=%s\n", e.Symbol)
+			select {
+			case got <- struct{}{}:
+			default:
+			}
+		}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_CombinedKlineMultiInterval(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCombinedKlineServeMultiInterval(
+		map[string][]string{"BTCUSDT": {"1m", "5m"}},
+		func(e *WsKlineEvent) {
+			fmt.Printf("  [demo/market] combined kline multi symbol=%s interval=%s\n", e.Symbol, e.Kline.Interval)
+			select {
+			case got <- struct{}{}:
+			default:
+			}
+		}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_ContinuousKline(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsContinuousKlineServe(
+		&WsContinuousKlineSubscribeArgs{Pair: "BTCUSDT", ContractType: "perpetual", Interval: "1m"},
+		func(e *WsContinuousKlineEvent) {
+			fmt.Printf("  [demo/market] continuousKline pair=%s interval=%s\n", e.PairSymbol, e.Kline.Interval)
+			select {
+			case got <- struct{}{}:
+			default:
+			}
+		}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_CombinedContinuousKline(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCombinedContinuousKlineServe(
+		[]*WsContinuousKlineSubscribeArgs{
+			{Pair: "BTCUSDT", ContractType: "perpetual", Interval: "1m"},
+			{Pair: "ETHUSDT", ContractType: "perpetual", Interval: "1m"},
+		},
+		func(e *WsContinuousKlineEvent) {
+			fmt.Printf("  [demo/market] combined continuousKline pair=%s\n", e.PairSymbol)
+			select {
+			case got <- struct{}{}:
+			default:
+			}
+		}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_AllMiniTicker(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsAllMiniMarketTickerServe(func(events WsAllMiniMarketTickerEvent) {
+		if len(events) > 0 {
+			fmt.Printf("  [demo/market] !miniTicker@arr count=%d\n", len(events))
+		}
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_AllTicker(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsAllMarketTickerServe(func(events WsAllMarketTickerEvent) {
+		if len(events) > 0 {
+			fmt.Printf("  [demo/market] !ticker@arr count=%d\n", len(events))
+		}
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_AllBookTicker(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsAllBookTickerServe(func(e *WsBookTickerEvent) {
+		fmt.Printf("  [demo/public] !bookTicker symbol=%s\n", e.Symbol)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_PartialDepthWithRate(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	rate := 500 * time.Millisecond
+	doneC, stopC, err := WsPartialDepthServeWithRate("BTCUSDT", 10, rate, func(e *WsDepthEvent) {
+		fmt.Printf("  [demo/public] depth@10@500ms bids=%d asks=%d\n", len(e.Bids), len(e.Asks))
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_DiffDepthWithRate(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	rate := 500 * time.Millisecond
+	doneC, stopC, err := WsDiffDepthServeWithRate("BTCUSDT", rate, func(e *WsDepthEvent) {
+		fmt.Printf("  [demo/public] diffDepth@500ms bids=%d asks=%d\n", len(e.Bids), len(e.Asks))
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_LiquidationOrder(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	// Liquidation events are rare; just verify the connection holds without error
+	doneC, stopC, err := WsLiquidationOrderServe("BTCUSDT", func(e *WsLiquidationOrderEvent) {
+		fmt.Printf("  [demo/market] forceOrder symbol=%s\n", e.Event)
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(3 * time.Second):
+		t.Log("PASS: connection held 3s (liquidation events are rare)")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_AllLiquidationOrder(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	doneC, stopC, err := WsAllLiquidationOrderServe(func(e *WsLiquidationOrderEvent) {
+		fmt.Printf("  [demo/market] !forceOrder@arr\n")
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(3 * time.Second):
+		t.Log("PASS: connection held 3s")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_CompositeIndex(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCompositiveIndexServe("DEFIUSDT", func(e *WsCompositeIndexEvent) {
+		fmt.Printf("  [demo/market] compositeIndex symbol=%s price=%s baseAssetType=%s compositions=%d\n",
+			e.Symbol, e.Price, e.BaseAssetType, len(e.Composition))
+		if len(e.Composition) > 0 {
+			fmt.Printf("    first: base=%s quote=%s weight=%s indexPrice=%s\n",
+				e.Composition[0].BaseAsset, e.Composition[0].QuoteAsset, e.Composition[0].WeighPercent, e.Composition[0].IndexPrice)
+		}
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Fatalf("compositeIndex unmarshal error (should not happen): %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(10 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+// --- Testnet environment tests ---
+
+func TestWsEndpoints_Testnet_Market_AggTrade(t *testing.T) {
+	requireKeys(t)
+	UseTestnet = true
+	defer func() { UseTestnet = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsAggTradeServe("BTCUSDT", func(e *WsAggTradeEvent) {
+		fmt.Printf("  [testnet/market] aggTrade symbol=%s price=%s\n", e.Symbol, e.Price)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Testnet_Market_CombinedMarkPrice(t *testing.T) {
+	requireKeys(t)
+	UseTestnet = true
+	defer func() { UseTestnet = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCombinedMarkPriceServe([]string{"BTCUSDT", "ETHUSDT"}, func(e *WsMarkPriceEvent) {
+		fmt.Printf("  [testnet/market] combined markPrice symbol=%s\n", e.Symbol)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Testnet_Public_BookTicker(t *testing.T) {
+	requireKeys(t)
+	UseTestnet = true
+	defer func() { UseTestnet = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsBookTickerServe("BTCUSDT", func(e *WsBookTickerEvent) {
+		fmt.Printf("  [testnet/public] bookTicker symbol=%s bid=%s ask=%s\n", e.Symbol, e.BestBidPrice, e.BestAskPrice)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Testnet_Public_CombinedDepth(t *testing.T) {
+	requireKeys(t)
+	UseTestnet = true
+	defer func() { UseTestnet = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsCombinedDepthServe(map[string]string{"BTCUSDT": "5"}, func(e *WsDepthEvent) {
+		fmt.Printf("  [testnet/public] combined depth bids=%d\n", len(e.Bids))
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Testnet_Private_UserData(t *testing.T) {
+	requireKeys(t)
+	UseTestnet = true
+	defer func() { UseTestnet = false }()
+
+	client := NewClient(os.Getenv("BINANCE_APIKEY"), os.Getenv("BINANCE_SECRET"))
+	listenKey, err := client.NewStartUserStreamService().Do(context.Background())
+	if err != nil {
+		t.Fatalf("listen key: %v (testnet may need different keys)", err)
+	}
+	t.Logf("listen key: %s...", listenKey[:8])
+
+	doneC, stopC, err := WsUserDataServe(listenKey, func(e *WsUserDataEvent) {
+		fmt.Printf("  [testnet/private] userData event\n")
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+
+	select {
+	case <-doneC:
+		t.Fatal("connection closed unexpectedly")
+	case <-time.After(3 * time.Second):
+		t.Log("PASS: connection held for 3s")
+	}
+}
+
+// --- New stream tests: contractInfo, assetIndex, private query-param ---
+
+func TestWsEndpoints_Demo_Market_ContractInfo(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	// contractInfo only fires when contract info changes — rare event, just verify connection holds
+	doneC, stopC, err := WsContractInfoServe(func(e *WsContractInfoEvent) {
+		fmt.Printf("  [demo/market] contractInfo event=%s dataLen=%d\n", e.Event, len(e.Data))
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(3 * time.Second):
+		t.Log("PASS: connection held 3s (contractInfo is a rare event)")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_AssetIndex(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsAssetIndexServe("BTCUSD", func(e *WsAssetIndexEvent) {
+		fmt.Printf("  [demo/market] assetIndex symbol=%s index=%s bidRate=%s askRate=%s\n",
+			e.Symbol, e.Index, e.BidRate, e.AskRate)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_AllAssetIndex(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsAllAssetIndexServe(func(events WsAllAssetIndexEvent) {
+		if len(events) > 0 {
+			fmt.Printf("  [demo/market] !assetIndex@arr count=%d first=%s index=%s\n",
+				len(events), events[0].Symbol, events[0].Index)
+		}
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Private_UserDataWithEvents(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	client := NewClient(os.Getenv("BINANCE_APIKEY"), os.Getenv("BINANCE_SECRET"))
+	listenKey, err := client.NewStartUserStreamService().Do(context.Background())
+	if err != nil {
+		t.Fatalf("listen key: %v", err)
+	}
+	t.Logf("listen key: %s...", listenKey[:8])
+
+	events := []string{"ORDER_TRADE_UPDATE", "ACCOUNT_UPDATE"}
+	doneC, stopC, err := WsUserDataServeWithEvents(listenKey, events, func(e *WsUserDataEvent) {
+		fmt.Printf("  [demo/private] userData with events: %s\n", e.Event)
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+
+	select {
+	case <-doneC:
+		t.Fatal("connection closed unexpectedly")
+	case <-time.After(3 * time.Second):
+		t.Log("PASS: connection held 3s with events filter")
+	}
+}
+
+func TestWsEndpoints_Demo_Private_UserDataMultiple(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	client := NewClient(os.Getenv("BINANCE_APIKEY"), os.Getenv("BINANCE_SECRET"))
+	listenKey, err := client.NewStartUserStreamService().Do(context.Background())
+	if err != nil {
+		t.Fatalf("listen key: %v", err)
+	}
+	t.Logf("listen key: %s...", listenKey[:8])
+
+	configs := []WsPrivateStreamConfig{
+		{ListenKey: listenKey, Events: []string{"ORDER_TRADE_UPDATE"}},
+		{ListenKey: listenKey, Events: []string{"ACCOUNT_UPDATE"}},
+	}
+	doneC, stopC, err := WsUserDataServeMultiple(configs, func(e *WsUserDataEvent) {
+		fmt.Printf("  [demo/private] userData multiple: %s\n", e.Event)
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+
+	select {
+	case <-doneC:
+		t.Fatal("connection closed unexpectedly")
+	case <-time.After(3 * time.Second):
+		t.Log("PASS: stream mode connection held 3s with multiple configs")
+	}
+}
+
+func TestWsEndpoints_Testnet_Market_AssetIndex(t *testing.T) {
+	requireKeys(t)
+	UseTestnet = true
+	defer func() { UseTestnet = false }()
+
+	got := make(chan struct{}, 1)
+	doneC, stopC, err := WsAssetIndexServe("BTCUSD", func(e *WsAssetIndexEvent) {
+		fmt.Printf("  [testnet/market] assetIndex symbol=%s index=%s\n", e.Symbol, e.Index)
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Testnet_Private_UserDataWithEvents(t *testing.T) {
+	requireKeys(t)
+	UseTestnet = true
+	defer func() { UseTestnet = false }()
+
+	client := NewClient(os.Getenv("BINANCE_APIKEY"), os.Getenv("BINANCE_SECRET"))
+	listenKey, err := client.NewStartUserStreamService().Do(context.Background())
+	if err != nil {
+		t.Fatalf("listen key: %v", err)
+	}
+	t.Logf("listen key: %s...", listenKey[:8])
+
+	events := []string{"ORDER_TRADE_UPDATE", "ACCOUNT_UPDATE"}
+	doneC, stopC, err := WsUserDataServeWithEvents(listenKey, events, func(e *WsUserDataEvent) {
+		fmt.Printf("  [testnet/private] userData with events: %s\n", e.Event)
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+
+	select {
+	case <-doneC:
+		t.Fatal("connection closed unexpectedly")
+	case <-time.After(3 * time.Second):
+		t.Log("PASS: connection held 3s")
+	}
+}
+
+// --- BLVT and rate variant tests ---
+
+func TestWsEndpoints_Demo_Market_BLVTInfo(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	// BLVT tokens may not be active on demo; just verify connection holds
+	doneC, stopC, err := WsBLVTInfoServe("BTCDOWN", func(e *WsBLVTInfoEvent) {
+		fmt.Printf("  [demo/market] BLVT tokenNav symbol=%s nav=%f leverage=%f\n",
+			e.Symbol, e.Nav, e.Leverage)
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(3 * time.Second):
+		t.Log("PASS: connection held 3s (BLVT may not be active on demo)")
+	}
+}
+
+func TestWsEndpoints_Demo_Market_BLVTKline(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	doneC, stopC, err := WsBLVTKlineServe("BTCDOWN", "1m", func(e *WsBLVTKlineEvent) {
+		fmt.Printf("  [demo/market] BLVT kline symbol=%s open=%s close=%s\n",
+			e.Symbol, e.Kline.OpenPrice, e.Kline.ClosePrice)
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(3 * time.Second):
+		t.Log("PASS: connection held 3s (BLVT may not be active on demo)")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_PartialDepthWith100ms(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	rate := 100 * time.Millisecond
+	doneC, stopC, err := WsPartialDepthServeWithRate("BTCUSDT", 5, rate, func(e *WsDepthEvent) {
+		fmt.Printf("  [demo/public] depth@5@100ms bids=%d asks=%d\n", len(e.Bids), len(e.Asks))
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestWsEndpoints_Demo_Public_DiffDepthWith100ms(t *testing.T) {
+	requireKeys(t)
+	UseDemo = true
+	defer func() { UseDemo = false }()
+
+	got := make(chan struct{}, 1)
+	rate := 100 * time.Millisecond
+	doneC, stopC, err := WsDiffDepthServeWithRate("BTCUSDT", rate, func(e *WsDepthEvent) {
+		fmt.Printf("  [demo/public] diffDepth@100ms bids=%d asks=%d\n", len(e.Bids), len(e.Asks))
+		select {
+		case got <- struct{}{}:
+		default:
+		}
+	}, func(err error) { t.Logf("err: %v", err) })
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer close(stopC)
+	select {
+	case <-got:
+		t.Log("PASS")
+	case <-doneC:
+		t.Fatal("closed")
+	case <-time.After(15 * time.Second):
+		t.Fatal("timeout")
+	}
+}
diff --git a/v2/futures/websocket_service.go b/v2/futures/websocket_service.go
index c14eb002..a078d704 100644
--- a/v2/futures/websocket_service.go
+++ b/v2/futures/websocket_service.go
@@ -21,6 +21,27 @@ var (
 	BaseWsApiMainURL       = "wss://ws-fapi.binance.com/ws-fapi/v1"
 	BaseWsApiTestnetURL    = "wss://testnet.binancefuture.com/ws-fapi/v1"
 	BaseWsApiDemoURL       = "wss://testnet.binancefuture.com/ws-fapi/v1"
+
+	// Public endpoints (high-frequency public market data: bookTicker, depth)
+	BaseWsPublicMainUrl          = "wss://fstream.binance.com/public/ws"
+	BaseWsPublicTestnetUrl       = "wss://stream.binancefuture.com/public/ws"
+	BaseWsPublicDemoURL          = "wss://fstream.binancefuture.com/public/ws"
+	BaseCombinedPublicMainURL    = "wss://fstream.binance.com/public/stream?streams="
+	BaseCombinedPublicTestnetURL = "wss://stream.binancefuture.com/public/stream?streams="
+	BaseCombinedPublicDemoURL    = "wss://fstream.binancefuture.com/public/stream?streams="
+
+	// Market endpoints (regular market data: aggTrade, markPrice, kline, ticker, etc.)
+	BaseWsMarketMainUrl          = "wss://fstream.binance.com/market/ws"
+	BaseWsMarketTestnetUrl       = "wss://stream.binancefuture.com/market/ws"
+	BaseWsMarketDemoURL          = "wss://fstream.binancefuture.com/market/ws"
+	BaseCombinedMarketMainURL    = "wss://fstream.binance.com/market/stream?streams="
+	BaseCombinedMarketTestnetURL = "wss://stream.binancefuture.com/market/stream?streams="
+	BaseCombinedMarketDemoURL    = "wss://fstream.binancefuture.com/market/stream?streams="
+
+	// Private endpoints (user data streams)
+	BaseWsPrivateMainUrl    = "wss://fstream.binance.com/private/ws"
+	BaseWsPrivateTestnetUrl = "wss://stream.binancefuture.com/private/ws"
+	BaseWsPrivateDemoURL    = "wss://fstream.binancefuture.com/private/ws"
 )
 
 var (
@@ -73,6 +94,61 @@ func getCombinedEndpoint() string {
 	return BaseCombinedMainURL
 }
 
+// getWsPublicEndpoint return the base endpoint for high-frequency public data (bookTicker, depth)
+func getWsPublicEndpoint() string {
+	if UseTestnet {
+		return BaseWsPublicTestnetUrl
+	}
+	if UseDemo {
+		return BaseWsPublicDemoURL
+	}
+	return BaseWsPublicMainUrl
+}
+
+// getCombinedPublicEndpoint return the combined stream endpoint for public data
+func getCombinedPublicEndpoint() string {
+	if UseTestnet {
+		return BaseCombinedPublicTestnetURL
+	}
+	if UseDemo {
+		return BaseCombinedPublicDemoURL
+	}
+	return BaseCombinedPublicMainURL
+}
+
+// getWsMarketEndpoint return the base endpoint for regular market data (aggTrade, markPrice, kline, ticker, etc.)
+func getWsMarketEndpoint() string {
+	if UseTestnet {
+		return BaseWsMarketTestnetUrl
+	}
+	if UseDemo {
+		return BaseWsMarketDemoURL
+	}
+	return BaseWsMarketMainUrl
+}
+
+// getCombinedMarketEndpoint return the combined stream endpoint for market data
+func getCombinedMarketEndpoint() string {
+	if UseTestnet {
+		return BaseCombinedMarketTestnetURL
+	}
+	if UseDemo {
+		return BaseCombinedMarketDemoURL
+	}
+	return BaseCombinedMarketMainURL
+}
+
+// getWsPrivateEndpoint return the base endpoint for private user data streams
+func getWsPrivateEndpoint() string {
+	if UseTestnet {
+		return BaseWsPrivateTestnetUrl
+	}
+	if UseDemo {
+		return BaseWsPrivateDemoURL
+	}
+	return BaseWsPrivateMainUrl
+}
+
 // WsAggTradeEvent define websocket aggTrde event.
 type WsAggTradeEvent struct {
 	Event            string `json:"e"`
@@ -92,7 +168,7 @@ type WsAggTradeHandler func(event *WsAggTradeEvent)
 
 // WsAggTradeServe serve websocket that push trade information that is aggregated for a single taker order.
 func WsAggTradeServe(symbol string, handler WsAggTradeHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s@aggTrade", getWsEndpoint(), strings.ToLower(symbol))
+	endpoint := fmt.Sprintf("%s/%s@aggTrade", getWsMarketEndpoint(), strings.ToLower(symbol))
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsAggTradeEvent)
@@ -108,7 +184,7 @@ func WsAggTradeServe(symbol string, handler WsAggTradeHandler, errHandler ErrHan
 
 // WsCombinedAggTradeServe is similar to WsAggTradeServe, but it handles multiple symbols
 func WsCombinedAggTradeServe(symbols []string, handler WsAggTradeHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := getCombinedEndpoint()
+	endpoint := getCombinedMarketEndpoint()
 	for _, s := range symbols {
 		endpoint += fmt.Sprintf("%s@aggTrade", strings.ToLower(s)) + "/"
 	}
@@ -172,7 +248,7 @@ func wsMarkPriceServe(endpoint string, handler WsMarkPriceHandler, errHandler Er
 
 // WsMarkPriceServe serve websocket that pushes price and funding rate for a single symbol.
 func WsMarkPriceServe(symbol string, handler WsMarkPriceHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s@markPrice", getWsEndpoint(), strings.ToLower(symbol))
+	endpoint := fmt.Sprintf("%s/%s@markPrice", getWsMarketEndpoint(), strings.ToLower(symbol))
 	return wsMarkPriceServe(endpoint, handler, errHandler)
 }
 
@@ -187,7 +263,7 @@ func WsMarkPriceServeWithRate(symbol string, rate time.Duration, handler WsMarkP
 	default:
 		return nil, nil, errors.New("Invalid rate")
 	}
-	endpoint := fmt.Sprintf("%s/%s@markPrice%s", getWsEndpoint(), strings.ToLower(symbol), rateStr)
+	endpoint := fmt.Sprintf("%s/%s@markPrice%s", getWsMarketEndpoint(), strings.ToLower(symbol), rateStr)
 	return wsMarkPriceServe(endpoint, handler, errHandler)
 }
 
@@ -218,7 +294,7 @@ func wsCombinedMarkPriceServe(endpoint string, handler WsMarkPriceHandler, errHa
 
 // WsCombinedMarkPriceServe is similar to WsMarkPriceServe, but it handles multiple symbols
 func WsCombinedMarkPriceServe(symbols []string, handler WsMarkPriceHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := getCombinedEndpoint()
+	endpoint := getCombinedMarketEndpoint()
 	for _, s := range symbols {
 		endpoint += fmt.Sprintf("%s@markPrice", strings.ToLower(s)) + "/"
 	}
@@ -229,7 +305,7 @@ func WsCombinedMarkPriceServe(symbols []string, handler WsMarkPriceHandler, errH
 
 // WsCombinedMarkPriceServeWithRate is similar to WsMarkPriceServeWithRate, but it for multiple symbols
 func WsCombinedMarkPriceServeWithRate(symbolLevels map[string]time.Duration, handler WsMarkPriceHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := getCombinedEndpoint()
+	endpoint := getCombinedMarketEndpoint()
 	for symbol, rate := range symbolLevels {
 		var rateStr string
 		switch rate {
@@ -271,7 +347,7 @@ func wsAllMarkPriceServe(endpoint string, handler WsAllMarkPriceHandler, errHand
 
 // WsAllMarkPriceServe serve websocket that pushes price and funding rate for all symbol.
 func WsAllMarkPriceServe(handler WsAllMarkPriceHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/!markPrice@arr", getWsEndpoint())
+	endpoint := fmt.Sprintf("%s/!markPrice@arr", getWsMarketEndpoint())
 	return wsAllMarkPriceServe(endpoint, handler, errHandler)
 }
 
@@ -286,7 +362,7 @@ func WsAllMarkPriceServeWithRate(rate time.Duration, handler WsAllMarkPriceHandl
 	default:
 		return nil, nil, errors.New("Invalid rate")
 	}
-	endpoint := fmt.Sprintf("%s/!markPrice@arr%s", getWsEndpoint(), rateStr)
+	endpoint := fmt.Sprintf("%s/!markPrice@arr%s", getWsMarketEndpoint(), rateStr)
 	return wsAllMarkPriceServe(endpoint, handler, errHandler)
 }
 
@@ -323,7 +399,7 @@ type WsKlineHandler func(event *WsKlineEvent)
 
 // WsKlineServe serve websocket kline handler with a symbol and interval like 15m, 30s
 func WsKlineServe(symbol string, interval string, handler WsKlineHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s@kline_%s", getWsEndpoint(), strings.ToLower(symbol), interval)
+	endpoint := fmt.Sprintf("%s/%s@kline_%s", getWsMarketEndpoint(), strings.ToLower(symbol), interval)
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsKlineEvent)
@@ -339,7 +415,7 @@ func WsKlineServe(symbol string, interval string, handler WsKlineHandler, errHan
 
 // WsCombinedKlineServe is similar to WsKlineServe, but it handles multiple symbols with it interval
 func WsCombinedKlineServe(symbolIntervalPair map[string]string, handler WsKlineHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := getCombinedEndpoint()
+	endpoint := getCombinedMarketEndpoint()
 	for symbol, interval := range symbolIntervalPair {
 		endpoint += fmt.Sprintf("%s@kline_%s", strings.ToLower(symbol), interval) + "/"
 	}
@@ -374,7 +450,7 @@ func WsCombinedKlineServe(symbolIntervalPair map[string]string, handler WsKlineH
 
 // WsCombinedKlineServeMultiInterval is similar to WsCombinedKlineServe, but it supports multiple intervals per symbol
 func WsCombinedKlineServeMultiInterval(symbolIntervals map[string][]string, handler WsKlineHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := getCombinedEndpoint()
+	endpoint := getCombinedMarketEndpoint()
 	for symbol, intervals := range symbolIntervals {
 		for _, interval := range intervals {
 			endpoint += fmt.Sprintf("%s@kline_%s", strings.ToLower(symbol), interval) + "/"
@@ -450,7 +526,7 @@ type WsContinuousKlineHandler func(event *WsContinuousKlineEvent)
 // WsContinuousKlineServe serve websocket continuous kline handler with a pair and contractType and interval like 15m, 30s
 func WsContinuousKlineServe(subscribeArgs *WsContinuousKlineSubscribeArgs, handler WsContinuousKlineHandler,
 	errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s_%s@continuousKline_%s", getWsEndpoint(), strings.ToLower(subscribeArgs.Pair),
+	endpoint := fmt.Sprintf("%s/%s_%s@continuousKline_%s", getWsMarketEndpoint(), strings.ToLower(subscribeArgs.Pair),
 		strings.ToLower(subscribeArgs.ContractType), subscribeArgs.Interval)
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
@@ -468,7 +544,7 @@ func WsContinuousKlineServe(subscribeArgs *WsContinuousKlineSubscribeArgs, handl
 // WsCombinedContinuousKlineServe is similar to WsContinuousKlineServe, but it handles multiple pairs of different contractType with its interval
 func WsCombinedContinuousKlineServe(subscribeArgsList []*WsContinuousKlineSubscribeArgs,
 	handler WsContinuousKlineHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := getCombinedEndpoint()
+	endpoint := getCombinedMarketEndpoint()
 	for _, val := range subscribeArgsList {
 		endpoint += fmt.Sprintf("%s_%s@continuousKline_%s", strings.ToLower(val.Pair),
 			strings.ToLower(val.ContractType), val.Interval) + "/"
@@ -516,7 +592,7 @@ type WsMiniMarketTickerHandler func(event *WsMiniMarketTickerEvent)
 
 // WsMiniMarketTickerServe serve websocket that pushes 24hr rolling window mini-ticker statistics for a single symbol.
 func WsMiniMarketTickerServe(symbol string, handler WsMiniMarketTickerHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s@miniTicker", getWsEndpoint(), strings.ToLower(symbol))
+	endpoint := fmt.Sprintf("%s/%s@miniTicker", getWsMarketEndpoint(), strings.ToLower(symbol))
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsMiniMarketTickerEvent)
@@ -538,7 +614,7 @@ type WsAllMiniMarketTickerHandler func(event WsAllMiniMarketTickerEvent)
 
 // WsAllMiniMarketTickerServe serve websocket that pushes price and funding rate for all markets.
 func WsAllMiniMarketTickerServe(handler WsAllMiniMarketTickerHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/!miniTicker@arr", getWsEndpoint())
+	endpoint := fmt.Sprintf("%s/!miniTicker@arr", getWsMarketEndpoint())
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		var event WsAllMiniMarketTickerEvent
@@ -579,7 +655,7 @@ type WsMarketTickerHandler func(event *WsMarketTickerEvent)
 
 // WsMarketTickerServe serve websocket that pushes 24hr rolling window mini-ticker statistics for a single symbol.
 func WsMarketTickerServe(symbol string, handler WsMarketTickerHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s@ticker", getWsEndpoint(), strings.ToLower(symbol))
+	endpoint := fmt.Sprintf("%s/%s@ticker", getWsMarketEndpoint(), strings.ToLower(symbol))
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsMarketTickerEvent)
@@ -601,7 +677,7 @@ type WsAllMarketTickerHandler func(event WsAllMarketTickerEvent)
 
 // WsAllMarketTickerServe serve websocket that pushes price and funding rate for all markets.
 func WsAllMarketTickerServe(handler WsAllMarketTickerHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/!ticker@arr", getWsEndpoint())
+	endpoint := fmt.Sprintf("%s/!ticker@arr", getWsMarketEndpoint())
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		var event WsAllMarketTickerEvent
@@ -638,7 +714,7 @@ type WsBookTickerHandler func(event *WsBookTickerEvent)
 
 // WsBookTickerServe serve websocket that pushes updates to the best bid or ask price or quantity in real-time for a specified symbol.
 func WsBookTickerServe(symbol string, handler WsBookTickerHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s@bookTicker", getWsEndpoint(), strings.ToLower(symbol))
+	endpoint := fmt.Sprintf("%s/%s@bookTicker", getWsPublicEndpoint(), strings.ToLower(symbol))
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsBookTickerEvent)
@@ -653,7 +729,7 @@ func WsBookTickerServe(symbol string, handler WsBookTickerHandler, errHandler Er
 }
 
 func WsCombinedBookTickerServe(symbols []string, handler WsBookTickerHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := getCombinedEndpoint()
+	endpoint := getCombinedPublicEndpoint()
 	for _, s := range symbols {
 		endpoint += fmt.Sprintf("%s@bookTicker", strings.ToLower(s)) + "/"
 	}
@@ -673,7 +749,7 @@ func WsCombinedBookTickerServe(symbols []string, handler WsBookTickerHandler, er
 
 // WsAllBookTickerServe serve websocket that pushes updates to the best bid or ask price or quantity in real-time for all symbols.
 func WsAllBookTickerServe(handler WsBookTickerHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/!bookTicker", getWsEndpoint())
+	endpoint := fmt.Sprintf("%s/!bookTicker", getWsPublicEndpoint())
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsBookTickerEvent)
@@ -714,7 +790,7 @@ type WsLiquidationOrderHandler func(event *WsLiquidationOrderEvent)
 
 // WsLiquidationOrderServe serve websocket that pushes force liquidation order information for specific symbol.
 func WsLiquidationOrderServe(symbol string, handler WsLiquidationOrderHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s@forceOrder", getWsEndpoint(), strings.ToLower(symbol))
+	endpoint := fmt.Sprintf("%s/%s@forceOrder", getWsMarketEndpoint(), strings.ToLower(symbol))
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsLiquidationOrderEvent)
@@ -730,7 +806,7 @@ func WsLiquidationOrderServe(symbol string, handler WsLiquidationOrderHandler, e
 
 // WsAllLiquidationOrderServe serve websocket that pushes force liquidation order information for all symbols.
 func WsAllLiquidationOrderServe(handler WsLiquidationOrderHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/!forceOrder@arr", getWsEndpoint())
+	endpoint := fmt.Sprintf("%s/!forceOrder@arr", getWsMarketEndpoint())
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsLiquidationOrderEvent)
@@ -785,7 +861,7 @@ func WsDiffDepthServe(symbol string, handler WsDepthHandler, errHandler ErrHandl
 
 // WsCombinedDepthServe is similar to WsPartialDepthServe, but it for multiple symbols
 func WsCombinedDepthServe(symbolLevels map[string]string, handler WsDepthHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := getCombinedEndpoint()
+	endpoint := getCombinedPublicEndpoint()
 	for s, l := range symbolLevels {
 		endpoint += fmt.Sprintf("%s@depth%s", strings.ToLower(s), l) + "/"
 	}
@@ -831,7 +907,7 @@ func WsCombinedDepthServe(symbolLevels map[string]string, handler WsDepthHandler
 
 // WsCombinedDiffDepthServe is similar to WsDiffDepthServe, but it for multiple symbols
 func WsCombinedDiffDepthServe(symbols []string, handler WsDepthHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := getCombinedEndpoint()
+	endpoint := getCombinedPublicEndpoint()
 	for _, s := range symbols {
 		endpoint += fmt.Sprintf("%s@depth", strings.ToLower(s)) + "/"
 	}
@@ -894,7 +970,7 @@ func wsDepthServe(symbol string, levels string, rate *time.Duration, handler WsD
 			return nil, nil, errors.New("Invalid rate")
 		}
 	}
-	endpoint := fmt.Sprintf("%s/%s@depth%s%s", getWsEndpoint(), strings.ToLower(symbol), levels, rateStr)
+	endpoint := fmt.Sprintf("%s/%s@depth%s%s", getWsPublicEndpoint(), strings.ToLower(symbol), levels, rateStr)
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		j, err := newJSON(message)
@@ -957,7 +1033,7 @@ type WsBLVTInfoHandler func(event *WsBLVTInfoEvent)
 
 // WsBLVTInfoServe serve BLVT info stream
 func WsBLVTInfoServe(name string, handler WsBLVTInfoHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s@tokenNav", getWsEndpoint(), strings.ToUpper(name))
+	endpoint := fmt.Sprintf("%s/%s@tokenNav", getWsMarketEndpoint(), strings.ToUpper(name))
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsBLVTInfoEvent)
@@ -1000,7 +1076,7 @@ type WsBLVTKlineHandler func(event *WsBLVTKlineEvent)
 
 // WsBLVTKlineServe serve BLVT kline stream
 func WsBLVTKlineServe(name string, interval string, handler WsBLVTKlineHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s@nav_Kline_%s", getWsEndpoint(), strings.ToUpper(name), interval)
+	endpoint := fmt.Sprintf("%s/%s@nav_Kline_%s", getWsMarketEndpoint(), strings.ToUpper(name), interval)
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsBLVTKlineEvent)
@@ -1016,18 +1092,21 @@ func WsBLVTKlineServe(name string, interval string, handler WsBLVTKlineHandler,
 
 // WsCompositeIndexEvent websocket composite index event
 type WsCompositeIndexEvent struct {
-	Event       string          `json:"e"`
-	Time        int64           `json:"E"`
-	Symbol      string          `json:"s"`
-	Price       string          `json:"p"`
-	Composition []WsComposition `json:"c"`
+	Event         string          `json:"e"`
+	Time          int64           `json:"E"`
+	Symbol        string          `json:"s"`
+	Price         string          `json:"p"`
+	BaseAssetType string          `json:"C"`
+	Composition   []WsComposition `json:"c"`
 }
 
 // WsComposition websocket composite index event composition
 type WsComposition struct {
 	BaseAsset    string `json:"b"`
+	QuoteAsset   string `json:"q"`
 	WeightQty    string `json:"w"`
 	WeighPercent string `json:"W"`
+	IndexPrice   string `json:"i"`
 }
 
 // WsCompositeIndexHandler websocket composite index handler
@@ -1035,7 +1114,7 @@ type WsCompositeIndexHandler func(event *WsCompositeIndexEvent)
 
 // WsCompositiveIndexServe serve composite index information for index symbols
 func WsCompositiveIndexServe(symbol string, handler WsCompositeIndexHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s@compositeIndex", getWsEndpoint(), strings.ToLower(symbol))
+	endpoint := fmt.Sprintf("%s/%s@compositeIndex", getWsMarketEndpoint(), strings.ToLower(symbol))
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsCompositeIndexEvent)
@@ -1049,6 +1128,99 @@ func WsCompositiveIndexServe(symbol string, handler WsCompositeIndexHandler, err
 	return wsServe(cfg, wsHandler, errHandler)
 }
 
+// WsContractInfoEvent define websocket contract info event
+type WsContractInfoEvent struct {
+	Event string               `json:"e"`
+	Time  int64                `json:"E"`
+	Data  []WsContractInfoData `json:"data"`
+}
+
+// WsContractInfoData define contract info data
+type WsContractInfoData struct {
+	Symbol        string `json:"s"`
+	Pair          string `json:"ps"`
+	ContractType  string `json:"ct"`
+	DeliveryDate  int64  `json:"dt"`
+	OnboardDate   int64  `json:"ot"`
+	ContractState string `json:"cs"`
+}
+
+// WsContractInfoHandler handle WsContractInfoEvent
+type WsContractInfoHandler func(event *WsContractInfoEvent)
+
+// WsContractInfoServe serve websocket that pushes contract info updates
+func WsContractInfoServe(handler WsContractInfoHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
+	endpoint := fmt.Sprintf("%s/!contractInfo", getWsMarketEndpoint())
+	cfg := newWsConfig(endpoint)
+	wsHandler := func(message []byte) {
+		event := new(WsContractInfoEvent)
+		err := json.Unmarshal(message, event)
+		if err != nil {
+			errHandler(err)
+			return
+		}
+		handler(event)
+	}
+	return wsServe(cfg, wsHandler, errHandler)
+}
+
+// WsAssetIndexEvent define websocket asset index event
+type WsAssetIndexEvent struct {
+	Event                 string `json:"e"`
+	Time                  int64  `json:"E"`
+	Symbol                string `json:"s"`
+	Index                 string `json:"i"`
+	BidBuffer             string `json:"b"`
+	AskBuffer             string `json:"a"`
+	BidRate               string `json:"B"`
+	AskRate               string `json:"A"`
+	AutoExchangeBidBuffer string `json:"q"`
+	AutoExchangeAskBuffer string `json:"g"`
+	AutoExchangeBidRate   string `json:"Q"`
+	AutoExchangeAskRate   string `json:"G"`
+}
+
+// WsAllAssetIndexEvent define all asset index event
+type WsAllAssetIndexEvent []*WsAssetIndexEvent
+
+// WsAssetIndexHandler handle WsAssetIndexEvent
+type WsAssetIndexHandler func(event *WsAssetIndexEvent)
+
+// WsAssetIndexServe serve websocket that pushes asset index for a single symbol
+func WsAssetIndexServe(symbol string, handler WsAssetIndexHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
+	endpoint := fmt.Sprintf("%s/%s@assetIndex", getWsMarketEndpoint(), strings.ToLower(symbol))
+	cfg := newWsConfig(endpoint)
+	wsHandler := func(message []byte) {
+		event := new(WsAssetIndexEvent)
+		err := json.Unmarshal(message, event)
+		if err != nil {
+			errHandler(err)
+			return
+		}
+		handler(event)
+	}
+	return wsServe(cfg, wsHandler, errHandler)
+}
+
+// WsAllAssetIndexHandler handle WsAllAssetIndexEvent
+type WsAllAssetIndexHandler func(event WsAllAssetIndexEvent)
+
+// WsAllAssetIndexServe serve websocket that pushes asset index for all symbols
+func WsAllAssetIndexServe(handler WsAllAssetIndexHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
+	endpoint := fmt.Sprintf("%s/!assetIndex@arr", getWsMarketEndpoint())
+	cfg := newWsConfig(endpoint)
+	wsHandler := func(message []byte) {
+		var event WsAllAssetIndexEvent
+		err := json.Unmarshal(message, &event)
+		if err != nil {
+			errHandler(err)
+			return
+		}
+		handler(event)
+	}
+	return wsServe(cfg, wsHandler, errHandler)
+}
+
 // WsUserDataEvent define user data event
 type WsUserDataEvent struct {
 	Event           UserDataEventType `json:"e"`
@@ -1266,7 +1438,62 @@ type WsUserDataHandler func(event *WsUserDataEvent)
 
 // WsUserDataServe serve user data handler with listen key
 func WsUserDataServe(listenKey string, handler WsUserDataHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
-	endpoint := fmt.Sprintf("%s/%s", getWsEndpoint(), listenKey)
+	endpoint := fmt.Sprintf("%s/%s", getWsPrivateEndpoint(), listenKey)
+	cfg := newWsConfig(endpoint)
+	wsHandler := func(message []byte) {
+		event := new(WsUserDataEvent)
+		err := json.Unmarshal(message, event)
+		if err != nil {
+			errHandler(err)
+			return
+		}
+		handler(event)
+	}
+	return wsServe(cfg, wsHandler, errHandler)
+}
+
+// WsUserDataServeWithEvents serve user data handler with listen key and event filter using the
+// new query-param format: /private/ws?listenKey=<key>&events=EVENT1/EVENT2
+func WsUserDataServeWithEvents(listenKey string, events []string, handler WsUserDataHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
+	endpoint := fmt.Sprintf("%s?listenKey=%s", getWsPrivateEndpoint(), listenKey)
+	if len(events) > 0 {
+		endpoint += "&events=" + strings.Join(events, "/")
+	}
+	cfg := newWsConfig(endpoint)
+	wsHandler := func(message []byte) {
+		event := new(WsUserDataEvent)
+		err := json.Unmarshal(message, event)
+		if err != nil {
+			errHandler(err)
+			return
+		}
+		handler(event)
+	}
+	return wsServe(cfg, wsHandler, errHandler)
+}
+
+// WsPrivateStreamConfig defines a listen key and its event subscriptions for the multi-key stream mode
+type WsPrivateStreamConfig struct {
+	ListenKey string
+	Events    []string
+}
+
+// WsUserDataServeMultiple serve user data handler with multiple listen keys using the stream mode:
+// /private/stream?listenKey=<key1>&events=EVENT1&listenKey=<key2>&events=EVENT2
+func WsUserDataServeMultiple(configs []WsPrivateStreamConfig, handler WsUserDataHandler, errHandler ErrHandler) (doneC, stopC chan struct{}, err error) {
+	endpoint := getWsPrivateEndpoint()
+	endpoint = strings.Replace(endpoint, "/ws", "/stream", 1)
+	params := ""
+	for _, c := range configs {
+		if params != "" {
+			params += "&"
+		}
+		params += "listenKey=" + c.ListenKey
+		if len(c.Events) > 0 {
+			params += "&events=" + strings.Join(c.Events, "/")
+		}
+	}
+	endpoint += "?" + params
 	cfg := newWsConfig(endpoint)
 	wsHandler := func(message []byte) {
 		event := new(WsUserDataEvent)