From aa24995d283a5a22a18c69293dc2213cb75de57d Mon Sep 17 00:00:00 2001 From: discountry Date: Fri, 16 Jan 2026 10:49:16 +0800 Subject: [PATCH] Enhance WebSocket and API documentation; implement connection protection features - Added a note in the HTTP API documentation regarding the non-guaranteed sequence of price levels in order book responses. - Updated WebSocket documentation to include a connection duration limit and a note on local sorting requirements for price levels. - Introduced connection event handling in the ExchangeAdapter interface, allowing for disconnection and reconnection events. - Implemented connection protection logic in the StandxExchangeAdapter and MakerPointsEngine to manage order states during connection disruptions. - Enhanced the StandxGateway with methods for querying open orders and forcefully canceling all orders, improving reliability during network issues. --- docs/standx/http.md | 6 +- docs/standx/websocket.md | 9 ++ src/exchanges/adapter.ts | 10 ++ src/exchanges/standx/adapter.ts | 36 +++++- src/exchanges/standx/gateway.ts | 174 ++++++++++++++++++++++++++ src/strategy/maker-points-engine.ts | 181 +++++++++++++++++++++++++++- 6 files changed, 411 insertions(+), 5 deletions(-) diff --git a/docs/standx/http.md b/docs/standx/http.md index 91c11ae..b1da388 100644 --- a/docs/standx/http.md +++ b/docs/standx/http.md @@ -679,6 +679,8 @@ To receive order updates via [Order Response Stream](https://docs.standx.com/sta `GET /api/query_depth_book` +**⚠️ Note: The sequence of price levels in the asks and bids arrays is not guaranteed. Please implement local sorting on the client side based on your specific requirements.** + **Required Parameters** | Parameter | Type | Description | @@ -791,7 +793,7 @@ To receive order updates via [Order Response Stream](https://docs.standx.com/sta | Parameter | Type | Description | | --- | --- | --- | -| countBack | u64 | The required amount of bars to load | +| countback | u64 | The required amount of bars to load | **Response Example**: @@ -840,4 +842,4 @@ For enums, constants, and error codes, see [API Reference](https://docs.standx.c Last updated on -[Perps Auth](https://docs.standx.com/standx-api/perps-auth "Perps Auth") [Perps WebSocket API](https://docs.standx.com/standx-api/perps-ws "Perps WebSocket API") \ No newline at end of file +[Perps Auth SVM Example](https://docs.standx.com/standx-api/perps-auth-svm-example "Perps Auth SVM Example") [Perps WebSocket API](https://docs.standx.com/standx-api/perps-ws "Perps WebSocket API") \ No newline at end of file diff --git a/docs/standx/websocket.md b/docs/standx/websocket.md index c72d984..a659f54 100644 --- a/docs/standx/websocket.md +++ b/docs/standx/websocket.md @@ -8,6 +8,12 @@ The WebSocket API provides two streams: **Market Stream** for market data and us Both WebSocket streams implement the following connection management behavior: +### Connection Duration Limit + +- **Maximum Duration**: A single WebSocket connection can be maintained for a maximum of **24 hours** +- After 24 hours, the connection will be automatically terminated +- Clients should implement reconnection logic to handle this gracefully + ### Ping/Pong Mechanism - **Server Ping Interval**: The server sends a WebSocket Ping frame every 10 seconds @@ -45,6 +51,7 @@ Base Endpoint: `wss://perps.standx.com/ws-stream/v1` // public channels { channel: "price", symbol: "" }, { channel: "depth_book", symbol: "" }, + { channel: "public_trade", symbol: "" }, // user-level authenticated channels { channel: "order" }, { channel: "position" }, @@ -55,6 +62,8 @@ Base Endpoint: `wss://perps.standx.com/ws-stream/v1` ### Subscribe to Depth Book +**⚠️ Note: The sequence of price levels in the asks and bids arrays is not guaranteed. Please implement local sorting on the client side based on your specific requirements.** + - Request: - Response: ``` diff --git a/src/exchanges/adapter.ts b/src/exchanges/adapter.ts index a102f59..d88640a 100644 --- a/src/exchanges/adapter.ts +++ b/src/exchanges/adapter.ts @@ -47,6 +47,11 @@ export interface ExchangePrecision { minQuoteAmount?: number; } +export type ConnectionEventType = "disconnected" | "reconnected"; +export interface ConnectionEventListener { + (event: ConnectionEventType, symbol: string): void; +} + export interface ExchangeAdapter { readonly id: string; supportsTrailingStops(): boolean; @@ -61,4 +66,9 @@ export interface ExchangeAdapter { cancelOrders(params: { symbol: string; orderIdList: Array }): Promise; cancelAllOrders(params: { symbol: string }): Promise; getPrecision?(): Promise; + // 连接保护相关方法（可选，仅 StandX 支持） + onConnectionEvent?(listener: ConnectionEventListener): void; + offConnectionEvent?(listener: ConnectionEventListener): void; + queryOpenOrders?(): Promise; + forceCancelAllOrders?(): Promise; } diff --git a/src/exchanges/standx/adapter.ts b/src/exchanges/standx/adapter.ts index 90bce2f..450d599 100644 --- a/src/exchanges/standx/adapter.ts +++ b/src/exchanges/standx/adapter.ts @@ -11,7 +11,9 @@ import type { } from "../adapter"; import type { AsterOrder, CreateOrderParams } from "../types"; import { extractMessage } from "../../utils/errors"; -import { StandxGateway, type StandxGatewayOptions } from "./gateway"; +import { StandxGateway, type StandxGatewayOptions, type ConnectionEventListener, type ConnectionEventType } from "./gateway"; + +export type { ConnectionEventListener, ConnectionEventType }; export interface StandxCredentials { token?: string; @@ -122,6 +124,38 @@ export class StandxExchangeAdapter implements ExchangeAdapter { } } + /** + * 监听连接事件（断连/重连） + */ + onConnectionEvent(listener: ConnectionEventListener): void { + this.gateway.onConnectionEvent(listener); + } + + /** + * 取消连接事件监听 + */ + offConnectionEvent(listener: ConnectionEventListener): void { + this.gateway.offConnectionEvent(listener); + } + + /** + * 查询当前真实的挂单状态（通过 HTTP API） + * 用于验证实际挂单情况，防止取消请求丢失 + */ + async queryOpenOrders(): Promise { + await this.ensureInitialized("queryOpenOrders"); + return this.gateway.queryOpenOrders(this.symbol); + } + + /** + * 强制取消所有挂单 + * 会查询当前挂单然后取消，并验证取消成功 + */ + async forceCancelAllOrders(): Promise { + await this.ensureInitialized("forceCancelAllOrders"); + return this.gateway.forceCancelAllOrders(this.symbol); + } + private safeInvoke void>(context: string, cb: T): T { const wrapped = ((...args: any[]) => { try { diff --git a/src/exchanges/standx/gateway.ts b/src/exchanges/standx/gateway.ts index 0bc6a47..23ca04f 100644 --- a/src/exchanges/standx/gateway.ts +++ b/src/exchanges/standx/gateway.ts @@ -80,6 +80,9 @@ export interface StandxGatewayOptions { logger?: (context: string, error: unknown) => void; } +export type ConnectionEventType = "disconnected" | "reconnected"; +export type ConnectionEventListener = (event: ConnectionEventType, symbol: string) => void; + class StandxRequestSigner { private readonly privateKey: Uint8Array | null; @@ -396,6 +399,7 @@ export class StandxGateway { private readonly tickerListeners = new Map>(); private readonly klineListeners = new Map>(); private readonly fundingListeners = new Map>(); + private readonly connectionListeners = new Set(); private readonly openOrders = new Map(); private readonly positions = new Map(); @@ -417,6 +421,12 @@ export class StandxGateway { private lastPriceBySymbol = new Map(); + // 断连保护相关 + private disconnectCancelRetryTimer: ReturnType | null = null; + private disconnectCancelRetryActive = false; + private lastKnownOpenOrders: Array<{ orderId: string; clOrdId?: string }> = []; + private disconnectedSymbol: string | null = null; + constructor(options: StandxGatewayOptions) { this.token = options.token ?? process.env.STANDX_TOKEN ?? ""; if (!this.token) { @@ -511,6 +521,54 @@ export class StandxGateway { this.startFundingPolling(key); } + onConnectionEvent(listener: ConnectionEventListener): void { + this.connectionListeners.add(listener); + } + + offConnectionEvent(listener: ConnectionEventListener): void { + this.connectionListeners.delete(listener); + } + + /** + * 查询当前真实的挂单状态（通过 HTTP API） + * 用于在网络恢复后验证实际挂单情况 + */ + async queryOpenOrders(symbol: string): Promise { + const normalized = normalizeSymbol(symbol); + const ordersPayload = await this.requestJson("/api/query_open_orders", { + method: "GET", + params: { symbol: normalized }, + }); + const orders = extractOrders(ordersPayload); + const result: AsterOrder[] = []; + for (const raw of orders) { + const order = this.mapOrder(raw); + result.push(order); + } + return result; + } + + /** + * 强制取消所有挂单（用于断连保护） + * 会不断重试直到成功或确认没有挂单 + */ + async forceCancelAllOrders(symbol: string): Promise { + const normalized = normalizeSymbol(symbol); + try { + const currentOrders = await this.queryOpenOrders(normalized); + if (currentOrders.length === 0) { + return true; + } + await this.cancelAllOrders({ symbol: normalized }); + // 再次查询确认 + const afterCancel = await this.queryOpenOrders(normalized); + return afterCancel.length === 0; + } catch (error) { + this.logger("forceCancelAllOrders", error); + return false; + } + } + async createOrder(params: CreateOrderParams): Promise { const normalizedSymbol = normalizeSymbol(params.symbol); if (params.type === "STOP_MARKET") { @@ -791,11 +849,16 @@ export class StandxGateway { this.sendAuthIfNeeded(); }; const handleClose = () => { + const wasReady = this.marketWsReady; this.marketWsReady = false; this.marketWsAuthed = false; this.marketWsAuthRequested = false; this.marketWs = null; this.logDebug("ws close"); + // 触发断连事件，启动断连保护 + if (wasReady) { + this.onDisconnect(); + } this.scheduleReconnect(); }; const handleError = (error: unknown) => { @@ -854,6 +917,8 @@ export class StandxGateway { this.marketWsAuthed = true; this.marketWsAuthRequested = false; this.flushSubscriptions(); + // 触发重连事件 + this.onReconnect(); } return; } @@ -1393,4 +1458,113 @@ export class StandxGateway { return text as unknown as T; } } + + /** + * 断连时触发，记录当前挂单状态并启动持久重试取消 + */ + private onDisconnect(): void { + // 记录最后已知的挂单状态 + this.lastKnownOpenOrders = Array.from(this.openOrders.values()).map((order) => ({ + orderId: String(order.orderId), + clOrdId: order.clientOrderId, + })); + + // 获取当前订阅的 symbol + const symbols = new Set(); + for (const key of this.subscriptions) { + const [, symbol] = key.split(":"); + if (symbol) symbols.add(symbol); + } + this.disconnectedSymbol = symbols.size > 0 ? Array.from(symbols)[0] ?? null : null; + + this.logDebug("disconnect protection", { + openOrderCount: this.lastKnownOpenOrders.length, + symbol: this.disconnectedSymbol, + }); + + // 触发断连事件 + for (const listener of this.connectionListeners) { + try { + listener("disconnected", this.disconnectedSymbol ?? ""); + } catch (error) { + this.logger("connectionListener", error); + } + } + + // 启动断连保护：持续重试取消所有挂单 + if (this.lastKnownOpenOrders.length > 0 && this.disconnectedSymbol) { + this.startDisconnectCancelRetry(this.disconnectedSymbol); + } + } + + /** + * 重连成功时触发，停止断连保护并通知监听器 + */ + private onReconnect(): void { + this.logDebug("reconnect protection", { + wasRetrying: this.disconnectCancelRetryActive, + symbol: this.disconnectedSymbol, + }); + + // 停止断连保护重试 + this.stopDisconnectCancelRetry(); + + // 触发重连事件 + for (const listener of this.connectionListeners) { + try { + listener("reconnected", this.disconnectedSymbol ?? ""); + } catch (error) { + this.logger("connectionListener", error); + } + } + + this.disconnectedSymbol = null; + this.lastKnownOpenOrders = []; + } + + /** + * 启动断连保护：持续重试取消所有挂单 + * 即使网络不通也不停止重试 + */ + private startDisconnectCancelRetry(symbol: string): void { + if (this.disconnectCancelRetryActive) return; + this.disconnectCancelRetryActive = true; + + const retryCancel = async () => { + if (!this.disconnectCancelRetryActive) return; + + this.logDebug("disconnect cancel retry attempt", { symbol }); + + try { + const success = await this.forceCancelAllOrders(symbol); + if (success) { + this.logDebug("disconnect cancel retry success"); + this.stopDisconnectCancelRetry(); + return; + } + } catch (error) { + this.logger("disconnectCancelRetry", error); + } + + // 如果仍在重试状态，继续下一次重试 + if (this.disconnectCancelRetryActive) { + this.disconnectCancelRetryTimer = setTimeout(() => { + void retryCancel(); + }, 2000); // 每 2 秒重试一次 + } + }; + + void retryCancel(); + } + + /** + * 停止断连保护重试 + */ + private stopDisconnectCancelRetry(): void { + this.disconnectCancelRetryActive = false; + if (this.disconnectCancelRetryTimer) { + clearTimeout(this.disconnectCancelRetryTimer); + this.disconnectCancelRetryTimer = null; + } + } } diff --git a/src/strategy/maker-points-engine.ts b/src/strategy/maker-points-engine.ts index 5ed9978..324bc1a 100644 --- a/src/strategy/maker-points-engine.ts +++ b/src/strategy/maker-points-engine.ts @@ -1,5 +1,5 @@ import type { MakerPointsConfig } from "../config"; -import type { ExchangeAdapter } from "../exchanges/adapter"; +import type { ExchangeAdapter, ConnectionEventType } from "../exchanges/adapter"; import type { AsterAccountSnapshot, AsterDepth, @@ -149,6 +149,12 @@ export class MakerPointsEngine { private lastPositionAmt = 0; private lastPositionSide: "LONG" | "SHORT" | "FLAT" = "FLAT"; + // 连接保护相关状态 + private connectionState: "connected" | "disconnected" = "connected"; + private reconnectResetPending = false; + private lastRepriceQueryTime = 0; + private readonly repriceQueryIntervalMs = 3000; // 最小查询间隔 + constructor(private readonly config: MakerPointsConfig, private readonly exchange: ExchangeAdapter) { this.tradeLog = createTradeLog(this.config.maxLogEntries); this.rateLimit = new RateLimitController(this.config.refreshIntervalMs, (type, detail) => @@ -290,6 +296,101 @@ export class MakerPointsEngine { processFail: (error) => t("log.process.tickerError", { error: String(error) }), } ); + + // 注册连接事件监听（如果交易所支持） + this.setupConnectionProtection(); + } + + /** + * 设置连接保护机制 + * 监听断连/重连事件，实现保护逻辑 + */ + private setupConnectionProtection(): void { + if (!this.exchange.onConnectionEvent) return; + + this.exchange.onConnectionEvent((event, symbol) => { + if (event === "disconnected") { + this.handleDisconnect(symbol); + } else if (event === "reconnected") { + this.handleReconnect(symbol); + } + }); + } + + /** + * 处理断连事件 + */ + private handleDisconnect(symbol: string): void { + this.connectionState = "disconnected"; + this.tradeLog.push("warn", `WebSocket 断连 (${symbol})，启动断连保护`); + this.notify({ + type: "token_expired", + level: "warn", + symbol: this.config.symbol, + title: "连接断开", + message: "WebSocket 断连，正在尝试取消所有挂单", + details: { symbol }, + }); + } + + /** + * 处理重连事件 + * 重连后需要重新查询挂单并取消所有挂单 + */ + private async handleReconnect(symbol: string): Promise { + this.connectionState = "connected"; + this.reconnectResetPending = true; + this.tradeLog.push("info", `WebSocket 重连成功 (${symbol})，开始重连保护流程`); + + try { + // 查询真实挂单状态 + if (this.exchange.queryOpenOrders) { + const realOrders = await this.exchange.queryOpenOrders(); + this.tradeLog.push("info", `重连后查询到 ${realOrders.length} 个挂单`); + + if (realOrders.length > 0) { + // 取消所有挂单 + if (this.exchange.forceCancelAllOrders) { + const success = await this.exchange.forceCancelAllOrders(); + if (success) { + this.tradeLog.push("order", "重连保护：已取消所有挂单"); + } else { + this.tradeLog.push("warn", "重连保护：取消挂单未完全成功，将在下次循环重试"); + } + } else { + await this.exchange.cancelAllOrders({ symbol: this.config.symbol }); + this.tradeLog.push("order", "重连保护：已取消所有挂单"); + } + } + } + + // 重置本地挂单状态 + this.openOrders = []; + this.pendingCancelOrders.clear(); + unlockOperating(this.locks, this.timers, this.pending, "LIMIT"); + + // 重置 reprice 基准，强制下一次重新计算 + this.lastQuoteBid1 = null; + this.lastQuoteAsk1 = null; + this.desiredOrders = []; + this.lastDesiredSummary = null; + + // 标记启动重置需要重新执行 + this.initialOrderResetDone = false; + + this.notify({ + type: "position_opened", + level: "info", + symbol: this.config.symbol, + title: "重连完成", + message: "WebSocket 重连成功，已清理挂单状态", + details: { symbol }, + }); + } catch (error) { + this.tradeLog.push("error", `重连保护流程失败: ${extractMessage(error)}`); + } finally { + this.reconnectResetPending = false; + } } private syncLocksWithOrders(orders: AsterOrder[] | null | undefined): void { @@ -553,7 +654,14 @@ export class MakerPointsEngine { } } - private async syncOrders(targets: DesiredOrder[], closeOnly: boolean): Promise { + private async syncOrders(targets: DesiredOrder[], _closeOnly: boolean): Promise { + // 价格变化保护：如果需要 reprice 且距上次查询已过足够时间，先查询真实挂单 + const shouldVerifyOrders = await this.verifyOrdersIfNeeded(); + if (shouldVerifyOrders) { + // 如果发现有未预期的挂单，先取消所有挂单 + return; + } + const availableOrders = this.openOrders.filter((o) => !this.pendingCancelOrders.has(String(o.orderId))); const openOrders = availableOrders.filter((order) => isOrderActiveStatus(order.status)); const { toCancel, toPlace } = makeOrderPlan(openOrders, targets); @@ -629,6 +737,75 @@ export class MakerPointsEngine { } } + /** + * 验证真实挂单状态，防止取消请求丢失 + * 在每次 reprice 时查询真实挂单，发现未预期的挂单时取消所有挂单 + * @returns true 表示发现问题并执行了取消操作，调用方应跳过本轮挂单 + */ + private async verifyOrdersIfNeeded(): Promise { + // 如果交易所不支持查询挂单，跳过验证 + if (!this.exchange.queryOpenOrders) return false; + + // 限制查询频率 + const now = Date.now(); + if (now - this.lastRepriceQueryTime < this.repriceQueryIntervalMs) { + return false; + } + + try { + const realOrders = await this.exchange.queryOpenOrders(); + this.lastRepriceQueryTime = now; + + // 比较真实挂单与本地记录 + const realOrderIds = new Set(realOrders.map((o) => String(o.orderId))); + const localOrderIds = new Set(this.openOrders.map((o) => String(o.orderId))); + + // 查找本地以为已取消但实际还存在的订单 + const unexpectedOrders = realOrders.filter((order) => { + const orderId = String(order.orderId); + // 如果本地没有这个订单，说明我们以为它已经被取消了 + if (!localOrderIds.has(orderId)) { + return true; + } + // 如果本地记录这个订单在等待取消，但实际还存在 + if (this.pendingCancelOrders.has(orderId)) { + return true; + } + return false; + }); + + if (unexpectedOrders.length > 0) { + this.tradeLog.push( + "warn", + `发现 ${unexpectedOrders.length} 个未预期挂单，执行强制取消` + ); + + // 强制取消所有挂单 + if (this.exchange.forceCancelAllOrders) { + await this.exchange.forceCancelAllOrders(); + } else { + await this.exchange.cancelAllOrders({ symbol: this.config.symbol }); + } + + // 重置本地状态 + this.openOrders = []; + this.pendingCancelOrders.clear(); + this.tradeLog.push("order", "已强制取消所有挂单，重置本地状态"); + return true; + } + + // 更新本地挂单状态以匹配真实状态 + if (realOrders.length !== this.openOrders.length) { + // 移除本地记录中不存在于服务器的订单 + this.openOrders = this.openOrders.filter((o) => realOrderIds.has(String(o.orderId))); + } + } catch (error) { + this.tradeLog.push("error", `验证挂单状态失败: ${extractMessage(error)}`); + } + + return false; + } + private async checkStopLoss(): Promise { if (this.stopLossProcessing) return; const lossLimit = Number(this.config.stopLossUsd);