feat(activity-ws): 同时监听 trades 和 orders_matched，添加 txHash 去重

- 订阅消息同时包含 trades 和 orders_matched 两种类型
- 添加 processedTxHashes Cache (LRU 100 条，10 分钟过期)
- 防止同一交易被重复处理
- 新增 duplicateTxHashMessages 统计指标
- 更新日志信息，明确标注监听范围

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<String, Long>()
 
+    // 存储已处理的交易哈希，用于去重（LRU 缓存，保留最近 100 条）
+    // 因为同时订阅 trades 和 orders_matched，同一个交易可能被推送两次
+    private val processedTxHashes: Cache<String, Long> = 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<ActivityTradeMessage>() ?: 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<String, Any> {
+        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()
     }