diff --git a/src/core/lib/rate-limit.ts b/src/core/lib/rate-limit.ts new file mode 100644 index 0000000..815c613 --- /dev/null +++ b/src/core/lib/rate-limit.ts @@ -0,0 +1,122 @@ +import type { LogHandler } from "../order-coordinator"; + +type RateLimitState = "normal" | "degraded" | "paused"; + +export type RateLimitDecision = "run" | "skip" | "paused"; + +const DEFAULT_PAUSE_MS = 30_000; +const DEFAULT_RECOVERY_MS = 60_000; + +export class RateLimitController { + private state: RateLimitState = "normal"; + private pausedUntil: number | null = null; + private lastCycleAt = 0; + private lastRateLimitAt = 0; + private readonly pauseMs: number; + private readonly recoveryMs: number; + private entriesSuppressed = false; + + constructor( + private readonly baseInterval: number, + private readonly log: LogHandler, + options?: { pauseMs?: number; recoveryMs?: number } + ) { + this.pauseMs = options?.pauseMs ?? DEFAULT_PAUSE_MS; + this.recoveryMs = options?.recoveryMs ?? DEFAULT_RECOVERY_MS; + } + + shouldBlockEntries(): boolean { + return this.entriesSuppressed || this.state === "paused"; + } + + private suppressEntries(source?: string): void { + if (this.entriesSuppressed) return; + this.entriesSuppressed = true; + this.log("info", `${source ? `${source} ` : ""}限频期间暂停新开仓`); + } + + private allowEntries(): void { + if (!this.entriesSuppressed) return; + this.entriesSuppressed = false; + this.log("info", "限频恢复,允许重新开仓"); + } + + beforeCycle(): RateLimitDecision { + const now = Date.now(); + if (this.state === "paused") { + if (this.pausedUntil != null && now >= this.pausedUntil) { + this.state = "degraded"; + this.pausedUntil = null; + this.log("info", "限频暂停结束,继续以降频模式运行"); + } else { + this.lastCycleAt = now; + return "paused"; + } + } + + const interval = this.currentInterval(); + if (now - this.lastCycleAt < interval) { + return "skip"; + } + this.lastCycleAt = now; + return "run"; + } + + registerRateLimit(source?: string): void { + const now = Date.now(); + this.lastRateLimitAt = now; + if (this.state === "normal") { + this.state = "degraded"; + this.log( + "warn", + `${source ? `${source} ` : ""}触发 429,降频至 ${(this.currentInterval() / 1000).toFixed(2)}s` + ); + this.lastCycleAt = now; + this.suppressEntries(source); + return; + } + if (this.state === "degraded") { + this.state = "paused"; + this.pausedUntil = now + this.pauseMs; + this.log( + "warn", + `${source ? `${source} ` : ""}连续 429,暂停请求 ${(this.pauseMs / 1000).toFixed(0)}s` + ); + this.suppressEntries(source); + return; + } + this.pausedUntil = now + this.pauseMs; + this.log( + "warn", + `${source ? `${source} ` : ""}限频仍在持续,延长暂停 ${(this.pauseMs / 1000).toFixed(0)}s` + ); + this.suppressEntries(source); + } + + onCycleComplete(hadRateLimit: boolean): void { + if (hadRateLimit) return; + if (this.state === "degraded" && this.lastRateLimitAt > 0) { + const now = Date.now(); + if (now - this.lastRateLimitAt >= this.recoveryMs) { + this.state = "normal"; + this.log("info", "限频恢复,重置为正常请求频率"); + this.allowEntries(); + this.lastRateLimitAt = 0; + } + } + if (this.state === "normal" && this.entriesSuppressed && this.lastRateLimitAt === 0) { + this.allowEntries(); + } + } + + private currentInterval(): number { + if (this.state === "degraded") { + return this.baseInterval * 2; + } + if (this.state === "paused") { + return this.baseInterval * 2; + } + return this.baseInterval; + } +} + diff --git a/src/core/maker-engine.ts b/src/core/maker-engine.ts index 9f9e869..215a5a0 100644 --- a/src/core/maker-engine.ts +++ b/src/core/maker-engine.ts @@ -8,7 +8,7 @@ import type { } from "../exchanges/types"; import { roundDownToTick } from "../utils/math"; import { createTradeLog, type TradeLogEntry } from "../state/trade-log"; -import { isUnknownOrderError } from "../utils/errors"; +import { isUnknownOrderError, isRateLimitError } from "../utils/errors"; import { getPosition, type PositionSnapshot } from "../utils/strategy"; import { computePositionPnl } from "../utils/pnl"; import { getTopPrices, getMidOrLast } from "../utils/price"; @@ -21,6 +21,7 @@ import { import type { OrderLockMap, OrderPendingMap, OrderTimerMap } from "./order-coordinator"; import { makeOrderPlan } from "./lib/order-plan"; import { safeCancelOrder } from "./lib/orders"; +import { RateLimitController } from "./lib/rate-limit"; interface DesiredOrder { side: "BUY" | "SELL"; @@ -74,9 +75,13 @@ export class MakerEngine { private initialOrderSnapshotReady = false; private initialOrderResetDone = false; private entryPricePendingLogged = false; + private readonly rateLimit: RateLimitController; constructor(private readonly config: MakerConfig, private readonly exchange: ExchangeAdapter) { this.tradeLog = createTradeLog(this.config.maxLogEntries); + this.rateLimit = new RateLimitController(this.config.refreshIntervalMs, (type, detail) => + this.tradeLog.push(type, detail) + ); this.bootstrap(); } @@ -215,7 +220,16 @@ export class MakerEngine { private async tick(): Promise { if (this.processing) return; this.processing = true; + let hadRateLimit = false; try { + const decision = this.rateLimit.beforeCycle(); + if (decision === "paused") { + this.emitUpdate(); + return; + } + if (decision === "skip") { + return; + } if (!this.isReady()) { this.emitUpdate(); return; @@ -237,11 +251,14 @@ export class MakerEngine { const position = getPosition(this.accountSnapshot, this.config.symbol); const absPosition = Math.abs(position.positionAmt); const desired: DesiredOrder[] = []; + const canEnter = !this.rateLimit.shouldBlockEntries(); if (absPosition < EPS) { this.entryPricePendingLogged = false; - desired.push({ side: "BUY", price: bidPrice, amount: this.config.tradeAmount, reduceOnly: false }); - desired.push({ side: "SELL", price: askPrice, amount: this.config.tradeAmount, reduceOnly: false }); + if (canEnter) { + desired.push({ side: "BUY", price: bidPrice, amount: this.config.tradeAmount, reduceOnly: false }); + desired.push({ side: "SELL", price: askPrice, amount: this.config.tradeAmount, reduceOnly: false }); + } } else { const closeSide: "BUY" | "SELL" = position.positionAmt > 0 ? "SELL" : "BUY"; const closePrice = closeSide === "SELL" ? askPrice : bidPrice; @@ -254,13 +271,32 @@ export class MakerEngine { await this.checkRisk(position, bidPrice, askPrice); this.emitUpdate(); } catch (error) { - this.tradeLog.push("error", `做市循环异常: ${String(error)}`); + if (isRateLimitError(error)) { + hadRateLimit = true; + this.rateLimit.registerRateLimit("maker"); + await this.enforceRateLimitStop(); + this.tradeLog.push("warn", `MakerEngine 429: ${String(error)}`); + } else { + this.tradeLog.push("error", `做市循环异常: ${String(error)}`); + } this.emitUpdate(); } finally { + this.rateLimit.onCycleComplete(hadRateLimit); this.processing = false; } } + private async enforceRateLimitStop(): Promise { + const position = getPosition(this.accountSnapshot, this.config.symbol); + if (Math.abs(position.positionAmt) < EPS) return; + const { topBid, topAsk } = getTopPrices(this.depthSnapshot); + if (topBid == null || topAsk == null) return; + const bidPrice = roundDownToTick(topBid - this.config.bidOffset, this.config.priceTick); + const askPrice = roundDownToTick(topAsk + this.config.askOffset, this.config.priceTick); + await this.checkRisk(position, bidPrice, askPrice); + await this.flushOrders(); + } + private async ensureStartupOrderReset(): Promise { if (this.initialOrderResetDone) return true; if (!this.initialOrderSnapshotReady) return false; diff --git a/src/core/offset-maker-engine.ts b/src/core/offset-maker-engine.ts index 02e0850..d95e1db 100644 --- a/src/core/offset-maker-engine.ts +++ b/src/core/offset-maker-engine.ts @@ -8,7 +8,7 @@ import type { } from "../exchanges/types"; import { roundDownToTick } from "../utils/math"; import { createTradeLog } from "../state/trade-log"; -import { isUnknownOrderError } from "../utils/errors"; +import { isUnknownOrderError, isRateLimitError } from "../utils/errors"; import { getPosition, type PositionSnapshot } from "../utils/strategy"; import { computeDepthStats } from "../utils/depth"; import { computePositionPnl } from "../utils/pnl"; @@ -23,6 +23,7 @@ import type { OrderLockMap, OrderPendingMap, OrderTimerMap } from "./order-coord import type { MakerEngineSnapshot } from "./maker-engine"; import { makeOrderPlan } from "./lib/order-plan"; import { safeCancelOrder } from "./lib/orders"; +import { RateLimitController } from "./lib/rate-limit"; interface DesiredOrder { side: "BUY" | "SELL"; @@ -68,6 +69,7 @@ export class OffsetMakerEngine { private initialOrderSnapshotReady = false; private initialOrderResetDone = false; private entryPricePendingLogged = false; + private readonly rateLimit: RateLimitController; private lastBuyDepthSum10 = 0; private lastSellDepthSum10 = 0; @@ -77,6 +79,9 @@ export class OffsetMakerEngine { constructor(private readonly config: MakerConfig, private readonly exchange: ExchangeAdapter) { this.tradeLog = createTradeLog(this.config.maxLogEntries); + this.rateLimit = new RateLimitController(this.config.refreshIntervalMs, (type, detail) => + this.tradeLog.push(type, detail) + ); this.bootstrap(); } @@ -214,7 +219,16 @@ export class OffsetMakerEngine { private async tick(): Promise { if (this.processing) return; this.processing = true; + let hadRateLimit = false; try { + const decision = this.rateLimit.beforeCycle(); + if (decision === "paused") { + this.emitUpdate(); + return; + } + if (decision === "skip") { + return; + } if (!this.isReady()) { this.emitUpdate(); return; @@ -249,13 +263,14 @@ export class OffsetMakerEngine { const askPrice = roundDownToTick(topAsk! + this.config.askOffset, this.config.priceTick); const absPosition = Math.abs(position.positionAmt); const desired: DesiredOrder[] = []; + const canEnter = !this.rateLimit.shouldBlockEntries(); if (absPosition < EPS) { this.entryPricePendingLogged = false; - if (!skipBuySide) { + if (!skipBuySide && canEnter) { desired.push({ side: "BUY", price: bidPrice, amount: this.config.tradeAmount, reduceOnly: false }); } - if (!skipSellSide) { + if (!skipSellSide && canEnter) { desired.push({ side: "SELL", price: askPrice, amount: this.config.tradeAmount, reduceOnly: false }); } } else { @@ -270,13 +285,53 @@ export class OffsetMakerEngine { await this.checkRisk(position, bidPrice, askPrice); this.emitUpdate(); } catch (error) { - this.tradeLog.push("error", `偏移做市循环异常: ${String(error)}`); + if (isRateLimitError(error)) { + hadRateLimit = true; + this.rateLimit.registerRateLimit("offset-maker"); + await this.enforceRateLimitStop(); + this.tradeLog.push("warn", `OffsetMakerEngine 429: ${String(error)}`); + } else { + this.tradeLog.push("error", `偏移做市循环异常: ${String(error)}`); + } this.emitUpdate(); } finally { + this.rateLimit.onCycleComplete(hadRateLimit); this.processing = false; } } + private async enforceRateLimitStop(): Promise { + const position = getPosition(this.accountSnapshot, this.config.symbol); + if (Math.abs(position.positionAmt) < EPS) return; + await this.flushOrders(); + const absPosition = Math.abs(position.positionAmt); + const side: "BUY" | "SELL" = position.positionAmt > 0 ? "SELL" : "BUY"; + try { + await marketClose( + this.exchange, + this.config.symbol, + this.openOrders, + this.locks, + this.timers, + this.pending, + side, + absPosition, + (type, detail) => this.tradeLog.push(type, detail), + { + markPrice: position.markPrice, + expectedPrice: Number(side === "SELL" ? this.depthSnapshot?.bids?.[0]?.[0] : this.depthSnapshot?.asks?.[0]?.[0]) || null, + maxPct: this.config.maxCloseSlippagePct, + } + ); + } catch (error) { + if (isUnknownOrderError(error)) { + this.tradeLog.push("order", "限频强制平仓时订单已不存在"); + } else { + this.tradeLog.push("error", `限频强制平仓失败: ${String(error)}`); + } + } + } + private async ensureStartupOrderReset(): Promise { if (this.initialOrderResetDone) return true; if (!this.initialOrderSnapshotReady) return false; diff --git a/src/core/trend-engine.ts b/src/core/trend-engine.ts index 09e12ba..e97eed0 100644 --- a/src/core/trend-engine.ts +++ b/src/core/trend-engine.ts @@ -29,6 +29,8 @@ import { isUnknownOrderError } from "../utils/errors"; import { roundDownToTick } from "../utils/math"; import { createTradeLog, type TradeLogEntry } from "../state/trade-log"; import { decryptCopyright } from "../utils/copyright"; +import { isRateLimitError } from "../utils/errors"; +import { RateLimitController } from "./lib/rate-limit"; export interface TrendEngineSnapshot { ready: boolean; @@ -84,6 +86,7 @@ export class TrendEngine { private initializedPosition = false; private cancelAllRequested = false; private readonly pendingCancelOrders = new Set(); + private readonly rateLimit: RateLimitController; // 控制入场频率:同一分钟内最多入场一次 private lastEntryMinute: number | null = null; @@ -102,6 +105,9 @@ export class TrendEngine { constructor(private readonly config: TradingConfig, private readonly exchange: ExchangeAdapter) { this.tradeLog = createTradeLog(this.config.maxLogEntries); + this.rateLimit = new RateLimitController(this.config.pollIntervalMs, (type, detail) => + this.tradeLog.push(type, detail) + ); this.bootstrap(); } @@ -240,7 +246,16 @@ export class TrendEngine { private async tick(): Promise { if (this.processing) return; this.processing = true; + let hadRateLimit = false; try { + const decision = this.rateLimit.beforeCycle(); + if (decision === "paused") { + this.emitUpdate(); + return; + } + if (decision === "skip") { + return; + } if (!this.ordersSnapshotReady) { this.emitUpdate(); return; @@ -259,7 +274,9 @@ export class TrendEngine { const position = getPosition(this.accountSnapshot, this.config.symbol); if (Math.abs(position.positionAmt) < 1e-5) { - await this.handleOpenPosition(price, sma30); + if (!this.rateLimit.shouldBlockEntries()) { + await this.handleOpenPosition(price, sma30); + } } else { const result = await this.handlePositionManagement(position, price); if (result.closed) { @@ -273,13 +290,33 @@ export class TrendEngine { this.lastPrice = price; this.emitUpdate(); } catch (error) { - this.tradeLog.push("error", `策略循环异常: ${String(error)}`); + if (isRateLimitError(error)) { + hadRateLimit = true; + this.rateLimit.registerRateLimit("trend"); + await this.enforceRateLimitStop(); + this.tradeLog.push("warn", `TrendEngine 429: ${String(error)}`); + } else { + this.tradeLog.push("error", `策略循环异常: ${String(error)}`); + } this.emitUpdate(); } finally { + this.rateLimit.onCycleComplete(hadRateLimit); this.processing = false; } } + private async enforceRateLimitStop(): Promise { + const position = getPosition(this.accountSnapshot, this.config.symbol); + if (Math.abs(position.positionAmt) < 1e-5) return; + const price = this.getReferencePrice() ?? Number(this.tickerSnapshot?.lastPrice) ?? this.lastPrice; + if (!Number.isFinite(price) || price == null) return; + const result = await this.handlePositionManagement(position, Number(price)); + if (result.closed) { + this.totalTrades += 1; + this.totalProfit += result.pnl; + } + } + private logStartupState(): void { if (this.startupLogged) return; const position = getPosition(this.accountSnapshot, this.config.symbol); diff --git a/src/exchanges/types.ts b/src/exchanges/types.ts index c35bd61..bb676d7 100644 --- a/src/exchanges/types.ts +++ b/src/exchanges/types.ts @@ -154,6 +154,7 @@ export interface AsterOrder { reduceOnly: boolean; closePosition: boolean; workingType?: string; + activationPrice?: string; avgPrice?: string; cumQuote?: string; origType?: string; diff --git a/src/utils/errors.ts b/src/utils/errors.ts index dda74cc..2825321 100644 --- a/src/utils/errors.ts +++ b/src/utils/errors.ts @@ -13,3 +13,26 @@ export function extractMessage(error: unknown): string { return String(error); } } + +export function isRateLimitError(error: unknown): boolean { + if (!error) return false; + if (typeof error === "object" && "status" in error) { + const status = Number((error as { status?: unknown }).status); + if (Number.isFinite(status) && status === 429) { + return true; + } + } + if (typeof error === "object" && "code" in error) { + const code = (error as { code?: unknown }).code; + if (code === 429 || code === "429") { + return true; + } + } + const message = extractMessage(error).toLowerCase(); + return ( + message.includes("429") || + message.includes("too many requests") || + message.includes("rate limit") || + message.includes("request rate") + ); +}