From 377da4fff6d03db8257d6c5861ac6167cc91f1d4 Mon Sep 17 00:00:00 2001 From: WrBug Date: Wed, 25 Feb 2026 19:27:49 +0800 Subject: [PATCH] =?UTF-8?q?fix(cryptotail):=20=E4=BC=98=E5=8C=96=20WebSock?= =?UTF-8?q?et=20=E8=AE=A2=E9=98=85=E7=AE=A1=E7=90=86=E4=B8=8E=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E6=B8=85=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 使用 createClient() 懒加载 OkHttpClient - 修复订阅计数归零时未移除策略的问题 - 简化 handleMessage，移除 isFromCurrentPeriod 参数和 book 事件处理 - 清理调试日志，保留关键业务日志 Co-authored-by: Cursor --- .../common/WebSocketSubscriptionService.kt | 7 +- .../cryptotail/CryptoTailMonitorService.kt | 109 ++++++++++++------ .../websocket/UnifiedWebSocketHandler.kt | 3 +- 3 files changed, 79 insertions(+), 40 deletions(-) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/WebSocketSubscriptionService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/WebSocketSubscriptionService.kt index 8723b71..286aac6 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/WebSocketSubscriptionService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/WebSocketSubscriptionService.kt @@ -66,22 +66,21 @@ class WebSocketSubscriptionService( * 注销会话 */ fun unregisterSession(sessionId: String) { - // 取消所有订阅 val channels = sessionSubscriptions.remove(sessionId) ?: emptySet() channels.forEach { channel -> unsubscribe(sessionId, channel) } - + // 清理 order 频道的回调 orderChannelCallbacks.remove(sessionId) - + // 清理尾盘监控频道的回调 val monitorCallbacks = monitorChannelCallbacks.remove(sessionId) monitorCallbacks?.keys?.forEach { strategyId -> cryptoTailMonitorService?.unsubscribe(sessionId, strategyId) } - + sessionCallbacks.remove(sessionId) } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailMonitorService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailMonitorService.kt index 64fddec..ed903cd 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailMonitorService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailMonitorService.kt @@ -12,6 +12,7 @@ import com.wrbug.polymarketbot.service.binance.BinanceKlineAutoSpreadService import com.wrbug.polymarketbot.service.binance.BinanceKlineService import com.wrbug.polymarketbot.service.common.WebSocketSubscriptionService import com.wrbug.polymarketbot.util.RetrofitFactory +import com.wrbug.polymarketbot.util.createClient import com.wrbug.polymarketbot.util.fromJson import com.wrbug.polymarketbot.util.toJson import com.wrbug.polymarketbot.util.toSafeBigDecimal @@ -58,6 +59,7 @@ class CryptoTailMonitorService( /** 当前周期 token 映射 */ private val currentPeriodTokenToStrategy = AtomicReference>>(emptyMap()) + /** 下一周期 token 映射 */ private val nextPeriodTokenToStrategy = AtomicReference>>(emptyMap()) @@ -70,7 +72,10 @@ class CryptoTailMonitorService( private var currentPeriodWebSocket: WebSocket? = null private var nextPeriodWebSocket: WebSocket? = null private val wsUrl = PolymarketConstants.RTDS_WS_URL + "/ws/market" - private val client = OkHttpClient.Builder().build() + + private val client by lazy { + createClient().build() + } private val reconnectDelayMs = 3_000L private var reconnectJob: Job? = null @@ -144,14 +149,22 @@ class CryptoTailMonitorService( val tokenIds = parseClobTokenIds(market?.clobTokenIds) // 获取开盘价（币安 K 线 open = BTC 价格 USDC） - val openClose = binanceKlineService.getCurrentOpenClose(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix) + val openClose = binanceKlineService.getCurrentOpenClose( + strategy.marketSlugPrefix, + strategy.intervalSeconds, + periodStartUnix + ) val openPriceBtc = openClose?.first // 获取自动计算的最小价差 var autoMinSpreadUp: BigDecimal? = null var autoMinSpreadDown: BigDecimal? = null if (strategy.spreadMode.name.uppercase() == "AUTO") { - val autoSpreads = binanceKlineAutoSpreadService.computeAndCache(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix) + val autoSpreads = binanceKlineAutoSpreadService.computeAndCache( + strategy.marketSlugPrefix, + strategy.intervalSeconds, + periodStartUnix + ) autoMinSpreadUp = autoSpreads?.first autoMinSpreadDown = autoSpreads?.second } @@ -202,7 +215,6 @@ class CryptoTailMonitorService( fun subscribe(sessionId: String, strategyId: Long, callback: (CryptoTailMonitorPushData) -> Unit) { // 增加订阅计数 val count = strategySubscribers.merge(strategyId, 1) { old, inc -> old + inc } ?: 1 - logger.info("策略 $strategyId 订阅数: $count") // 注册推送回调 webSocketSubscriptionService.registerMonitorCallback(sessionId, strategyId, callback) @@ -230,15 +242,20 @@ class CryptoTailMonitorService( */ fun unsubscribe(sessionId: String, strategyId: Long) { // 减少订阅计数 - val count = strategySubscribers.merge(strategyId, -1) { old, dec -> (old - dec).coerceAtLeast(0) } ?: 0 - logger.info("策略 $strategyId 订阅数: $count") + val currentCount = strategySubscribers[strategyId] ?: 0 + val newCount = (currentCount - 1).coerceAtLeast(0) + + if (newCount == 0) { + strategySubscribers.remove(strategyId) + } else { + strategySubscribers[strategyId] = newCount + } // 移除回调 webSocketSubscriptionService.unregisterMonitorCallback(sessionId, strategyId) // 如果没有订阅者，关闭 WebSocket 和定时推送 - if (count == 0) { - strategySubscribers.remove(strategyId) + if (newCount == 0) { scope.launch { refreshSubscription() } @@ -277,7 +294,11 @@ class CryptoTailMonitorService( /** * 发送当前数据（含历史补全，用于中途进入时填充分时图） */ - private suspend fun sendCurrentData(sessionId: String, strategyId: Long, callback: (CryptoTailMonitorPushData) -> Unit) { + private suspend fun sendCurrentData( + sessionId: String, + strategyId: Long, + callback: (CryptoTailMonitorPushData) -> Unit + ) { val strategy = strategyRepository.findById(strategyId).orElse(null) ?: return val priceData = strategyPriceData[strategyId] ?: StrategyPriceData() @@ -298,7 +319,6 @@ class CryptoTailMonitorService( */ private suspend fun refreshSubscription() { if (!refreshSubscriptionMutex.tryLock()) { - logger.debug("refreshSubscription 正在执行，跳过本次并发调用") return } try { @@ -334,7 +354,8 @@ class CryptoTailMonitorService( nextPeriodWebSocket = null val nextMap = nextPeriodTokenToStrategy.get() currentPeriodTokenToStrategy.set(nextMap) - val nextPeriodByStrategy = nextMap.values.flatten().distinctBy { it.strategyId }.associate { it.strategyId to it.periodStartUnix } + val nextPeriodByStrategy = + nextMap.values.flatten().distinctBy { it.strategyId }.associate { it.strategyId to it.periodStartUnix } logger.info("周期切换：下一周期连接晋升为当前") for ((strategyId, periodStartUnix) in nextPeriodByStrategy) { updateStrategyPriceDataForPeriod(listOf(strategyId), periodStartUnix, pushDefault = true) @@ -405,14 +426,24 @@ class CryptoTailMonitorService( val map = mutableMapOf>() for (strategy in strategies) { - if (!strategy.enabled || strategy.id == null) continue + if (!strategy.enabled || strategy.id == null) { + continue + } val currentPeriod = (nowSeconds / strategy.intervalSeconds) * strategy.intervalSeconds val nextPeriod = currentPeriod + strategy.intervalSeconds val slug = "${strategy.marketSlugPrefix}-$nextPeriod" - val event = fetchEventBySlug(slug).getOrNull() ?: continue - val market = event.markets?.firstOrNull() ?: continue + val event = fetchEventBySlug(slug).getOrNull() + if (event == null) { + continue + } + val market = event.markets?.firstOrNull() + if (market == null) { + continue + } val tokenIds = parseClobTokenIds(market.clobTokenIds) - if (tokenIds.size < 2) continue + if (tokenIds.size < 2) { + continue + } for (i in tokenIds.indices) { tokenIdSet.add(tokenIds[i]) map.getOrPut(tokenIds[i]) { mutableListOf() }.add( @@ -424,11 +455,19 @@ class CryptoTailMonitorService( } /** 更新策略价格数据为指定周期（开盘价、价差线等），可选是否推送默认 0.5 */ - private suspend fun updateStrategyPriceDataForPeriod(strategyIds: List, periodStartUnix: Long, pushDefault: Boolean) { + private suspend fun updateStrategyPriceDataForPeriod( + strategyIds: List, + periodStartUnix: Long, + pushDefault: Boolean + ) { val strategies = strategyRepository.findAllById(strategyIds) for (strategy in strategies) { if (strategy.id == null) continue - val openClose = binanceKlineService.getCurrentOpenClose(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix) + val openClose = binanceKlineService.getCurrentOpenClose( + strategy.marketSlugPrefix, + strategy.intervalSeconds, + periodStartUnix + ) val openPriceBtc = openClose?.first var minSpreadLineUp: BigDecimal? = null var minSpreadLineDown: BigDecimal? = null @@ -437,8 +476,13 @@ class CryptoTailMonitorService( minSpreadLineUp = strategy.spreadValue?.toSafeBigDecimal() minSpreadLineDown = strategy.spreadValue?.toSafeBigDecimal() } + "AUTO" -> { - val autoSpreads = binanceKlineAutoSpreadService.computeAndCache(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix) + val autoSpreads = binanceKlineAutoSpreadService.computeAndCache( + strategy.marketSlugPrefix, + strategy.intervalSeconds, + periodStartUnix + ) minSpreadLineUp = autoSpreads?.first minSpreadLineDown = autoSpreads?.second } @@ -479,7 +523,7 @@ class CryptoTailMonitorService( } override fun onMessage(webSocket: WebSocket, text: String) { - handleMessage(webSocket, text, isFromCurrentPeriod = true) + handleMessage(webSocket, text) } override fun onClosing(webSocket: WebSocket, code: Int, reason: String) { @@ -499,7 +543,9 @@ class CryptoTailMonitorService( } private fun connectNextPeriod(tokenIds: List, map: Map>) { - if (nextPeriodWebSocket != null) return + if (nextPeriodWebSocket != null) { + return + } val request = Request.Builder().url(wsUrl).build() nextPeriodWebSocket = client.newWebSocket(request, object : WebSocketListener() { override fun onOpen(webSocket: WebSocket, response: okhttp3.Response) { @@ -517,7 +563,7 @@ class CryptoTailMonitorService( } override fun onMessage(webSocket: WebSocket, text: String) { - handleMessage(webSocket, text, isFromCurrentPeriod = false) + handleMessage(webSocket, text) } override fun onClosing(webSocket: WebSocket, code: Int, reason: String) { @@ -551,10 +597,9 @@ class CryptoTailMonitorService( logger.info("尾盘监控 WebSocket 已全部关闭（无订阅者）") } - private fun handleMessage(webSocket: WebSocket, text: String, isFromCurrentPeriod: Boolean) { + private fun handleMessage(webSocket: WebSocket, text: String) { if (text == "pong" || text.isEmpty()) return if (closedForNoSubscribers.get()) return - if (!isFromCurrentPeriod) return maybeRefreshSubscriptionIfPeriodChanged() @@ -563,14 +608,6 @@ class CryptoTailMonitorService( val map = currentPeriodTokenToStrategy.get() when (eventType) { - "book" -> { - val assetId = (json.get("asset_id") as? com.google.gson.JsonPrimitive)?.asString ?: return - val bids = json.get("bids") as? com.google.gson.JsonArray - if (bids == null || bids.isEmpty) return - val firstBid = bids.get(0) as? com.google.gson.JsonObject - val bestBid = (firstBid?.get("price") as? com.google.gson.JsonPrimitive)?.asString?.toSafeBigDecimal() - if (bestBid != null) onPriceUpdate(assetId, bestBid, map) - } "price_change" -> { val priceChanges = json.get("price_changes") as? com.google.gson.JsonArray ?: return for (i in 0 until priceChanges.size()) { @@ -657,7 +694,11 @@ class CryptoTailMonitorService( val inTimeWindow = nowSeconds >= windowStart && nowSeconds < windowEnd // 币安 K 线：open = 周期开盘价，close = 当前最新价（实时更新） - val openClose = binanceKlineService.getCurrentOpenClose(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix) + val openClose = binanceKlineService.getCurrentOpenClose( + strategy.marketSlugPrefix, + strategy.intervalSeconds, + periodStartUnix + ) val openPriceBtc = priceData.openPriceBtc ?: openClose?.first val currentPriceBtc = openClose?.second // K 线数据回来后更新缓存，供后续使用 @@ -672,9 +713,9 @@ class CryptoTailMonitorService( val currentUp = priceData.currentPriceUp val currentDown = priceData.currentPriceDown val inPriceRangeUp = currentUp != null && - currentUp >= strategy.minPrice && currentUp <= strategy.maxPrice + currentUp >= strategy.minPrice && currentUp <= strategy.maxPrice val inPriceRangeDown = currentDown != null && - currentDown >= strategy.minPrice && currentDown <= strategy.maxPrice + currentDown >= strategy.minPrice && currentDown <= strategy.maxPrice return CryptoTailMonitorPushData( strategyId = strategy.id!!, diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/websocket/UnifiedWebSocketHandler.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/websocket/UnifiedWebSocketHandler.kt index d002764..b37e8de 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/websocket/UnifiedWebSocketHandler.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/websocket/UnifiedWebSocketHandler.kt @@ -193,7 +193,7 @@ class UnifiedWebSocketHandler( lastActivityTime.remove(sessionId) sessionLocks.remove(sessionId) // 清理同步锁 subscriptionService.unregisterSession(sessionId) - + if (session != null && session.isOpen) { try { session.close(CloseStatus.NORMAL) @@ -201,7 +201,6 @@ class UnifiedWebSocketHandler( // 忽略关闭时的异常 } } - } catch (e: Exception) { logger.error("清理 WebSocket 资源时发生错误: $sessionId, ${e.message}", e) }