From 20855c2d1dc22f55ae58e83108e16d9e6ea4b959 Mon Sep 17 00:00:00 2001 From: discountry Date: Mon, 6 Apr 2026 18:36:56 +0800 Subject: [PATCH] refactor: extract shared adapter utilities (safeInvoke + initManager) Created src/exchanges/adapter-utils.ts with createSafeInvoke() and createInitManager() factories. Converted 7/8 adapters to use shared utilities, eliminating ~350 lines of duplicated retry/init/error-handling boilerplate. Lighter adapter kept as-is (no retry logic by design). Also translated Chinese comments in adapter.ts and order-schema.ts. --- src/exchanges/adapter-utils.ts | 91 ++++++++++++++++++++++ src/exchanges/adapter.ts | 2 +- src/exchanges/aster/adapter.ts | 89 ++++----------------- src/exchanges/backpack/adapter.ts | 97 ++++------------------- src/exchanges/binance/adapter.ts | 124 +++++++----------------------- src/exchanges/grvt/adapter.ts | 96 ++++------------------- src/exchanges/nado/adapter.ts | 96 ++++------------------- src/exchanges/order-schema.ts | 5 +- src/exchanges/paradex/adapter.ts | 94 ++++------------------ src/exchanges/standx/adapter.ts | 103 +++++-------------------- 10 files changed, 217 insertions(+), 580 deletions(-) create mode 100644 src/exchanges/adapter-utils.ts diff --git a/src/exchanges/adapter-utils.ts b/src/exchanges/adapter-utils.ts new file mode 100644 index 0000000..751e967 --- /dev/null +++ b/src/exchanges/adapter-utils.ts @@ -0,0 +1,91 @@ +import { extractMessage } from "../utils/errors"; + +/** + * Wrap a callback so exceptions inside it are swallowed and logged. + * Used by adapter watch* methods to prevent gateway callback errors + * from killing the event loop. + */ +export function createSafeInvoke(adapterName: string) { + return function safeInvoke void>(context: string, cb: T): T { + const wrapped = ((...args: any[]) => { + try { + cb(...args); + } catch (error) { + console.error(`[${adapterName}] ${context} handler failed: ${extractMessage(error)}`); + } + }) as T; + return wrapped; + }; +} + +export interface InitManager { + ensureInitialized(context?: string): Promise; +} + +/** + * Creates a reusable init-once-with-retry manager. + * Every adapter uses the same pattern: cache the init promise, retry on + * failure with exponential back-off, and deduplicate context error logs. + */ +export function createInitManager( + adapterName: string, + doInitialize: () => Promise, +): InitManager { + let initPromise: Promise | null = null; + const initContexts = new Set(); + let retryTimer: ReturnType | null = null; + let retryDelayMs = 3000; + let lastInitErrorAt = 0; + + function clearRetry(): void { + if (retryTimer) { + clearTimeout(retryTimer); + retryTimer = null; + } + retryDelayMs = 3000; + } + + function handleInitError(context: string, error: unknown): void { + const now = Date.now(); + if (now - lastInitErrorAt < 5000) return; + lastInitErrorAt = now; + console.error(`[${adapterName}] ${context} failed`, error); + } + + function scheduleRetry(): void { + if (retryTimer) return; + retryTimer = setTimeout(() => { + retryTimer = null; + if (initPromise) return; + retryDelayMs = Math.min(retryDelayMs * 2, 60_000); + void ensureInitialized("retry"); + }, retryDelayMs); + } + + function ensureInitialized(context?: string): Promise { + if (!initPromise) { + initContexts.clear(); + initPromise = doInitialize() + .then((value) => { + clearRetry(); + return value; + }) + .catch((error) => { + handleInitError("initialize", error); + initPromise = null; + scheduleRetry(); + throw error; + }); + } + if (context && !initContexts.has(context)) { + initContexts.add(context); + initPromise.catch((error) => { + handleInitError(context, error); + scheduleRetry(); + }); + } + return initPromise; + } + + return { ensureInitialized }; +} diff --git a/src/exchanges/adapter.ts b/src/exchanges/adapter.ts index 75efde9..a2cdd3c 100644 --- a/src/exchanges/adapter.ts +++ b/src/exchanges/adapter.ts @@ -77,7 +77,7 @@ export interface ExchangeAdapter { cancelOrders(params: { symbol: string; orderIdList: Array }): Promise; cancelAllOrders(params: { symbol: string }): Promise; getPrecision?(): Promise; - // 连接保护相关方法（可选，仅 StandX 支持） + // Connection protection methods (optional) onConnectionEvent?(listener: ConnectionEventListener): void; offConnectionEvent?(listener: ConnectionEventListener): void; onRestHealthEvent?(listener: RestHealthListener): void; diff --git a/src/exchanges/aster/adapter.ts b/src/exchanges/aster/adapter.ts index a12d862..45bd96f 100644 --- a/src/exchanges/aster/adapter.ts +++ b/src/exchanges/aster/adapter.ts @@ -8,7 +8,7 @@ import type { TickerListener, } from "../adapter"; import type { Order, CreateOrderParams, Depth, Ticker, Kline } from "../types"; -import { extractMessage } from "../../utils/errors"; +import { createSafeInvoke, createInitManager } from "../adapter-utils"; import { AsterGateway } from "./gateway"; export interface AsterCredentials { @@ -21,132 +21,73 @@ export class AsterExchangeAdapter implements ExchangeAdapter { readonly id = "aster"; private readonly gateway: AsterGateway; private readonly symbol: string; - private initPromise: Promise | null = null; - private lastInitErrorAt = 0; - private readonly initContexts = new Set(); - private retryTimer: ReturnType | null = null; - private retryDelayMs = 3000; + private readonly safeInvoke = createSafeInvoke("AsterExchangeAdapter"); + private readonly init: ReturnType; constructor(credentials: AsterCredentials = {}) { this.gateway = new AsterGateway({ apiKey: credentials.apiKey, apiSecret: credentials.apiSecret }); this.symbol = (credentials.symbol ?? process.env.TRADE_SYMBOL ?? "BTCUSDT").toUpperCase(); + this.init = createInitManager("AsterExchangeAdapter", () => + this.gateway.ensureInitialized(this.symbol), + ); } supportsTrailingStops(): boolean { return true; } - private safeInvoke void>(context: string, cb: T): T { - const wrapped = ((...args: any[]) => { - try { - cb(...args); - } catch (error) { - console.error(`[AsterExchangeAdapter] ${context} handler failed: ${extractMessage(error)}`); - } - }) as T; - return wrapped; - } - - private ensureInitialized(context?: string): Promise { - if (!this.initPromise) { - this.initContexts.clear(); - this.initPromise = this.gateway.ensureInitialized(this.symbol).then((value) => { - this.clearRetry(); - return value; - }).catch((error) => { - this.handleInitError("initialize", error); - this.initPromise = null; - this.scheduleRetry(); - throw error; - }); - } - if (context && !this.initContexts.has(context)) { - this.initContexts.add(context); - this.initPromise.catch((error) => { - this.handleInitError(context, error); - this.scheduleRetry(); - }); - } - return this.initPromise; - } - - private scheduleRetry(): void { - if (this.retryTimer) return; - this.retryTimer = setTimeout(() => { - this.retryTimer = null; - if (this.initPromise) return; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, 60_000); - void this.ensureInitialized("retry"); - }, this.retryDelayMs); - } - - private clearRetry(): void { - if (this.retryTimer) { - clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = 3000; - } - - private handleInitError(context: string, error: unknown): void { - const now = Date.now(); - if (now - this.lastInitErrorAt < 5000) return; - this.lastInitErrorAt = now; - console.error(`[AsterExchangeAdapter] ${context} failed`, error); - } - watchAccount(cb: AccountListener): void { - void this.ensureInitialized("watchAccount"); + void this.init.ensureInitialized("watchAccount"); this.gateway.onAccount(this.safeInvoke("watchAccount", (snapshot) => { cb(snapshot); })); } watchOrders(cb: OrderListener): void { - void this.ensureInitialized("watchOrders"); + void this.init.ensureInitialized("watchOrders"); this.gateway.onOrders(this.safeInvoke("watchOrders", (orders) => { cb(orders); })); } watchDepth(symbol: string, cb: DepthListener): void { - void this.ensureInitialized("watchDepth"); + void this.init.ensureInitialized("watchDepth"); this.gateway.onDepth(symbol, this.safeInvoke("watchDepth", (depth: Depth) => { cb(depth); })); } watchTicker(symbol: string, cb: TickerListener): void { - void this.ensureInitialized("watchTicker"); + void this.init.ensureInitialized("watchTicker"); this.gateway.onTicker(symbol, this.safeInvoke("watchTicker", (ticker: Ticker) => { cb(ticker); })); } watchKlines(symbol: string, interval: string, cb: KlineListener): void { - void this.ensureInitialized("watchKlines"); + void this.init.ensureInitialized("watchKlines"); this.gateway.onKlines(symbol, interval, this.safeInvoke("watchKlines", (klines: Kline[]) => { cb(klines); })); } async createOrder(params: CreateOrderParams): Promise { - await this.ensureInitialized("createOrder"); + await this.init.ensureInitialized("createOrder"); return this.gateway.createOrder(params); } async cancelOrder(params: { symbol: string; orderId: number | string }): Promise { - await this.ensureInitialized("cancelOrder"); + await this.init.ensureInitialized("cancelOrder"); await this.gateway.cancelOrder({ symbol: params.symbol, orderId: Number(params.orderId) }); } async cancelOrders(params: { symbol: string; orderIdList: Array }): Promise { - await this.ensureInitialized("cancelOrders"); + await this.init.ensureInitialized("cancelOrders"); await this.gateway.cancelOrders({ symbol: params.symbol, orderIdList: params.orderIdList }); } async cancelAllOrders(params: { symbol: string }): Promise { - await this.ensureInitialized("cancelAllOrders"); + await this.init.ensureInitialized("cancelAllOrders"); await this.gateway.cancelAllOrders(params); } diff --git a/src/exchanges/backpack/adapter.ts b/src/exchanges/backpack/adapter.ts index 8c977d4..a0285ac 100644 --- a/src/exchanges/backpack/adapter.ts +++ b/src/exchanges/backpack/adapter.ts @@ -9,6 +9,7 @@ import type { import type { Order, CreateOrderParams } from "../types"; import { extractMessage } from "../../utils/errors"; import { BackpackGateway, type BackpackGatewayOptions } from "./gateway"; +import { createSafeInvoke, createInitManager } from "../adapter-utils"; export interface BackpackCredentials { apiKey?: string; @@ -23,11 +24,8 @@ export class BackpackExchangeAdapter implements ExchangeAdapter { readonly id = "backpack"; private readonly gateway: BackpackGateway; private readonly symbol: string; - private initPromise: Promise | null = null; - private readonly initContexts = new Set(); - private retryTimer: ReturnType | null = null; - private retryDelayMs = 3000; - private lastInitErrorAt = 0; + private readonly safeInvoke = createSafeInvoke("BackpackExchangeAdapter"); + private readonly init: ReturnType; constructor(credentials: BackpackCredentials = {}) { const apiKey = credentials.apiKey ?? process.env.BACKPACK_API_KEY; @@ -54,6 +52,9 @@ export class BackpackExchangeAdapter implements ExchangeAdapter { this.gateway = new BackpackGateway(gatewayOptions); this.symbol = symbol; + this.init = createInitManager("BackpackExchangeAdapter", () => + this.gateway.ensureInitialized(this.symbol), + ); } supportsTrailingStops(): boolean { @@ -61,118 +62,50 @@ export class BackpackExchangeAdapter implements ExchangeAdapter { } watchAccount(cb: AccountListener): void { - void this.ensureInitialized("watchAccount"); + void this.init.ensureInitialized("watchAccount"); this.gateway.onAccount(this.safeInvoke("watchAccount", cb)); } watchOrders(cb: OrderListener): void { - void this.ensureInitialized("watchOrders"); + void this.init.ensureInitialized("watchOrders"); this.gateway.onOrders(this.safeInvoke("watchOrders", cb)); } watchDepth(_symbol: string, cb: DepthListener): void { - void this.ensureInitialized("watchDepth"); + void this.init.ensureInitialized("watchDepth"); this.gateway.onDepth(this.safeInvoke("watchDepth", cb)); } watchTicker(_symbol: string, cb: TickerListener): void { - void this.ensureInitialized("watchTicker"); + void this.init.ensureInitialized("watchTicker"); this.gateway.onTicker(this.safeInvoke("watchTicker", cb)); } watchKlines(_symbol: string, interval: string, cb: KlineListener): void { - void this.ensureInitialized(`watchKlines:${interval}`); + void this.init.ensureInitialized(`watchKlines:${interval}`); this.gateway.watchKlines(interval, this.safeInvoke("watchKlines", cb)); } async createOrder(params: CreateOrderParams): Promise { - await this.ensureInitialized("createOrder"); + await this.init.ensureInitialized("createOrder"); return this.gateway.createOrder(params); } async cancelOrder(params: { symbol: string; orderId: number | string }): Promise { - await this.ensureInitialized("cancelOrder"); + await this.init.ensureInitialized("cancelOrder"); await this.gateway.cancelOrder({ orderId: params.orderId }); } async cancelOrders(params: { symbol: string; orderIdList: Array }): Promise { - await this.ensureInitialized("cancelOrders"); + await this.init.ensureInitialized("cancelOrders"); await this.gateway.cancelOrders({ orderIdList: params.orderIdList }); } async cancelAllOrders(_params: { symbol: string }): Promise { - await this.ensureInitialized("cancelAllOrders"); + await this.init.ensureInitialized("cancelAllOrders"); await this.gateway.cancelAllOrders(); } - private safeInvoke void>(context: string, cb: T): T { - const wrapped = ((...args: any[]) => { - try { - cb(...args); - } catch (error) { - console.error(`[BackpackExchangeAdapter] ${context} handler failed: ${extractMessage(error)}`); - } - }) as T; - return wrapped; - } - - private ensureInitialized(context?: string): Promise { - if (!this.initPromise) { - this.initContexts.clear(); - this.initPromise = this.gateway - .ensureInitialized(this.symbol) - .then((value) => { - if (process.env.BACKPACK_DEBUG === "1") { - console.error(`[BackpackExchangeAdapter] initialize succeeded`); - } - if (process.env.BACKPACK_DEBUG === "1") { - console.error(`[BackpackExchangeAdapter] initialize succeeded`); - } - this.clearRetry(); - return value; - }) - .catch((error) => { - this.handleInitError("initialize", error); - this.initPromise = null; - this.scheduleRetry(); - throw error; - }); - } - if (context && !this.initContexts.has(context)) { - this.initContexts.add(context); - this.initPromise.catch((error) => { - this.handleInitError(context, error); - this.scheduleRetry(); - }); - } - return this.initPromise; - } - - private scheduleRetry(): void { - if (this.retryTimer) return; - this.retryTimer = setTimeout(() => { - this.retryTimer = null; - if (this.initPromise) return; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, 60_000); - void this.ensureInitialized("retry"); - }, this.retryDelayMs); - } - - private clearRetry(): void { - if (this.retryTimer) { - clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = 3000; - } - - private handleInitError(context: string, error: unknown): void { - const now = Date.now(); - if (now - this.lastInitErrorAt < 5000) return; - this.lastInitErrorAt = now; - console.error(`[BackpackExchangeAdapter] ${context} failed`, error); - } - private logError(context: string, error: unknown): void { if (process.env.BACKPACK_DEBUG === "1" || process.env.BACKPACK_DEBUG === "true") { console.error(`[BackpackExchangeAdapter] ${context} failed: ${extractMessage(error)}`); diff --git a/src/exchanges/binance/adapter.ts b/src/exchanges/binance/adapter.ts index 71dcf69..80bde81 100644 --- a/src/exchanges/binance/adapter.ts +++ b/src/exchanges/binance/adapter.ts @@ -1,4 +1,3 @@ -import { clearTimeout, setTimeout } from "timers"; import type { AccountListener, DepthListener, @@ -10,8 +9,8 @@ import type { TickerListener, } from "../adapter"; import type { Order, CreateOrderParams } from "../types"; -import { extractMessage } from "../../utils/errors"; import { BinanceGateway, type BinanceGatewayOptions } from "./gateway"; +import { createSafeInvoke, createInitManager } from "../adapter-utils"; export interface BinanceCredentials { apiKey?: string; @@ -32,11 +31,8 @@ export class BinanceExchangeAdapter implements ExchangeAdapter { private readonly gateway: BinanceGateway; private readonly symbol: string; private readonly marketType: "spot" | "perp" | "auto"; - private initPromise: Promise | null = null; - private readonly initContexts = new Set(); - private retryTimer: ReturnType | null = null; - private retryDelayMs = 3000; - private lastInitErrorAt = 0; + private readonly safeInvoke = createSafeInvoke("BinanceExchangeAdapter"); + private readonly init: ReturnType; constructor(credentials: BinanceCredentials = {}) { const apiKey = credentials.apiKey ?? process.env.BINANCE_API_KEY; @@ -61,6 +57,9 @@ export class BinanceExchangeAdapter implements ExchangeAdapter { futuresWsUrl: credentials.futuresWsUrl, logger: credentials.logger, }); + this.init = createInitManager("BinanceExchangeAdapter", () => + this.gateway.ensureInitialized(this.symbol), + ); } supportsTrailingStops(): boolean { @@ -69,163 +68,94 @@ export class BinanceExchangeAdapter implements ExchangeAdapter { watchAccount(cb: AccountListener): void { const safe = this.safeInvoke("watchAccount", cb); - void this.ensureInitialized("watchAccount") + void this.init.ensureInitialized("watchAccount") .then(() => { this.gateway.onAccount(safe); - }) - .catch((error) => this.handleInitError("watchAccount", error)); + }); } watchOrders(cb: OrderListener): void { const safe = this.safeInvoke("watchOrders", cb); - void this.ensureInitialized("watchOrders") + void this.init.ensureInitialized("watchOrders") .then(() => { this.gateway.onOrders(safe); - }) - .catch((error) => this.handleInitError("watchOrders", error)); + }); } watchDepth(symbol: string, cb: DepthListener): void { const safe = this.safeInvoke("watchDepth", cb); - void this.ensureInitialized(`watchDepth:${symbol}`) + void this.init.ensureInitialized(`watchDepth:${symbol}`) .then(() => { this.gateway.onDepth(symbol, safe); - }) - .catch((error) => this.handleInitError("watchDepth", error)); + }); } watchTicker(symbol: string, cb: TickerListener): void { const safe = this.safeInvoke("watchTicker", cb); - void this.ensureInitialized(`watchTicker:${symbol}`) + void this.init.ensureInitialized(`watchTicker:${symbol}`) .then(() => { this.gateway.onTicker(symbol, safe); - }) - .catch((error) => this.handleInitError("watchTicker", error)); + }); } watchKlines(symbol: string, interval: string, cb: KlineListener): void { const safe = this.safeInvoke("watchKlines", cb); - void this.ensureInitialized(`watchKlines:${symbol}:${interval}`) + void this.init.ensureInitialized(`watchKlines:${symbol}:${interval}`) .then(() => { this.gateway.onKlines(symbol, interval, safe); - }) - .catch((error) => this.handleInitError("watchKlines", error)); + }); } watchFundingRate(symbol: string, cb: FundingRateListener): void { const safe = this.safeInvoke("watchFundingRate", cb); - void this.ensureInitialized(`watchFundingRate:${symbol}`) + void this.init.ensureInitialized(`watchFundingRate:${symbol}`) .then(() => { this.gateway.onFundingRate(symbol, safe); - }) - .catch((error) => this.handleInitError("watchFundingRate", error)); + }); } async createOrder(params: CreateOrderParams): Promise { - await this.ensureInitialized("createOrder"); + await this.init.ensureInitialized("createOrder"); return this.gateway.createOrder(params); } async cancelOrder(params: { symbol: string; orderId: number | string }): Promise { - await this.ensureInitialized("cancelOrder"); + await this.init.ensureInitialized("cancelOrder"); await this.gateway.cancelOrder(params); } async cancelOrders(params: { symbol: string; orderIdList: Array }): Promise { - await this.ensureInitialized("cancelOrders"); + await this.init.ensureInitialized("cancelOrders"); await this.gateway.cancelOrders(params); } async cancelAllOrders(params: { symbol: string }): Promise { - await this.ensureInitialized("cancelAllOrders"); + await this.init.ensureInitialized("cancelAllOrders"); await this.gateway.cancelAllOrders(params); } async getPrecision(): Promise { - await this.ensureInitialized("getPrecision"); + await this.init.ensureInitialized("getPrecision"); return this.gateway.getPrecision(this.symbol); } async queryOpenOrders(): Promise { - await this.ensureInitialized("queryOpenOrders"); + await this.init.ensureInitialized("queryOpenOrders"); return this.gateway.queryOpenOrders(); } async queryAccountSnapshot() { - await this.ensureInitialized("queryAccountSnapshot"); + await this.init.ensureInitialized("queryAccountSnapshot"); return this.gateway.queryAccountSnapshot(); } async changeMarginMode(params: { symbol: string; marginMode: "isolated" | "cross" }): Promise { - await this.ensureInitialized("changeMarginMode"); + await this.init.ensureInitialized("changeMarginMode"); await this.gateway.changeMarginMode(params.symbol, params.marginMode); } async forceCancelAllOrders(): Promise { - await this.ensureInitialized("forceCancelAllOrders"); + await this.init.ensureInitialized("forceCancelAllOrders"); return this.gateway.forceCancelAllOrders(); } - - private safeInvoke void>(context: string, cb: T): T { - const wrapped = ((...args: any[]) => { - try { - cb(...args); - } catch (error) { - console.error(`[BinanceExchangeAdapter] ${context} handler failed: ${extractMessage(error)}`); - } - }) as T; - return wrapped; - } - - private ensureInitialized(context?: string): Promise { - if (!this.initPromise) { - this.initContexts.clear(); - this.initPromise = this.gateway - .ensureInitialized(this.symbol) - .then((value) => { - this.clearRetry(); - return value; - }) - .catch((error) => { - this.handleInitError("initialize", error); - this.initPromise = null; - this.scheduleRetry(); - throw error; - }); - } - - if (context && !this.initContexts.has(context)) { - this.initContexts.add(context); - this.initPromise.catch((error) => { - this.handleInitError(context, error); - this.scheduleRetry(); - }); - } - return this.initPromise; - } - - private scheduleRetry(): void { - if (this.retryTimer) return; - this.retryTimer = setTimeout(() => { - this.retryTimer = null; - if (this.initPromise) return; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, 60_000); - void this.ensureInitialized("retry"); - }, this.retryDelayMs); - } - - private clearRetry(): void { - if (this.retryTimer) { - clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = 3000; - } - - private handleInitError(context: string, error: unknown): void { - const now = Date.now(); - if (now - this.lastInitErrorAt < 5000) return; - this.lastInitErrorAt = now; - console.error(`[BinanceExchangeAdapter] ${context} failed`, error); - } } diff --git a/src/exchanges/grvt/adapter.ts b/src/exchanges/grvt/adapter.ts index 53a4b96..7f67bc0 100644 --- a/src/exchanges/grvt/adapter.ts +++ b/src/exchanges/grvt/adapter.ts @@ -1,4 +1,3 @@ -import { setTimeout, clearTimeout } from "timers"; import path from "path"; import { createRequire } from "module"; import type { @@ -10,7 +9,6 @@ import type { TickerListener, } from "../adapter"; import type { Order, CreateOrderParams } from "../types"; -import { extractMessage } from "../../utils/errors"; import { GrvtGateway, type GrvtEnvironment, @@ -18,6 +16,7 @@ import { type GrvtHostsOverride, type GrvtSignatureProvider, } from "./gateway"; +import { createSafeInvoke, createInitManager } from "../adapter-utils"; export interface GrvtCredentials { cookie?: string; @@ -40,11 +39,8 @@ export class GrvtExchangeAdapter implements ExchangeAdapter { private readonly gateway: GrvtGateway; private readonly symbol: string; private readonly instrument: string; - private initPromise: Promise | null = null; - private readonly initContexts = new Set(); - private retryTimer: ReturnType | null = null; - private retryDelayMs = 3000; - private lastInitErrorAt = 0; + private readonly safeInvoke = createSafeInvoke("GrvtExchangeAdapter"); + private readonly init: ReturnType; private klineInterval = "1m"; constructor(credentials: GrvtCredentials = {}) { @@ -91,6 +87,9 @@ export class GrvtExchangeAdapter implements ExchangeAdapter { pollIntervals: credentials.pollIntervals, logger: credentials.logger, }); + this.init = createInitManager("GrvtExchangeAdapter", () => + this.gateway.ensureInitialized(this.klineInterval), + ); } supportsTrailingStops(): boolean { @@ -98,115 +97,50 @@ export class GrvtExchangeAdapter implements ExchangeAdapter { } watchAccount(cb: AccountListener): void { - void this.ensureInitialized("watchAccount"); + void this.init.ensureInitialized("watchAccount"); this.gateway.onAccount(this.safeInvoke("watchAccount", cb)); } watchOrders(cb: OrderListener): void { - void this.ensureInitialized("watchOrders"); + void this.init.ensureInitialized("watchOrders"); this.gateway.onOrders(this.safeInvoke("watchOrders", cb)); } watchDepth(_symbol: string, cb: DepthListener): void { - void this.ensureInitialized("watchDepth"); + void this.init.ensureInitialized("watchDepth"); this.gateway.onDepth(this.safeInvoke("watchDepth", cb)); } watchTicker(_symbol: string, cb: TickerListener): void { - void this.ensureInitialized("watchTicker"); + void this.init.ensureInitialized("watchTicker"); this.gateway.onTicker(this.safeInvoke("watchTicker", cb)); } watchKlines(_symbol: string, interval: string, cb: KlineListener): void { this.klineInterval = interval ?? this.klineInterval; - void this.ensureInitialized("watchKlines", this.klineInterval); + void this.init.ensureInitialized("watchKlines"); this.gateway.onKlines(this.safeInvoke("watchKlines", cb)); } async createOrder(params: CreateOrderParams): Promise { - await this.ensureInitialized("createOrder"); + await this.init.ensureInitialized("createOrder"); return this.gateway.createOrder(params); } async cancelOrder(params: { symbol: string; orderId: number | string }): Promise { - await this.ensureInitialized("cancelOrder"); + await this.init.ensureInitialized("cancelOrder"); await this.gateway.cancelOrder(params); } async cancelOrders(params: { symbol: string; orderIdList: Array }): Promise { - await this.ensureInitialized("cancelOrders"); + await this.init.ensureInitialized("cancelOrders"); await this.gateway.cancelOrders(params); } async cancelAllOrders(_params: { symbol: string }): Promise { - await this.ensureInitialized("cancelAllOrders"); + await this.init.ensureInitialized("cancelAllOrders"); await this.gateway.cancelAllOrders(); } - - private safeInvoke void>(context: string, cb: T): T { - const wrapped = ((...args: any[]) => { - try { - cb(...args); - } catch (error) { - console.error(`[GrvtExchangeAdapter] ${context} handler failed: ${extractMessage(error)}`); - } - }) as T; - return wrapped; - } - - private ensureInitialized(context?: string, interval?: string): Promise { - if (interval) { - this.klineInterval = interval; - } - if (!this.initPromise) { - this.initContexts.clear(); - this.initPromise = this.gateway - .ensureInitialized(this.klineInterval) - .then((value) => { - this.clearRetry(); - return value; - }) - .catch((error) => { - this.handleInitError("initialize", error); - this.initPromise = null; - this.scheduleRetry(); - throw error; - }); - } - if (context && !this.initContexts.has(context)) { - this.initContexts.add(context); - this.initPromise.catch((error) => { - this.handleInitError(context, error); - this.scheduleRetry(); - }); - } - return this.initPromise; - } - - private scheduleRetry(): void { - if (this.retryTimer) return; - this.retryTimer = setTimeout(() => { - this.retryTimer = null; - if (this.initPromise) return; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, 60_000); - void this.ensureInitialized("retry"); - }, this.retryDelayMs); - } - - private clearRetry(): void { - if (this.retryTimer) { - clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = 3000; - } - - private handleInitError(context: string, error: unknown): void { - const now = Date.now(); - if (now - this.lastInitErrorAt < 5000) return; - this.lastInitErrorAt = now; - console.error(`[GrvtExchangeAdapter] ${context} failed`, error); - } } function requireValue(value: T | undefined | null, key: string): T { diff --git a/src/exchanges/nado/adapter.ts b/src/exchanges/nado/adapter.ts index 667c1a9..d382720 100644 --- a/src/exchanges/nado/adapter.ts +++ b/src/exchanges/nado/adapter.ts @@ -1,4 +1,3 @@ -import { setTimeout, clearTimeout } from "timers"; import type { AccountListener, DepthListener, @@ -12,6 +11,7 @@ import type { import type { Order, CreateOrderParams } from "../types"; import { extractMessage } from "../../utils/errors"; import { NadoGateway, type NadoGatewayOptions } from "./gateway"; +import { createSafeInvoke, createInitManager } from "../adapter-utils"; import type { ChainEnv } from "@nadohq/shared"; import type { Address } from "viem"; @@ -35,11 +35,8 @@ export class NadoExchangeAdapter implements ExchangeAdapter { private readonly gateway: NadoGateway; private readonly symbol: string; - private initPromise: Promise | null = null; - private readonly initContexts = new Set(); - private retryTimer: ReturnType | null = null; - private retryDelayMs = 3000; - private lastInitErrorAt = 0; + private readonly safeInvoke = createSafeInvoke("NadoExchangeAdapter"); + private readonly init: ReturnType; constructor(credentials: NadoCredentials = {}) { const signerPrivateKey = credentials.signerPrivateKey ?? process.env.NADO_SIGNER_PRIVATE_KEY; @@ -79,6 +76,9 @@ export class NadoExchangeAdapter implements ExchangeAdapter { (process.env.NADO_STOP_TRIGGER_SOURCE as NadoGatewayOptions["stopTriggerSource"] | undefined), logger: (context, error) => this.logError(context, error), }); + this.init = createInitManager("NadoExchangeAdapter", () => + this.gateway.ensureInitialized(this.symbol), + ); } supportsTrailingStops(): boolean { @@ -86,52 +86,52 @@ export class NadoExchangeAdapter implements ExchangeAdapter { } watchAccount(cb: AccountListener): void { - void this.ensureInitialized("watchAccount"); + void this.init.ensureInitialized("watchAccount"); this.gateway.onAccount(this.safeInvoke("watchAccount", cb)); } watchOrders(cb: OrderListener): void { - void this.ensureInitialized("watchOrders"); + void this.init.ensureInitialized("watchOrders"); this.gateway.onOrders(this.safeInvoke("watchOrders", cb)); } watchDepth(symbol: string, cb: DepthListener): void { - void this.ensureInitialized(`watchDepth:${symbol}`); + void this.init.ensureInitialized(`watchDepth:${symbol}`); this.gateway.onDepth(symbol, this.safeInvoke("watchDepth", cb)); } watchTicker(symbol: string, cb: TickerListener): void { - void this.ensureInitialized(`watchTicker:${symbol}`); + void this.init.ensureInitialized(`watchTicker:${symbol}`); this.gateway.onTicker(symbol, this.safeInvoke("watchTicker", cb)); } watchKlines(symbol: string, interval: string, cb: KlineListener): void { - void this.ensureInitialized(`watchKlines:${symbol}:${interval}`); + void this.init.ensureInitialized(`watchKlines:${symbol}:${interval}`); this.gateway.onKlines(symbol, interval, this.safeInvoke("watchKlines", cb)); } watchFundingRate(symbol: string, cb: FundingRateListener): void { - void this.ensureInitialized(`watchFundingRate:${symbol}`); + void this.init.ensureInitialized(`watchFundingRate:${symbol}`); this.gateway.onFundingRate(symbol, this.safeInvoke("watchFundingRate", cb)); } async createOrder(params: CreateOrderParams): Promise { - await this.ensureInitialized("createOrder"); + await this.init.ensureInitialized("createOrder"); return this.gateway.createOrder(params); } async cancelOrder(params: { symbol: string; orderId: number | string }): Promise { - await this.ensureInitialized("cancelOrder"); + await this.init.ensureInitialized("cancelOrder"); await this.gateway.cancelOrder(params); } async cancelOrders(params: { symbol: string; orderIdList: Array }): Promise { - await this.ensureInitialized("cancelOrders"); + await this.init.ensureInitialized("cancelOrders"); await this.gateway.cancelOrders(params); } async cancelAllOrders(params: { symbol: string }): Promise { - await this.ensureInitialized("cancelAllOrders"); + await this.init.ensureInitialized("cancelAllOrders"); await this.gateway.cancelAllOrders(params); } @@ -144,70 +144,6 @@ export class NadoExchangeAdapter implements ExchangeAdapter { } } - private safeInvoke void>(context: string, cb: T): T { - const wrapped = ((...args: any[]) => { - try { - cb(...args); - } catch (error) { - console.error(`[NadoExchangeAdapter] ${context} handler failed: ${extractMessage(error)}`); - } - }) as T; - return wrapped; - } - - private ensureInitialized(context?: string): Promise { - if (!this.initPromise) { - this.initContexts.clear(); - this.initPromise = this.gateway - .ensureInitialized(this.symbol) - .then((value) => { - this.clearRetry(); - return value; - }) - .catch((error) => { - this.handleInitError("initialize", error); - this.initPromise = null; - this.scheduleRetry(); - throw error; - }); - } - - if (context && !this.initContexts.has(context)) { - this.initContexts.add(context); - this.initPromise.catch((error) => { - this.handleInitError(context, error); - this.scheduleRetry(); - }); - } - - return this.initPromise; - } - - private scheduleRetry(): void { - if (this.retryTimer) return; - this.retryTimer = setTimeout(() => { - this.retryTimer = null; - if (this.initPromise) return; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, 60_000); - void this.ensureInitialized("retry"); - }, this.retryDelayMs); - } - - private clearRetry(): void { - if (this.retryTimer) { - clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = 3000; - } - - private handleInitError(context: string, error: unknown): void { - const now = Date.now(); - if (now - this.lastInitErrorAt < 5000) return; - this.lastInitErrorAt = now; - console.error(`[NadoExchangeAdapter] ${context} failed`, error); - } - private logError(context: string, error: unknown): void { const detail = extractMessage(error); const message = `[NadoExchangeAdapter] ${context} failed: ${detail}`; diff --git a/src/exchanges/order-schema.ts b/src/exchanges/order-schema.ts index 0c787ed..75cc3f2 100644 --- a/src/exchanges/order-schema.ts +++ b/src/exchanges/order-schema.ts @@ -13,9 +13,8 @@ export interface BaseOrderIntent { export interface LimitOrderIntent extends BaseOrderIntent { price: number; - // StandX TPSL 参数 - slPrice?: number; // 止损价格 - tpPrice?: number; // 止盈价格 + slPrice?: number; + tpPrice?: number; } export interface MarketOrderIntent extends BaseOrderIntent { diff --git a/src/exchanges/paradex/adapter.ts b/src/exchanges/paradex/adapter.ts index b21d1f7..cd0f419 100644 --- a/src/exchanges/paradex/adapter.ts +++ b/src/exchanges/paradex/adapter.ts @@ -1,4 +1,3 @@ -import { setTimeout, clearTimeout } from "timers"; import type { AccountListener, DepthListener, @@ -10,6 +9,7 @@ import type { import type { Order, CreateOrderParams } from "../types"; import { extractMessage } from "../../utils/errors"; import { ParadexGateway, type ParadexGatewayOptions } from "./gateway"; +import { createSafeInvoke, createInitManager } from "../adapter-utils"; export interface ParadexCredentials { privateKey?: string; @@ -26,11 +26,8 @@ export class ParadexExchangeAdapter implements ExchangeAdapter { private readonly gateway: ParadexGateway; private readonly symbol: string; - private initPromise: Promise | null = null; - private readonly initContexts = new Set(); - private retryTimer: ReturnType | null = null; - private retryDelayMs = 3000; - private lastInitErrorAt = 0; + private readonly safeInvoke = createSafeInvoke("ParadexExchangeAdapter"); + private readonly init: ReturnType; constructor(credentials: ParadexCredentials = {}) { const privateKey = credentials.privateKey ?? process.env.PARADEX_PRIVATE_KEY; @@ -54,6 +51,9 @@ export class ParadexExchangeAdapter implements ExchangeAdapter { }); this.symbol = symbol; + this.init = createInitManager("ParadexExchangeAdapter", () => + this.gateway.ensureInitialized(this.symbol), + ); } supportsTrailingStops(): boolean { @@ -61,114 +61,50 @@ export class ParadexExchangeAdapter implements ExchangeAdapter { } watchAccount(cb: AccountListener): void { - void this.ensureInitialized("watchAccount"); + void this.init.ensureInitialized("watchAccount"); this.gateway.onAccount(this.safeInvoke("watchAccount", cb)); } watchOrders(cb: OrderListener): void { - void this.ensureInitialized("watchOrders"); + void this.init.ensureInitialized("watchOrders"); this.gateway.onOrders(this.safeInvoke("watchOrders", cb)); } watchDepth(symbol: string, cb: DepthListener): void { - void this.ensureInitialized(`watchDepth:${symbol}`); + void this.init.ensureInitialized(`watchDepth:${symbol}`); this.gateway.onDepth(this.safeInvoke("watchDepth", cb)); } watchTicker(symbol: string, cb: TickerListener): void { - void this.ensureInitialized(`watchTicker:${symbol}`); + void this.init.ensureInitialized(`watchTicker:${symbol}`); this.gateway.onTicker(this.safeInvoke("watchTicker", cb)); } watchKlines(symbol: string, interval: string, cb: KlineListener): void { - void this.ensureInitialized(`watchKlines:${symbol}:${interval}`); + void this.init.ensureInitialized(`watchKlines:${symbol}:${interval}`); this.gateway.watchKlines(interval, this.safeInvoke("watchKlines", cb)); } async createOrder(params: CreateOrderParams): Promise { - await this.ensureInitialized("createOrder"); + await this.init.ensureInitialized("createOrder"); return this.gateway.createOrder(params); } async cancelOrder(params: { symbol: string; orderId: number | string }): Promise { - await this.ensureInitialized("cancelOrder"); + await this.init.ensureInitialized("cancelOrder"); await this.gateway.cancelOrder(params); } async cancelOrders(params: { symbol: string; orderIdList: Array }): Promise { - await this.ensureInitialized("cancelOrders"); + await this.init.ensureInitialized("cancelOrders"); await this.gateway.cancelOrders(params); } async cancelAllOrders(params: { symbol: string }): Promise { - await this.ensureInitialized("cancelAllOrders"); + await this.init.ensureInitialized("cancelAllOrders"); await this.gateway.cancelAllOrders(params); } - private safeInvoke void>(context: string, cb: T): T { - const wrapped = ((...args: any[]) => { - try { - cb(...args); - } catch (error) { - console.error(`[ParadexExchangeAdapter] ${context} handler failed: ${extractMessage(error)}`); - } - }) as T; - return wrapped; - } - - private ensureInitialized(context?: string): Promise { - if (!this.initPromise) { - this.initContexts.clear(); - this.initPromise = this.gateway - .ensureInitialized(this.symbol) - .then((value) => { - this.clearRetry(); - return value; - }) - .catch((error) => { - this.handleInitError("initialize", error); - this.initPromise = null; - this.scheduleRetry(); - throw error; - }); - } - - if (context && !this.initContexts.has(context)) { - this.initContexts.add(context); - this.initPromise.catch((error) => { - this.handleInitError(context, error); - this.scheduleRetry(); - }); - } - - return this.initPromise; - } - - private scheduleRetry(): void { - if (this.retryTimer) return; - this.retryTimer = setTimeout(() => { - this.retryTimer = null; - if (this.initPromise) return; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, 60_000); - void this.ensureInitialized("retry"); - }, this.retryDelayMs); - } - - private clearRetry(): void { - if (this.retryTimer) { - clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = 3000; - } - - private handleInitError(context: string, error: unknown): void { - const now = Date.now(); - if (now - this.lastInitErrorAt < 5000) return; - this.lastInitErrorAt = now; - console.error(`[ParadexExchangeAdapter] ${context} failed`, error); - } - private logError(context: string, error: unknown): void { const detail = extractMessage(error); if (context === "initialize" && typeof error === "string" && /initialized/i.test(error)) { diff --git a/src/exchanges/standx/adapter.ts b/src/exchanges/standx/adapter.ts index baedd0f..ec22163 100644 --- a/src/exchanges/standx/adapter.ts +++ b/src/exchanges/standx/adapter.ts @@ -1,4 +1,3 @@ -import { setTimeout, clearTimeout } from "timers"; import type { AccountListener, DepthListener, @@ -11,8 +10,8 @@ import type { TickerListener, } from "../adapter"; import type { Order, CreateOrderParams } from "../types"; -import { extractMessage } from "../../utils/errors"; import { StandxGateway, type StandxGatewayOptions, type ConnectionEventListener, type ConnectionEventType } from "./gateway"; +import { createSafeInvoke, createInitManager } from "../adapter-utils"; export type { ConnectionEventListener, ConnectionEventType }; @@ -31,11 +30,8 @@ export class StandxExchangeAdapter implements ExchangeAdapter { private readonly gateway: StandxGateway; private readonly symbol: string; - private initPromise: Promise | null = null; - private readonly initContexts = new Set(); - private retryTimer: ReturnType | null = null; - private retryDelayMs = 3000; - private lastInitErrorAt = 0; + private readonly safeInvoke = createSafeInvoke("StandxExchangeAdapter"); + private readonly init: ReturnType; constructor(credentials: StandxCredentials = {}) { const token = credentials.token ?? process.env.STANDX_TOKEN; @@ -52,6 +48,9 @@ export class StandxExchangeAdapter implements ExchangeAdapter { signingKey: credentials.signingKey, logger: credentials.logger, }); + this.init = createInitManager("StandxExchangeAdapter", () => + this.gateway.ensureInitialized(this.symbol), + ); } supportsTrailingStops(): boolean { @@ -59,52 +58,52 @@ export class StandxExchangeAdapter implements ExchangeAdapter { } watchAccount(cb: AccountListener): void { - void this.ensureInitialized("watchAccount"); + void this.init.ensureInitialized("watchAccount"); this.gateway.onAccount(this.safeInvoke("watchAccount", cb)); } watchOrders(cb: OrderListener): void { - void this.ensureInitialized("watchOrders"); + void this.init.ensureInitialized("watchOrders"); this.gateway.onOrders(this.safeInvoke("watchOrders", cb)); } watchDepth(symbol: string, cb: DepthListener): void { - void this.ensureInitialized("watchDepth"); + void this.init.ensureInitialized("watchDepth"); this.gateway.onDepth(symbol, this.safeInvoke("watchDepth", cb)); } watchTicker(symbol: string, cb: TickerListener): void { - void this.ensureInitialized("watchTicker"); + void this.init.ensureInitialized("watchTicker"); this.gateway.onTicker(symbol, this.safeInvoke("watchTicker", cb)); } watchKlines(symbol: string, interval: string, cb: KlineListener): void { - void this.ensureInitialized("watchKlines"); + void this.init.ensureInitialized("watchKlines"); this.gateway.onKlines(symbol, interval, this.safeInvoke("watchKlines", cb)); } watchFundingRate(symbol: string, cb: FundingRateListener): void { - void this.ensureInitialized("watchFundingRate"); + void this.init.ensureInitialized("watchFundingRate"); this.gateway.onFundingRate(symbol, this.safeInvoke("watchFundingRate", cb)); } async createOrder(params: CreateOrderParams): Promise { - await this.ensureInitialized("createOrder"); + await this.init.ensureInitialized("createOrder"); return this.gateway.createOrder(params); } async cancelOrder(params: { symbol: string; orderId: number | string }): Promise { - await this.ensureInitialized("cancelOrder"); + await this.init.ensureInitialized("cancelOrder"); await this.gateway.cancelOrder(params); } async cancelOrders(params: { symbol: string; orderIdList: Array }): Promise { - await this.ensureInitialized("cancelOrders"); + await this.init.ensureInitialized("cancelOrders"); await this.gateway.cancelOrders(params); } async cancelAllOrders(params: { symbol: string }): Promise { - await this.ensureInitialized("cancelAllOrders"); + await this.init.ensureInitialized("cancelAllOrders"); await this.gateway.cancelAllOrders(params); } @@ -152,17 +151,17 @@ export class StandxExchangeAdapter implements ExchangeAdapter { * 用于验证实际挂单情况，防止取消请求丢失 */ async queryOpenOrders(): Promise { - await this.ensureInitialized("queryOpenOrders"); + await this.init.ensureInitialized("queryOpenOrders"); return this.gateway.queryOpenOrders(this.symbol); } async queryAccountSnapshot() { - await this.ensureInitialized("queryAccountSnapshot"); + await this.init.ensureInitialized("queryAccountSnapshot"); return this.gateway.queryAccountSnapshot(); } async changeMarginMode(params: { symbol: string; marginMode: "isolated" | "cross" }): Promise { - await this.ensureInitialized("changeMarginMode"); + await this.init.ensureInitialized("changeMarginMode"); await this.gateway.changeMarginMode(params.symbol, params.marginMode); } @@ -171,69 +170,7 @@ export class StandxExchangeAdapter implements ExchangeAdapter { * 会查询当前挂单然后取消，并验证取消成功 */ async forceCancelAllOrders(): Promise { - await this.ensureInitialized("forceCancelAllOrders"); + await this.init.ensureInitialized("forceCancelAllOrders"); return this.gateway.forceCancelAllOrders(this.symbol); } - - private safeInvoke void>(context: string, cb: T): T { - const wrapped = ((...args: any[]) => { - try { - cb(...args); - } catch (error) { - console.error(`[StandxExchangeAdapter] ${context} handler failed: ${extractMessage(error)}`); - } - }) as T; - return wrapped; - } - - private ensureInitialized(context?: string): Promise { - if (!this.initPromise) { - this.initContexts.clear(); - this.initPromise = this.gateway - .ensureInitialized(this.symbol) - .then((value) => { - this.clearRetry(); - return value; - }) - .catch((error) => { - this.handleInitError("initialize", error); - this.initPromise = null; - this.scheduleRetry(); - throw error; - }); - } - if (context && !this.initContexts.has(context)) { - this.initContexts.add(context); - this.initPromise.catch((error) => { - this.handleInitError(context, error); - this.scheduleRetry(); - }); - } - return this.initPromise; - } - - private scheduleRetry(): void { - if (this.retryTimer) return; - this.retryTimer = setTimeout(() => { - this.retryTimer = null; - if (this.initPromise) return; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, 60_000); - void this.ensureInitialized("retry"); - }, this.retryDelayMs); - } - - private clearRetry(): void { - if (this.retryTimer) { - clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = 3000; - } - - private handleInitError(context: string, error: unknown): void { - const now = Date.now(); - if (now - this.lastInitErrorAt < 5000) return; - this.lastInitErrorAt = now; - console.error(`[StandxExchangeAdapter] ${context} failed`, error); - } }