feat: 实现 Polymarket Activity WebSocket 双重监听方案

主要功能： - 新增 PolymarketActivityWsService：通过 Activity WebSocket 实时监听 Leader 交易（延迟 < 100ms） - 实现双重监听机制：Activity WS（低延迟）+ On-Chain WS（高可靠性） - 支持动态添加/移除 Leader 监听 - 地址筛选忽略大小写修复和改进： - 修复订阅消息格式：使用 'action' 而不是 'type'（与 @polymarket/real-time-data-client 保持一致） - 修复 PING 消息格式：使用小写 'ping'（与官方客户端保持一致） - 添加 outcomeIndex 字段支持：优先使用消息中的 outcomeIndex，fallback 到从 outcome 解析 - 添加 connection_id 字段到 DTO - 改进日志记录：增加 INFO 级别日志以便调试代码重构： - JsonUtils 重构：将 fromJson 和 toJson 改为扩展方法 - 更新后端开发规范：JSON 解析和 Data Class 规范前端修复： - 修复 InputNumber parser 类型错误：返回 number 而不是 string 文档： - 添加双重监听方案文档 - 添加 Polymarket Activity WebSocket API 格式文档
2026-01-09 10:41:15 +08:00
parent a16b6fce54
commit 19508dcd19
12 changed files with 1566 additions and 29 deletions
@@ -0,0 +1,64 @@
+package com.wrbug.polymarketbot.dto
+
+import com.google.gson.annotations.SerializedName
+
+/**
+ * Activity WebSocket 交易消息 DTO
+ * 根据 Polymarket RTDS Activity WebSocket API 格式定义
+ */
+data class ActivityTradeMessage(
+    val topic: String = "",                    // "activity"
+    val type: String = "",                     // "trades"
+    val timestamp: Long? = null,               // 消息时间戳（可选）
+    @SerializedName("connection_id")
+    val connectionId: String? = null,          // 连接 ID（可选，由服务器返回）
+    val payload: ActivityTradePayload = ActivityTradePayload()  // 交易数据
+)
+
+/**
+ * Activity Trade Payload
+ */
+data class ActivityTradePayload(
+    val asset: String = "",                    // Token ID (用于下单)
+    
+    @SerializedName("conditionId")
+    val conditionId: String = "",              // Market condition ID
+    
+    @SerializedName("eventSlug")
+    val eventSlug: String? = null,             // 事件 slug
+    
+    val slug: String? = null,                  // 市场 slug
+    
+    val outcome: String? = null,               // 结果方向 (Yes/No/Up/Down)
+    
+    @SerializedName("outcomeIndex")
+    val outcomeIndex: Int? = null,             // 结果索引 (0=Yes/Up, 1=No/Down) - 优先使用此字段
+    
+    val side: String = "",                     // 交易方向 (BUY/SELL)
+    
+    // price 和 size 可能是数字或字符串，使用 Any 类型，后续转换为 String
+    val price: Any? = null,                    // 交易价格
+    
+    val size: Any? = null,                     // 交易数量 (shares)
+    
+    val timestamp: Any? = null,                // Unix 时间戳（可能是秒或毫秒，可能是数字或字符串）
+    
+    @SerializedName("transactionHash")
+    val transactionHash: String? = null,       // 交易哈希
+    
+    val trader: ActivityTrader? = null,        // 交易者信息对象（优先）
+    
+    @SerializedName("proxyWallet")
+    val proxyWallet: String? = null,           // 交易者地址（fallback，如果 trader 不存在）
+    
+    val name: String? = null                   // 交易者名称（fallback，如果 trader 不存在）
+)
+
+/**
+ * Activity Trader 信息
+ */
+data class ActivityTrader(
+    val name: String? = null,             // 交易者用户名（可选）
+    val address: String? = null           // 交易者钱包地址 ⭐ 关键字段
+)
+
@@ -14,7 +14,9 @@ import org.springframework.stereotype.Service
 /**
 * 跟单监听服务（主服务）
 * 管理所有Leader的交易监听
- * 使用链上 WebSocket 监听 Leader 的交易（实时，秒级延迟）
+ * 使用双重监听机制：
+ * 1. Activity WebSocket - 低延迟（< 100ms），优先使用
+ * 2. OnChain WebSocket - 高可靠性（~2-3s），作为兜底
 * 同时监听跟单账户的卖出/赎回事件（通过链上 WebSocket）
 */
@Service
@@ -22,6 +24,7 @@ class CopyTradingMonitorService(
    private val copyTradingRepository: CopyTradingRepository,
    private val leaderRepository: LeaderRepository,
    private val accountRepository: AccountRepository,
+    private val activityWsService: PolymarketActivityWsService,
    private val onChainWsService: OnChainWsService,
    private val accountOnChainMonitorService: AccountOnChainMonitorService
 ) {
@@ -50,14 +53,17 @@ class CopyTradingMonitorService(
    @PreDestroy
    fun destroy() {
        scope.cancel()
-        // 停止链上 WS 监听
+        // 停止所有监听
+        activityWsService.stop()
        onChainWsService.stop()
        accountOnChainMonitorService.stop()
    }
    
    /**
     * 启动监听
-     * 启动链上 WebSocket 监听 Leader 的交易（实时，秒级延迟）
+     * 启动双重监听机制：
+     * 1. Activity WebSocket - 低延迟（< 100ms），优先使用
+     * 2. OnChain WebSocket - 高可靠性（~2-3s），作为兜底
     * 同时启动跟单账户的链上 WebSocket 监听（用于检测卖出/赎回事件）
     */
    suspend fun startMonitoring() {
@@ -80,10 +86,13 @@ class CopyTradingMonitorService(
            accountRepository.findById(accountId).orElse(null)
        }
        
-        // 4. 启动链上 WebSocket 监听 Leader 的交易（实时，秒级延迟）
+        // 4. 启动 Activity WebSocket 监听（优先，低延迟）
+        activityWsService.start(leaders)
+        
+        // 5. 启动链上 WebSocket 监听（兜底，高可靠性）
        onChainWsService.start(leaders)
        
-        // 5. 启动跟单账户的链上 WebSocket 监听（用于检测卖出/赎回事件）
+        // 6. 启动跟单账户的链上 WebSocket 监听（用于检测卖出/赎回事件）
        accountOnChainMonitorService.start(accounts)
    }
    
@@ -100,7 +109,8 @@ class CopyTradingMonitorService(
            return
        }
        
-        // 添加到链上 WS 监听（如果不在列表中才添加）
+        // 同时添加到两种监听
+        activityWsService.addLeader(leader)
        onChainWsService.addLeader(leader)
    }
    
@@ -115,7 +125,8 @@ class CopyTradingMonitorService(
            return
        }
        
-        // 没有启用的跟单配置了，移除监听
+        // 没有启用的跟单配置了，同时从两种监听移除
+        activityWsService.removeLeader(leaderId)
        onChainWsService.removeLeader(leaderId)
    }
    
@@ -130,6 +141,7 @@ class CopyTradingMonitorService(
        
        if (copyTradings.isNotEmpty()) {
            // 有启用的跟单配置，确保在监听列表中
+            activityWsService.addLeader(leader)
            onChainWsService.addLeader(leader)
            
            // 更新账户监听（添加该配置关联的账户）
@@ -141,7 +153,8 @@ class CopyTradingMonitorService(
                }
            }
        } else {
-            // 没有启用的跟单配置，移除监听
+            // 没有启用的跟单配置，同时从两种监听移除
+            activityWsService.removeLeader(leaderId)
            onChainWsService.removeLeader(leaderId)
        }
    }
@@ -171,6 +184,7 @@ class CopyTradingMonitorService(
     */
    suspend fun restartMonitoring() {
        // 停止所有监听
+        activityWsService.stop()
        onChainWsService.stop()
        delay(1000)  // 等待1秒
        startMonitoring()
@@ -0,0 +1,412 @@
+package com.wrbug.polymarketbot.service.copytrading.monitor
+
+import com.wrbug.polymarketbot.api.TradeResponse
+import com.wrbug.polymarketbot.dto.ActivityTradeMessage
+import com.wrbug.polymarketbot.dto.ActivityTradePayload
+import com.wrbug.polymarketbot.entity.Leader
+import com.wrbug.polymarketbot.repository.LeaderRepository
+import com.wrbug.polymarketbot.service.copytrading.statistics.CopyOrderTrackingService
+import com.wrbug.polymarketbot.util.fromJson
+import com.wrbug.polymarketbot.websocket.PolymarketWebSocketClient
+import jakarta.annotation.PreDestroy
+import kotlinx.coroutines.*
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.stereotype.Service
+import java.math.BigDecimal
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+ * Polymarket Activity WebSocket 监听服务
+ * 通过订阅全局 activity 交易流，客户端过滤 Leader 地址，实现实时交易检测
+ * 延迟 < 100ms，适合快速跟单场景
+ */
+@Service
+class PolymarketActivityWsService(
+    private val copyOrderTrackingService: CopyOrderTrackingService,
+    private val leaderRepository: LeaderRepository
+) {
+
+    private val logger = LoggerFactory.getLogger(PolymarketActivityWsService::class.java)
+
+    @Value("\${polymarket.websocket.activity.url:wss://ws-live-data.polymarket.com}")
+    private var websocketUrl: String = "wss://ws-live-data.polymarket.com"
+
+    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+
+    // 单例 WebSocket 客户端
+    private var wsClient: PolymarketWebSocketClient? = null
+
+    // 要监听的 Leader 地址集合（小写地址 -> leaderId）
+    private val monitoredAddresses = ConcurrentHashMap<String, Long>()
+
+    // 是否已订阅
+    @Volatile
+    private var isSubscribed = false
+
+    /**
+     * 启动监听
+     */
+    fun start(leaders: List<Leader>) {
+        monitoredAddresses.clear()
+        leaders.forEach { leader ->
+            val leaderId = leader.id
+            if (leaderId != null) {
+                monitoredAddresses[leader.leaderAddress.lowercase()] = leaderId
+            }
+        }
+
+        if (monitoredAddresses.isEmpty()) {
+            logger.info("没有需要监听的 Leader，停止 Activity WebSocket")
+            stop()
+            return
+        }
+
+        logger.info("启动 Activity WebSocket 监听，监控 ${monitoredAddresses.size} 个 Leader 地址")
+        connectAndSubscribe()
+    }
+
+    /**
+     * 添加 Leader
+     */
+    fun addLeader(leader: Leader) {
+        if (leader.id == null) {
+            logger.warn("Leader ID 为空，跳过: ${leader.leaderAddress}")
+            return
+        }
+
+        val address = leader.leaderAddress.lowercase()
+        val existingLeaderId = monitoredAddresses[address]
+
+        if (existingLeaderId != null && existingLeaderId == leader.id) {
+            logger.debug("Leader 已在监听列表中: ${leader.leaderName} (${address})")
+            return
+        }
+
+        val leaderId = leader.id
+        if (leaderId != null) {
+            monitoredAddresses[address] = leaderId
+            logger.info("添加 Leader 到 Activity WS 监听: ${leader.leaderName} (${address})")
+
+            // 如果 WebSocket 未连接，连接
+            val client = wsClient
+            if (client == null || !client.isConnected()) {
+                connectAndSubscribe()
+            }
+        }
+    }
+
+    /**
+     * 移除 Leader
+     */
+    fun removeLeader(leaderId: Long) {
+        val addressToRemove = monitoredAddresses.entries
+            .find { it.value == leaderId }?.key
+
+        if (addressToRemove != null) {
+            monitoredAddresses.remove(addressToRemove)
+            logger.info("从 Activity WS 监听移除 Leader: leaderId=$leaderId, address=$addressToRemove")
+        }
+
+        // 如果没有 Leader 了，停止监听
+        if (monitoredAddresses.isEmpty()) {
+            logger.info("没有 Leader 需要监听了，停止 Activity WebSocket")
+            stop()
+        }
+    }
+
+    /**
+     * 连接并订阅
+     */
+    private fun connectAndSubscribe() {
+        val existingClient = wsClient
+        if (existingClient != null && existingClient.isConnected()) {
+            // 如果已连接但未订阅，发送订阅
+            if (!isSubscribed) {
+                subscribeAllActivity()
+            }
+            return
+        }
+
+        logger.info("连接 Activity WebSocket: $websocketUrl")
+
+        val newClient = PolymarketWebSocketClient(
+            url = websocketUrl,
+            sessionId = "copy-trading-activity",
+            onMessage = { message -> handleMessage(message) },
+            onOpen = {
+                logger.info("Activity WebSocket 连接成功")
+                subscribeAllActivity()
+            },
+            onReconnect = {
+                logger.info("Activity WebSocket 重连成功，重新订阅")
+                subscribeAllActivity()
+            }
+        )
+
+        wsClient = newClient
+
+        scope.launch {
+            try {
+                newClient.connect()
+            } catch (e: Exception) {
+                logger.error("连接 Activity WebSocket 失败", e)
+            }
+        }
+    }
+
+    /**
+     * 订阅全局 activity
+     * 根据 @polymarket/real-time-data-client 的协议格式
+     * 使用 "action": "subscribe" 而不是 "type": "subscribe"
+     */
+    private fun subscribeAllActivity() {
+        val client = wsClient
+        if (client == null || !client.isConnected()) {
+            logger.warn("WebSocket 未连接，无法订阅")
+            return
+        }
+
+        try {
+            // 根据 real-time-data-client 的协议格式
+            // 订阅消息应包含 "action": "subscribe" 和 "subscriptions" 数组
+            val subscribeMessage = """
+            {
+                "action": "subscribe",
+                "subscriptions": [
+                    {
+                        "topic": "activity",
+                        "type": "trades"
+                    }
+                ]
+            }
+            """.trimIndent()
+
+            client.sendMessage(subscribeMessage)
+            isSubscribed = true
+            logger.info("Activity WebSocket 订阅成功（全局交易流）")
+        } catch (e: Exception) {
+            logger.error("订阅 Activity WebSocket 失败", e)
+            isSubscribed = false
+        }
+    }
+
+    /**
+     * 处理消息
+     */
+    private fun handleMessage(message: String) {
+        try {
+            // 处理 PONG 响应
+            if (message.trim() == "PONG" || message.trim() == "pong") {
+                return
+            }
+            
+            // 使用扩展函数解析消息
+            val tradeMessage = message.fromJson<ActivityTradeMessage>() ?: run {
+                // 不是有效的 JSON 或格式不匹配，跳过
+                logger.warn("无法解析为 ActivityTradeMessage，可能不是 activity 消息: ${message.take(200)}")
+                return
+            }
+            
+            // 检查是否是 activity trade 消息
+            if (tradeMessage.topic != "activity" || tradeMessage.type != "trades") {
+                // 不是我们关心的消息，直接返回
+                return
+            }
+            
+            val payload = tradeMessage.payload
+            
+            // 提取交易者地址
+            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
+            }
+
+            // 解析交易数据
+            val trade = parseActivityTrade(payload, leaderId)
+            if (trade != null) {
+                logger.info("✅ 检测到 Leader 交易: leaderId=$leaderId, address=$traderAddress, side=${trade.side}, market=${trade.market}, size=${trade.size}")
+
+                // 异步处理交易（避免阻塞消息处理）
+                scope.launch {
+                    try {
+                        copyOrderTrackingService.processTrade(
+                            leaderId = leaderId,
+                            trade = trade,
+                            source = "activity-ws"
+                        )
+                    } catch (e: Exception) {
+                        logger.error("处理 Activity WS 交易失败: leaderId=$leaderId, tradeId=${trade.id}", e)
+                    }
+                }
+            } else {
+                logger.warn("解析交易数据失败: leaderId=$leaderId, address=$traderAddress, asset=${payload.asset}, side=${payload.side}")
+            }
+        } catch (e: Exception) {
+            logger.error("处理 Activity WebSocket 消息失败: ${e.message}", e)
+        }
+    }
+
+    /**
+     * 提取交易者地址
+     * 优先检查 trader.address，fallback 到 proxyWallet
+     */
+    private fun extractTraderAddress(payload: ActivityTradePayload): String? {
+        // 优先检查 trader.address
+        val address = payload.trader?.address
+            ?: payload.proxyWallet
+
+        return address
+    }
+
+    /**
+     * 解析 Activity Trade 为 TradeResponse
+     */
+    private fun parseActivityTrade(payload: ActivityTradePayload, leaderId: Long): TradeResponse? {
+        return try {
+            // 提取必需字段并验证
+            val asset = payload.asset
+            val conditionId = payload.conditionId
+            val sideRaw = payload.side
+
+            if (asset.isBlank() || conditionId.isBlank() || sideRaw.isBlank()) {
+                logger.warn("Activity Trade 消息缺少必需字段: asset=$asset, conditionId=$conditionId, side=$sideRaw")
+                return null
+            }
+
+            val side = sideRaw.uppercase()
+
+            // 验证 side 必须是 BUY 或 SELL
+            if (side != "BUY" && side != "SELL") {
+                logger.warn("Activity Trade 消息 side 字段无效: side=$side")
+                return null
+            }
+
+            // price 和 size 可能是数字或字符串，统一转换为字符串
+            val price = convertToString(payload.price) ?: run {
+                logger.warn("Activity Trade 消息 price 字段无效: ${payload.price}")
+                return null
+            }
+
+            val size = convertToString(payload.size) ?: run {
+                logger.warn("Activity Trade 消息 size 字段无效: ${payload.size}")
+                return null
+            }
+
+            // 时间戳处理：可能是秒或毫秒，可能是数字或字符串
+            val timestamp = when {
+                payload.timestamp == null -> System.currentTimeMillis().toString()
+                payload.timestamp is Number -> {
+                    val ts = (payload.timestamp as Number).toLong()
+                    // 如果时间戳小于 1e12（秒级），转换为毫秒
+                    if (ts < 1e12) (ts * 1000).toString() else ts.toString()
+                }
+
+                payload.timestamp is String -> {
+                    val tsStr = payload.timestamp as String
+                    val ts = tsStr.toLongOrNull() ?: System.currentTimeMillis()
+                    if (ts < 1e12) (ts * 1000).toString() else tsStr
+                }
+
+                else -> System.currentTimeMillis().toString()
+            }
+
+            // 解析 outcome 和 outcomeIndex
+            // 优先使用消息中的 outcomeIndex，如果没有则从 outcome 字符串解析
+            val outcome = payload.outcome
+            val outcomeIndex = payload.outcomeIndex
+                ?: parseOutcomeIndex(outcome)
+
+            // 使用 transactionHash 作为 trade ID，如果没有则生成 fallback ID
+            val tradeId = payload.transactionHash ?: "${leaderId}_${System.currentTimeMillis()}_${asset.take(10)}"
+
+            TradeResponse(
+                id = tradeId,
+                market = conditionId,
+                side = side,
+                price = price,
+                size = size,
+                timestamp = timestamp,
+                user = null, // Activity WS 中不需要
+                outcomeIndex = outcomeIndex,
+                outcome = outcome
+            )
+        } catch (e: Exception) {
+            logger.error("解析 Activity Trade 失败: ${e.message}", e)
+            null
+        }
+    }
+
+    /**
+     * 将 Any 类型的值转换为 String
+     * 支持 Number、String、BigDecimal 等类型
+     */
+    private fun convertToString(value: Any?): String? {
+        if (value == null) return null
+
+        return when (value) {
+            is String -> value
+            is Number -> {
+                // 如果是数字，转换为 BigDecimal 再转为字符串（保持精度）
+                try {
+                    BigDecimal(value.toString()).toPlainString()
+                } catch (e: Exception) {
+                    value.toString()
+                }
+            }
+
+            is BigDecimal -> value.toPlainString()
+            else -> value.toString()
+        }
+    }
+
+    /**
+     * 解析 outcome 为 outcomeIndex
+     */
+    private fun parseOutcomeIndex(outcome: String?): Int? {
+        return when (outcome?.uppercase()) {
+            "YES", "UP", "TRUE" -> 0
+            "NO", "DOWN", "FALSE" -> 1
+            else -> null
+        }
+    }
+
+    /**
+     * 停止监听
+     */
+    fun stop() {
+        logger.info("停止 Activity WebSocket 监听")
+        wsClient?.closeConnection()
+        wsClient = null
+        isSubscribed = false
+        monitoredAddresses.clear()
+    }
+
+    /**
+     * 获取连接状态
+     */
+    fun isConnected(): Boolean {
+        return wsClient?.isConnected() ?: false
+    }
+
+    /**
+     * 获取监听的 Leader 数量
+     */
+    fun getMonitoredCount(): Int {
+        return monitoredAddresses.size
+    }
+
+    @PreDestroy
+    fun destroy() {
+        stop()
+        scope.cancel()
+    }
+}
+
@@ -124,13 +124,14 @@ open class CopyOrderTrackingService(
    suspend fun processTrade(leaderId: Long, trade: TradeResponse, source: String): Result<Unit> {
        // 获取该交易的 Mutex（按交易ID锁定，不同交易可以并行处理）
        val mutex = getMutex(leaderId, trade.id)
-
+        logger.debug("processTrade: ${trade.id}, $source")
        return mutex.withLock {
            try {
                // 1. 检查是否已处理（去重，包括失败状态）
                val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id)

                if (existingProcessed != null) {
+                    logger.debug("processTrade: 重复 ${trade.id}, $source")
                    if (existingProcessed.status == "FAILED") {
                        return@withLock Result.success(Unit)
                    }
@@ -269,8 +270,9 @@ open class CopyOrderTrackingService(
                    // 如果启用了关键字过滤或市场截止时间过滤，需要先获取市场信息
                    var marketTitle: String? = null
                    var marketEndDate: Long? = null
-                    val needMarketInfo = copyTrading.keywordFilterMode != "DISABLED" || copyTrading.maxMarketEndDate != null
-                    
+                    val needMarketInfo =
+                        copyTrading.keywordFilterMode != "DISABLED" || copyTrading.maxMarketEndDate != null
+
                    if (needMarketInfo) {
                        try {
                            val market = marketService.getMarket(trade.market)
@@ -1,34 +1,154 @@
 package com.wrbug.polymarketbot.util

 import com.google.gson.Gson
+import com.google.gson.GsonBuilder
+import com.google.gson.JsonElement
 import com.google.gson.reflect.TypeToken
+import jakarta.annotation.PostConstruct
 import org.springframework.stereotype.Component

 /**
 * JSON 工具类
- * 用于解析 JSON 字符串
+ * 用于初始化全局 Gson 实例，供扩展函数使用
+ *
+ * 使用 Spring 注入的 Gson Bean（已在 GsonConfig 中配置为 lenient 模式）
+ * 初始化后设置全局 gson 实例，供扩展函数使用
 */
@Component
 class JsonUtils(
-    private val gson: Gson
+    private val injectedGson: Gson
 ) {
-    
+
+    @PostConstruct
+    fun init() {
+        // 设置全局 Gson 实例，供扩展函数使用
+        _gson = injectedGson
+    }
+
    /**
     * 解析 JSON 字符串数组
     * @param jsonString JSON 字符串，如 "[\"Yes\", \"No\"]"
     * @return 字符串列表，如果解析失败返回空列表
     */
    fun parseStringArray(jsonString: String?): List<String> {
-        if (jsonString.isNullOrBlank()) {
-            return emptyList()
-        }
-        
-        return try {
-            val listType = object : TypeToken<List<String>>() {}.type
-            gson.fromJson<List<String>>(jsonString, listType) ?: emptyList()
+        return jsonString?.parseStringArray() ?: emptyList()
+    }
+}
+
+/**
+ * 全局 Gson 实例（用于扩展函数）
+ * 在 JsonUtils 初始化时设置
+ */
+@Volatile
+private var _gson: Gson? = null
+
+/**
+ * 获取全局 Gson 实例
+ * 如果尚未初始化，创建默认实例
+ */
+val gson: Gson
+    get() = _gson ?: GsonBuilder().setLenient().create().also { _gson = it }
+
+// ============================================================================
+// 扩展函数：JSON 序列化和反序列化
+// ============================================================================
+
+/**
+ * 将对象转换为 JSON 字符串
+ *
+ * @return JSON 字符串，如果对象为 null 则返回空字符串
+ *
+ * @example
+ * ```kotlin
+ * val obj = MyDataClass(name = "test", value = 123)
+ * val json = obj.toJson()  // {"name":"test","value":123}
+ * ```
+ */
+fun Any?.toJson(): String {
+    return if (this == null) {
+        ""
+    } else {
+        try {
+            gson.toJson(this)
        } catch (e: Exception) {
-            emptyList()
+            ""
        }
    }
 }

+/**
+ * 将 JSON 字符串解析为对象（支持泛型）
+ *
+ * @return 解析后的对象，如果解析失败返回 null
+ *
+ * @example
+ * ```kotlin
+ * val json = "{\"name\":\"test\",\"value\":123}"
+ * val obj = json.fromJson<MyDataClass>()  // MyDataClass(name="test", value=123)
+ *
+ * // 对于泛型类型，自动使用 TypeToken
+ * val json = "[\"a\", \"b\", \"c\"]"
+ * val list = json.fromJson<List<String>>()  // ["a", "b", "c"]
+ * ```
+ */
+inline fun <reified T> String?.fromJson(): T? {
+    if (this.isNullOrBlank()) {
+        return null
+    }
+
+    return try {
+        val typeToken = object : TypeToken<T>() {}
+        gson.fromJson(this, typeToken.type)
+    } catch (e: Exception) {
+        null
+    }
+}
+
+/**
+ * 将 JsonElement 解析为对象（支持泛型）
+ *
+ * @return 解析后的对象，如果解析失败返回 null
+ *
+ * @example
+ * ```kotlin
+ * val jsonElement: JsonElement = ...
+ * val obj = jsonElement.fromJson<MyDataClass>()
+ * ```
+ */
+inline fun <reified T> JsonElement?.fromJson(): T? {
+    if (this == null) {
+        return null
+    }
+
+    return try {
+        val typeToken = object : TypeToken<T>() {}
+        gson.fromJson(this, typeToken.type)
+    } catch (e: Exception) {
+        null
+    }
+}
+
+/**
+ * 解析 JSON 字符串数组
+ *
+ * @return 字符串列表，如果解析失败返回空列表
+ *
+ * @example
+ * ```kotlin
+ * val json = "[\"Yes\", \"No\"]"
+ * val list = json.parseStringArray()  // ["Yes", "No"]
+ * ```
+ */
+fun String?.parseStringArray(): List<String> {
+    if (this.isNullOrBlank()) {
+        return emptyList()
+    }
+
+    return try {
+        val listType = object : TypeToken<List<String>>() {}.type
+        gson.fromJson<List<String>>(this, listType) ?: emptyList()
+    } catch (e: Exception) {
+        emptyList()
+    }
+}
+
@@ -139,7 +139,8 @@ class PolymarketWebSocketClient(
    
    /**
     * 启动 PING 保活机制
-     * 根据官方文档，每 10 秒发送一次 "PING"
+     * 根据 @polymarket/real-time-data-client，发送小写 "ping"
+     * 每 10 秒发送一次
     */
    private fun startPing() {
        stopPing()  // 先停止之前的 PING 任务
@@ -149,7 +150,7 @@ class PolymarketWebSocketClient(
                delay(10000)  // 10 秒
                if (isConnected) {
                    try {
-                        sendMessage("PING")
+                        sendMessage("ping")  // 使用小写，与官方客户端保持一致
                    } catch (e: Exception) {
                        logger.warn("发送 PING 失败: $sessionId, ${e.message}")
                        break
@@ -37,6 +37,8 @@ logging.pattern.console=%d{yyyy-MM-dd HH:mm:ss} - %msg%n
 # Polymarket API 配置
 polymarket.clob.base-url=https://clob.polymarket.com
 polymarket.rtds.ws-url=wss://ws-subscriptions-clob.polymarket.com
+polymarket.websocket.url=wss://ws-live-data.polymarket.com
+polymarket.websocket.activity.url=${POLYMARKET_WEBSOCKET_ACTIVITY_URL:wss://ws-live-data.polymarket.com}
 polymarket.data-api.base-url=https://data-api.polymarket.com
 polymarket.gamma.base-url=https://gamma-api.polymarket.com

@@ -52,6 +54,10 @@ copy.trading.polling.interval=${COPY_TRADING_POLLING_INTERVAL:2000}
 # 是否启用轮询，默认true
 copy.trading.polling.enabled=${COPY_TRADING_POLLING_ENABLED:true}

+# 跟单监听配置
+# 链上 WebSocket 重连延迟（毫秒），默认3秒
+copy.trading.onchain.ws.reconnect.delay=${COPY_TRADING_ONCHAIN_WS_RECONNECT_DELAY:3000}
+
 # WebSocket 配置
 websocket.heartbeat-timeout=${WEBSOCKET_HEARTBEAT_TIMEOUT:60000}