4 Commits
Author SHA1 Message Date
discountry ed855f6859 Refine data staleness checks in MakerPointsEngine
- Updated the logic to only consider depth data for staleness checks, excluding account data from the criteria.
- Removed unnecessary account staleness checks from defense mode activation, streamlining the data validation process.
- Enhanced comments for clarity on the rationale behind the changes.
2026-01-21 16:30:51 +08:00
discountry e144c1822f Implement data staleness defense mode in MakerPointsEngine
- Introduced a defense mode that activates when data from StandX or Binance is stale for over 5 seconds.
- Added methods to check data freshness, enter and exit defense mode, and cancel all orders during defense mode.
- Enhanced logging to provide insights into data staleness and defense mode transitions.
- Updated connection state management for clarity and consistency.
2026-01-21 16:24:17 +08:00
discountry 69271d33ca Refactor MakerPointsEngine and BinanceDepthTracker for improved connection management
- Renamed connection state variable in MakerPointsEngine for clarity.
- Added connection state change listeners in BinanceDepthTracker to handle connection status updates.
- Implemented heartbeat monitoring and connection duration checks in BinanceDepthTracker to enhance WebSocket reliability.
- Introduced data staleness checks and improved error handling for WebSocket connections.
- Enhanced logging for connection events to provide better insights into connection status changes.
2026-01-21 16:03:34 +08:00
discountry f1140f106a Enhance WebSocket connection management and data handling
- Introduced constants for WebSocket reconnection delays, heartbeat timeout, and data staleness thresholds.
- Implemented heartbeat monitoring to ensure timely reconnections on inactivity.
- Added data staleness checks to trigger REST API calls when market or account data is outdated.
- Enhanced the StandxGateway class with methods for managing heartbeat and data checks, improving overall connection reliability and data integrity.
2026-01-21 15:40:28 +08:00
3 changed files with 729 additions and 30 deletions
+263 -6
View File
@@ -48,7 +48,20 @@ const DEFAULT_WS_URL = "wss://perps.standx.com/ws-stream/v1";
const DEFAULT_KLINE_LIMIT = 200; const DEFAULT_KLINE_LIMIT = 200;
const KLINE_REFRESH_MS = 30_000; const KLINE_REFRESH_MS = 30_000;
const FUNDING_REFRESH_MS = 60_000; const FUNDING_REFRESH_MS = 60_000;
const WS_RECONNECT_DELAY = 2000;
// ========== WebSocket 连接管理常量 ==========
// 基础重连延迟(毫秒)
const WS_RECONNECT_DELAY_BASE = 2000;
// 最大重连延迟(毫秒)- 指数退避上限
const WS_RECONNECT_DELAY_MAX = 30_000;
// 心跳超时(毫秒)- StandX 服务器 5 分钟无 pong 会断连,我们设置 2 分钟作为安全阈值
const WS_HEARTBEAT_TIMEOUT = 120_000;
// 心跳检查间隔(毫秒)- 每 30 秒检查一次是否收到消息
const WS_HEARTBEAT_CHECK_INTERVAL = 30_000;
// 数据过时阈值(毫秒)- 超过此时间未收到行情/仓位数据,启动 REST 主动拉取
const WS_DATA_STALE_THRESHOLD = 3000;
// REST 轮询间隔(毫秒)- WS 断连或数据过时时的 REST 拉取间隔
const REST_POLL_INTERVAL = 2000;
const SUPPORTED_QUOTES = ["USD", "USDT", "USDC", "DUSD"]; const SUPPORTED_QUOTES = ["USD", "USDT", "USDC", "DUSD"];
@@ -100,7 +113,7 @@ class StandxRequestSigner {
const requestId = crypto.randomUUID(); const requestId = crypto.randomUUID();
const timestamp = Date.now(); const timestamp = Date.now();
const signMessage = `${version},${requestId},${timestamp},${payload}`; const signMessage = `${version},${requestId},${timestamp},${payload}`;
const signatureBytes = await sign(Buffer.from(signMessage, "utf-8"), this.privateKey); const signatureBytes = sign(Buffer.from(signMessage, "utf-8"), this.privateKey);
return { return {
"x-request-sign-version": version, "x-request-sign-version": version,
"x-request-id": requestId, "x-request-id": requestId,
@@ -416,6 +429,26 @@ export class StandxGateway {
private marketReconnectTimer: ReturnType<typeof setTimeout> | null = null; private marketReconnectTimer: ReturnType<typeof setTimeout> | null = null;
private readonly subscriptions = new Set<string>(); private readonly subscriptions = new Set<string>();
// ========== 心跳与连接管理 ==========
// 上次收到消息的时间戳
private lastMessageTime = 0;
// 心跳检查定时器
private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
// 重连次数(用于指数退避)
private reconnectAttempts = 0;
// ========== 数据过时检测与 REST 备用 ==========
// 上次收到行情数据(price/depth)的时间戳
private lastMarketDataTime = 0;
// 上次收到账户数据(position/balance)的时间戳
private lastAccountDataTime = 0;
// 数据过时检查定时器
private dataStaleCheckTimer: ReturnType<typeof setInterval> | null = null;
// REST 轮询定时器(WS 断连时启用)
private restPollTimer: ReturnType<typeof setInterval> | null = null;
// REST 轮询是否激活
private restPollActive = false;
private readonly klineTimers = new Map<string, PollTimer>(); private readonly klineTimers = new Map<string, PollTimer>();
private readonly fundingTimers = new Map<string, PollTimer>(); private readonly fundingTimers = new Map<string, PollTimer>();
@@ -852,7 +885,18 @@ export class StandxGateway {
const handleOpen = () => { const handleOpen = () => {
this.marketWsReady = true; this.marketWsReady = true;
this.marketWsAuthed = false; this.marketWsAuthed = false;
// 重置重连计数和时间戳
this.reconnectAttempts = 0;
this.lastMessageTime = Date.now();
this.lastMarketDataTime = Date.now();
this.lastAccountDataTime = Date.now();
this.logDebug("ws open"); this.logDebug("ws open");
// 启动心跳监控
this.startHeartbeatMonitor();
// 启动数据过时检测
this.startDataStaleCheck();
// 停止 REST 轮询(WS 恢复后不再需要)
this.stopRestPoll();
this.sendAuthIfNeeded(); this.sendAuthIfNeeded();
}; };
const handleClose = () => { const handleClose = () => {
@@ -861,6 +905,9 @@ export class StandxGateway {
this.marketWsAuthed = false; this.marketWsAuthed = false;
this.marketWsAuthRequested = false; this.marketWsAuthRequested = false;
this.marketWs = null; this.marketWs = null;
// 停止心跳监控和数据过时检测
this.stopHeartbeatMonitor();
this.stopDataStaleCheck();
this.logDebug("ws close"); this.logDebug("ws close");
// 触发断连事件,启动断连保护 // 触发断连事件,启动断连保护
if (wasReady) { if (wasReady) {
@@ -899,15 +946,23 @@ export class StandxGateway {
private scheduleReconnect(): void { private scheduleReconnect(): void {
if (this.marketReconnectTimer) return; if (this.marketReconnectTimer) return;
this.logDebug("scheduling reconnect in 2s"); // 指数退避:delay = min(base * 2^attempts, max)
const delay = Math.min(
WS_RECONNECT_DELAY_BASE * Math.pow(2, this.reconnectAttempts),
WS_RECONNECT_DELAY_MAX
);
this.reconnectAttempts += 1;
this.logDebug(`scheduling reconnect in ${delay}ms (attempt ${this.reconnectAttempts})`);
this.marketReconnectTimer = setTimeout(() => { this.marketReconnectTimer = setTimeout(() => {
this.marketReconnectTimer = null; this.marketReconnectTimer = null;
this.logDebug("attempting reconnect"); this.logDebug("attempting reconnect");
this.connectMarketWs(); this.connectMarketWs();
}, WS_RECONNECT_DELAY); }, delay);
} }
private handleMarketMessage(event: { data: any }): void { private handleMarketMessage(event: { data: any }): void {
// 更新最后收到消息的时间(心跳监控)
this.lastMessageTime = Date.now();
this.logRawPayload(event.data); this.logRawPayload(event.data);
const payloads = parseJsonPayloads(event.data); const payloads = parseJsonPayloads(event.data);
if (payloads.length === 0) return; if (payloads.length === 0) return;
@@ -938,9 +993,11 @@ export class StandxGateway {
return; return;
} }
if (channel === "depth_book") { if (channel === "depth_book") {
// 更新行情数据时间戳
this.lastMarketDataTime = Date.now();
const data = message.data as StandxDepthBook | undefined; const data = message.data as StandxDepthBook | undefined;
const rawSymbol = data?.symbol ?? message?.symbol; const rawSymbol = data?.symbol ?? message?.symbol;
if (!rawSymbol) return; if (!rawSymbol || !data) return;
const bids = normalizeDepthLevels((data.bids ?? []).map(([price, qty]) => [String(price), String(qty)]), "bid"); const bids = normalizeDepthLevels((data.bids ?? []).map(([price, qty]) => [String(price), String(qty)]), "bid");
const asks = normalizeDepthLevels((data.asks ?? []).map(([price, qty]) => [String(price), String(qty)]), "ask"); const asks = normalizeDepthLevels((data.asks ?? []).map(([price, qty]) => [String(price), String(qty)]), "ask");
const decrossed = decrossDepthBook(bids, asks); const decrossed = decrossDepthBook(bids, asks);
@@ -969,13 +1026,15 @@ export class StandxGateway {
lastUpdateId: Number(message.seq ?? Date.now()), lastUpdateId: Number(message.seq ?? Date.now()),
bids: finalBids, bids: finalBids,
asks: finalAsks, asks: finalAsks,
eventTime: toTimestamp(data.time), eventTime: Date.now(),
symbol: rawSymbol, symbol: rawSymbol,
}; };
this.emitDepth(rawSymbol, depth); this.emitDepth(rawSymbol, depth);
return; return;
} }
if (channel === "price") { if (channel === "price") {
// 更新行情数据时间戳
this.lastMarketDataTime = Date.now();
const data = message.data as StandxPrice | undefined; const data = message.data as StandxPrice | undefined;
if (!data?.symbol) return; if (!data?.symbol) return;
const ticker = this.mapTicker(data); const ticker = this.mapTicker(data);
@@ -983,6 +1042,8 @@ export class StandxGateway {
return; return;
} }
if (channel === "order") { if (channel === "order") {
// 更新账户数据时间戳
this.lastAccountDataTime = Date.now();
const payload = message.data as StandxOrder | StandxOrder[] | undefined; const payload = message.data as StandxOrder | StandxOrder[] | undefined;
if (!payload) return; if (!payload) return;
const items = Array.isArray(payload) ? payload : [payload]; const items = Array.isArray(payload) ? payload : [payload];
@@ -994,6 +1055,8 @@ export class StandxGateway {
return; return;
} }
if (channel === "position") { if (channel === "position") {
// 更新账户数据时间戳
this.lastAccountDataTime = Date.now();
const payload = message.data as StandxPosition | StandxPosition[] | undefined; const payload = message.data as StandxPosition | StandxPosition[] | undefined;
if (!payload) return; if (!payload) return;
const items = Array.isArray(payload) ? payload : [payload]; const items = Array.isArray(payload) ? payload : [payload];
@@ -1006,6 +1069,8 @@ export class StandxGateway {
return; return;
} }
if (channel === "balance") { if (channel === "balance") {
// 更新账户数据时间戳
this.lastAccountDataTime = Date.now();
const payload = message.data as StandxBalance | StandxBalance[] | undefined; const payload = message.data as StandxBalance | StandxBalance[] | undefined;
if (!payload) return; if (!payload) return;
const items = Array.isArray(payload) ? payload : [payload]; const items = Array.isArray(payload) ? payload : [payload];
@@ -1048,6 +1113,7 @@ export class StandxGateway {
if (!this.marketWsAuthed) return; if (!this.marketWsAuthed) return;
for (const entry of this.subscriptions) { for (const entry of this.subscriptions) {
const [channel, symbol] = entry.split(":"); const [channel, symbol] = entry.split(":");
if (!channel) continue;
this.sendSubscribe({ channel, ...(symbol ? { symbol } : {}) }); this.sendSubscribe({ channel, ...(symbol ? { symbol } : {}) });
} }
} }
@@ -1057,6 +1123,194 @@ export class StandxGateway {
this.marketWs?.send(JSON.stringify({ subscribe: stream })); this.marketWs?.send(JSON.stringify({ subscribe: stream }));
} }
// ========== 心跳监控 ==========
/**
* 启动心跳监控
* 定期检查是否收到消息,如果超时则主动触发重连
*/
private startHeartbeatMonitor(): void {
this.stopHeartbeatMonitor();
this.heartbeatTimer = setInterval(() => {
const now = Date.now();
const elapsed = now - this.lastMessageTime;
if (elapsed > WS_HEARTBEAT_TIMEOUT) {
this.logDebug(`heartbeat timeout (${elapsed}ms since last message), forcing reconnect`);
this.forceReconnect("heartbeat_timeout");
}
}, WS_HEARTBEAT_CHECK_INTERVAL);
}
/**
* 停止心跳监控
*/
private stopHeartbeatMonitor(): void {
if (this.heartbeatTimer) {
clearInterval(this.heartbeatTimer);
this.heartbeatTimer = null;
}
}
// ========== 数据过时检测与 REST 备用拉取 ==========
/**
* 启动数据过时检测
* 即使 WS 连接正常,如果超过 3 秒未收到行情/账户数据,也主动通过 REST 拉取
*/
private startDataStaleCheck(): void {
this.stopDataStaleCheck();
this.dataStaleCheckTimer = setInterval(() => {
const now = Date.now();
const marketStale = now - this.lastMarketDataTime > WS_DATA_STALE_THRESHOLD;
const accountStale = now - this.lastAccountDataTime > WS_DATA_STALE_THRESHOLD;
if (marketStale || accountStale) {
this.logDebug("data stale detected", {
marketStaleMs: now - this.lastMarketDataTime,
accountStaleMs: now - this.lastAccountDataTime,
marketStale,
accountStale,
});
// 主动通过 REST 拉取数据
this.fetchStaleData(marketStale, accountStale);
}
}, 1000); // 每秒检查一次
}
/**
* 停止数据过时检测
*/
private stopDataStaleCheck(): void {
if (this.dataStaleCheckTimer) {
clearInterval(this.dataStaleCheckTimer);
this.dataStaleCheckTimer = null;
}
}
/**
* 主动拉取过时的数据
*/
private fetchStaleData(marketStale: boolean, accountStale: boolean): void {
// 获取当前订阅的 symbols
const symbols = new Set<string>();
for (const key of this.subscriptions) {
const [, symbol] = key.split(":");
if (symbol) symbols.add(symbol);
}
if (marketStale) {
for (const symbol of symbols) {
void this.fetchTickerSnapshot(symbol).catch((e) => this.logger("staleTickerFetch", e));
void this.fetchDepthSnapshot(symbol).catch((e) => this.logger("staleDepthFetch", e));
}
// 更新时间戳避免重复拉取
this.lastMarketDataTime = Date.now();
}
if (accountStale) {
void this.refreshAccountSnapshot().catch((e) => this.logger("staleAccountFetch", e));
for (const symbol of symbols) {
void this.refreshOpenOrders(symbol).catch((e) => this.logger("staleOrdersFetch", e));
}
// 更新时间戳避免重复拉取
this.lastAccountDataTime = Date.now();
}
}
/**
* 启动 REST 轮询(WS 断连时使用)
* 持续通过 REST API 拉取行情和账户数据,确保止损等逻辑能正常工作
*/
private startRestPoll(): void {
if (this.restPollActive) return;
this.restPollActive = true;
this.logDebug("REST poll started (WS disconnected)");
const poll = async () => {
if (!this.restPollActive) return;
// 获取当前订阅的 symbols
const symbols = new Set<string>();
for (const key of this.subscriptions) {
const [, symbol] = key.split(":");
if (symbol) symbols.add(symbol);
}
// 拉取行情数据
for (const symbol of symbols) {
try {
await this.fetchTickerSnapshot(symbol);
this.lastMarketDataTime = Date.now();
} catch (e) {
this.logger("restPollTicker", e);
}
try {
await this.fetchDepthSnapshot(symbol);
} catch (e) {
this.logger("restPollDepth", e);
}
}
// 拉取账户数据
try {
await this.refreshAccountSnapshot();
this.lastAccountDataTime = Date.now();
} catch (e) {
this.logger("restPollAccount", e);
}
// 继续下一次轮询
if (this.restPollActive) {
this.restPollTimer = setTimeout(() => void poll(), REST_POLL_INTERVAL);
}
};
void poll();
}
/**
* 停止 REST 轮询
*/
private stopRestPoll(): void {
if (!this.restPollActive) return;
this.restPollActive = false;
if (this.restPollTimer) {
clearTimeout(this.restPollTimer);
this.restPollTimer = null;
}
this.logDebug("REST poll stopped (WS restored)");
}
/**
* 强制重连(用于心跳超时)
* 与普通断连不同,这里不增加重连计数(因为是主动行为)
*/
private forceReconnect(reason: string): void {
this.logDebug(`force reconnect: ${reason}`);
// 停止监控
this.stopHeartbeatMonitor();
this.stopDataStaleCheck();
// 关闭现有连接
if (this.marketWs) {
try {
this.marketWs.close();
} catch {
// ignore close errors
}
this.marketWs = null;
}
// 重置状态
const wasReady = this.marketWsReady;
this.marketWsReady = false;
this.marketWsAuthed = false;
this.marketWsAuthRequested = false;
// 触发断连事件
if (wasReady) {
this.onDisconnect();
}
// 立即重连(不使用指数退避,因为是主动行为)
this.reconnectAttempts = 0;
this.scheduleReconnect();
}
private logDebug(context: string, detail?: unknown): void { private logDebug(context: string, detail?: unknown): void {
if (!this.debugWs) return; if (!this.debugWs) return;
if (detail === undefined) { if (detail === undefined) {
@@ -1509,6 +1763,9 @@ export class StandxGateway {
if (this.lastKnownOpenOrders.length > 0 && this.disconnectedSymbol) { if (this.lastKnownOpenOrders.length > 0 && this.disconnectedSymbol) {
this.startDisconnectCancelRetry(this.disconnectedSymbol); this.startDisconnectCancelRetry(this.disconnectedSymbol);
} }
// 启动 REST 轮询,确保断连期间仍能获取行情和账户数据(用于止损等逻辑)
this.startRestPoll();
} }
/** /**
+216 -16
View File
@@ -8,6 +8,23 @@ const WebSocketCtor: typeof globalThis.WebSocket =
const DEFAULT_BASE_URL = "wss://fstream.binance.com/ws"; const DEFAULT_BASE_URL = "wss://fstream.binance.com/ws";
// ========== Binance WebSocket 连接管理常量 ==========
// Binance 服务器每 3 分钟发送 ping,10 分钟无 pong 会断连
// 我们设置 5 分钟作为心跳超时阈值(保守值)
const HEARTBEAT_TIMEOUT_MS = 5 * 60 * 1000;
// 心跳检查间隔(每 30 秒检查一次)
const HEARTBEAT_CHECK_INTERVAL_MS = 30_000;
// Binance 连接最长有效期 24 小时,我们设置 23 小时主动重连
const MAX_CONNECTION_DURATION_MS = 23 * 60 * 60 * 1000;
// 数据过时阈值(毫秒)- 超过此时间未收到数据,标记为不可用
const DATA_STALE_THRESHOLD_MS = 5_000;
// 基础重连延迟
const RECONNECT_DELAY_BASE_MS = 3000;
// 最大重连延迟
const RECONNECT_DELAY_MAX_MS = 60_000;
export type BinanceConnectionState = "connected" | "disconnected" | "stale";
export interface BinanceDepthSnapshot { export interface BinanceDepthSnapshot {
symbol: string; symbol: string;
buySum: number; buySum: number;
@@ -18,13 +35,29 @@ export interface BinanceDepthSnapshot {
updatedAt: number; updatedAt: number;
} }
export type BinanceConnectionListener = (state: BinanceConnectionState) => void;
export class BinanceDepthTracker { export class BinanceDepthTracker {
private ws: WebSocket | null = null; private ws: WebSocket | null = null;
private reconnectTimer: ReturnType<typeof setTimeout> | null = null; private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
private reconnectDelayMs = 3000; private reconnectDelayMs = RECONNECT_DELAY_BASE_MS;
private stopped = false; private stopped = false;
private snapshot: BinanceDepthSnapshot | null = null; private snapshot: BinanceDepthSnapshot | null = null;
private listeners = new Set<(snapshot: BinanceDepthSnapshot) => void>(); private listeners = new Set<(snapshot: BinanceDepthSnapshot) => void>();
private connectionListeners = new Set<BinanceConnectionListener>();
// ========== 心跳与连接管理 ==========
// 上次收到消息的时间戳
private lastMessageTime = 0;
// 心跳检查定时器
private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
// 连接建立时间(用于日志记录)
// eslint-disable-next-line @typescript-eslint/no-unused-vars
private connectionStartTime = 0;
// 24 小时重连定时器
private maxDurationTimer: ReturnType<typeof setTimeout> | null = null;
// 当前连接状态
private connectionState: BinanceConnectionState = "disconnected";
constructor( constructor(
private readonly symbol: string, private readonly symbol: string,
@@ -43,18 +76,7 @@ export class BinanceDepthTracker {
stop(): void { stop(): void {
this.stopped = true; this.stopped = true;
if (this.reconnectTimer) { this.cleanup();
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
}
if (this.ws) {
try {
this.ws.close();
} catch {
// Ignore close errors
}
this.ws = null;
}
} }
onUpdate(handler: (snapshot: BinanceDepthSnapshot) => void): void { onUpdate(handler: (snapshot: BinanceDepthSnapshot) => void): void {
@@ -65,37 +87,122 @@ export class BinanceDepthTracker {
this.listeners.delete(handler); this.listeners.delete(handler);
} }
/**
* 监听连接状态变化
*/
onConnectionChange(handler: BinanceConnectionListener): void {
this.connectionListeners.add(handler);
}
offConnectionChange(handler: BinanceConnectionListener): void {
this.connectionListeners.delete(handler);
}
getSnapshot(): BinanceDepthSnapshot | null { getSnapshot(): BinanceDepthSnapshot | null {
return this.snapshot ? { ...this.snapshot } : null; return this.snapshot ? { ...this.snapshot } : null;
} }
/**
* 获取当前连接状态
*/
getConnectionState(): BinanceConnectionState {
return this.connectionState;
}
/**
* 检查数据是否过时
*/
isDataStale(): boolean {
if (!this.snapshot) return true;
return Date.now() - this.snapshot.updatedAt > DATA_STALE_THRESHOLD_MS;
}
private cleanup(): void {
// 停止心跳监控
if (this.heartbeatTimer) {
clearInterval(this.heartbeatTimer);
this.heartbeatTimer = null;
}
// 停止 24 小时重连定时器
if (this.maxDurationTimer) {
clearTimeout(this.maxDurationTimer);
this.maxDurationTimer = null;
}
// 停止重连定时器
if (this.reconnectTimer) {
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
}
// 关闭 WebSocket
if (this.ws) {
try {
this.ws.close();
} catch {
// Ignore close errors
}
this.ws = null;
}
}
private connect(): void { private connect(): void {
if (this.ws || this.stopped) return; if (this.ws || this.stopped) return;
const url = this.buildUrl(); const url = this.buildUrl();
this.ws = new WebSocketCtor(url); this.ws = new WebSocketCtor(url);
const handleOpen = () => { const handleOpen = () => {
this.reconnectDelayMs = 3000; this.reconnectDelayMs = RECONNECT_DELAY_BASE_MS;
this.connectionStartTime = Date.now();
this.lastMessageTime = Date.now();
this.updateConnectionState("connected");
// 启动心跳监控
this.startHeartbeatMonitor();
// 启动 24 小时自动重连定时器
this.startMaxDurationTimer();
this.options?.logger?.("binanceDepth", "WebSocket connected");
}; };
const handleClose = () => { const handleClose = () => {
this.ws = null; this.ws = null;
this.stopHeartbeatMonitor();
this.stopMaxDurationTimer();
this.updateConnectionState("disconnected");
if (!this.stopped) { if (!this.stopped) {
this.options?.logger?.("binanceDepth", "WebSocket closed, scheduling reconnect");
this.scheduleReconnect(); this.scheduleReconnect();
} }
}; };
const handleError = (error: unknown) => { const handleError = (error: unknown) => {
this.options?.logger?.("binanceDepth", error); this.options?.logger?.("binanceDepth", error);
// 如果连接从未成功建立,需要清理并重连
if (this.ws && this.connectionState === "disconnected") {
this.ws = null;
this.scheduleReconnect();
}
}; };
const handleMessage = (event: { data: unknown }) => { const handleMessage = (event: { data: unknown }) => {
this.lastMessageTime = Date.now();
// 如果之前是 stale 状态,恢复为 connected
if (this.connectionState === "stale") {
this.updateConnectionState("connected");
}
this.handlePayload(event.data); this.handlePayload(event.data);
}; };
// 处理 Binance 服务器的 ping 帧
// 根据文档:必须尽快回复 pongpayload 为 ping 的 payload 副本
const handlePing = (data: unknown) => { const handlePing = (data: unknown) => {
this.lastMessageTime = Date.now();
if (this.ws && "pong" in this.ws && typeof this.ws.pong === "function") { if (this.ws && "pong" in this.ws && typeof this.ws.pong === "function") {
try {
this.ws.pong(data as any); this.ws.pong(data as any);
} catch (error) {
this.options?.logger?.("binanceDepth pong", error);
}
} }
}; };
@@ -130,11 +237,101 @@ export class BinanceDepthTracker {
if (this.reconnectTimer || this.stopped) return; if (this.reconnectTimer || this.stopped) return;
this.reconnectTimer = setTimeout(() => { this.reconnectTimer = setTimeout(() => {
this.reconnectTimer = null; this.reconnectTimer = null;
this.reconnectDelayMs = Math.min(this.reconnectDelayMs * 2, 60_000); this.reconnectDelayMs = Math.min(this.reconnectDelayMs * 2, RECONNECT_DELAY_MAX_MS);
this.connect(); this.connect();
}, this.reconnectDelayMs); }, this.reconnectDelayMs);
} }
/**
* 启动心跳监控
* 根据 Binance 文档:10 分钟无 pong 会断连
* 我们设置 5 分钟作为心跳超时阈值
*/
private startHeartbeatMonitor(): void {
this.stopHeartbeatMonitor();
this.heartbeatTimer = setInterval(() => {
const now = Date.now();
const elapsed = now - this.lastMessageTime;
// 检查数据是否过时(5 秒无数据)
if (elapsed > DATA_STALE_THRESHOLD_MS && this.connectionState === "connected") {
this.updateConnectionState("stale");
this.options?.logger?.("binanceDepth", `Data stale: ${elapsed}ms since last message`);
}
// 检查心跳超时(5 分钟无消息)
if (elapsed > HEARTBEAT_TIMEOUT_MS) {
this.options?.logger?.("binanceDepth", `Heartbeat timeout: ${elapsed}ms, forcing reconnect`);
this.forceReconnect("heartbeat_timeout");
}
}, HEARTBEAT_CHECK_INTERVAL_MS);
}
private stopHeartbeatMonitor(): void {
if (this.heartbeatTimer) {
clearInterval(this.heartbeatTimer);
this.heartbeatTimer = null;
}
}
/**
* 启动 24 小时自动重连定时器
* 根据 Binance 文档:连接最长有效期 24 小时
* 我们设置 23 小时主动重连,避免被服务器断开
*/
private startMaxDurationTimer(): void {
this.stopMaxDurationTimer();
this.maxDurationTimer = setTimeout(() => {
this.options?.logger?.("binanceDepth", "Max connection duration reached (23h), reconnecting");
this.forceReconnect("max_duration");
}, MAX_CONNECTION_DURATION_MS);
}
private stopMaxDurationTimer(): void {
if (this.maxDurationTimer) {
clearTimeout(this.maxDurationTimer);
this.maxDurationTimer = null;
}
}
/**
* 强制重连
*/
private forceReconnect(reason: string): void {
this.options?.logger?.("binanceDepth", `Force reconnect: ${reason}`);
this.stopHeartbeatMonitor();
this.stopMaxDurationTimer();
if (this.ws) {
try {
this.ws.close();
} catch {
// ignore
}
this.ws = null;
}
this.updateConnectionState("disconnected");
// 立即重连(不使用指数退避)
this.reconnectDelayMs = RECONNECT_DELAY_BASE_MS;
this.scheduleReconnect();
}
/**
* 更新连接状态并通知监听器
*/
private updateConnectionState(state: BinanceConnectionState): void {
if (this.connectionState === state) return;
this.connectionState = state;
for (const listener of this.connectionListeners) {
try {
listener(state);
} catch (error) {
this.options?.logger?.("binanceDepth connectionListener", error);
}
}
}
private handlePayload(data: unknown): void { private handlePayload(data: unknown): void {
const payload = this.parsePayload(data); const payload = this.parsePayload(data);
if (!payload) return; if (!payload) return;
@@ -158,7 +355,11 @@ export class BinanceDepthTracker {
updatedAt: Date.now(), updatedAt: Date.now(),
}; };
for (const listener of this.listeners) { for (const listener of this.listeners) {
try {
listener({ ...this.snapshot }); listener({ ...this.snapshot });
} catch (error) {
this.options?.logger?.("binanceDepth listener", error);
}
} }
} }
@@ -174,4 +375,3 @@ export class BinanceDepthTracker {
} }
} }
} }
+248 -6
View File
@@ -1,5 +1,5 @@
import type { MakerPointsConfig } from "../config"; import type { MakerPointsConfig } from "../config";
import type { ExchangeAdapter, ConnectionEventType } from "../exchanges/adapter"; import type { ExchangeAdapter } from "../exchanges/adapter";
import type { import type {
AsterAccountSnapshot, AsterAccountSnapshot,
AsterDepth, AsterDepth,
@@ -86,6 +86,8 @@ const INSUFFICIENT_BALANCE_COOLDOWN_MS = 15_000;
const STOP_LOSS_COOLDOWN_MS = 5_000; const STOP_LOSS_COOLDOWN_MS = 5_000;
const STOP_LOSS_CHECK_INTERVAL_MS = 250; // 止损检查最大间隔 const STOP_LOSS_CHECK_INTERVAL_MS = 250; // 止损检查最大间隔
const STOP_LOSS_RETRY_INTERVAL_MS = 500; // 止损失败后重试间隔 const STOP_LOSS_RETRY_INTERVAL_MS = 500; // 止损失败后重试间隔
const DATA_STALE_THRESHOLD_MS = 5_000; // 数据过时阈值(5秒)
const DEFENSE_MODE_CHECK_INTERVAL_MS = 1000; // 防御模式检查间隔
export class MakerPointsEngine { export class MakerPointsEngine {
private accountSnapshot: AsterAccountSnapshot | null = null; private accountSnapshot: AsterAccountSnapshot | null = null;
@@ -153,12 +155,25 @@ export class MakerPointsEngine {
private lastPositionAmt = 0; private lastPositionAmt = 0;
private lastPositionSide: "LONG" | "SHORT" | "FLAT" = "FLAT"; private lastPositionSide: "LONG" | "SHORT" | "FLAT" = "FLAT";
// 连接保护相关状态 // 连接保护相关状态(用于断连/重连事件处理)
private connectionState: "connected" | "disconnected" = "connected"; private _standxConnectionState: "connected" | "disconnected" = "connected";
private reconnectResetPending = false; private reconnectResetPending = false;
private lastRepriceQueryTime = 0; private lastRepriceQueryTime = 0;
private readonly repriceQueryIntervalMs = 3000; // 最小查询间隔 private readonly repriceQueryIntervalMs = 3000; // 最小查询间隔
// ========== 数据过时防御模式 ==========
// 各数据源最后更新时间
private lastStandxDepthTime = 0;
private lastStandxAccountTime = 0;
private lastBinanceDepthTime = 0;
// 防御模式状态
private defenseMode = false;
private defenseModeNotified = false;
private defenseModeTimer: ReturnType<typeof setInterval> | null = null;
// 防御模式下的 REST 轮询定时器
private defenseRestPollTimer: ReturnType<typeof setTimeout> | null = null;
private defenseRestPollActive = false;
constructor(private readonly config: MakerPointsConfig, private readonly exchange: ExchangeAdapter) { constructor(private readonly config: MakerPointsConfig, private readonly exchange: ExchangeAdapter) {
this.tradeLog = createTradeLog(this.config.maxLogEntries); this.tradeLog = createTradeLog(this.config.maxLogEntries);
this.rateLimit = new RateLimitController(this.config.refreshIntervalMs, (type, detail) => this.rateLimit = new RateLimitController(this.config.refreshIntervalMs, (type, detail) =>
@@ -177,6 +192,20 @@ export class MakerPointsEngine {
}); });
this.binanceDepth.onUpdate(() => { this.binanceDepth.onUpdate(() => {
this.feedStatus.binance = true; this.feedStatus.binance = true;
this.lastBinanceDepthTime = Date.now();
this.emitUpdate();
});
// 监听 Binance 连接状态变化
this.binanceDepth.onConnectionChange((state) => {
if (state === "disconnected") {
this.feedStatus.binance = false;
this.tradeLog.push("warn", "Binance 深度连接断开");
} else if (state === "stale") {
this.tradeLog.push("warn", "Binance 深度数据过时");
} else if (state === "connected") {
this.feedStatus.binance = true;
this.tradeLog.push("info", "Binance 深度连接恢复");
}
this.emitUpdate(); this.emitUpdate();
}); });
this.syncPrecision(); this.syncPrecision();
@@ -185,6 +214,12 @@ export class MakerPointsEngine {
start(): void { start(): void {
if (this.timer) return; if (this.timer) return;
// 初始化数据时间戳
const now = Date.now();
this.lastStandxDepthTime = now;
this.lastStandxAccountTime = now;
this.lastBinanceDepthTime = now;
this.timer = setInterval(() => { this.timer = setInterval(() => {
void this.tick(); void this.tick();
}, this.config.refreshIntervalMs); }, this.config.refreshIntervalMs);
@@ -193,6 +228,12 @@ export class MakerPointsEngine {
void this.checkStopLoss(); void this.checkStopLoss();
}, Math.min(STOP_LOSS_CHECK_INTERVAL_MS, this.config.refreshIntervalMs)); }, Math.min(STOP_LOSS_CHECK_INTERVAL_MS, this.config.refreshIntervalMs));
} }
// 启动防御模式检测定时器
if (!this.defenseModeTimer) {
this.defenseModeTimer = setInterval(() => {
this.checkDataStaleAndDefense();
}, DEFENSE_MODE_CHECK_INTERVAL_MS);
}
this.binanceDepth.start(); this.binanceDepth.start();
} }
@@ -205,6 +246,11 @@ export class MakerPointsEngine {
clearInterval(this.stopLossTimer); clearInterval(this.stopLossTimer);
this.stopLossTimer = null; this.stopLossTimer = null;
} }
if (this.defenseModeTimer) {
clearInterval(this.defenseModeTimer);
this.defenseModeTimer = null;
}
this.stopDefenseRestPoll();
this.binanceDepth.stop(); this.binanceDepth.stop();
} }
@@ -227,6 +273,7 @@ export class MakerPointsEngine {
this.exchange.watchAccount.bind(this.exchange), this.exchange.watchAccount.bind(this.exchange),
(snapshot) => { (snapshot) => {
this.accountSnapshot = snapshot; this.accountSnapshot = snapshot;
this.lastStandxAccountTime = Date.now();
const totalUnrealized = Number(snapshot.totalUnrealizedProfit ?? "0"); const totalUnrealized = Number(snapshot.totalUnrealizedProfit ?? "0");
if (Number.isFinite(totalUnrealized)) { if (Number.isFinite(totalUnrealized)) {
this.accountUnrealized = totalUnrealized; this.accountUnrealized = totalUnrealized;
@@ -277,6 +324,7 @@ export class MakerPointsEngine {
this.exchange.watchDepth.bind(this.exchange, this.config.symbol), this.exchange.watchDepth.bind(this.exchange, this.config.symbol),
(depth) => { (depth) => {
this.depthSnapshot = depth; this.depthSnapshot = depth;
this.lastStandxDepthTime = Date.now();
this.feedStatus.depth = true; this.feedStatus.depth = true;
this.emitUpdate(); this.emitUpdate();
}, },
@@ -325,7 +373,7 @@ export class MakerPointsEngine {
* *
*/ */
private handleDisconnect(symbol: string): void { private handleDisconnect(symbol: string): void {
this.connectionState = "disconnected"; this._standxConnectionState = "disconnected";
this.tradeLog.push("warn", `WebSocket 断连 (${symbol}),启动断连保护`); this.tradeLog.push("warn", `WebSocket 断连 (${symbol}),启动断连保护`);
this.notify({ this.notify({
type: "token_expired", type: "token_expired",
@@ -342,7 +390,7 @@ export class MakerPointsEngine {
* *
*/ */
private async handleReconnect(symbol: string): Promise<void> { private async handleReconnect(symbol: string): Promise<void> {
this.connectionState = "connected"; this._standxConnectionState = "connected";
this.reconnectResetPending = true; this.reconnectResetPending = true;
this.tradeLog.push("info", `WebSocket 重连成功 (${symbol}),开始重连保护流程`); this.tradeLog.push("info", `WebSocket 重连成功 (${symbol}),开始重连保护流程`);
@@ -420,6 +468,12 @@ export class MakerPointsEngine {
private async tick(): Promise<void> { private async tick(): Promise<void> {
if (this.processing) return; if (this.processing) return;
// 重连处理期间不执行主循环,避免状态竞争
if (this.reconnectResetPending) return;
// 止损执行期间不执行主循环,避免订单冲突
if (this.stopLossProcessing) return;
// 防御模式下不执行正常挂单逻辑
if (this.defenseMode) return;
this.processing = true; this.processing = true;
let hadRateLimit = false; let hadRateLimit = false;
try { try {
@@ -747,6 +801,11 @@ export class MakerPointsEngine {
} }
private async syncOrders(targets: DesiredOrder[], _closeOnly: boolean): Promise<void> { private async syncOrders(targets: DesiredOrder[], _closeOnly: boolean): Promise<void> {
// 止损执行期间不进行挂单操作,避免订单冲突
if (this.stopLossProcessing) return;
// 重连处理期间不进行挂单操作
if (this.reconnectResetPending) return;
// 价格变化保护:如果需要 reprice 且距上次查询已过足够时间,先查询真实挂单 // 价格变化保护:如果需要 reprice 且距上次查询已过足够时间,先查询真实挂单
const shouldVerifyOrders = await this.verifyOrdersIfNeeded(); const shouldVerifyOrders = await this.verifyOrdersIfNeeded();
if (shouldVerifyOrders) { if (shouldVerifyOrders) {
@@ -1356,7 +1415,7 @@ export class MakerPointsEngine {
this.lastPositionSide = currentSide; this.lastPositionSide = currentSide;
} }
private async handleTokenExpiry(position: PositionSnapshot, absPosition: number): Promise<boolean> { private async handleTokenExpiry(position: PositionSnapshot, _absPosition: number): Promise<boolean> {
if (!isTokenExpiryConfigured()) { if (!isTokenExpiryConfigured()) {
return false; return false;
} }
@@ -1439,6 +1498,189 @@ export class MakerPointsEngine {
return true; return true;
} }
// ========== 数据过时防御模式方法 ==========
/**
* 退
*
*/
private checkDataStaleAndDefense(): void {
const now = Date.now();
const standxDepthStale = this.lastStandxDepthTime > 0 && (now - this.lastStandxDepthTime) > DATA_STALE_THRESHOLD_MS;
// 账户数据只在有变化时推送,不作为过时判断依据
// const standxAccountStale = this.lastStandxAccountTime > 0 && (now - this.lastStandxAccountTime) > DATA_STALE_THRESHOLD_MS;
const binanceStale = this.lastBinanceDepthTime > 0 && (now - this.lastBinanceDepthTime) > DATA_STALE_THRESHOLD_MS;
const shouldDefend = standxDepthStale || binanceStale;
if (shouldDefend && !this.defenseMode) {
// 进入防御模式
this.enterDefenseMode({
standxDepthStale,
binanceStale,
standxDepthAge: now - this.lastStandxDepthTime,
binanceAge: now - this.lastBinanceDepthTime,
});
} else if (!shouldDefend && this.defenseMode) {
// 退出防御模式
this.exitDefenseMode();
}
}
/**
*
* REST
*/
private enterDefenseMode(staleInfo: {
standxDepthStale: boolean;
binanceStale: boolean;
standxDepthAge: number;
binanceAge: number;
}): void {
this.defenseMode = true;
// 构建过时信息描述
const staleItems: string[] = [];
if (staleInfo.standxDepthStale) {
staleItems.push(`StandX深度(${Math.round(staleInfo.standxDepthAge / 1000)}s)`);
}
if (staleInfo.binanceStale) {
staleItems.push(`Binance深度(${Math.round(staleInfo.binanceAge / 1000)}s)`);
}
this.tradeLog.push("warn", `数据过时检测: ${staleItems.join(", ")},进入防御模式`);
// 发送通知
if (!this.defenseModeNotified) {
this.notify({
type: "token_expired",
level: "warn",
symbol: this.config.symbol,
title: "防御模式",
message: `数据推送中断: ${staleItems.join(", ")},已取消所有挂单`,
details: staleInfo,
});
this.defenseModeNotified = true;
}
// 立即取消所有挂单
void this.defenseCancelAllOrders();
// 启动 REST 轮询保护仓位
this.startDefenseRestPoll();
}
/**
* 退
*/
private exitDefenseMode(): void {
this.defenseMode = false;
this.defenseModeNotified = false;
this.tradeLog.push("info", "数据推送恢复正常,退出防御模式");
this.notify({
type: "position_opened",
level: "info",
symbol: this.config.symbol,
title: "防御模式解除",
message: "数据推送恢复正常,恢复正常交易",
details: {},
});
// 停止 REST 轮询
this.stopDefenseRestPoll();
// 重置本地状态,强制下一轮重新计算挂单
this.desiredOrders = [];
this.lastDesiredSummary = null;
this.lastQuoteBid1 = null;
this.lastQuoteAsk1 = null;
}
/**
*
*/
private async defenseCancelAllOrders(): Promise<void> {
try {
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");
} catch (error) {
if (isUnknownOrderError(error)) {
this.tradeLog.push("order", "防御模式: 挂单已不存在");
this.openOrders = [];
this.pendingCancelOrders.clear();
} else {
this.tradeLog.push("error", `防御模式取消挂单失败: ${extractMessage(error)}`);
}
}
}
/**
* REST
* 使 REST API
*/
private startDefenseRestPoll(): void {
if (this.defenseRestPollActive) return;
this.defenseRestPollActive = true;
this.tradeLog.push("info", "防御模式: 启动 REST 数据轮询");
const poll = async () => {
if (!this.defenseRestPollActive || !this.defenseMode) return;
try {
// 如果有查询挂单的方法,定期检查并取消
if (this.exchange.queryOpenOrders) {
const realOrders = await this.exchange.queryOpenOrders();
if (realOrders.length > 0) {
this.tradeLog.push("warn", `防御模式: 发现 ${realOrders.length} 个挂单,执行取消`);
await this.defenseCancelAllOrders();
}
}
// 检查止损条件(使用当前账户快照中的数据)
// checkStopLoss 会继续运行,使用最后收到的数据进行止损判断
} catch (error) {
this.tradeLog.push("error", `防御模式 REST 轮询失败: ${extractMessage(error)}`);
}
// 继续下一次轮询
if (this.defenseRestPollActive && this.defenseMode) {
this.defenseRestPollTimer = setTimeout(() => void poll(), 2000);
}
};
void poll();
}
/**
* REST
*/
private stopDefenseRestPoll(): void {
if (!this.defenseRestPollActive) return;
this.defenseRestPollActive = false;
if (this.defenseRestPollTimer) {
clearTimeout(this.defenseRestPollTimer);
this.defenseRestPollTimer = null;
}
this.tradeLog.push("info", "防御模式: 停止 REST 数据轮询");
}
} }
function resolveBinanceSymbol(symbol: string): string { function resolveBinanceSymbol(symbol: string): string {