diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/PolymarketActivityWsService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/PolymarketActivityWsService.kt index b2d8989..3d8180f 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/PolymarketActivityWsService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/PolymarketActivityWsService.kt @@ -1,5 +1,7 @@ package com.wrbug.polymarketbot.service.copytrading.monitor +import com.github.benmanes.caffeine.cache.Cache +import com.github.benmanes.caffeine.cache.Caffeine import com.wrbug.polymarketbot.api.TradeResponse import com.wrbug.polymarketbot.dto.ActivityTradeMessage import com.wrbug.polymarketbot.dto.ActivityTradePayload @@ -15,10 +17,11 @@ import org.slf4j.LoggerFactory import org.springframework.stereotype.Service import java.math.BigDecimal import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.TimeUnit /** * Polymarket Activity WebSocket 监听服务 - * 通过订阅全局 activity 交易流，客户端过滤 Leader 地址，实现实时交易检测 + * 通过订阅全局 activity 交易流（trades + orders_matched），客户端过滤 Leader 地址，实现实时交易检测 * 延迟 < 100ms，适合快速跟单场景 */ @Service @@ -39,6 +42,13 @@ class PolymarketActivityWsService( // 要监听的 Leader 地址集合（小写地址 -> leaderId） private val monitoredAddresses = ConcurrentHashMap() + // 存储已处理的交易哈希，用于去重（LRU 缓存，保留最近 100 条） + // 因为同时订阅 trades 和 orders_matched，同一个交易可能被推送两次 + private val processedTxHashes: Cache = Caffeine.newBuilder() + .maximumSize(100) + .expireAfterWrite(10, TimeUnit.MINUTES) + .build() + // 是否已订阅 @Volatile private var isSubscribed = false @@ -50,6 +60,12 @@ class PolymarketActivityWsService( // Activity 消息超时检测任务 private var activityTimeoutJob: Job? = null + // 性能统计 + private var totalMessagesProcessed = 0L + private var addressMatchMessages = 0L + private var jsonParseMessages = 0L + private var duplicateTxHashMessages = 0L + /** * 启动监听 */ @@ -68,7 +84,7 @@ class PolymarketActivityWsService( return } - logger.info("启动 Activity WebSocket 监听，监控 ${monitoredAddresses.size} 个 Leader 地址") + logger.info("启动 Activity WebSocket 监听（trades + orders_matched），监控 ${monitoredAddresses.size} 个 Leader 地址") connectAndSubscribe() } @@ -165,6 +181,7 @@ class PolymarketActivityWsService( * 订阅全局 activity * 根据 @polymarket/real-time-data-client 的协议格式 * 使用 "action": "subscribe" 而不是 "type": "subscribe" + * 同时订阅 trades 和 orders_matched 两种类型 */ private fun subscribeAllActivity() { val client = wsClient @@ -176,6 +193,7 @@ class PolymarketActivityWsService( try { // 根据 real-time-data-client 的协议格式 // 订阅消息应包含 "action": "subscribe" 和 "subscriptions" 数组 + // 同时订阅 trades 和 orders_matched 两种类型 val subscribeMessage = """ { "action": "subscribe", @@ -183,6 +201,10 @@ class PolymarketActivityWsService( { "topic": "activity", "type": "trades" + }, + { + "topic": "activity", + "type": "orders_matched" } ] } @@ -193,8 +215,8 @@ class PolymarketActivityWsService( // 重置最后一次收到 activity 消息的时间 lastActivityTime = System.currentTimeMillis() // 启动 Activity 消息超时检测 - startActivityTimeoutCheck() - logger.info("Activity WebSocket 订阅成功（全局交易流）") +// startActivityTimeoutCheck() + logger.info("Activity WebSocket 订阅成功（全局交易流: trades + orders_matched）") } catch (e: Exception) { logger.error("订阅 Activity WebSocket 失败", e) isSubscribed = false @@ -208,24 +230,24 @@ class PolymarketActivityWsService( private fun startActivityTimeoutCheck() { // 先停止之前的检测任务 stopActivityTimeoutCheck() - + activityTimeoutJob = scope.launch { while (isActive && isSubscribed) { delay(30000) // 每30秒检查一次 - + // 如果已经取消订阅，停止检测 if (!isSubscribed) { break } - + // 如果 lastActivityTime 为 0，说明还没有收到过消息，跳过本次检测 if (lastActivityTime == 0L) { continue } - + val currentTime = System.currentTimeMillis() val timeSinceLastActivity = currentTime - lastActivityTime - + // 如果超过30秒没有收到activity消息，触发重连 if (timeSinceLastActivity >= 30000) { logger.warn("超过30秒未收到 Activity 消息，触发重连。距离上次消息: ${timeSinceLastActivity}ms") @@ -240,7 +262,7 @@ class PolymarketActivityWsService( } } } - + /** * 停止 Activity 消息超时检测 */ @@ -249,42 +271,95 @@ class PolymarketActivityWsService( activityTimeoutJob = null } + /** + * 检查消息是否包含监听的 Leader 地址 + * 快速过滤，避免不必要的 JSON 解析 + * 只需要检查 "proxyWallet":"0x..." 或 "trader":{"address":"0x..."} 格式 + */ + private fun containsMonitoredAddress(message: String): Boolean { + // 快速检查：如果消息很短，不可能包含地址 + if (message.length < 50) { + return false + } + + // 遍历所有监听的地址 + for ((address, leaderId) in monitoredAddresses) { + // 检查 proxyWallet：格式为 "proxyWallet":"0x..." + if (message.contains("\"proxyWallet\":\"$address\"", ignoreCase = true)) { + addressMatchMessages++ + return true + } + + // 检查 trader.address：格式为 "trader":{"address":"0x..."} + if (message.contains("\"trader\"", ignoreCase = true) && + message.contains("\"address\":\"$address\"", ignoreCase = true) + ) { + addressMatchMessages++ + return true + } + } + + return false + } + /** * 处理消息 */ private fun handleMessage(message: String) { try { + totalMessagesProcessed++ + // 处理 PONG 响应 if (message.trim() == "PONG" || message.trim() == "pong") { return } - - // 使用扩展函数解析消息 + + // 快速预检查：检查是否包含监听地址 + // 绝大部分消息会在这一步被过滤掉，避免不必要的 JSON 解析 + if (!containsMonitoredAddress(message)) { + return + } + logger.info("发现leader交易：${message}") + // 使用扩展函数解析消息（只对包含监听地址的消息） val tradeMessage = message.fromJson() ?: run { // 不是有效的 JSON 或格式不匹配，跳过 - logger.warn("无法解析为 ActivityTradeMessage，可能不是 activity 消息: ${message.take(200)}") + logger.warn("无法解析为 ActivityTradeMessage: ${message.take(200)}") return } - - // 检查是否是 activity trade 消息 - if (tradeMessage.topic != "activity" || tradeMessage.type != "trades") { - // 不是我们关心的消息，直接返回 + + jsonParseMessages++ + + // 检查是否是 activity 消息（trades 或 orders_matched） + if (tradeMessage.topic != "activity" || + (tradeMessage.type != "trades" && tradeMessage.type != "orders_matched")) { return } - + // 更新最后一次收到 activity 消息的时间（即使不是我们监听的 Leader 的交易） lastActivityTime = System.currentTimeMillis() - + val payload = tradeMessage.payload - + + // 根据 txHash 去重（使用原子操作避免竞态条件） + val txHash = payload.transactionHash + if (txHash != null && txHash.isNotBlank()) { + val currentTime = System.currentTimeMillis() + val existingTimestamp = processedTxHashes.asMap().putIfAbsent(txHash, currentTime) + if (existingTimestamp != null) { + duplicateTxHashMessages++ + logger.debug("交易已处理过，跳过: txHash=$txHash, firstProcessedAt=$existingTimestamp, type=${tradeMessage.type}") + return + } + } + // 提取交易者地址 val traderAddress = extractTraderAddress(payload) ?: run { // 没有交易者地址，跳过 logger.warn("Activity Trade 消息中没有交易者地址: trader=${payload.trader}, proxyWallet=${payload.proxyWallet}, asset=${payload.asset}") return } - - // 检查是否是我们监听的 Leader + + // 二次验证：确认地址匹配 val normalizedAddress = traderAddress.lowercase() val leaderId = monitoredAddresses[normalizedAddress] ?: run { return @@ -449,6 +524,7 @@ class PolymarketActivityWsService( wsClient = null isSubscribed = false monitoredAddresses.clear() + processedTxHashes.invalidateAll() // 清空去重缓存 lastActivityTime = 0 } @@ -466,8 +542,33 @@ class PolymarketActivityWsService( return monitoredAddresses.size } + /** + * 获取性能统计信息 + */ + fun getPerformanceStats(): Map { + val jsonParseRate = if (totalMessagesProcessed > 0) { + (jsonParseMessages.toDouble() / totalMessagesProcessed * 100).toInt() + } else { + 0 + } + + return mapOf( + "totalMessages" to totalMessagesProcessed, + "addressMatches" to addressMatchMessages, + "jsonParses" to jsonParseMessages, + "duplicateTxHashes" to duplicateTxHashMessages, + "jsonParseRate" to "$jsonParseRate%", + "filteringEfficiency" to if (totalMessagesProcessed > 0) { + ((1.0 - jsonParseMessages.toDouble() / totalMessagesProcessed) * 100).toInt() + } else { + 0 + } + ) + } + @PreDestroy fun destroy() { + logger.info("Activity WS 性能统计: ${getPerformanceStats()}") stop() scope.cancel() }