diff --git a/src/strategy/common/event-emitter.ts b/src/strategy/common/event-emitter.ts new file mode 100644 index 0000000..5e31682 --- /dev/null +++ b/src/strategy/common/event-emitter.ts @@ -0,0 +1,30 @@ +export class StrategyEventEmitter { + private readonly listeners = new Map void>>(); + + on(event: TEvent, handler: (payload: TPayload) => void): void { + const handlers = this.listeners.get(event) ?? new Set<(payload: TPayload) => void>(); + handlers.add(handler); + this.listeners.set(event, handlers); + } + + off(event: TEvent, handler: (payload: TPayload) => void): void { + const handlers = this.listeners.get(event); + if (!handlers) return; + handlers.delete(handler); + if (handlers.size === 0) { + this.listeners.delete(event); + } + } + + emit(event: TEvent, payload: TPayload, onError?: (error: unknown) => void): void { + const handlers = this.listeners.get(event); + if (!handlers) return; + for (const handler of handlers) { + try { + handler(payload); + } catch (error) { + onError?.(error); + } + } + } +} diff --git a/src/strategy/common/session-volume.ts b/src/strategy/common/session-volume.ts new file mode 100644 index 0000000..e5384f0 --- /dev/null +++ b/src/strategy/common/session-volume.ts @@ -0,0 +1,28 @@ +import type { PositionSnapshot } from "../../utils/strategy"; + +export class SessionVolumeTracker { + private initialized = false; + private previousPositionAmt = 0; + private total = 0; + + update(position: PositionSnapshot, referencePrice: number | null): void { + if (!this.initialized) { + this.previousPositionAmt = position.positionAmt; + this.initialized = true; + return; + } + if (referencePrice == null) { + this.previousPositionAmt = position.positionAmt; + return; + } + const delta = Math.abs(position.positionAmt - this.previousPositionAmt); + if (delta > 0) { + this.total += delta * referencePrice; + } + this.previousPositionAmt = position.positionAmt; + } + + get value(): number { + return this.total; + } +} diff --git a/src/strategy/common/subscriptions.ts b/src/strategy/common/subscriptions.ts new file mode 100644 index 0000000..15ec55d --- /dev/null +++ b/src/strategy/common/subscriptions.ts @@ -0,0 +1,25 @@ +export type LogHandler = (type: string, detail: string) => void; + +interface SubscriptionMessages { + subscribeFail: (error: unknown) => string; + processFail: (error: unknown) => string; +} + +export function safeSubscribe( + subscribe: (cb: (payload: T) => void) => void, + handler: (payload: T) => void, + log: LogHandler, + messages: SubscriptionMessages +): void { + try { + subscribe((payload) => { + try { + handler(payload); + } catch (error) { + log("error", messages.processFail(error)); + } + }); + } catch (error) { + log("error", messages.subscribeFail(error)); + } +} diff --git a/src/strategy/maker-engine.ts b/src/strategy/maker-engine.ts index 199dcd1..4c61746 100644 --- a/src/strategy/maker-engine.ts +++ b/src/strategy/maker-engine.ts @@ -3,13 +3,15 @@ import type { ExchangeAdapter } from "../exchanges/adapter"; import type { AsterAccountSnapshot, AsterDepth, + AsterKline, AsterOrder, AsterTicker, } from "../exchanges/types"; import { roundDownToTick } from "../utils/math"; import { createTradeLog, type TradeLogEntry } from "../logging/trade-log"; import { isUnknownOrderError, isRateLimitError } from "../utils/errors"; -import { getPosition, type PositionSnapshot } from "../utils/strategy"; +import { getPosition } from "../utils/strategy"; +import type { PositionSnapshot } from "../utils/strategy"; import { computePositionPnl } from "../utils/pnl"; import { getTopPrices, getMidOrLast } from "../utils/price"; import { shouldStopLoss } from "../utils/risk"; @@ -22,6 +24,9 @@ import type { OrderLockMap, OrderPendingMap, OrderTimerMap } from "../core/order import { makeOrderPlan } from "../core/lib/order-plan"; import { safeCancelOrder } from "../core/lib/orders"; import { RateLimitController } from "../core/lib/rate-limit"; +import { StrategyEventEmitter } from "./common/event-emitter"; +import { safeSubscribe, type LogHandler } from "./common/subscriptions"; +import { SessionVolumeTracker } from "./common/session-volume"; interface DesiredOrder { side: "BUY" | "SELL"; @@ -63,15 +68,13 @@ export class MakerEngine { private readonly pendingCancelOrders = new Set(); private readonly tradeLog: ReturnType; - private readonly listeners = new Map>(); + private readonly events = new StrategyEventEmitter(); + private readonly sessionVolume = new SessionVolumeTracker(); private timer: ReturnType | null = null; private processing = false; private desiredOrders: DesiredOrder[] = []; private accountUnrealized = 0; - private sessionQuoteVolume = 0; - private prevPositionAmt = 0; - private initializedPosition = false; private initialOrderSnapshotReady = false; private initialOrderResetDone = false; private entryPricePendingLogged = false; @@ -100,18 +103,11 @@ export class MakerEngine { } on(event: MakerEvent, handler: MakerListener): void { - const handlers = this.listeners.get(event) ?? new Set(); - handlers.add(handler); - this.listeners.set(event, handlers); + this.events.on(event, handler); } off(event: MakerEvent, handler: MakerListener): void { - const handlers = this.listeners.get(event); - if (!handlers) return; - handlers.delete(handler); - if (handlers.size === 0) { - this.listeners.delete(event); - } + this.events.off(event, handler); } getSnapshot(): MakerEngineSnapshot { @@ -119,86 +115,88 @@ export class MakerEngine { } private bootstrap(): void { - try { - this.exchange.watchAccount((snapshot) => { - try { - this.accountSnapshot = snapshot; - const totalUnrealized = Number(snapshot.totalUnrealizedProfit ?? "0"); - if (Number.isFinite(totalUnrealized)) { - this.accountUnrealized = totalUnrealized; + const log: LogHandler = (type, detail) => this.tradeLog.push(type, detail); + + safeSubscribe( + this.exchange.watchAccount.bind(this.exchange), + (snapshot) => { + this.accountSnapshot = snapshot; + const totalUnrealized = Number(snapshot.totalUnrealizedProfit ?? "0"); + if (Number.isFinite(totalUnrealized)) { + this.accountUnrealized = totalUnrealized; + } + const position = getPosition(snapshot, this.config.symbol); + this.sessionVolume.update(position, this.getReferencePrice()); + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅账户失败: ${String(error)}`, + processFail: (error) => `账户推送处理异常: ${String(error)}`, + } + ); + + safeSubscribe( + this.exchange.watchOrders.bind(this.exchange), + (orders) => { + this.syncLocksWithOrders(orders); + this.openOrders = Array.isArray(orders) + ? orders.filter((order) => order.type !== "MARKET" && order.symbol === this.config.symbol) + : []; + const currentIds = new Set(this.openOrders.map((order) => String(order.orderId))); + for (const id of Array.from(this.pendingCancelOrders)) { + if (!currentIds.has(id)) { + this.pendingCancelOrders.delete(id); } - const position = getPosition(snapshot, this.config.symbol); - this.updateSessionVolume(position); - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `账户推送处理异常: ${String(err)}`); } - }); - } catch (err) { - this.tradeLog.push("error", `订阅账户失败: ${String(err)}`); - } + this.initialOrderSnapshotReady = true; + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅订单失败: ${String(error)}`, + processFail: (error) => `订单推送处理异常: ${String(error)}`, + } + ); - try { - this.exchange.watchOrders((orders) => { - try { - this.syncLocksWithOrders(orders); - this.openOrders = Array.isArray(orders) - ? orders.filter((order) => order.type !== "MARKET" && order.symbol === this.config.symbol) - : []; - const currentIds = new Set(this.openOrders.map((order) => String(order.orderId))); - for (const id of Array.from(this.pendingCancelOrders)) { - if (!currentIds.has(id)) { - this.pendingCancelOrders.delete(id); - } - } - this.initialOrderSnapshotReady = true; - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `订单推送处理异常: ${String(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅订单失败: ${String(err)}`); - } + safeSubscribe( + this.exchange.watchDepth.bind(this.exchange, this.config.symbol), + (depth) => { + this.depthSnapshot = depth; + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅深度失败: ${String(error)}`, + processFail: (error) => `深度推送处理异常: ${String(error)}`, + } + ); - try { - this.exchange.watchDepth(this.config.symbol, (depth) => { - try { - this.depthSnapshot = depth; - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `深度推送处理异常: ${String(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅深度失败: ${String(err)}`); - } - - try { - this.exchange.watchTicker(this.config.symbol, (ticker) => { - try { - this.tickerSnapshot = ticker; - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `价格推送处理异常: ${String(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅Ticker失败: ${String(err)}`); - } + safeSubscribe( + this.exchange.watchTicker.bind(this.exchange, this.config.symbol), + (ticker) => { + this.tickerSnapshot = ticker; + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅Ticker失败: ${String(error)}`, + processFail: (error) => `价格推送处理异常: ${String(error)}`, + } + ); // Maker strategy does not consume klines, but subscribe to keep parity with other modules - try { - this.exchange.watchKlines(this.config.symbol, "1m", () => { - try { - /* no-op */ - } catch (err) { - this.tradeLog.push("error", `K线推送处理异常: ${String(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅K线失败: ${String(err)}`); - } + safeSubscribe( + this.exchange.watchKlines.bind(this.exchange, this.config.symbol, "1m"), + (_klines) => { + /* no-op */ + }, + log, + { + subscribeFail: (error) => `订阅K线失败: ${String(error)}`, + processFail: (error) => `K线推送处理异常: ${String(error)}`, + } + ); } private syncLocksWithOrders(orders: AsterOrder[] | null | undefined): void { @@ -268,7 +266,7 @@ export class MakerEngine { } this.desiredOrders = desired; - this.updateSessionVolume(position); + this.sessionVolume.update(position, this.getReferencePrice()); await this.syncOrders(desired); await this.checkRisk(position, closeBidPrice, closeAskPrice); this.emitUpdate(); @@ -468,16 +466,9 @@ export class MakerEngine { private emitUpdate(): void { try { const snapshot = this.buildSnapshot(); - const handlers = this.listeners.get("update"); - if (handlers) { - handlers.forEach((handler) => { - try { - handler(snapshot); - } catch (err) { - this.tradeLog.push("error", `更新回调处理异常: ${String(err)}`); - } - }); - } + this.events.emit("update", snapshot, (error) => { + this.tradeLog.push("error", `更新回调处理异常: ${String(error)}`); + }); } catch (err) { this.tradeLog.push("error", `快照或更新分发异常: ${String(err)}`); } @@ -498,7 +489,7 @@ export class MakerEngine { position, pnl, accountUnrealized: this.accountUnrealized, - sessionVolume: this.sessionQuoteVolume, + sessionVolume: this.sessionVolume.value, openOrders: this.openOrders, desiredOrders: this.desiredOrders, tradeLog: this.tradeLog.all(), @@ -506,24 +497,6 @@ export class MakerEngine { }; } - private updateSessionVolume(position: PositionSnapshot): void { - const price = this.getReferencePrice(); - if (!this.initializedPosition) { - this.prevPositionAmt = position.positionAmt; - this.initializedPosition = true; - return; - } - if (price == null) { - this.prevPositionAmt = position.positionAmt; - return; - } - const delta = Math.abs(position.positionAmt - this.prevPositionAmt); - if (delta > 0) { - this.sessionQuoteVolume += delta * price; - } - this.prevPositionAmt = position.positionAmt; - } - private getReferencePrice(): number | null { return getMidOrLast(this.depthSnapshot, this.tickerSnapshot); } diff --git a/src/strategy/offset-maker-engine.ts b/src/strategy/offset-maker-engine.ts index adedf5e..46e8506 100644 --- a/src/strategy/offset-maker-engine.ts +++ b/src/strategy/offset-maker-engine.ts @@ -3,13 +3,15 @@ import type { ExchangeAdapter } from "../exchanges/adapter"; import type { AsterAccountSnapshot, AsterDepth, + AsterKline, AsterOrder, AsterTicker, } from "../exchanges/types"; import { roundDownToTick } from "../utils/math"; import { createTradeLog } from "../logging/trade-log"; import { isUnknownOrderError, isRateLimitError } from "../utils/errors"; -import { getPosition, type PositionSnapshot } from "../utils/strategy"; +import { getPosition } from "../utils/strategy"; +import type { PositionSnapshot } from "../utils/strategy"; import { computeDepthStats } from "../utils/depth"; import { computePositionPnl } from "../utils/pnl"; import { getTopPrices, getMidOrLast } from "../utils/price"; @@ -24,6 +26,9 @@ import type { MakerEngineSnapshot } from "./maker-engine"; import { makeOrderPlan } from "../core/lib/order-plan"; import { safeCancelOrder } from "../core/lib/orders"; import { RateLimitController } from "../core/lib/rate-limit"; +import { StrategyEventEmitter } from "./common/event-emitter"; +import { safeSubscribe, type LogHandler } from "./common/subscriptions"; +import { SessionVolumeTracker } from "./common/session-volume"; interface DesiredOrder { side: "BUY" | "SELL"; @@ -57,15 +62,13 @@ export class OffsetMakerEngine { private readonly pendingCancelOrders = new Set(); private readonly tradeLog: ReturnType; - private readonly listeners = new Map>(); + private readonly events = new StrategyEventEmitter(); + private readonly sessionVolume = new SessionVolumeTracker(); private timer: ReturnType | null = null; private processing = false; private desiredOrders: DesiredOrder[] = []; private accountUnrealized = 0; - private sessionQuoteVolume = 0; - private prevPositionAmt = 0; - private initializedPosition = false; private initialOrderSnapshotReady = false; private initialOrderResetDone = false; private entryPricePendingLogged = false; @@ -100,18 +103,11 @@ export class OffsetMakerEngine { } on(event: MakerEvent, handler: MakerListener): void { - const handlers = this.listeners.get(event) ?? new Set(); - handlers.add(handler); - this.listeners.set(event, handlers); + this.events.on(event, handler); } off(event: MakerEvent, handler: MakerListener): void { - const handlers = this.listeners.get(event); - if (!handlers) return; - handlers.delete(handler); - if (handlers.size === 0) { - this.listeners.delete(event); - } + this.events.off(event, handler); } getSnapshot(): OffsetMakerEngineSnapshot { @@ -119,85 +115,87 @@ export class OffsetMakerEngine { } private bootstrap(): void { - try { - this.exchange.watchAccount((snapshot) => { - try { - this.accountSnapshot = snapshot; - const totalUnrealized = Number(snapshot.totalUnrealizedProfit ?? "0"); - if (Number.isFinite(totalUnrealized)) { - this.accountUnrealized = totalUnrealized; + const log: LogHandler = (type, detail) => this.tradeLog.push(type, detail); + + safeSubscribe( + this.exchange.watchAccount.bind(this.exchange), + (snapshot) => { + this.accountSnapshot = snapshot; + const totalUnrealized = Number(snapshot.totalUnrealizedProfit ?? "0"); + if (Number.isFinite(totalUnrealized)) { + this.accountUnrealized = totalUnrealized; + } + const position = getPosition(snapshot, this.config.symbol); + this.sessionVolume.update(position, this.getReferencePrice()); + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅账户失败: ${String(error)}`, + processFail: (error) => `账户推送处理异常: ${String(error)}`, + } + ); + + safeSubscribe( + this.exchange.watchOrders.bind(this.exchange), + (orders) => { + this.syncLocksWithOrders(orders); + this.openOrders = Array.isArray(orders) + ? orders.filter((order) => order.type !== "MARKET" && order.symbol === this.config.symbol) + : []; + const currentIds = new Set(this.openOrders.map((order) => String(order.orderId))); + for (const id of Array.from(this.pendingCancelOrders)) { + if (!currentIds.has(id)) { + this.pendingCancelOrders.delete(id); } - const position = getPosition(snapshot, this.config.symbol); - this.updateSessionVolume(position); - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `账户推送处理异常: ${String(err)}`); } - }); - } catch (err) { - this.tradeLog.push("error", `订阅账户失败: ${String(err)}`); - } + this.initialOrderSnapshotReady = true; + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅订单失败: ${String(error)}`, + processFail: (error) => `订单推送处理异常: ${String(error)}`, + } + ); - try { - this.exchange.watchOrders((orders) => { - try { - this.syncLocksWithOrders(orders); - this.openOrders = Array.isArray(orders) - ? orders.filter((order) => order.type !== "MARKET" && order.symbol === this.config.symbol) - : []; - const currentIds = new Set(this.openOrders.map((order) => String(order.orderId))); - for (const id of Array.from(this.pendingCancelOrders)) { - if (!currentIds.has(id)) { - this.pendingCancelOrders.delete(id); - } - } - this.initialOrderSnapshotReady = true; - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `订单推送处理异常: ${String(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅订单失败: ${String(err)}`); - } + safeSubscribe( + this.exchange.watchDepth.bind(this.exchange, this.config.symbol), + (depth) => { + this.depthSnapshot = depth; + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅深度失败: ${String(error)}`, + processFail: (error) => `深度推送处理异常: ${String(error)}`, + } + ); - try { - this.exchange.watchDepth(this.config.symbol, (depth) => { - try { - this.depthSnapshot = depth; - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `深度推送处理异常: ${String(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅深度失败: ${String(err)}`); - } + safeSubscribe( + this.exchange.watchTicker.bind(this.exchange, this.config.symbol), + (ticker) => { + this.tickerSnapshot = ticker; + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅Ticker失败: ${String(error)}`, + processFail: (error) => `价格推送处理异常: ${String(error)}`, + } + ); - try { - this.exchange.watchTicker(this.config.symbol, (ticker) => { - try { - this.tickerSnapshot = ticker; - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `价格推送处理异常: ${String(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅Ticker失败: ${String(err)}`); - } - - try { - this.exchange.watchKlines(this.config.symbol, "1m", () => { - try { - /* no-op */ - } catch (err) { - this.tradeLog.push("error", `K线推送处理异常: ${String(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅K线失败: ${String(err)}`); - } + safeSubscribe( + this.exchange.watchKlines.bind(this.exchange, this.config.symbol, "1m"), + (_klines) => { + /* no-op */ + }, + log, + { + subscribeFail: (error) => `订阅K线失败: ${String(error)}`, + processFail: (error) => `K线推送处理异常: ${String(error)}`, + } + ); } private syncLocksWithOrders(orders: AsterOrder[] | null | undefined): void { @@ -282,7 +280,7 @@ export class OffsetMakerEngine { } this.desiredOrders = desired; - this.updateSessionVolume(position); + this.sessionVolume.update(position, this.getReferencePrice()); await this.syncOrders(desired); await this.checkRisk(position, closeBidPrice, closeAskPrice); this.emitUpdate(); @@ -569,16 +567,9 @@ export class OffsetMakerEngine { private emitUpdate(): void { try { const snapshot = this.buildSnapshot(); - const handlers = this.listeners.get("update"); - if (handlers) { - handlers.forEach((handler) => { - try { - handler(snapshot); - } catch (err) { - this.tradeLog.push("error", `更新回调处理异常: ${String(err)}`); - } - }); - } + this.events.emit("update", snapshot, (error) => { + this.tradeLog.push("error", `更新回调处理异常: ${String(error)}`); + }); } catch (err) { this.tradeLog.push("error", `快照或更新分发异常: ${String(err)}`); } @@ -599,7 +590,7 @@ export class OffsetMakerEngine { position, pnl, accountUnrealized: this.accountUnrealized, - sessionVolume: this.sessionQuoteVolume, + sessionVolume: this.sessionVolume.value, openOrders: this.openOrders, desiredOrders: this.desiredOrders, tradeLog: this.tradeLog.all(), @@ -612,24 +603,6 @@ export class OffsetMakerEngine { }; } - private updateSessionVolume(position: PositionSnapshot): void { - const price = this.getReferencePrice(); - if (!this.initializedPosition) { - this.prevPositionAmt = position.positionAmt; - this.initializedPosition = true; - return; - } - if (price == null) { - this.prevPositionAmt = position.positionAmt; - return; - } - const delta = Math.abs(position.positionAmt - this.prevPositionAmt); - if (delta > 0) { - this.sessionQuoteVolume += delta * price; - } - this.prevPositionAmt = position.positionAmt; - } - private getReferencePrice(): number | null { return getMidOrLast(this.depthSnapshot, this.tickerSnapshot); } diff --git a/src/strategy/trend-engine.ts b/src/strategy/trend-engine.ts index 652bdb8..3bad5f2 100644 --- a/src/strategy/trend-engine.ts +++ b/src/strategy/trend-engine.ts @@ -32,6 +32,9 @@ import { createTradeLog, type TradeLogEntry } from "../logging/trade-log"; import { decryptCopyright } from "../utils/copyright"; import { isRateLimitError } from "../utils/errors"; import { RateLimitController } from "../core/lib/rate-limit"; +import { StrategyEventEmitter } from "./common/event-emitter"; +import { safeSubscribe, type LogHandler } from "./common/subscriptions"; +import { SessionVolumeTracker } from "./common/session-volume"; export interface TrendEngineSnapshot { ready: boolean; @@ -75,6 +78,8 @@ export class TrendEngine { private readonly pending: OrderPendingMap = {}; private readonly tradeLog: ReturnType; + private readonly events = new StrategyEventEmitter(); + private readonly sessionVolume = new SessionVolumeTracker(); private timer: ReturnType | null = null; private processing = false; @@ -84,9 +89,6 @@ export class TrendEngine { private totalProfit = 0; private totalTrades = 0; private lastOpenPlan: OpenOrderPlan = { side: null, price: null }; - private sessionQuoteVolume = 0; - private prevPositionAmt = 0; - private initializedPosition = false; private cancelAllRequested = false; private readonly pendingCancelOrders = new Set(); private readonly rateLimit: RateLimitController; @@ -137,18 +139,11 @@ export class TrendEngine { } on(event: TrendEngineEvent, handler: TrendEngineListener): void { - const handlers = this.listeners.get(event) ?? new Set(); - handlers.add(handler); - this.listeners.set(event, handlers); + this.events.on(event, handler); } off(event: TrendEngineEvent, handler: TrendEngineListener): void { - const handlers = this.listeners.get(event); - if (!handlers) return; - handlers.delete(handler); - if (handlers.size === 0) { - this.listeners.delete(event); - } + this.events.off(event, handler); } getSnapshot(): TrendEngineSnapshot { @@ -156,82 +151,89 @@ export class TrendEngine { } private bootstrap(): void { - try { - this.exchange.watchAccount((snapshot) => { - try { - this.accountSnapshot = snapshot; - const position = getPosition(snapshot, this.config.symbol); - this.updateSessionVolume(position); - this.trackPositionLifecycle(position, this.getReferencePrice()); - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `账户推送处理异常: ${extractMessage(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅账户失败: ${String(err)}`); - } - try { - this.exchange.watchOrders((orders) => { - try { - this.synchronizeLocks(orders); - this.openOrders = Array.isArray(orders) - ? orders.filter((order) => order.type !== "MARKET" && order.symbol === this.config.symbol) - : []; - const currentIds = new Set(this.openOrders.map((order) => String(order.orderId))); - for (const id of Array.from(this.pendingCancelOrders)) { - if (!currentIds.has(id)) { - this.pendingCancelOrders.delete(id); - } + const log: LogHandler = (type, detail) => this.tradeLog.push(type, detail); + + safeSubscribe( + this.exchange.watchAccount.bind(this.exchange), + (snapshot) => { + this.accountSnapshot = snapshot; + const position = getPosition(snapshot, this.config.symbol); + const reference = this.getReferencePrice(); + this.sessionVolume.update(position, reference); + this.trackPositionLifecycle(position, reference); + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅账户失败: ${String(error)}`, + processFail: (error) => `账户推送处理异常: ${extractMessage(error)}`, + } + ); + + safeSubscribe( + this.exchange.watchOrders.bind(this.exchange), + (orders) => { + this.synchronizeLocks(orders); + this.openOrders = Array.isArray(orders) + ? orders.filter((order) => order.type !== "MARKET" && order.symbol === this.config.symbol) + : []; + const currentIds = new Set(this.openOrders.map((order) => String(order.orderId))); + for (const id of Array.from(this.pendingCancelOrders)) { + if (!currentIds.has(id)) { + this.pendingCancelOrders.delete(id); } - if (this.openOrders.length === 0 || this.pendingCancelOrders.size === 0) { - this.cancelAllRequested = false; - } - this.ordersSnapshotReady = true; - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `订单推送处理异常: ${extractMessage(err)}`); } - }); - } catch (err) { - this.tradeLog.push("error", `订阅订单失败: ${String(err)}`); - } - try { - this.exchange.watchDepth(this.config.symbol, (depth) => { - try { - this.depthSnapshot = depth; - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `深度推送处理异常: ${extractMessage(err)}`); + if (this.openOrders.length === 0 || this.pendingCancelOrders.size === 0) { + this.cancelAllRequested = false; } - }); - } catch (err) { - this.tradeLog.push("error", `订阅深度失败: ${String(err)}`); - } - try { - this.exchange.watchTicker(this.config.symbol, (ticker) => { - try { - this.tickerSnapshot = ticker; - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `价格推送处理异常: ${extractMessage(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅Ticker失败: ${String(err)}`); - } - try { - this.exchange.watchKlines(this.config.symbol, this.config.klineInterval, (klines) => { - try { - this.klineSnapshot = Array.isArray(klines) ? klines : []; - this.emitUpdate(); - } catch (err) { - this.tradeLog.push("error", `K线推送处理异常: ${extractMessage(err)}`); - } - }); - } catch (err) { - this.tradeLog.push("error", `订阅K线失败: ${String(err)}`); - } + this.ordersSnapshotReady = true; + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅订单失败: ${String(error)}`, + processFail: (error) => `订单推送处理异常: ${extractMessage(error)}`, + } + ); + + safeSubscribe( + this.exchange.watchDepth.bind(this.exchange, this.config.symbol), + (depth) => { + this.depthSnapshot = depth; + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅深度失败: ${String(error)}`, + processFail: (error) => `深度推送处理异常: ${extractMessage(error)}`, + } + ); + + safeSubscribe( + this.exchange.watchTicker.bind(this.exchange, this.config.symbol), + (ticker) => { + this.tickerSnapshot = ticker; + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅Ticker失败: ${String(error)}`, + processFail: (error) => `价格推送处理异常: ${extractMessage(error)}`, + } + ); + + safeSubscribe( + this.exchange.watchKlines.bind(this.exchange, this.config.symbol, this.config.klineInterval), + (klines) => { + this.klineSnapshot = Array.isArray(klines) ? klines : []; + this.emitUpdate(); + }, + log, + { + subscribeFail: (error) => `订阅K线失败: ${String(error)}`, + processFail: (error) => `K线推送处理异常: ${extractMessage(error)}`, + } + ); } private synchronizeLocks(orders: AsterOrder[] | null | undefined): void { @@ -303,7 +305,7 @@ export class TrendEngine { } } - this.updateSessionVolume(position); + this.sessionVolume.update(position, price); this.trackPositionLifecycle(position, price); this.lastSma30 = sma30; this.lastPrice = price; @@ -849,16 +851,9 @@ export class TrendEngine { private emitUpdate(): void { try { const snapshot = this.buildSnapshot(); - const handlers = this.listeners.get("update"); - if (handlers) { - handlers.forEach((handler) => { - try { - handler(snapshot); - } catch (err) { - this.tradeLog.push("error", `更新回调处理异常: ${String(err)}`); - } - }); - } + this.events.emit("update", snapshot, (error) => { + this.tradeLog.push("error", `更新回调处理异常: ${String(error)}`); + }); } catch (err) { this.tradeLog.push("error", `快照或更新分发异常: ${String(err)}`); } @@ -888,7 +883,7 @@ export class TrendEngine { unrealized: position.unrealizedProfit, totalProfit: this.totalProfit, totalTrades: this.totalTrades, - sessionVolume: this.sessionQuoteVolume, + sessionVolume: this.sessionVolume.value, tradeLog: this.tradeLog.all(), openOrders: this.openOrders, depth: this.depthSnapshot, @@ -898,24 +893,6 @@ export class TrendEngine { }; } - private updateSessionVolume(position: PositionSnapshot): void { - const price = this.getReferencePrice(); - if (!this.initializedPosition) { - this.prevPositionAmt = position.positionAmt; - this.initializedPosition = true; - return; - } - if (price == null) { - this.prevPositionAmt = position.positionAmt; - return; - } - const delta = Math.abs(position.positionAmt - this.prevPositionAmt); - if (delta > 0) { - this.sessionQuoteVolume += delta * price; - } - this.prevPositionAmt = position.positionAmt; - } - private getReferencePrice(): number | null { return getMidOrLast(this.depthSnapshot, this.tickerSnapshot) ?? (this.lastPrice != null && Number.isFinite(this.lastPrice) ? this.lastPrice : null); }