diff --git a/htx/spot/marketws/client.go b/htx/spot/marketws/client.go
index 7367266..7c46bfd 100644
--- a/htx/spot/marketws/client.go
+++ b/htx/spot/marketws/client.go
@@ -231,7 +231,9 @@ func (m *MarketWsClient) readMessages() {
 
 				switch {
 				case msg.Ping != nil:
-					err := m.pong(msg.Ping)
+					err := m.pong(&PongMessage{
+						Pong: msg.Ping.Ping,
+					})
 					if err != nil {
 						m.logger.Error(fmt.Sprintf("handle ping error: %s", err.Error()))
 					}
@@ -327,7 +329,7 @@ func (m *MarketWsClient) send(req *Request) error {
 	return m.conn.WriteJSON(req)
 }
 
-func (m *MarketWsClient) pong(ping *PingMessage) error {
+func (m *MarketWsClient) pong(ping *PongMessage) error {
 	m.sending.Lock()
 
 	// Rate Limit: https://www.htx.com/en-us/opend/newApiPages/?id=662
diff --git a/htx/spot/marketws/response.go b/htx/spot/marketws/response.go
index 36053e2..69b7b8c 100644
--- a/htx/spot/marketws/response.go
+++ b/htx/spot/marketws/response.go
@@ -41,6 +41,10 @@ type PingMessage struct {
 	Ping int64 `json:"ping,omitempty"`
 }
 
+type PongMessage struct {
+	Pong int64 `json:"pong,omitempty"`
+}
+
 type Response struct {
 	ID     string `json:"id,omitempty"`
 	Status string `json:"status,omitempty"`
diff --git a/htx/usdm/accountws/client.go b/htx/usdm/accountws/client.go
new file mode 100644
index 0000000..7a84b49
--- /dev/null
+++ b/htx/usdm/accountws/client.go
@@ -0,0 +1,421 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accountws
+
+import (
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"net/url"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/chuckpreslar/emission"
+	"github.com/go-playground/validator"
+	"github.com/gorilla/websocket"
+	htxutils "github.com/linstohu/nexapi/htx/utils"
+	cmap "github.com/orcaman/concurrent-map/v2"
+)
+
+type AccountWsClient struct {
+	host, path string
+	baseURL    string
+	// debug mode
+	debug bool
+	// logger
+	logger *slog.Logger
+
+	key, secret string
+
+	ctx         context.Context
+	conn        *websocket.Conn
+	mu          sync.RWMutex
+	isConnected bool
+
+	autoReconnect bool
+	disconnect    chan struct{}
+
+	sending       sync.Mutex
+	subscriptions cmap.ConcurrentMap[string, struct{}]
+
+	emitter *emission.Emitter
+}
+
+type AccountWsClientCfg struct {
+	Debug bool
+	// Logger
+	Logger  *slog.Logger
+	BaseURL string `validate:"required"`
+	Key     string `validate:"required"`
+	Secret  string `validate:"required"`
+}
+
+func NewAccountWsClient(ctx context.Context, cfg *AccountWsClientCfg) (*AccountWsClient, error) {
+	if err := validator.New().Struct(cfg); err != nil {
+		return nil, err
+	}
+
+	baseURL, err := url.Parse(cfg.BaseURL)
+	if err != nil {
+		return nil, err
+	}
+
+	cli := &AccountWsClient{
+		debug:   cfg.Debug,
+		logger:  cfg.Logger,
+		baseURL: cfg.BaseURL,
+		host:    baseURL.Host,
+		path:    baseURL.Path,
+
+		key:    cfg.Key,
+		secret: cfg.Secret,
+
+		ctx:           ctx,
+		autoReconnect: true,
+
+		subscriptions: cmap.New[struct{}](),
+		emitter:       emission.NewEmitter(),
+	}
+
+	if cli.logger == nil {
+		cli.logger = slog.Default()
+	}
+
+	err = cli.start()
+	if err != nil {
+		return nil, err
+	}
+
+	time.Sleep(100 * time.Millisecond)
+
+	return cli, nil
+}
+
+func (m *AccountWsClient) start() error {
+	m.conn = nil
+	m.setIsConnected(false)
+	m.disconnect = make(chan struct{})
+
+	for i := 0; i < MaxTryTimes; i++ {
+		conn, _, err := m.connect()
+		if err != nil {
+			m.logger.Info(fmt.Sprintf("connect error, times(%v), error: %s", i, err.Error()))
+			tm := (i + 1) * 5
+			time.Sleep(time.Duration(tm) * time.Second)
+			continue
+		}
+		m.conn = conn
+		break
+	}
+	if m.conn == nil {
+		return errors.New("connect failed")
+	}
+
+	m.setIsConnected(true)
+
+	m.resubscribe()
+
+	if m.autoReconnect {
+		go m.reconnect()
+	}
+
+	m.auth()
+
+	go m.readMessages()
+
+	return nil
+}
+
+func (m *AccountWsClient) connect() (*websocket.Conn, *http.Response, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	conn, resp, err := websocket.DefaultDialer.DialContext(ctx, m.baseURL, nil)
+	if err == nil {
+		conn.SetReadLimit(32768 * 64)
+	}
+
+	return conn, resp, err
+}
+
+func (m *AccountWsClient) reconnect() {
+	<-m.disconnect
+
+	m.setIsConnected(false)
+
+	m.logger.Info("disconnect, then reconnect...")
+
+	time.Sleep(1 * time.Second)
+
+	select {
+	case <-m.ctx.Done():
+		m.logger.Info(fmt.Sprintf("never reconnect, %s", m.ctx.Err()))
+		return
+	default:
+		m.start()
+	}
+}
+
+// close closes the websocket connection
+func (m *AccountWsClient) close() error {
+	close(m.disconnect)
+
+	err := m.conn.Close()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// setIsConnected sets state for isConnected
+func (m *AccountWsClient) setIsConnected(state bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	m.isConnected = state
+}
+
+// IsConnected returns the WebSocket connection state
+func (m *AccountWsClient) IsConnected() bool {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	return m.isConnected
+}
+
+func (m *AccountWsClient) auth() error {
+	parameters := url.Values{}
+
+	parameters.Add("AccessKeyId", m.key)
+	parameters.Add("SignatureMethod", "HmacSHA256")
+	parameters.Add("SignatureVersion", "2")
+
+	timestamp := time.Now().UTC().Format("2006-01-02T15:04:05")
+	parameters.Add("Timestamp", timestamp)
+
+	var sb strings.Builder
+	sb.WriteString(http.MethodGet)
+	sb.WriteString("\n")
+	sb.WriteString(m.host)
+	sb.WriteString("\n")
+	sb.WriteString(m.path)
+	sb.WriteString("\n")
+	sb.WriteString(parameters.Encode())
+
+	hm := hmac.New(sha256.New, []byte(m.secret))
+	hm.Write([]byte(sb.String()))
+	sign := base64.StdEncoding.EncodeToString(hm.Sum(nil))
+
+	msg := AuthRequest{
+		Operation:        "auth",
+		Type:             "api",
+		AccessKeyId:      m.key,
+		SignatureMethod:  "HmacSHA256",
+		SignatureVersion: "2",
+		Timestamp:        timestamp,
+		Signature:        sign,
+	}
+
+	m.sending.Lock()
+	defer m.sending.Unlock()
+
+	if !m.IsConnected() {
+		return errors.New("connection is closed")
+	}
+
+	return m.conn.WriteJSON(msg)
+}
+
+func (m *AccountWsClient) readMessages() {
+	for {
+		select {
+		case <-m.ctx.Done():
+			m.logger.Info(fmt.Sprintf("context done, error: %s", m.ctx.Err().Error()))
+
+			if err := m.close(); err != nil {
+				m.logger.Info(fmt.Sprintf("websocket connection closed error, %s", err.Error()))
+			}
+
+			return
+		default:
+			msgType, buf, err := m.conn.ReadMessage()
+			if err != nil {
+				m.logger.Error(fmt.Sprintf("read message error, %s", err))
+				time.Sleep(TimerIntervalSecond * time.Second)
+				continue
+			}
+
+			// decompress gzip data if it is binary message
+			if msgType == websocket.BinaryMessage {
+				message, err := htxutils.GZipDecompress(buf)
+				if err != nil {
+					m.logger.Error(fmt.Sprintf("ungzip data error: %s", err))
+
+					if err := m.close(); err != nil {
+						m.logger.Error(fmt.Sprintf("websocket connection closed error, %s", err.Error()))
+					}
+
+					return
+				}
+
+				var msg Message
+
+				err = json.Unmarshal([]byte(message), &msg)
+				if err != nil {
+					m.logger.Error(fmt.Sprintf("read object error, %s", err))
+
+					if err := m.close(); err != nil {
+						m.logger.Error(fmt.Sprintf("websocket connection closed error, %s", err.Error()))
+					}
+
+					return
+				}
+
+				switch {
+				case msg.Operation == PING:
+					err := m.pong(&Message{
+						Operation: PONG,
+						Ts:        msg.Ts,
+					})
+					if err != nil {
+						m.logger.Error(fmt.Sprintf("handle ping error: %s", err.Error()))
+					}
+				case msg.Operation == SUB:
+					if msg.ErrCode != 0 {
+						m.logger.Error(fmt.Sprintf("sub websocket error, op: %s, topic: %s, err-code: %v, err-msg: %v", msg.Operation, msg.Topic, msg.ErrCode, msg.ErrMsg))
+					}
+				case msg.Operation == AUTH:
+					if msg.ErrCode != 0 {
+						m.logger.Error(fmt.Sprintf("auth websocket error, op: %s, err-code: %v, err-msg: %v", msg.Operation, msg.ErrCode, msg.ErrMsg))
+
+						if err := m.close(); err != nil {
+							m.logger.Error(fmt.Sprintf("websocket connection closed error, %s", err.Error()))
+						}
+
+						return
+					} else {
+						m.logger.Info(fmt.Sprintf("auth websocket success, op: %s, err-code: %v", msg.Operation, msg.ErrCode))
+					}
+				case msg.Operation == "notify":
+					err := m.handle(&msg)
+					if err != nil {
+						m.logger.Error(fmt.Sprintf("handle message error: %s", err.Error()))
+					}
+				}
+			}
+		}
+	}
+}
+
+func (m *AccountWsClient) resubscribe() error {
+	topics := m.subscriptions.Keys()
+
+	if len(topics) == 0 {
+		return nil
+	}
+
+	redo := make([]string, 0)
+
+	for _, v := range topics {
+		// do subscription
+		err := m.send(&Message{
+			Operation: SUB,
+			Topic:     v,
+		})
+
+		if err != nil {
+			redo = append(redo, v)
+			continue
+		}
+	}
+
+	if len(redo) != 0 {
+		return fmt.Errorf("resubscribe error: %s", strings.Join(redo, ","))
+	}
+
+	return nil
+}
+
+func (m *AccountWsClient) subscribe(topic string) error {
+	if m.subscriptions.Has(topic) {
+		return nil
+	}
+
+	// do subscription
+
+	err := m.send(&Message{
+		Operation: SUB,
+		Topic:     topic,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	m.subscriptions.Set(topic, struct{}{})
+
+	return nil
+}
+
+func (m *AccountWsClient) unsubscribe(topic string) error {
+	err := m.send(&Message{
+		Operation: SUB,
+		Topic:     topic,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	m.subscriptions.Remove(topic)
+
+	return nil
+}
+
+func (m *AccountWsClient) send(req *Message) error {
+	m.sending.Lock()
+	defer m.sending.Unlock()
+
+	if !m.IsConnected() {
+		return errors.New("connection is closed")
+	}
+
+	return m.conn.WriteJSON(req)
+}
+
+func (m *AccountWsClient) pong(ping *Message) error {
+	m.sending.Lock()
+
+	// Rate Limit: https://www.htx.com/en-us/opend/newApiPages/?id=662
+	defer time.Sleep(100 * time.Millisecond)
+	defer m.sending.Unlock()
+
+	if !m.IsConnected() {
+		return errors.New("connection is closed")
+	}
+
+	return m.conn.WriteJSON(ping)
+}
diff --git a/htx/usdm/accountws/client_test.go b/htx/usdm/accountws/client_test.go
new file mode 100644
index 0000000..0e5f543
--- /dev/null
+++ b/htx/usdm/accountws/client_test.go
@@ -0,0 +1,67 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accountws
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/linstohu/nexapi/htx/usdm/accountws/types"
+	"github.com/stretchr/testify/assert"
+)
+
+func testNewAccountWsClient(ctx context.Context, t *testing.T, url string) *AccountWsClient {
+	cli, err := NewAccountWsClient(ctx, &AccountWsClientCfg{
+		BaseURL: url,
+		Debug:   true,
+		Key:     os.Getenv("HTX_KEY"),
+		Secret:  os.Getenv("HTX_SECRET"),
+	})
+
+	if err != nil {
+		t.Fatalf("Could not create websocket client, %s", err)
+	}
+
+	return cli
+}
+
+func TestSubscribeAccountUpdate(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+
+	cli := testNewAccountWsClient(ctx, t, GlobalOrderWsBaseURL)
+
+	topic, err := cli.GetCrossAccountUpdateTopic("ETH-USDT")
+	assert.Nil(t, err)
+
+	cli.AddListener(topic, func(e any) {
+		acc, ok := e.(*types.CrossAccount)
+		if !ok {
+			return
+		}
+
+		fmt.Printf("Topic: %s, Data: %+v\n",
+			topic, acc.Data)
+	})
+
+	cli.Subscribe(topic)
+
+	select {}
+}
diff --git a/htx/usdm/accountws/events.go b/htx/usdm/accountws/events.go
new file mode 100644
index 0000000..bf30e78
--- /dev/null
+++ b/htx/usdm/accountws/events.go
@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accountws
+
+import "github.com/chuckpreslar/emission"
+
+type Listener func(any)
+
+func (m *AccountWsClient) AddListener(event string, listener Listener) *emission.Emitter {
+	return m.emitter.On(event, listener)
+}
+
+func (m *AccountWsClient) RemoveListener(event string, listener Listener) *emission.Emitter {
+	return m.emitter.Off(m, listener)
+}
+
+func (m *AccountWsClient) GetListeners(event string, argument any) *emission.Emitter {
+	return m.emitter.Emit(event, argument)
+}
diff --git a/htx/usdm/accountws/request.go b/htx/usdm/accountws/request.go
new file mode 100644
index 0000000..0118307
--- /dev/null
+++ b/htx/usdm/accountws/request.go
@@ -0,0 +1,35 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accountws
+
+const (
+	SUB  = "sub"
+	PING = "ping"
+	PONG = "pong"
+	AUTH = "auth"
+)
+
+type AuthRequest struct {
+	Operation        string `json:"op,omitempty"`
+	Type             string `json:"type,omitempty"`
+	AccessKeyId      string `json:"AccessKeyId,omitempty"`
+	SignatureMethod  string `json:"SignatureMethod,omitempty"`
+	SignatureVersion string `json:"SignatureVersion,omitempty"`
+	Timestamp        string `json:"Timestamp,omitempty"`
+	Signature        string `json:"Signature,omitempty"`
+}
diff --git a/htx/usdm/accountws/response.go b/htx/usdm/accountws/response.go
new file mode 100644
index 0000000..a846ac8
--- /dev/null
+++ b/htx/usdm/accountws/response.go
@@ -0,0 +1,54 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accountws
+
+import (
+	"encoding/json"
+
+	"github.com/valyala/fastjson"
+)
+
+type Message struct {
+	Operation string `json:"op,omitempty"`
+	Topic     string `json:"topic,omitempty"`
+	Ts        int64  `json:"ts,omitempty"`
+
+	ErrCode int    `json:"err-code,omitempty"`
+	ErrMsg  string `json:"err-msg,omitempty"`
+
+	Raw json.RawMessage `json:"-"`
+}
+
+func (m *Message) UnmarshalJSON(data []byte) error {
+	var p fastjson.Parser
+	v, err := p.ParseBytes(data)
+	if err != nil {
+		return err
+	}
+
+	m.Operation = string(v.GetStringBytes("op"))
+	m.Topic = string(v.GetStringBytes("topic"))
+	m.Ts = v.GetInt64("ts")
+
+	m.ErrCode = v.GetInt("err-code")
+	m.ErrMsg = string(v.GetStringBytes("err-msg"))
+
+	m.Raw = data
+
+	return nil
+}
diff --git a/htx/usdm/accountws/subscriptions.go b/htx/usdm/accountws/subscriptions.go
new file mode 100644
index 0000000..403731c
--- /dev/null
+++ b/htx/usdm/accountws/subscriptions.go
@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accountws
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/linstohu/nexapi/htx/usdm/accountws/types"
+)
+
+func (m *AccountWsClient) Subscribe(topic string) error {
+	return m.subscribe(topic)
+}
+
+func (m *AccountWsClient) UnSubscribe(topic string) error {
+	return m.unsubscribe(topic)
+}
+
+func (m *AccountWsClient) handle(msg *Message) error {
+	if m.debug {
+		m.logger.Info(fmt.Sprintf("subscribed message, channel: %s", msg.Topic))
+	}
+
+	switch {
+	case strings.HasPrefix(msg.Topic, "accounts."):
+		var data types.IsoAccount
+		err := json.Unmarshal(msg.Raw, &data)
+		if err != nil {
+			return err
+		}
+		m.GetListeners(msg.Topic, &data)
+	case strings.HasPrefix(msg.Topic, "accounts_cross."):
+		var data types.CrossAccount
+		err := json.Unmarshal(msg.Raw, &data)
+		if err != nil {
+			return err
+		}
+		m.GetListeners(msg.Topic, &data)
+	default:
+		return fmt.Errorf("unknown message, topic: %s", msg.Topic)
+	}
+
+	return nil
+}
diff --git a/htx/usdm/accountws/topics.go b/htx/usdm/accountws/topics.go
new file mode 100644
index 0000000..60633c5
--- /dev/null
+++ b/htx/usdm/accountws/topics.go
@@ -0,0 +1,38 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accountws
+
+import (
+	"fmt"
+)
+
+func (m *AccountWsClient) GetIsolatedAccountUpdateTopic(contractCode string) (string, error) {
+	if contractCode == "" {
+		return "", fmt.Errorf("the contract_code field must be provided")
+	}
+
+	return fmt.Sprintf("accounts.%s", contractCode), nil
+}
+
+func (m *AccountWsClient) GetCrossAccountUpdateTopic(marginAccount string) (string, error) {
+	if marginAccount == "" {
+		return "", fmt.Errorf("the margin_account field must be provided")
+	}
+
+	return fmt.Sprintf("accounts_cross.%s", marginAccount), nil
+}
diff --git a/htx/usdm/accountws/types/messages.go b/htx/usdm/accountws/types/messages.go
new file mode 100644
index 0000000..6d0e3ab
--- /dev/null
+++ b/htx/usdm/accountws/types/messages.go
@@ -0,0 +1,104 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package types
+
+type IsoAccount struct {
+	Op    string    `json:"op,omitempty"`
+	Topic string    `json:"topic,omitempty"`
+	Ts    int64     `json:"ts,omitempty"`
+	Event string    `json:"event,omitempty"`
+	Data  []IsoData `json:"data,omitempty"`
+	UID   string    `json:"uid,omitempty"`
+}
+
+type IsoData struct {
+	Symbol            string  `json:"symbol,omitempty"`
+	ContractCode      string  `json:"contract_code,omitempty"`
+	MarginBalance     float64 `json:"margin_balance,omitempty"`
+	MarginStatic      float64 `json:"margin_static,omitempty"`
+	MarginPosition    float64 `json:"margin_position,omitempty"`
+	MarginFrozen      float64 `json:"margin_frozen,omitempty"`
+	MarginAvailable   float64 `json:"margin_available,omitempty"`
+	ProfitReal        float64 `json:"profit_real,omitempty"`
+	ProfitUnreal      float64 `json:"profit_unreal,omitempty"`
+	WithdrawAvailable float64 `json:"withdraw_available,omitempty"`
+	RiskRate          float64 `json:"risk_rate,omitempty"`
+	LiquidationPrice  float64 `json:"liquidation_price,omitempty"`
+	LeverRate         int     `json:"lever_rate,omitempty"`
+	AdjustFactor      float64 `json:"adjust_factor,omitempty"`
+	MarginAsset       string  `json:"margin_asset,omitempty"`
+	MarginMode        string  `json:"margin_mode,omitempty"`
+	MarginAccount     string  `json:"margin_account,omitempty"`
+	PositionMode      string  `json:"position_mode,omitempty"`
+}
+
+type CrossAccount struct {
+	Op    string      `json:"op,omitempty"`
+	Topic string      `json:"topic,omitempty"`
+	Ts    int64       `json:"ts,omitempty"`
+	Event string      `json:"event,omitempty"`
+	Data  []CrossData `json:"data,omitempty"`
+	UID   string      `json:"uid,omitempty"`
+}
+
+type ContractDetail struct {
+	Symbol           string  `json:"symbol,omitempty"`
+	ContractCode     string  `json:"contract_code,omitempty"`
+	MarginPosition   float64 `json:"margin_position,omitempty"`
+	MarginFrozen     float64 `json:"margin_frozen,omitempty"`
+	MarginAvailable  float64 `json:"margin_available,omitempty"`
+	ProfitUnreal     float64 `json:"profit_unreal,omitempty"`
+	LiquidationPrice float64 `json:"liquidation_price,omitempty"`
+	LeverRate        int     `json:"lever_rate,omitempty"`
+	AdjustFactor     float64 `json:"adjust_factor,omitempty"`
+	ContractType     string  `json:"contract_type,omitempty"`
+	Pair             string  `json:"pair,omitempty"`
+	BusinessType     string  `json:"business_type,omitempty"`
+}
+
+type FuturesContractDetail struct {
+	Symbol           string  `json:"symbol,omitempty"`
+	ContractCode     string  `json:"contract_code,omitempty"`
+	MarginPosition   float64 `json:"margin_position,omitempty"`
+	MarginFrozen     float64 `json:"margin_frozen,omitempty"`
+	MarginAvailable  float64 `json:"margin_available,omitempty"`
+	ProfitUnreal     float64 `json:"profit_unreal,omitempty"`
+	LiquidationPrice float64 `json:"liquidation_price,omitempty"`
+	LeverRate        int     `json:"lever_rate,omitempty"`
+	AdjustFactor     float64 `json:"adjust_factor,omitempty"`
+	ContractType     string  `json:"contract_type,omitempty"`
+	Pair             string  `json:"pair,omitempty"`
+	BusinessType     string  `json:"business_type,omitempty"`
+}
+
+type CrossData struct {
+	MarginMode            string                  `json:"margin_mode,omitempty"`
+	MarginAccount         string                  `json:"margin_account,omitempty"`
+	MarginAsset           string                  `json:"margin_asset,omitempty"`
+	MarginBalance         float64                 `json:"margin_balance,omitempty"`
+	MarginStatic          float64                 `json:"margin_static,omitempty"`
+	MarginPosition        float64                 `json:"margin_position,omitempty"`
+	MarginFrozen          float64                 `json:"margin_frozen,omitempty"`
+	ProfitReal            float64                 `json:"profit_real,omitempty"`
+	ProfitUnreal          float64                 `json:"profit_unreal,omitempty"`
+	WithdrawAvailable     float64                 `json:"withdraw_available,omitempty"`
+	RiskRate              float64                 `json:"risk_rate,omitempty"`
+	PositionMode          string                  `json:"position_mode,omitempty"`
+	ContractDetail        []ContractDetail        `json:"contract_detail,omitempty"`
+	FuturesContractDetail []FuturesContractDetail `json:"futures_contract_detail,omitempty"`
+}
diff --git a/htx/usdm/websocket/client.go b/htx/usdm/accountws/vars.go
similarity index 80%
rename from htx/usdm/websocket/client.go
rename to htx/usdm/accountws/vars.go
index d783d40..64cae89 100644
--- a/htx/usdm/websocket/client.go
+++ b/htx/usdm/accountws/vars.go
@@ -15,4 +15,14 @@
  * limitations under the License.
  */
 
-package websocket
+package accountws
+
+const (
+	GlobalOrderWsBaseURL = "wss://api.hbdm.com/linear-swap-notification"
+)
+
+const (
+	MaxTryTimes = 5
+
+	TimerIntervalSecond = 5
+)
diff --git a/htx/usdm/marketws/client.go b/htx/usdm/marketws/client.go
new file mode 100644
index 0000000..7c46bfd
--- /dev/null
+++ b/htx/usdm/marketws/client.go
@@ -0,0 +1,344 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package marketws
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"math/rand"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/chuckpreslar/emission"
+	"github.com/go-playground/validator"
+	"github.com/gorilla/websocket"
+	htxutils "github.com/linstohu/nexapi/htx/utils"
+	cmap "github.com/orcaman/concurrent-map/v2"
+)
+
+type MarketWsClient struct {
+	baseURL string
+	// debug mode
+	debug bool
+	// logger
+	logger *slog.Logger
+
+	ctx         context.Context
+	conn        *websocket.Conn
+	mu          sync.RWMutex
+	isConnected bool
+
+	autoReconnect bool
+	disconnect    chan struct{}
+
+	sending       sync.Mutex
+	subscriptions cmap.ConcurrentMap[string, struct{}]
+
+	emitter *emission.Emitter
+}
+
+type MarketWsClientCfg struct {
+	BaseURL string `validate:"required"`
+	Debug   bool
+	// Logger
+	Logger *slog.Logger
+}
+
+func NewMarketWsClient(ctx context.Context, cfg *MarketWsClientCfg) (*MarketWsClient, error) {
+	if err := validator.New().Struct(cfg); err != nil {
+		return nil, err
+	}
+
+	cli := &MarketWsClient{
+		baseURL: cfg.BaseURL,
+		debug:   cfg.Debug,
+		logger:  cfg.Logger,
+
+		ctx:           ctx,
+		autoReconnect: true,
+
+		subscriptions: cmap.New[struct{}](),
+		emitter:       emission.NewEmitter(),
+	}
+
+	if cli.logger == nil {
+		cli.logger = slog.Default()
+	}
+
+	err := cli.start()
+	if err != nil {
+		return nil, err
+	}
+
+	return cli, nil
+}
+
+func (m *MarketWsClient) start() error {
+	m.conn = nil
+	m.setIsConnected(false)
+	m.disconnect = make(chan struct{})
+
+	for i := 0; i < MaxTryTimes; i++ {
+		conn, _, err := m.connect()
+		if err != nil {
+			m.logger.Info(fmt.Sprintf("connect error, times(%v), error: %s", i, err.Error()))
+			tm := (i + 1) * 5
+			time.Sleep(time.Duration(tm) * time.Second)
+			continue
+		}
+		m.conn = conn
+		break
+	}
+	if m.conn == nil {
+		return errors.New("connect failed")
+	}
+
+	m.setIsConnected(true)
+
+	m.resubscribe()
+
+	if m.autoReconnect {
+		go m.reconnect()
+	}
+
+	go m.readMessages()
+
+	return nil
+}
+
+func (m *MarketWsClient) connect() (*websocket.Conn, *http.Response, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	conn, resp, err := websocket.DefaultDialer.DialContext(ctx, m.baseURL, nil)
+	if err == nil {
+		conn.SetReadLimit(32768 * 64)
+	}
+
+	return conn, resp, err
+}
+
+func (m *MarketWsClient) reconnect() {
+	<-m.disconnect
+
+	m.setIsConnected(false)
+
+	m.logger.Info("disconnect, then reconnect...")
+
+	time.Sleep(1 * time.Second)
+
+	select {
+	case <-m.ctx.Done():
+		m.logger.Info(fmt.Sprintf("never reconnect, %s", m.ctx.Err()))
+		return
+	default:
+		m.start()
+	}
+}
+
+// close closes the websocket connection
+func (m *MarketWsClient) close() error {
+	close(m.disconnect)
+
+	err := m.conn.Close()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// setIsConnected sets state for isConnected
+func (m *MarketWsClient) setIsConnected(state bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	m.isConnected = state
+}
+
+// IsConnected returns the WebSocket connection state
+func (m *MarketWsClient) IsConnected() bool {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	return m.isConnected
+}
+
+func (m *MarketWsClient) readMessages() {
+	for {
+		select {
+		case <-m.ctx.Done():
+			m.logger.Info(fmt.Sprintf("context done, error: %s", m.ctx.Err().Error()))
+
+			if err := m.close(); err != nil {
+				m.logger.Info(fmt.Sprintf("websocket connection closed error, %s", err.Error()))
+			}
+
+			return
+		default:
+			msgType, buf, err := m.conn.ReadMessage()
+			if err != nil {
+				m.logger.Error(fmt.Sprintf("read message error, %s", err))
+				time.Sleep(TimerIntervalSecond * time.Second)
+				continue
+			}
+
+			// decompress gzip data if it is binary message
+			if msgType == websocket.BinaryMessage {
+				message, err := htxutils.GZipDecompress(buf)
+				if err != nil {
+					m.logger.Error(fmt.Sprintf("ungzip data error: %s", err))
+
+					if err := m.close(); err != nil {
+						m.logger.Error(fmt.Sprintf("websocket connection closed error, %s", err.Error()))
+					}
+
+					return
+				}
+
+				var msg AnyMessage
+
+				err = json.Unmarshal([]byte(message), &msg)
+				if err != nil {
+					m.logger.Error(fmt.Sprintf("read object error, %s", err))
+
+					if err := m.close(); err != nil {
+						m.logger.Error(fmt.Sprintf("websocket connection closed error, %s", err.Error()))
+					}
+
+					return
+				}
+
+				switch {
+				case msg.Ping != nil:
+					err := m.pong(&PongMessage{
+						Pong: msg.Ping.Ping,
+					})
+					if err != nil {
+						m.logger.Error(fmt.Sprintf("handle ping error: %s", err.Error()))
+					}
+				case msg.Response != nil:
+					// todo
+				case msg.SubscribedMessage != nil:
+					err := m.handle(msg.SubscribedMessage)
+					if err != nil {
+						m.logger.Error(fmt.Sprintf("handle message error: %s", err.Error()))
+					}
+				}
+			}
+		}
+	}
+}
+
+func (m *MarketWsClient) resubscribe() error {
+	topics := m.subscriptions.Keys()
+
+	if len(topics) == 0 {
+		return nil
+	}
+
+	redo := make([]string, 0)
+
+	for _, v := range topics {
+		// do subscription
+		err := m.send(&Request{
+			ID:  fmt.Sprintf("%v", rand.Uint32()),
+			Sub: v,
+		})
+
+		if err != nil {
+			redo = append(redo, v)
+			continue
+		}
+	}
+
+	if len(redo) != 0 {
+		return fmt.Errorf("resubscribe error: %s", strings.Join(redo, ","))
+	}
+
+	return nil
+}
+
+func (m *MarketWsClient) subscribe(topic string) error {
+	if m.subscriptions.Has(topic) {
+		return nil
+	}
+
+	// do subscription
+
+	err := m.send(&Request{
+		ID:  fmt.Sprintf("%v", rand.Uint32()),
+		Sub: topic,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	m.subscriptions.Set(topic, struct{}{})
+
+	return nil
+}
+
+func (m *MarketWsClient) unsubscribe(topic string) error {
+	err := m.send(&Request{
+		ID:    fmt.Sprintf("%v", rand.Uint32()),
+		UnSub: topic,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	m.subscriptions.Remove(topic)
+
+	return nil
+}
+
+func (m *MarketWsClient) send(req *Request) error {
+	m.sending.Lock()
+
+	// Rate Limit: https://www.htx.com/en-us/opend/newApiPages/?id=662
+	defer time.Sleep(100 * time.Millisecond)
+	defer m.sending.Unlock()
+
+	if !m.IsConnected() {
+		return errors.New("connection is closed")
+	}
+
+	return m.conn.WriteJSON(req)
+}
+
+func (m *MarketWsClient) pong(ping *PongMessage) error {
+	m.sending.Lock()
+
+	// Rate Limit: https://www.htx.com/en-us/opend/newApiPages/?id=662
+	defer time.Sleep(100 * time.Millisecond)
+	defer m.sending.Unlock()
+
+	if !m.IsConnected() {
+		return errors.New("connection is closed")
+	}
+
+	return m.conn.WriteJSON(ping)
+}
diff --git a/htx/usdm/marketws/client_test.go b/htx/usdm/marketws/client_test.go
new file mode 100644
index 0000000..822a260
--- /dev/null
+++ b/htx/usdm/marketws/client_test.go
@@ -0,0 +1,100 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package marketws
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/linstohu/nexapi/htx/usdm/marketws/types"
+	usdmtypes "github.com/linstohu/nexapi/htx/usdm/rest/types"
+	"github.com/stretchr/testify/assert"
+)
+
+func testNewMarketWsClient(ctx context.Context, t *testing.T, url string) *MarketWsClient {
+	cli, err := NewMarketWsClient(ctx, &MarketWsClientCfg{
+		BaseURL: url,
+		Debug:   true,
+	})
+
+	if err != nil {
+		t.Fatalf("Could not create websocket client, %s", err)
+	}
+
+	return cli
+}
+
+func TestSubscribeKline(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+
+	cli := testNewMarketWsClient(ctx, t, GlobalMarketWsBaseURL)
+
+	topic, err := cli.GetKlineTopic(&KlineTopicParam{
+		ContractCode: "BTC-USDT",
+		Interval:     usdmtypes.Minute1,
+	})
+	assert.Nil(t, err)
+
+	cli.AddListener(topic, func(e any) {
+		kline, ok := e.(*types.Kline)
+		if !ok {
+			return
+		}
+
+		fmt.Printf("Topic: %s, Open: %v, Close: %v, Low: %v, High: %v, Amount: %v\n",
+			topic, kline.Open, kline.Close, kline.Low, kline.High, kline.Amount)
+	})
+
+	cli.Subscribe(topic)
+
+	select {}
+}
+
+func TestSubscribeDepth(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+
+	cli := testNewMarketWsClient(ctx, t, GlobalMarketWsBaseURL)
+
+	topic, err := cli.GetDepthTopic(&DepthTopicParam{
+		ContractCode: "BTC-USDT",
+		Type:         "step0",
+	})
+	assert.Nil(t, err)
+
+	cli.AddListener(topic, func(e any) {
+		depth, ok := e.(*types.Depth)
+		if !ok {
+			return
+		}
+
+		fmt.Printf("Topic: %s, ", topic)
+		for _, v := range depth.Bids {
+			fmt.Printf("Bids: %v", v)
+		}
+		for _, v := range depth.Asks {
+			fmt.Printf("Bids: %v", v)
+		}
+	})
+
+	cli.Subscribe(topic)
+
+	select {}
+}
diff --git a/htx/usdm/marketws/events.go b/htx/usdm/marketws/events.go
new file mode 100644
index 0000000..095fe50
--- /dev/null
+++ b/htx/usdm/marketws/events.go
@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package marketws
+
+import "github.com/chuckpreslar/emission"
+
+type Listener func(any)
+
+func (m *MarketWsClient) AddListener(event string, listener Listener) *emission.Emitter {
+	return m.emitter.On(event, listener)
+}
+
+func (m *MarketWsClient) RemoveListener(event string, listener Listener) *emission.Emitter {
+	return m.emitter.Off(m, listener)
+}
+
+func (m *MarketWsClient) GetListeners(event string, argument any) *emission.Emitter {
+	return m.emitter.Emit(event, argument)
+}
diff --git a/htx/usdm/marketws/response.go b/htx/usdm/marketws/response.go
new file mode 100644
index 0000000..69b7b8c
--- /dev/null
+++ b/htx/usdm/marketws/response.go
@@ -0,0 +1,125 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package marketws
+
+import (
+	"encoding/json"
+	"errors"
+
+	"github.com/valyala/fastjson"
+)
+
+type Request struct {
+	ID    string `json:"id,omitempty"`
+	Sub   string `json:"sub,omitempty"`
+	UnSub string `json:"unsub,omitempty"`
+}
+
+// AnyMessage represents either a JSON Response or SubscribedMessage.
+type AnyMessage struct {
+	Ping              *PingMessage
+	Response          *Response
+	SubscribedMessage *SubscribedMessage
+}
+
+type PingMessage struct {
+	Ping int64 `json:"ping,omitempty"`
+}
+
+type PongMessage struct {
+	Pong int64 `json:"pong,omitempty"`
+}
+
+type Response struct {
+	ID     string `json:"id,omitempty"`
+	Status string `json:"status,omitempty"`
+	Subbed string `json:"subbed,omitempty"`
+	Ts     int64  `json:"ts,omitempty"`
+}
+
+type SubscribedMessage struct {
+	Channel string          `json:"ch,omitempty"`
+	Ts      int64           `json:"ts,omitempty"`
+	Data    json.RawMessage `json:"tick"`
+}
+
+func (m AnyMessage) MarshalJSON() ([]byte, error) {
+	var v any
+
+	switch {
+	case m.Response != nil && m.SubscribedMessage == nil:
+		v = m.Response
+	case m.Response == nil && m.SubscribedMessage != nil:
+		v = m.SubscribedMessage
+	}
+
+	if v != nil {
+		return json.Marshal(v)
+	}
+
+	return nil, errors.New("message must have exactly one of the Response or SubscribedMessage fields set")
+}
+
+func (m *AnyMessage) UnmarshalJSON(data []byte) error {
+	var p fastjson.Parser
+	v, err := p.ParseBytes(data)
+	if err != nil {
+		return err
+	}
+
+	if v.Exists("ping") {
+		var resp PingMessage
+
+		if err := json.Unmarshal(data, &resp); err != nil {
+			return err
+		}
+
+		m.Ping = &resp
+
+		return nil
+	}
+
+	if v.Exists("id") {
+		var resp Response
+
+		if err := json.Unmarshal(data, &resp); err != nil {
+			return err
+		}
+
+		m.Response = &resp
+
+		return nil
+	}
+
+	if v.Exists("ch") {
+		msg := &SubscribedMessage{
+			Channel: string(v.GetStringBytes("ch")),
+			Ts:      v.GetInt64("ts"),
+		}
+
+		if v.Get("tick") != nil {
+			msg.Data = v.Get("tick").MarshalTo(nil)
+		}
+
+		m.SubscribedMessage = msg
+
+		return nil
+	}
+
+	return nil
+}
diff --git a/htx/usdm/marketws/subscriptions.go b/htx/usdm/marketws/subscriptions.go
new file mode 100644
index 0000000..8aa5685
--- /dev/null
+++ b/htx/usdm/marketws/subscriptions.go
@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package marketws
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/linstohu/nexapi/htx/usdm/marketws/types"
+)
+
+func (m *MarketWsClient) Subscribe(topic string) error {
+	return m.subscribe(topic)
+}
+
+func (m *MarketWsClient) UnSubscribe(topic string) error {
+	return m.unsubscribe(topic)
+}
+
+func (m *MarketWsClient) handle(msg *SubscribedMessage) error {
+	if m.debug {
+		m.logger.Info(fmt.Sprintf("subscribed message, channel: %s", msg.Channel))
+	}
+
+	switch {
+	case strings.Contains(msg.Channel, "kline"):
+		var data types.Kline
+		err := json.Unmarshal(msg.Data, &data)
+		if err != nil {
+			return err
+		}
+		m.GetListeners(msg.Channel, &data)
+	case strings.Contains(msg.Channel, "depth"):
+		var data types.Depth
+		err := json.Unmarshal(msg.Data, &data)
+		if err != nil {
+			return err
+		}
+		m.GetListeners(msg.Channel, &data)
+	case strings.Contains(msg.Channel, "bbo"):
+		var data types.BBO
+		err := json.Unmarshal(msg.Data, &data)
+		if err != nil {
+			return err
+		}
+		m.GetListeners(msg.Channel, &data)
+	case strings.Contains(msg.Channel, "trade"):
+		var data types.MarketTradeMsg
+		err := json.Unmarshal(msg.Data, &data)
+		if err != nil {
+			return err
+		}
+		m.GetListeners(msg.Channel, &data)
+	default:
+		return fmt.Errorf("unknown message, topic: %s", msg.Channel)
+	}
+
+	return nil
+}
diff --git a/htx/usdm/marketws/topics.go b/htx/usdm/marketws/topics.go
new file mode 100644
index 0000000..f39111d
--- /dev/null
+++ b/htx/usdm/marketws/topics.go
@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package marketws
+
+import (
+	"fmt"
+
+	"github.com/go-playground/validator"
+	usdmtypes "github.com/linstohu/nexapi/htx/usdm/rest/types"
+)
+
+type KlineTopicParam struct {
+	ContractCode string                  `validate:"required"`
+	Interval     usdmtypes.KlineInterval `validate:"required,oneof=1min 5min 15min 30min 1hour 4hour 1day 1mon"`
+}
+
+func (m *MarketWsClient) GetKlineTopic(params *KlineTopicParam) (string, error) {
+	err := validator.New().Struct(params)
+	if err != nil {
+		return "", err
+	}
+
+	return fmt.Sprintf("market.%s.kline.%s", params.ContractCode, params.Interval), nil
+}
+
+type DepthTopicParam struct {
+	ContractCode string `validate:"required"`
+	Type         string `validate:"required,oneof=step0 step1 step2 step3 step4 step5 step6 ste7 step8 step9 step10 step11 step12 step13 step14 step15 step16 step17 step18 step19"`
+}
+
+func (m *MarketWsClient) GetDepthTopic(params *DepthTopicParam) (string, error) {
+	err := validator.New().Struct(params)
+	if err != nil {
+		return "", err
+	}
+	return fmt.Sprintf("market.%s.depth.%s", params.ContractCode, params.Type), nil
+}
+
+func (m *MarketWsClient) GetBBOTopic(contractCode string) (string, error) {
+	if contractCode == "" {
+		return "", fmt.Errorf("the contract_code field must be provided")
+	}
+	return fmt.Sprintf("market.%s.bbo", contractCode), nil
+}
+
+func (m *MarketWsClient) GetMarketTradeTopic(contractCode string) (string, error) {
+	if contractCode == "" {
+		return "", fmt.Errorf("the contract_code field must be provided")
+	}
+	return fmt.Sprintf("market.%s.trade.detail", contractCode), nil
+}
diff --git a/htx/usdm/marketws/types/messages.go b/htx/usdm/marketws/types/messages.go
new file mode 100644
index 0000000..498472f
--- /dev/null
+++ b/htx/usdm/marketws/types/messages.go
@@ -0,0 +1,67 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package types
+
+type Kline struct {
+	ID            int64   `json:"id,omitempty"`
+	MrID          int64   `json:"mrid,omitempty"`
+	Open          float64 `json:"open,omitempty"`
+	Close         float64 `json:"close,omitempty"`
+	Low           float64 `json:"low,omitempty"`
+	High          float64 `json:"high,omitempty"`
+	Amount        float64 `json:"amount,omitempty"`
+	Vol           float64 `json:"vol,omitempty"`
+	Count         int     `json:"count,omitempty"`
+	TradeTurnover float64 `json:"trade_turnover,omitempty"`
+}
+
+type Depth struct {
+	MrID    int64       `json:"mrid,omitempty"`
+	ID      int64       `json:"id,omitempty"`
+	Bids    [][]float64 `json:"bids,omitempty"`
+	Asks    [][]float64 `json:"asks,omitempty"`
+	Ts      int64       `json:"ts,omitempty"`
+	Version int64       `json:"version,omitempty"`
+	Ch      string      `json:"ch,omitempty"`
+}
+
+type BBO struct {
+	MrID    int64     `json:"mrid,omitempty"`
+	ID      int64     `json:"id,omitempty"`
+	Bid     []float64 `json:"bid,omitempty"`
+	Ask     []float64 `json:"ask,omitempty"`
+	Ts      int64     `json:"ts,omitempty"`
+	Version int64     `json:"version,omitempty"`
+	Ch      string    `json:"ch,omitempty"`
+}
+
+type MarketTradeMsg struct {
+	ID   int64         `json:"id,omitempty"`
+	Ts   int64         `json:"ts,omitempty"`
+	Data []MarketTrade `json:"data,omitempty"`
+}
+
+type MarketTrade struct {
+	ID            int64   `json:"id,omitempty"`
+	Ts            int64   `json:"ts,omitempty"`
+	Amount        float64 `json:"amount,omitempty"`
+	Price         float64 `json:"price,omitempty"`
+	Direction     string  `json:"direction,omitempty"`
+	Quantity      float64 `json:"quantity,omitempty"`
+	TradeTurnover float64 `json:"trade_turnover,omitempty"`
+}
diff --git a/htx/usdm/marketws/vars.go b/htx/usdm/marketws/vars.go
new file mode 100644
index 0000000..4cf7d62
--- /dev/null
+++ b/htx/usdm/marketws/vars.go
@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package marketws
+
+const (
+	GlobalMarketWsBaseURL       = "wss://api.hbdm.com/linear-swap-ws"
+	GlobalIndexWsBaseURL        = "wss://api.hbdm.com/ws_index"
+	GlobalSystemStatusWsBaseURL = "wss://api.hbdm.com/center-notification"
+)
+
+const (
+	MaxTryTimes = 5
+
+	TimerIntervalSecond = 5
+)
diff --git a/htx/usdm/rest/types/kline.go b/htx/usdm/rest/types/kline.go
new file mode 100644
index 0000000..be8fa41
--- /dev/null
+++ b/htx/usdm/rest/types/kline.go
@@ -0,0 +1,31 @@
+/*
+ * Copyright (c) 2023, LinstoHu
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package types
+
+type KlineInterval string
+
+var (
+	Minute1  KlineInterval = "1min"
+	Minute5  KlineInterval = "5min"
+	Minute15 KlineInterval = "15min"
+	Minute30 KlineInterval = "30min"
+	Hour1    KlineInterval = "1hour"
+	Hour4    KlineInterval = "4hour"
+	Day1     KlineInterval = "1day"
+	Month1   KlineInterval = "1mon"
+)