From c4ac3a8fe04e97f2bde4da10eca208f1ea1b839f Mon Sep 17 00:00:00 2001 From: WrBug Date: Fri, 26 Dec 2025 01:36:00 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E9=93=BE=E4=B8=8A=20?= =?UTF-8?q?WebSocket=20=E7=9B=91=E5=90=AC=E5=92=8C=E5=B9=B6=E8=A1=8C?= =?UTF-8?q?=E7=9B=91=E6=8E=A7=E7=AD=96=E7=95=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 创建 OnChainWsService，支持通过 Polygon RPC eth_subscribe 实时监听链上交易 - 实现并行监控策略：链上 WS 和轮询同时运行，哪个数据先返回用哪个 - 支持通过 eth_unsubscribe 取消单个 Leader 的订阅 - 在 CopyOrderTrackingService 中添加 Mutex 保证线程安全 - 实现链上交易解析逻辑（USDC Transfer + ERC1155 Transfer） - 实现 Gamma API 元数据补齐逻辑 - 优化 WebSocket 连接管理：只创建一个连接，没有跟单配置时自动取消 - 跟单配置生效/失效时及时更新 WebSocket 订阅 - 使用 Gson 替换所有 JSON 解析，JsonRpcResponse.result 使用 JsonElement 类型 - 添加相关文档：copy-trading-logic-summary.md 和 copy-trading-monitor-strategy.md --- .../wrbug/polymarketbot/api/EthereumRpcApi.kt | 4 +- .../service/common/BlockchainService.kt | 38 +- .../copytrading/configs/CopyTradingService.kt | 19 +- .../monitor/CopyTradingMonitorService.kt | 52 +- .../copytrading/monitor/OnChainWsService.kt | 925 ++++++++++++++++++ .../statistics/CopyOrderTrackingService.kt | 186 ++-- .../service/system/RelayClientService.kt | 8 +- .../service/system/RpcNodeService.kt | 2 +- docs/zh/copy-trading-logic-summary.md | 289 ++++++ docs/zh/copy-trading-monitor-strategy.md | 653 +++++++++++++ 10 files changed, 2056 insertions(+), 120 deletions(-) create mode 100644 backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/OnChainWsService.kt create mode 100644 docs/zh/copy-trading-logic-summary.md create mode 100644 docs/zh/copy-trading-monitor-strategy.md diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/api/EthereumRpcApi.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/api/EthereumRpcApi.kt index 34385fc..0a5be45 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/api/EthereumRpcApi.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/api/EthereumRpcApi.kt @@ -1,5 +1,6 @@ package com.wrbug.polymarketbot.api +import com.google.gson.JsonElement import retrofit2.Response import retrofit2.http.Body import retrofit2.http.POST @@ -29,10 +30,11 @@ data class JsonRpcRequest( /** * JSON-RPC 响应 + * 使用 JsonElement 类型处理 result 字段，可以灵活处理字符串、对象、数组等类型 */ data class JsonRpcResponse( val jsonrpc: String? = null, - val result: String? = null, + val result: JsonElement? = null, // 使用 JsonElement 类型，可以处理任意 JSON 类型 val error: JsonRpcError? = null, val id: Int? = null ) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/BlockchainService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/BlockchainService.kt index 1db0eec..58312ae 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/BlockchainService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/BlockchainService.kt @@ -117,7 +117,9 @@ class BlockchainService( throw Exception("RPC 错误: ${rpcResponse.error.message}") } - val hexResult = rpcResponse.result ?: throw Exception("RPC 响应格式错误: result 为空") + // 使用 Gson 解析 result（JsonElement） + val hexResult = rpcResponse.result?.asString + ?: throw Exception("RPC 响应格式错误: result 为空") // 解析代理地址 val proxyAddress = EthereumUtils.decodeAddress(hexResult) @@ -193,7 +195,9 @@ class BlockchainService( throw Exception("RPC 错误: ${rpcResponse.error.message}") } - val hexBalance = rpcResponse.result ?: throw Exception("RPC 响应格式错误: result 为空") + // 使用 Gson 解析 result（JsonElement） + val hexBalance = rpcResponse.result?.asString + ?: throw Exception("RPC 响应格式错误: result 为空") // 将十六进制转换为 BigDecimal（USDC 有 6 位小数） val balanceWei = BigInteger(hexBalance.removePrefix("0x"), 16) @@ -288,7 +292,9 @@ class BlockchainService( return Result.failure(Exception("调用 getCollectionId 失败: ${collectionIdResult.error}")) } - val collectionId = collectionIdResult.result ?: return Result.failure(Exception("getCollectionId 返回结果为空")) + // 使用 Gson 解析 result（JsonElement） + val collectionId = collectionIdResult.result?.asString + ?: return Result.failure(Exception("getCollectionId 返回结果为空")) // 2. 调用 getPositionId(collateralToken, collectionId) val getPositionIdSelector = EthereumUtils.getFunctionSelector("getPositionId(address,bytes32)") @@ -318,7 +324,9 @@ class BlockchainService( return Result.failure(Exception("调用 getPositionId 失败: ${positionIdResult.error}")) } - val tokenId = positionIdResult.result ?: return Result.failure(Exception("getPositionId 返回结果为空")) + // 使用 Gson 解析 result（JsonElement） + val tokenId = positionIdResult.result?.asString + ?: return Result.failure(Exception("getPositionId 返回结果为空")) val tokenIdBigInt = EthereumUtils.decodeUint256(tokenId) Result.success(tokenIdBigInt.toString()) @@ -454,7 +462,9 @@ class BlockchainService( return Result.failure(Exception("获取 Proxy nonce 失败: ${rpcResponse.error.message}")) } - val hexNonce = rpcResponse.result ?: return Result.failure(Exception("Proxy nonce 结果为空")) + // 使用 Gson 解析 result（JsonElement） + val hexNonce = rpcResponse.result?.asString + ?: return Result.failure(Exception("Proxy nonce 结果为空")) val nonce = EthereumUtils.decodeUint256(hexNonce) return Result.success(nonce) } @@ -482,7 +492,9 @@ class BlockchainService( return Result.failure(Exception("获取 nonce 失败: ${rpcResponse.error.message}")) } - val hexNonce = rpcResponse.result ?: return Result.failure(Exception("nonce 结果为空")) + // 使用 Gson 解析 result（JsonElement） + val hexNonce = rpcResponse.result?.asString + ?: return Result.failure(Exception("nonce 结果为空")) val nonce = EthereumUtils.decodeUint256(hexNonce) return Result.success(nonce) } @@ -508,7 +520,9 @@ class BlockchainService( return Result.failure(Exception("获取 gas price 失败: ${rpcResponse.error.message}")) } - val hexGasPrice = rpcResponse.result ?: return Result.failure(Exception("gas price 结果为空")) + // 使用 Gson 解析 result（JsonElement） + val hexGasPrice = rpcResponse.result?.asString + ?: return Result.failure(Exception("gas price 结果为空")) val gasPrice = EthereumUtils.decodeUint256(hexGasPrice) return Result.success(gasPrice) } @@ -582,7 +596,9 @@ class BlockchainService( return Result.failure(Exception("发送交易失败: ${rpcResponse.error.message}")) } - val txHash = rpcResponse.result ?: return Result.failure(Exception("交易哈希为空")) + // 使用 Gson 解析 result（JsonElement） + val txHash = rpcResponse.result?.asString + ?: return Result.failure(Exception("交易哈希为空")) return Result.success(txHash) } @@ -611,7 +627,9 @@ class BlockchainService( return Result.failure(Exception("查询交易失败: ${txRpcResponse.error.message}")) } - val txResult = txRpcResponse.result ?: return Result.failure(Exception("交易结果为空")) + // 使用 Gson 解析 result（JsonElement） + val txResult = txRpcResponse.result?.toString() + ?: return Result.failure(Exception("交易结果为空")) // 查询交易回执（包含内部调用和事件日志） val receiptRequest = JsonRpcRequest( @@ -628,7 +646,7 @@ class BlockchainService( val receiptResult = if (receiptRpcResponse.error != null) { "交易回执查询失败: ${receiptRpcResponse.error.message}" } else { - receiptRpcResponse.result ?: "交易回执为空（可能还在打包中）" + receiptRpcResponse.result?.toString() ?: "交易回执为空（可能还在打包中）" } Result.success("交易信息:\n$txResult\n\n交易回执:\n$receiptResult") diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/configs/CopyTradingService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/configs/CopyTradingService.kt index f337e5f..9fecc17 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/configs/CopyTradingService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/configs/CopyTradingService.kt @@ -151,13 +151,13 @@ class CopyTradingService( val saved = copyTradingRepository.save(copyTrading) - // 如果跟单已启用，重新启动监听（确保状态完全同步） + // 如果跟单已启用，更新 Leader 监听（增量更新，不重启所有监听） if (saved.enabled) { kotlinx.coroutines.runBlocking { try { - monitorService.restartMonitoring() + monitorService.updateLeaderMonitoring(saved.leaderId) } catch (e: Exception) { - logger.error("重新启动跟单监听失败", e) + logger.error("更新 Leader 监听失败", e) } } } @@ -219,12 +219,12 @@ class CopyTradingService( val saved = copyTradingRepository.save(updated) - // 重新启动监听（确保状态完全同步） + // 更新 Leader 监听（增量更新，根据 enabled 状态决定添加或移除） kotlinx.coroutines.runBlocking { try { - monitorService.restartMonitoring() + monitorService.updateLeaderMonitoring(saved.leaderId) } catch (e: Exception) { - logger.error("重新启动跟单监听失败", e) + logger.error("更新 Leader 监听失败", e) } } @@ -322,14 +322,15 @@ class CopyTradingService( val copyTrading = copyTradingRepository.findById(copyTradingId).orElse(null) ?: return Result.failure(IllegalArgumentException("跟单配置不存在")) + val leaderId = copyTrading.leaderId copyTradingRepository.delete(copyTrading) - // 重新启动监听（确保状态完全同步） + // 更新 Leader 监听（检查该 Leader 是否还有其他启用的跟单配置） kotlinx.coroutines.runBlocking { try { - monitorService.restartMonitoring() + monitorService.removeLeaderMonitoring(leaderId) } catch (e: Exception) { - logger.error("重新启动跟单监听失败", e) + logger.error("更新 Leader 监听失败", e) } } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingMonitorService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingMonitorService.kt index b622e16..937fef0 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingMonitorService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingMonitorService.kt @@ -12,14 +12,15 @@ import org.springframework.stereotype.Service /** * 跟单监听服务（主服务） - * 管理所有Leader的交易监听（使用轮询方式） - * 注意：WebSocket 需要认证才能订阅其他用户的交易，因此只使用轮询方式 + * 管理所有Leader的交易监听 + * 同时运行链上 WebSocket 监听和轮询监听（并行处理） */ @Service class CopyTradingMonitorService( private val copyTradingRepository: CopyTradingRepository, private val leaderRepository: LeaderRepository, - private val pollingService: CopyTradingPollingService + private val pollingService: CopyTradingPollingService, + private val onChainWsService: OnChainWsService ) { private val logger = LoggerFactory.getLogger(CopyTradingMonitorService::class.java) @@ -46,12 +47,14 @@ class CopyTradingMonitorService( @PreDestroy fun destroy() { scope.cancel() - // 只使用轮询，不使用WebSocket + // 停止轮询和链上 WS 监听 pollingService.stop() + onChainWsService.stop() } /** * 启动监听 + * 同时启动链上 WebSocket 监听和轮询监听（并行运行） */ suspend fun startMonitoring() { // 1. 获取所有启用的跟单关系 @@ -67,14 +70,17 @@ class CopyTradingMonitorService( leaderRepository.findById(leaderId).orElse(null) } + // 3. 同时启动链上 WebSocket 监听和轮询监听（并行运行） + // 链上 WS 监听（实时，秒级延迟） + onChainWsService.start(leaders) - // 3. 启动轮询监听（使用 /activity 接口，不需要认证） - // 注意：WebSocket 需要认证才能订阅其他用户的交易，因此禁用WebSocket，只使用轮询 + // 轮询监听（延迟，2秒间隔，作为备份） pollingService.start(leaders) } /** * 添加Leader监听（当创建新的跟单关系时调用） + * 如果 Leader 已经在监听列表中，不重复添加 */ suspend fun addLeaderMonitoring(leaderId: Long) { val leader = leaderRepository.findById(leaderId).orElse(null) @@ -85,28 +91,54 @@ class CopyTradingMonitorService( return } - // 只使用轮询，不使用WebSocket（需要认证） + // 同时添加到链上 WS 监听和轮询监听（如果不在列表中才添加） + onChainWsService.addLeader(leader) pollingService.addLeader(leader) } /** - * 移除Leader监听（当删除跟单关系时调用） + * 移除Leader监听（当删除跟单关系或禁用时调用） + * 检查该 Leader 是否还有其他启用的跟单配置 */ suspend fun removeLeaderMonitoring(leaderId: Long) { val copyTradings = copyTradingRepository.findByLeaderIdAndEnabledTrue(leaderId) + // 如果还有启用的跟单配置，不移除监听 if (copyTradings.isNotEmpty()) { return } - // 只使用轮询，不使用WebSocket + // 没有启用的跟单配置了，移除监听 + onChainWsService.removeLeader(leaderId) pollingService.removeLeader(leaderId) } + /** + * 更新Leader监听（当跟单配置状态改变时调用） + * 根据当前状态决定添加或移除监听 + */ + suspend fun updateLeaderMonitoring(leaderId: Long) { + val copyTradings = copyTradingRepository.findByLeaderIdAndEnabledTrue(leaderId) + val leader = leaderRepository.findById(leaderId).orElse(null) + ?: return + + if (copyTradings.isNotEmpty()) { + // 有启用的跟单配置，确保在监听列表中 + onChainWsService.addLeader(leader) + pollingService.addLeader(leader) + } else { + // 没有启用的跟单配置，移除监听 + onChainWsService.removeLeader(leaderId) + pollingService.removeLeader(leaderId) + } + } + /** * 重新启动监听（当跟单关系状态改变时调用） + * 注意：这个方法会停止所有监听并重新启动，建议使用 updateLeaderMonitoring 进行增量更新 */ suspend fun restartMonitoring() { - // 只使用轮询，不使用WebSocket + // 停止所有监听 + onChainWsService.stop() pollingService.stop() delay(1000) // 等待1秒 startMonitoring() diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/OnChainWsService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/OnChainWsService.kt new file mode 100644 index 0000000..18145cd --- /dev/null +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/OnChainWsService.kt @@ -0,0 +1,925 @@ +package com.wrbug.polymarketbot.service.copytrading.monitor + +import com.google.gson.Gson +import com.google.gson.JsonArray +import com.google.gson.JsonObject +import com.google.gson.reflect.TypeToken +import com.wrbug.polymarketbot.api.* +import com.wrbug.polymarketbot.entity.Leader +import com.wrbug.polymarketbot.repository.LeaderRepository +import com.wrbug.polymarketbot.service.copytrading.statistics.CopyOrderTrackingService +import com.wrbug.polymarketbot.service.system.RpcNodeService +import com.wrbug.polymarketbot.util.RetrofitFactory +import com.wrbug.polymarketbot.util.createClient +import com.wrbug.polymarketbot.util.getProxyConfig +import jakarta.annotation.PreDestroy +import kotlinx.coroutines.* +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.WebSocket +import okhttp3.WebSocketListener +import okio.ByteString +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Value +import org.springframework.stereotype.Service +import java.math.BigInteger +import java.util.concurrent.ConcurrentHashMap + +/** + * 链上 WebSocket 监听服务 + * 通过 Polygon RPC 的 eth_subscribe 实时监听链上交易 + */ +@Service +class OnChainWsService( + private val rpcNodeService: RpcNodeService, + private val retrofitFactory: RetrofitFactory, + private val copyOrderTrackingService: CopyOrderTrackingService, + private val leaderRepository: LeaderRepository +) { + + private val logger = LoggerFactory.getLogger(OnChainWsService::class.java) + + // Gson 实例，用于解析 JSON + private val gson = Gson() + + @Value("\${copy.trading.onchain.ws.reconnect.delay:3000}") + private var reconnectDelay: Long = 3000 // 重连延迟（毫秒），默认3秒 + + private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + + // 存储需要监听的Leader：leaderId -> Leader + private val monitoredLeaders = ConcurrentHashMap() + + // 存储每个 Leader 的订阅 ID：leaderId -> List + // 每个 Leader 有 6 个订阅：USDC from/to, ERC1155 TransferSingle from/to, ERC1155 TransferBatch from/to + private val leaderSubscriptions = ConcurrentHashMap>() + + // 存储请求 ID 到 Leader ID 的映射：requestId -> leaderId + // 用于在收到订阅响应时，将 subscription ID 关联到对应的 Leader + private val requestIdToLeaderId = ConcurrentHashMap() + + // WebSocket 连接 + private var webSocket: WebSocket? = null + @Volatile + private var isConnected = false + + // 订阅ID计数器（用于请求 ID） + private var requestIdCounter = 0 + + // 合约地址 + companion object { + private const val USDC_CONTRACT = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174" + private const val ERC1155_CONTRACT = "0x4d97dcd97ec945f40cf65f87097ace5ea0476045" + private const val ERC20_TRANSFER_TOPIC = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" + private const val ERC1155_TRANSFER_SINGLE_TOPIC = "0xc3d58168c5ae7397731d063d5bbf3d657854427343f4c083240f7aacaa2d0f62" + private const val ERC1155_TRANSFER_BATCH_TOPIC = "0x4a39dc06d4c0dbc64b70af90fd698a233a518aa5d07e595d983b8c0526c8f7fb" + } + + // 连接任务（确保只有一个连接任务在运行） + private var connectionJob: Job? = null + + /** + * 启动链上 WebSocket 监听 + * 只创建一个 WebSocket 连接，为所有 Leader 订阅 + */ + fun start(leaders: List) { + // 如果没有 Leader，不启动连接 + if (leaders.isEmpty()) { + logger.info("没有需要监听的 Leader，不启动链上 WebSocket 连接") + stop() + return + } + + // 如果连接任务已经在运行，先停止旧任务 + if (connectionJob != null && connectionJob!!.isActive) { + logger.info("停止旧的连接任务，准备重新启动") + connectionJob?.cancel() + connectionJob = null + // 关闭旧连接 + webSocket?.close(1000, "重新启动") + webSocket = null + isConnected = false + } + + // 更新 Leader 列表 + monitoredLeaders.clear() + leaders.forEach { leader -> + addLeader(leader) + } + + // 启动连接任务（只创建一个） + connectionJob = scope.launch { + startConnection() + } + } + + /** + * 添加Leader监听 + * 如果 Leader 已经在监听列表中，不重复添加 + * 如果已连接，立即订阅 + */ + fun addLeader(leader: Leader) { + if (leader.id == null) { + logger.warn("Leader ID为空，跳过: ${leader.leaderAddress}") + return + } + + val leaderId = leader.id!! + + // 如果已经在监听列表中，不重复添加 + if (monitoredLeaders.containsKey(leaderId)) { + logger.debug("Leader 已在监听列表中: ${leader.leaderName} (${leader.leaderAddress})") + return + } + + monitoredLeaders[leaderId] = leader + logger.info("添加 Leader 监听: ${leader.leaderName} (${leader.leaderAddress})") + + // 如果已连接，立即订阅 + if (isConnected && webSocket != null) { + scope.launch { + subscribeLeader(leader) + } + } else { + // 如果未连接，启动连接（如果连接任务未运行） + if (connectionJob == null || !connectionJob!!.isActive) { + connectionJob = scope.launch { + startConnection() + } + } + } + } + + /** + * 移除Leader监听 + * 通过 eth_unsubscribe 取消该 Leader 的所有订阅 + * 如果没有 Leader 了，关闭 WebSocket 连接 + */ + fun removeLeader(leaderId: Long) { + val leader = monitoredLeaders.remove(leaderId) + if (leader != null) { + logger.info("移除 Leader 监听: ${leader.leaderName} (${leader.leaderAddress})") + } + + // 取消该 Leader 的所有订阅 + val subscriptions = leaderSubscriptions.remove(leaderId) + if (subscriptions != null && subscriptions.isNotEmpty() && isConnected && webSocket != null) { + logger.info("取消 Leader ${leader?.leaderName} 的 ${subscriptions.size} 个订阅") + subscriptions.forEach { subscriptionId -> + unsubscribe(subscriptionId) + } + } + + // 如果没有 Leader 了，关闭连接 + if (monitoredLeaders.isEmpty()) { + logger.info("没有需要监听的 Leader，关闭链上 WebSocket 连接") + stop() + } + } + + /** + * 停止所有监听 + */ + fun stop() { + connectionJob?.cancel() + connectionJob = null + + // 取消所有订阅 + if (isConnected && webSocket != null) { + leaderSubscriptions.values.flatten().forEach { subscriptionId -> + unsubscribe(subscriptionId) + } + } + + webSocket?.close(1000, "正常关闭") + webSocket = null + isConnected = false + monitoredLeaders.clear() + leaderSubscriptions.clear() + requestIdToLeaderId.clear() + } + + /** + * 启动连接（带重连机制） + * 只创建一个 WebSocket 连接 + */ + private suspend fun startConnection() { + while (scope.isActive) { + try { + // 检查是否有需要监听的 Leader + if (monitoredLeaders.isEmpty()) { + logger.info("没有需要监听的 Leader，停止连接") + // 确保关闭连接 + webSocket?.close(1000, "没有 Leader") + webSocket = null + isConnected = false + break + } + + // 如果已经连接，不需要重新连接 + if (isConnected && webSocket != null) { + // 等待连接断开 + waitForDisconnect() + // 连接断开后继续重连循环 + continue + } + + // 从后台配置获取 WS RPC URL + val wsUrl = rpcNodeService.getWsUrl() + val httpUrl = rpcNodeService.getHttpUrl() + + logger.info("连接链上 WebSocket: $wsUrl (监听 ${monitoredLeaders.size} 个 Leader)") + + // 创建 HTTP 客户端（用于 RPC 调用） + val httpClient = createHttpClient() + + // 创建 RPC API 客户端 + val rpcApi = retrofitFactory.createEthereumRpcApi(httpUrl) + + // 连接 WebSocket（只创建一个连接，会先关闭旧连接） + connectWebSocket(wsUrl, httpClient, rpcApi) + + // 等待连接建立（最多等待 15 秒） + // 注意：onOpen 回调是异步的，需要等待一段时间 + var waitCount = 0 + val maxWait = 15 // 最多等待 15 秒 + while (!isConnected && waitCount < maxWait && scope.isActive) { + delay(1000) + waitCount++ + // 每 3 秒打印一次日志，方便调试 + if (waitCount % 3 == 0) { + logger.debug("等待 WebSocket 连接建立... (${waitCount}/${maxWait}秒)") + } + } + + // 检查连接状态（同时检查 isConnected 和 webSocket 状态） + val actuallyConnected = isConnected && webSocket != null + + // 如果连接失败，等待重连延迟后继续 + if (!actuallyConnected) { + logger.warn("WebSocket 连接超时或失败: isConnected=$isConnected, webSocket=${webSocket != null}, 等待重连") + delay(reconnectDelay) + continue + } + + logger.info("WebSocket 连接已建立，开始监听") + + // 连接成功后持续监听 + waitForDisconnect() + + // 连接断开后，如果没有 Leader 了，不再重连 + if (monitoredLeaders.isEmpty()) { + logger.info("没有需要监听的 Leader，停止重连") + break + } + + // 连接断开后，等待一下再重连（避免立即重连） + logger.info("WebSocket 连接断开，等待 ${reconnectDelay}ms 后重连") + delay(reconnectDelay) + } catch (e: Exception) { + // 如果没有 Leader 了，不再重连 + if (monitoredLeaders.isEmpty()) { + logger.info("没有需要监听的 Leader，停止重连") + // 确保关闭连接 + webSocket?.close(1000, "没有 Leader") + webSocket = null + isConnected = false + break + } + logger.warn("链上 WebSocket 连接失败，等待重连: ${e.message}") + // 确保关闭旧连接 + webSocket?.close(1000, "重连前关闭") + webSocket = null + isConnected = false + delay(reconnectDelay) + } + } + } + + /** + * 创建 HTTP 客户端 + */ + private fun createHttpClient(): OkHttpClient { + val proxy = getProxyConfig() + val builder = createClient() + + if (proxy != null) { + builder.proxy(proxy) + } + + return builder.build() + } + + /** + * 连接 WebSocket + * 确保只创建一个连接，创建新连接前先关闭旧连接 + */ + private fun connectWebSocket(wsUrl: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) { + // 先关闭旧连接（如果存在） + val oldWebSocket = webSocket + if (oldWebSocket != null) { + try { + oldWebSocket.close(1000, "重新连接") + } catch (e: Exception) { + logger.debug("关闭旧 WebSocket 连接时出错: ${e.message}") + } + } + webSocket = null + isConnected = false + + val request = Request.Builder() + .url(wsUrl) + .build() + + // 创建新连接（只创建一个） + webSocket = httpClient.newWebSocket(request, object : WebSocketListener() { + override fun onOpen(webSocket: WebSocket, response: okhttp3.Response) { + isConnected = true + logger.info("链上 WebSocket 连接成功") + + // 订阅所有 Leader + scope.launch { + monitoredLeaders.values.forEach { leader -> + subscribeLeader(leader) + } + } + } + + override fun onMessage(webSocket: WebSocket, text: String) { + scope.launch { + handleMessage(text, httpClient, rpcApi) + } + } + + override fun onMessage(webSocket: WebSocket, bytes: ByteString) { + scope.launch { + handleMessage(bytes.utf8(), httpClient, rpcApi) + } + } + + override fun onClosing(webSocket: WebSocket, code: Int, reason: String) { + isConnected = false + logger.warn("链上 WebSocket 连接关闭: code=$code, reason=$reason") + } + + override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { + isConnected = false + logger.warn("链上 WebSocket 连接已关闭: code=$code, reason=$reason") + } + + override fun onFailure(webSocket: WebSocket, t: Throwable, response: okhttp3.Response?) { + logger.error("链上 WebSocket 连接失败: ${t.message}", t) + isConnected = false + // 注意：这里不直接重连，由 startConnection 循环处理重连逻辑 + } + }) + } + + /** + * 等待连接断开 + */ + private suspend fun waitForDisconnect() { + while (isConnected && scope.isActive) { + delay(1000) + } + } + + /** + * 订阅 Leader 钱包地址 + * 每个 Leader 有 6 个订阅，保存订阅 ID 以便后续取消 + */ + private suspend fun subscribeLeader(leader: Leader) { + if (webSocket == null || !isConnected || leader.id == null) { + return + } + + val walletAddress = leader.leaderAddress.lowercase() + val walletTopic = addressToTopic32(walletAddress) + val leaderId = leader.id!! + + // 初始化该 Leader 的订阅列表 + if (!leaderSubscriptions.containsKey(leaderId)) { + leaderSubscriptions[leaderId] = mutableListOf() + } + + try { + // 订阅 USDC Transfer (from wallet) + subscribeLogs(USDC_CONTRACT, listOf(ERC20_TRANSFER_TOPIC, walletTopic), leaderId) + + // 订阅 USDC Transfer (to wallet) + subscribeLogs(USDC_CONTRACT, listOf(ERC20_TRANSFER_TOPIC, null, walletTopic), leaderId) + + // 订阅 ERC1155 TransferSingle (from wallet) + subscribeLogs(ERC1155_CONTRACT, listOf(ERC1155_TRANSFER_SINGLE_TOPIC, null, walletTopic), leaderId) + + // 订阅 ERC1155 TransferSingle (to wallet) + subscribeLogs(ERC1155_CONTRACT, listOf(ERC1155_TRANSFER_SINGLE_TOPIC, null, null, walletTopic), leaderId) + + // 订阅 ERC1155 TransferBatch (from wallet) + subscribeLogs(ERC1155_CONTRACT, listOf(ERC1155_TRANSFER_BATCH_TOPIC, null, walletTopic), leaderId) + + // 订阅 ERC1155 TransferBatch (to wallet) + subscribeLogs(ERC1155_CONTRACT, listOf(ERC1155_TRANSFER_BATCH_TOPIC, null, null, walletTopic), leaderId) + + logger.debug("已订阅 Leader 钱包地址: ${leader.leaderName} (${walletAddress})") + } catch (e: Exception) { + logger.error("订阅 Leader 失败: leaderId=$leaderId, address=$walletAddress", e) + } + } + + /** + * 订阅日志 + * @param address 合约地址 + * @param topics 主题列表 + * @param leaderId Leader ID，用于关联订阅响应 + */ + private fun subscribeLogs(address: String, topics: List, leaderId: Long) { + val ws = webSocket ?: return + + val params = mapOf( + "address" to address.lowercase(), + "topics" to topics + ) + + val subscribeParams = listOf("logs", params) + + val requestId = ++requestIdCounter + + // 保存请求 ID 到 Leader ID 的映射 + requestIdToLeaderId[requestId] = leaderId + + val request = mapOf( + "jsonrpc" to "2.0", + "id" to requestId, + "method" to "eth_subscribe", + "params" to subscribeParams + ) + + val json = gson.toJson(request) + ws.send(json) + } + + /** + * 取消订阅 + * @param subscriptionId 订阅 ID + */ + private fun unsubscribe(subscriptionId: String) { + val ws = webSocket ?: return + + val request = mapOf( + "jsonrpc" to "2.0", + "id" to (++requestIdCounter), + "method" to "eth_unsubscribe", + "params" to listOf(subscriptionId) + ) + + val json = gson.toJson(request) + ws.send(json) + logger.debug("已发送取消订阅请求: subscriptionId=$subscriptionId") + } + + /** + * 处理 WebSocket 消息 + */ + private suspend fun handleMessage(message: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) { + try { + // 使用 Gson 解析消息 + val messageJson = gson.fromJson(message, JsonObject::class.java) + + // 处理订阅响应（包含 subscription ID） + val id = messageJson.get("id") + if (id != null && !id.isJsonNull) { + val result = messageJson.get("result") + if (result != null && result.isJsonPrimitive && result.asJsonPrimitive.isString) { + // 这是订阅响应，result 是 subscription ID（字符串） + val requestId = id.asInt + val subscriptionId = result.asString + val leaderId = requestIdToLeaderId.remove(requestId) + if (leaderId != null) { + // 保存订阅 ID 到 Leader + leaderSubscriptions.getOrPut(leaderId) { mutableListOf() }.add(subscriptionId) + logger.debug("收到订阅响应: leaderId=$leaderId, subscriptionId=$subscriptionId") + } + } + return + } + + // 处理订阅通知（交易日志） + val method = messageJson.get("method")?.asString + if (method != "eth_subscription") { + return + } + + val params = messageJson.getAsJsonObject("params") ?: return + // result 是一个对象，包含日志信息 + val result = params.getAsJsonObject("result") ?: return + + // 从 result 对象中获取 transactionHash（关键数据） + val txHash = result.get("transactionHash")?.asString + if (txHash.isNullOrEmpty()) { + logger.debug("订阅通知中缺少 transactionHash，跳过处理") + return + } + + // 处理交易 + processTransaction(txHash, httpClient, rpcApi) + } catch (e: Exception) { + logger.error("处理 WebSocket 消息失败: ${e.message}", e) + logger.debug("消息内容: $message", e) + } + } + + /** + * 处理交易 + */ + private suspend fun processTransaction(txHash: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) { + try { + // 获取交易 receipt + val receiptRequest = JsonRpcRequest( + method = "eth_getTransactionReceipt", + params = listOf(txHash) + ) + + val receiptResponse = rpcApi.call(receiptRequest) + if (!receiptResponse.isSuccessful || receiptResponse.body() == null) { + return + } + + val receiptRpcResponse = receiptResponse.body()!! + if (receiptRpcResponse.error != null || receiptRpcResponse.result == null) { + return + } + + // 使用 Gson 解析 receipt JSON（result 是 JsonElement） + val receiptJson = receiptRpcResponse.result.asJsonObject + + // 获取区块号和时间戳 + val blockNumber = receiptJson.get("blockNumber")?.asString + val blockTimestamp = if (blockNumber != null) { + getBlockTimestamp(blockNumber, rpcApi) + } else { + null + } + + // 解析 receipt 中的 Transfer 日志 + val logs = receiptJson.getAsJsonArray("logs") ?: return + val (erc20Transfers, erc1155Transfers) = parseReceiptTransfers(logs) + + // 为每个 Leader 处理交易 + for (leader in monitoredLeaders.values) { + val trade = parseTradeFromTransfers( + txHash = txHash, + timestamp = blockTimestamp, + walletAddress = leader.leaderAddress, + erc20Transfers = erc20Transfers, + erc1155Transfers = erc1155Transfers + ) + + if (trade != null) { + // 调用 processTrade 处理交易（元数据已在 parseTradeFromTransfers 中补齐） + copyOrderTrackingService.processTrade( + leaderId = leader.id!!, + trade = trade, + source = "onchain-ws" + ) + } + } + } catch (e: Exception) { + logger.error("处理交易失败: txHash=$txHash, ${e.message}", e) + } + } + + /** + * 解析 receipt 中的 Transfer 日志 + */ + private fun parseReceiptTransfers(logs: com.google.gson.JsonArray): Pair, List> { + val erc20 = mutableListOf() + val erc1155 = mutableListOf() + + for (logElement in logs) { + val log = logElement.asJsonObject + val address = log.get("address")?.asString?.lowercase() ?: continue + val topicsArray = log.getAsJsonArray("topics") ?: continue + val topics = topicsArray.mapNotNull { it.asString } + if (topics.isEmpty()) continue + + val t0 = topics[0].lowercase() + val data = log.get("data")?.asString ?: "0x" + + // USDC ERC20 Transfer + if (address == USDC_CONTRACT.lowercase() && t0 == ERC20_TRANSFER_TOPIC && topics.size >= 3) { + val from = topicToAddress(topics[1]) + val to = topicToAddress(topics[2]) + val value = hexToBigInt(data) + erc20.add(Erc20Transfer(from, to, value)) + continue + } + + // ERC1155 TransferSingle + if (t0 == ERC1155_TRANSFER_SINGLE_TOPIC && topics.size >= 4) { + val from = topicToAddress(topics[2]) + val to = topicToAddress(topics[3]) + val bytes = bytesFromHex(data) + if (bytes.size >= 64) { + val tokenId = sliceBigInt32(bytes, 0) + val value = sliceBigInt32(bytes, 32) + erc1155.add(Erc1155Transfer(from, to, tokenId, value)) + } + continue + } + + // ERC1155 TransferBatch + if (t0 == ERC1155_TRANSFER_BATCH_TOPIC && topics.size >= 4) { + val from = topicToAddress(topics[2]) + val to = topicToAddress(topics[3]) + val bytes = bytesFromHex(data) + if (bytes.size < 64) continue + + val offIds = sliceBigInt32(bytes, 0).toInt() + val offVals = sliceBigInt32(bytes, 32).toInt() + if (offIds + 32 > bytes.size || offVals + 32 > bytes.size) continue + + val nIds = sliceBigInt32(bytes, offIds).toInt() + val nVals = sliceBigInt32(bytes, offVals).toInt() + if (nIds != nVals) continue + + val idsStart = offIds + 32 + val valsStart = offVals + 32 + for (i in 0 until nIds) { + val ib = idsStart + i * 32 + val vb = valsStart + i * 32 + if (ib + 32 > bytes.size || vb + 32 > bytes.size) break + val tokenId = sliceBigInt32(bytes, ib) + val value = sliceBigInt32(bytes, vb) + erc1155.add(Erc1155Transfer(from, to, tokenId, value)) + } + } + } + + return Pair(erc20, erc1155) + } + + /** + * 从 Transfer 日志解析交易信息 + */ + private suspend fun parseTradeFromTransfers( + txHash: String, + timestamp: Long?, + walletAddress: String, + erc20Transfers: List, + erc1155Transfers: List + ): TradeResponse? { + val wallet = walletAddress.lowercase() + + // 计算 USDC 流入和流出 + val usdcOut = erc20Transfers.filter { it.from.lowercase() == wallet } + .fold(BigInteger.ZERO) { acc, t -> acc + t.value } + val usdcIn = erc20Transfers.filter { it.to.lowercase() == wallet } + .fold(BigInteger.ZERO) { acc, t -> acc + t.value } + + // 计算 ERC1155 流入和流出（按 tokenId 聚合） + val inById = mutableMapOf() + val outById = mutableMapOf() + for (t in erc1155Transfers) { + if (t.to.lowercase() == wallet) { + inById[t.tokenId] = (inById[t.tokenId] ?: BigInteger.ZERO) + t.value + } + if (t.from.lowercase() == wallet) { + outById[t.tokenId] = (outById[t.tokenId] ?: BigInteger.ZERO) + t.value + } + } + + // 找到最大的流入和流出 tokenId + fun best(map: Map): Pair = + map.entries.maxByOrNull { it.value }?.let { it.key to it.value } ?: (null to BigInteger.ZERO) + + val (bestInId, bestInVal) = best(inById) + val (bestOutId, bestOutVal) = best(outById) + + // 判断交易方向 + var side: String? = null + var asset: BigInteger? = null + var sizeRaw = BigInteger.ZERO + var usdcRaw = BigInteger.ZERO + + if (bestInId != null && bestInVal > BigInteger.ZERO && usdcOut > BigInteger.ZERO) { + // BUY: 收到 token，支付 USDC + side = "BUY" + asset = bestInId + sizeRaw = bestInVal + usdcRaw = usdcOut + } else if (bestOutId != null && bestOutVal > BigInteger.ZERO && usdcIn > BigInteger.ZERO) { + // SELL: 卖出 token，收到 USDC + side = "SELL" + asset = bestOutId + sizeRaw = bestOutVal + usdcRaw = usdcIn + } else { + // 无法判断交易方向 + return null + } + + // 计算价格和数量（USDC 有 6 位小数，shares 也有 6 位小数） + val usdcSize = usdcRaw.toBigDecimal().divide(BigInteger("1000000").toBigDecimal(), 8, java.math.RoundingMode.DOWN) + val size = sizeRaw.toBigDecimal().divide(BigInteger("1000000").toBigDecimal(), 8, java.math.RoundingMode.DOWN) + val price = if (size.signum() > 0) { + usdcSize.divide(size, 8, java.math.RoundingMode.DOWN) + } else { + return null + } + + // 尝试通过 Gamma API 查询市场信息（通过 tokenId） + val marketInfo = fetchMarketByTokenId(asset.toString()) + + // 创建 TradeResponse + return TradeResponse( + id = txHash, + market = marketInfo?.conditionId ?: "", + side = side, + price = price.toPlainString(), + size = size.toPlainString(), + timestamp = (timestamp ?: System.currentTimeMillis() / 1000).toString(), + user = walletAddress, + outcomeIndex = marketInfo?.outcomeIndex, + outcome = marketInfo?.outcome + ) + } + + /** + * 通过 Gamma API 查询市场信息（通过 tokenId） + */ + private suspend fun fetchMarketByTokenId(tokenId: String): MarketInfo? { + return try { + // 使用 HTTP 请求直接调用 Gamma API（因为 Retrofit 接口可能不支持 clob_token_ids 参数） + val httpClient = createHttpClient() + val url = "https://gamma-api.polymarket.com/markets?clob_token_ids=$tokenId" + + val request = okhttp3.Request.Builder() + .url(url) + .get() + .build() + + val response = httpClient.newCall(request).execute() + if (!response.isSuccessful || response.body == null) { + return null + } + + val responseBody = response.body!!.string() + // 使用 Gson 解析市场列表 + val marketsType = object : TypeToken>() {}.type + val markets = gson.fromJson>(responseBody, marketsType) + + if (markets.isEmpty()) { + return null + } + + val market = markets.first() + + // 解析 clob_token_ids（可能是 JSON 字符串或数组） + val clobTokenIdsRaw = market.get("clobTokenIds") ?: market.get("clob_token_ids") + val clobTokenIds = when { + clobTokenIdsRaw == null || clobTokenIdsRaw.isJsonNull -> null + clobTokenIdsRaw.isJsonPrimitive && clobTokenIdsRaw.asJsonPrimitive.isString -> { + // 尝试解析 JSON 字符串 + try { + val listType = object : TypeToken>() {}.type + gson.fromJson>(clobTokenIdsRaw.asString, listType) + } catch (e: Exception) { + null + } + } + clobTokenIdsRaw.isJsonArray -> { + clobTokenIdsRaw.asJsonArray.mapNotNull { it.asString } + } + else -> null + } + + // 解析 outcomes（可能是 JSON 字符串或数组） + val outcomesRaw = market.get("outcomes") + val outcomes = when { + outcomesRaw == null || outcomesRaw.isJsonNull -> null + outcomesRaw.isJsonPrimitive && outcomesRaw.asJsonPrimitive.isString -> { + try { + val listType = object : TypeToken>() {}.type + gson.fromJson>(outcomesRaw.asString, listType) + } catch (e: Exception) { + null + } + } + outcomesRaw.isJsonArray -> { + outcomesRaw.asJsonArray.mapNotNull { it.asString } + } + else -> null + } + + // 查找 tokenId 在 clobTokenIds 中的索引 + val outcomeIndex = clobTokenIds?.indexOfFirst { + it.equals(tokenId, ignoreCase = true) + }?.takeIf { it >= 0 } + + // 获取 outcome 名称 + val outcome = if (outcomeIndex != null && outcomes != null && outcomeIndex < outcomes.size) { + outcomes[outcomeIndex] + } else { + null + } + + val conditionId = market.get("conditionId")?.asString ?: return null + + MarketInfo( + conditionId = conditionId, + outcomeIndex = outcomeIndex, + outcome = outcome + ) + } catch (e: Exception) { + logger.warn("通过 Gamma API 查询市场信息失败: tokenId=$tokenId, ${e.message}") + null + } + } + + /** + * 市场信息（从 Gamma API 获取） + */ + private data class MarketInfo( + val conditionId: String, + val outcomeIndex: Int?, + val outcome: String? + ) + + /** + * 获取区块时间戳 + */ + private suspend fun getBlockTimestamp(blockNumber: String, rpcApi: EthereumRpcApi): Long? { + return try { + val blockRequest = JsonRpcRequest( + method = "eth_getBlockByNumber", + params = listOf(blockNumber, false) + ) + + val blockResponse = rpcApi.call(blockRequest) + if (!blockResponse.isSuccessful || blockResponse.body() == null) { + return null + } + + val blockRpcResponse = blockResponse.body()!! + if (blockRpcResponse.error != null || blockRpcResponse.result == null) { + return null + } + + // 使用 Gson 解析 block JSON（result 是 JsonElement） + val blockJson = blockRpcResponse.result.asJsonObject + val timestampHex = blockJson.get("timestamp")?.asString ?: return null + hexToBigInt(timestampHex).toLong() + } catch (e: Exception) { + logger.warn("获取区块时间戳失败: ${e.message}") + null + } + } + + // 辅助函数 + private data class Erc20Transfer(val from: String, val to: String, val value: BigInteger) + private data class Erc1155Transfer(val from: String, val to: String, val tokenId: BigInteger, val value: BigInteger) + + private fun topicToAddress(topic: String): String { + val t = topic.removePrefix("0x").lowercase() + return "0x" + t.takeLast(40) + } + + private fun hexToBigInt(hex: String): BigInteger { + val h = hex.removePrefix("0x") + if (h.isEmpty()) return BigInteger.ZERO + return BigInteger(h, 16) + } + + private fun bytesFromHex(hex: String): ByteArray { + val s = hex.removePrefix("0x") + if (s.isEmpty()) return ByteArray(0) + val out = ByteArray(s.length / 2) + var i = 0 + while (i < s.length) { + out[i / 2] = s.substring(i, i + 2).toInt(16).toByte() + i += 2 + } + return out + } + + private fun sliceBigInt32(b: ByteArray, offset: Int): BigInteger { + val sub = b.copyOfRange(offset, offset + 32) + return BigInteger(1, sub) + } + + /** + * 地址转换为 32 字节 topic（前 24 字节为 0，后 8 字节为地址） + */ + private fun addressToTopic32(address: String): String { + val addr = address.removePrefix("0x").lowercase() + return "0x" + "0".repeat(24) + addr + } + + @PreDestroy + fun destroy() { + stop() + scope.cancel() + } +} + diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/CopyOrderTrackingService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/CopyOrderTrackingService.kt index db67019..efd219c 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/CopyOrderTrackingService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/CopyOrderTrackingService.kt @@ -8,10 +8,13 @@ import com.wrbug.polymarketbot.repository.* import com.wrbug.polymarketbot.util.RetrofitFactory import com.wrbug.polymarketbot.util.* import kotlinx.coroutines.* +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import org.slf4j.LoggerFactory import org.springframework.dao.DataIntegrityViolationException import org.springframework.dao.DuplicateKeyException import java.sql.SQLException +import java.util.concurrent.ConcurrentHashMap import com.wrbug.polymarketbot.service.copytrading.configs.CopyTradingFilterService import com.wrbug.polymarketbot.service.copytrading.configs.FilterStatus import com.wrbug.polymarketbot.service.copytrading.orders.OrderSigningService @@ -53,11 +56,22 @@ open class CopyOrderTrackingService( // 协程作用域（用于异步发送通知） private val notificationScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + // 使用 Mutex 保证线程安全（按交易ID锁定） + private val tradeMutexMap = ConcurrentHashMap() + // 订单创建重试配置 companion object { private const val MAX_RETRY_ATTEMPTS = 2 // 最多重试次数（首次 + 1次重试） private const val RETRY_DELAY_MS = 3000L // 重试前等待时间（毫秒，3秒） } + + /** + * 获取或创建 Mutex（按交易ID） + */ + private fun getMutex(leaderId: Long, tradeId: String): Mutex { + val key = "${leaderId}_${tradeId}" + return tradeMutexMap.getOrPut(key) { Mutex() } + } /** * 解密账户私钥 @@ -102,96 +116,98 @@ open class CopyOrderTrackingService( /** * 处理交易事件（WebSocket 或轮询） * 根据交易方向调用相应的处理方法 + * 使用 Mutex 保证线程安全（单实例部署） */ @Transactional suspend fun processTrade(leaderId: Long, trade: TradeResponse, source: String): Result { - return try { - // 1. 检查是否已处理（去重，包括失败状态） - val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) - - if (existingProcessed != null) { - if (existingProcessed.status == "FAILED") { - return Result.success(Unit) - } - return Result.success(Unit) - } - - // 检查是否已记录为失败交易 - val failedTrade = failedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) - if (failedTrade != null) { - return Result.success(Unit) - } - - // 2. 处理交易逻辑 - val result = when (trade.side.uppercase()) { - "BUY" -> processBuyTrade(leaderId, trade) - "SELL" -> processSellTrade(leaderId, trade) - else -> { - logger.warn("未知的交易方向: ${trade.side}") - Result.failure(IllegalArgumentException("未知的交易方向: ${trade.side}")) - } - } - - if (result.isFailure) { - logger.error( - "处理交易失败: leaderId=$leaderId, tradeId=${trade.id}, side=${trade.side}", - result.exceptionOrNull() - ) - return result - } - - // 3. 标记为已处理（成功状态） - // 注意：并发情况下可能多个请求同时处理同一笔交易，需要处理唯一约束冲突 + // 获取该交易的 Mutex（按交易ID锁定，不同交易可以并行处理） + val mutex = getMutex(leaderId, trade.id) + + return mutex.withLock { try { - val processed = ProcessedTrade( - leaderId = leaderId, - leaderTradeId = trade.id, - tradeType = trade.side.uppercase(), - source = source, - status = "SUCCESS", - processedAt = System.currentTimeMillis() - ) - processedTradeRepository.save(processed) - } catch (e: Exception) { - // 检查是否是唯一键冲突异常（可能是 DataIntegrityViolationException、DuplicateKeyException 或 SQLException） - if (isUniqueConstraintViolation(e)) { - // 唯一约束冲突，说明已经处理过了（可能是并发请求） - // 再次检查确认状态 - val existing = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) - if (existing != null) { - if (existing.status == "FAILED") { - logger.debug("交易已标记为失败，跳过处理: leaderId=$leaderId, tradeId=${trade.id}") - return Result.success(Unit) - } - logger.debug("交易已处理（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existing.status}") - return Result.success(Unit) - } else { - // 如果检查不到，可能是事务隔离级别问题，等待一下再查询 - delay(100) - val existingAfterDelay = - processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) - if (existingAfterDelay != null) { - logger.debug("延迟查询到记录（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existingAfterDelay.status}") - return Result.success(Unit) - } - // 如果还是查询不到，记录警告但不抛出异常（可能是其他约束冲突） - logger.warn( - "保存ProcessedTrade时发生唯一约束冲突，但查询不到记录: leaderId=$leaderId, tradeId=${trade.id}", - e - ) - // 不抛出异常，避免影响其他交易的处理 - return Result.success(Unit) - } - } else { - // 其他类型的异常，重新抛出 - throw e - } - } + // 1. 检查是否已处理（去重，包括失败状态） + val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) - Result.success(Unit) - } catch (e: Exception) { - logger.error("处理交易异常: leaderId=$leaderId, tradeId=${trade.id}", e) - Result.failure(e) + if (existingProcessed != null) { + if (existingProcessed.status == "FAILED") { + return@withLock Result.success(Unit) + } + return@withLock Result.success(Unit) + } + + // 检查是否已记录为失败交易 + val failedTrade = failedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) + if (failedTrade != null) { + return@withLock Result.success(Unit) + } + + // 2. 处理交易逻辑 + val result = when (trade.side.uppercase()) { + "BUY" -> processBuyTrade(leaderId, trade) + "SELL" -> processSellTrade(leaderId, trade) + else -> { + logger.warn("未知的交易方向: ${trade.side}") + Result.failure(IllegalArgumentException("未知的交易方向: ${trade.side}")) + } + } + + if (result.isFailure) { + logger.error( + "处理交易失败: leaderId=$leaderId, tradeId=${trade.id}, side=${trade.side}", + result.exceptionOrNull() + ) + return@withLock result + } + + // 3. 标记为已处理（成功状态） + // 由于使用了 Mutex，这里理论上不会出现并发冲突，但保留异常处理作为兜底 + try { + val processed = ProcessedTrade( + leaderId = leaderId, + leaderTradeId = trade.id, + tradeType = trade.side.uppercase(), + source = source, + status = "SUCCESS", + processedAt = System.currentTimeMillis() + ) + processedTradeRepository.save(processed) + } catch (e: Exception) { + // 检查是否是唯一键冲突异常（理论上不会发生，但保留作为兜底） + if (isUniqueConstraintViolation(e)) { + val existing = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) + if (existing != null) { + if (existing.status == "FAILED") { + logger.debug("交易已标记为失败，跳过处理: leaderId=$leaderId, tradeId=${trade.id}") + return@withLock Result.success(Unit) + } + logger.debug("交易已处理（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existing.status}") + return@withLock Result.success(Unit) + } else { + // 如果检查不到，可能是事务隔离级别问题，等待一下再查询 + delay(100) + val existingAfterDelay = + processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) + if (existingAfterDelay != null) { + logger.debug("延迟查询到记录（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existingAfterDelay.status}") + return@withLock Result.success(Unit) + } + logger.warn( + "保存ProcessedTrade时发生唯一约束冲突，但查询不到记录: leaderId=$leaderId, tradeId=${trade.id}", + e + ) + return@withLock Result.success(Unit) + } + } else { + // 其他类型的异常，重新抛出 + throw e + } + } + + Result.success(Unit) + } catch (e: Exception) { + logger.error("处理交易异常: leaderId=$leaderId, tradeId=${trade.id}", e) + Result.failure(e) + } } } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/RelayClientService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/RelayClientService.kt index 53fc696..89c2f0a 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/RelayClientService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/RelayClientService.kt @@ -655,7 +655,7 @@ class RelayClientService( } val hexNonce = rpcResponse.result ?: return Result.failure(Exception("Proxy nonce 结果为空")) - val nonce = EthereumUtils.decodeUint256(hexNonce) + val nonce = EthereumUtils.decodeUint256(hexNonce.asString) return Result.success(nonce) } @@ -679,7 +679,7 @@ class RelayClientService( } val hexNonce = rpcResponse.result ?: return Result.failure(Exception("nonce 结果为空")) - val nonce = EthereumUtils.decodeUint256(hexNonce) + val nonce = EthereumUtils.decodeUint256(hexNonce.asString) return Result.success(nonce) } @@ -703,7 +703,7 @@ class RelayClientService( } val hexGasPrice = rpcResponse.result ?: return Result.failure(Exception("gas price 结果为空")) - val gasPrice = EthereumUtils.decodeUint256(hexGasPrice) + val gasPrice = EthereumUtils.decodeUint256(hexGasPrice.asString) return Result.success(gasPrice) } @@ -774,7 +774,7 @@ class RelayClientService( } val txHash = rpcResponse.result ?: return Result.failure(Exception("交易哈希为空")) - return Result.success(txHash) + return Result.success(txHash.asString) } /** diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/RpcNodeService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/RpcNodeService.kt index 29b9888..014068a 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/RpcNodeService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/RpcNodeService.kt @@ -446,7 +446,7 @@ class RpcNodeService( )) } - val blockNumber = rpcResponse.result + val blockNumber = rpcResponse.result?.asString if (blockNumber.isNullOrBlank()) { return Result.success(NodeCheckResult( status = NodeHealthStatus.UNHEALTHY, diff --git a/docs/zh/copy-trading-logic-summary.md b/docs/zh/copy-trading-logic-summary.md new file mode 100644 index 0000000..78836d4 --- /dev/null +++ b/docs/zh/copy-trading-logic-summary.md @@ -0,0 +1,289 @@ +# 跟单买卖逻辑简易文档 + +## 一、跟单信号检测 + +### 1.1 监控方式 + +系统使用**轮询方式**监控 Leader 的交易活动： + +- **数据源**：Polymarket Data API 的 `/activity` 接口 +- **轮询间隔**：默认 2 秒（可配置） +- **查询参数**： + - `user`: Leader 钱包地址 + - `type`: `["TRADE"]`（只查询交易类型） + - `limit`: 100（每次最多查询 100 条） + - `sortBy`: `TIMESTAMP` + - `sortDirection`: `DESC`（按时间戳降序，最新的在前） + +### 1.2 增量检测机制 + +通过 **diff 算法**检测新增交易： + +1. **首次轮询**： + - 查询最近 100 条交易 + - 缓存所有交易 ID（不处理） + - 标记首次轮询完成 + +2. **后续轮询**： + - 查询最近 100 条交易 + - 与缓存的交易 ID 集合进行 diff + - 找出新增的交易 ID + - 处理新增交易 + - 更新缓存（添加新增的交易 ID） + +3. **去重机制**： + - 使用 `leaderId + tradeId` 作为唯一标识 + - 在 `processed_trade` 表中记录已处理的交易 + - 避免重复处理同一笔交易 + +### 1.3 交易数据转换 + +从 `UserActivityResponse` 转换为 `TradeResponse`： + +```kotlin +TradeResponse( + id = activity.transactionHash, // 交易ID（用于去重） + market = activity.conditionId, // 市场ID + side = activity.side, // "BUY" 或 "SELL" + price = activity.price, // 交易价格 + size = activity.size, // 交易数量 + timestamp = activity.timestamp, // 时间戳（秒） + user = activity.proxyWallet, // 用户钱包地址 + outcomeIndex = activity.outcomeIndex, // 结果索引（0=第一个outcome，1=第二个outcome） + outcome = activity.outcome // 结果名称 +) +``` + +### 1.4 信号触发流程 + +``` +轮询任务启动 + ↓ +定期轮询所有 Leader（每 2 秒） + ↓ +查询 Leader 活动（/activity 接口） + ↓ +转换为 TradeResponse + ↓ +diff 检测新增交易 + ↓ +调用 processTrade() 处理交易 + ↓ +根据 side 字段判断： + - "BUY" → processBuyTrade() + - "SELL" → processSellTrade() +``` + +## 二、订单构建 + +### 2.1 买入订单构建流程 + +#### 2.1.1 前置检查 + +1. **查找跟单关系**： + - 查询所有启用且支持该 Leader 的跟单配置 + - 验证账户 API 凭证是否配置 + - 验证账户是否启用 + +2. **获取 Token ID**： + - 使用 `outcomeIndex` 和 `market` 获取 tokenId + - 支持多元市场（不限于 YES/NO） + +3. **计算买入数量**： + - **RATIO 模式**：`买入数量 = Leader 数量 × 跟单比例` + - **FIXED 模式**：`买入数量 = 固定金额 / 买入价格` + +4. **过滤条件检查**： + - 价格区间检查 + - 仓位限制检查 + - 市场分类检查 + - 订单簿检查（获取最佳卖单价格） + +5. **价格调整**： + - 应用价格容忍度（`priceTolerance`） + - 买入价格 = Leader 价格 × (1 + 容忍度) + - 确保调整后的价格不低于最佳卖单价格 + +#### 2.1.2 订单签名 + +使用 `OrderSigningService.createAndSignOrder()` 创建并签名订单： + +1. **计算订单金额**： + - **BUY 订单**： + - `makerAmount = price × size`（USDC 金额，最多 2 位小数） + - `takerAmount = size`（shares 数量，最多 4 位小数） + - 转换为 wei（6 位小数） + +2. **生成订单参数**： + - `salt`: 时间戳（毫秒） + - `maker`: 代理钱包地址（proxyAddress） + - `signer`: 从私钥推导的签名地址 + - `taker`: 零地址（0x0000...） + - `tokenId`: 从 outcomeIndex 获取 + - `makerAmount`: 计算出的 maker 金额（wei） + - `takerAmount`: 计算出的 taker 金额（wei） + - `expiration`: "0"（永不过期） + - `nonce`: "0" + - `feeRateBps`: "0" + - `side`: "BUY" + - `signatureType`: 2（Browser Wallet） + +3. **EIP-712 签名**： + - 编码域分隔符（Exchange Contract + Chain ID） + - 编码订单消息哈希 + - 计算结构化数据哈希 + - 使用私钥签名（r + s + v） + +#### 2.1.3 创建订单请求 + +构建 `NewOrderRequest`： + +```kotlin +NewOrderRequest( + order = signedOrder, // 签名的订单对象 + owner = account.apiKey, // API Key + orderType = "FAK", // Fill-And-Kill（允许部分成交，未成交部分立即取消） + deferExec = false // 立即执行 +) +``` + +#### 2.1.4 提交订单 + +1. **创建 CLOB API 客户端**（带认证）： + - 使用账户的 API Key、Secret、Passphrase + - 解密 API 凭证 + +2. **调用 API 创建订单**： + - `POST /orders` + - 带重试机制（最多重试 2 次） + - 每次重试都重新生成 salt 并重新签名 + +3. **记录订单跟踪**： + - 保存到 `copy_order_tracking` 表 + - 记录买入订单 ID、数量、价格等信息 + - 状态：`filled` + +### 2.2 卖出订单构建流程 + +#### 2.2.1 前置检查 + +1. **查找跟单关系**： + - 查询所有启用且支持该 Leader 的跟单配置 + - 验证是否支持卖出（`supportSell = true`） + +2. **计算需要匹配的数量**： + - `需要匹配数量 = Leader 卖出数量 × 跟单比例` + +3. **查找未匹配的买入订单**： + - 使用 `outcomeIndex` 匹配（支持多元市场） + - 按 FIFO 顺序（先进先出） + - 查询 `copy_order_tracking` 表中未匹配的订单 + +4. **计算实际可卖出数量**： + - 按 FIFO 顺序匹配 + - 支持部分匹配（一个买入订单可以被多次卖出匹配） + +#### 2.2.2 价格计算 + +1. **优先使用订单簿 bestBid**： + - 查询订单簿（`getOrderbookByTokenId`） + - 获取最佳买单价格（bestBid） + - 使用 bestBid 作为卖出价格 + +2. **备选方案**： + - 如果获取订单簿失败，使用 Leader 价格 + - 卖出价格 = Leader 价格 × 0.9（固定按 90% 计算） + +#### 2.2.3 订单签名 + +与买入订单类似，但： +- `side`: "SELL" +- **SELL 订单金额计算**： + - `makerAmount = size`（shares 数量，最多 4 位小数） + - `takerAmount = price × size`（USDC 金额，使用原始价格计算） + +#### 2.2.4 创建订单请求 + +与买入订单相同： +- `orderType`: "FAK" +- `deferExec`: false + +#### 2.2.5 提交订单 + +1. **调用 API 创建卖出订单**（带重试机制） + +2. **更新买入订单状态**： + - 更新 `copy_order_tracking` 表中的 `remainingQuantity` + - 如果完全匹配，状态更新为 `fully_matched` + - 如果部分匹配，状态更新为 `partially_matched` + +3. **记录匹配关系**： + - 保存到 `sell_match_record` 表（卖出匹配记录） + - 保存到 `sell_match_detail` 表（匹配明细，包含盈亏计算） + +## 三、关键配置 + +### 3.1 跟单配置参数 + +- `copyMode`: 跟单模式（"RATIO" 或 "FIXED"） +- `copyRatio`: 跟单比例（RATIO 模式） +- `fixedAmount`: 固定金额（FIXED 模式） +- `priceTolerance`: 价格容忍度（买入时使用） +- `supportSell`: 是否支持卖出 +- `enabled`: 是否启用 + +### 3.2 订单类型 + +- **FAK (Fill-And-Kill)**： + - 允许部分成交 + - 未成交部分立即取消 + - 快速响应 Leader 交易，避免订单长期挂单导致价格不匹配 + +### 3.3 重试机制 + +- **最多重试次数**：2 次（首次 + 1 次重试） +- **重试延迟**：3 秒 +- **重试策略**：每次重试都重新生成 salt 并重新签名，确保签名唯一性 + +## 四、数据流向 + +``` +Leader 交易（链上） + ↓ +Polymarket Data API (/activity) + ↓ +轮询服务（CopyTradingPollingService） + ↓ +交易处理服务（CopyOrderTrackingService） + ↓ +订单签名服务（OrderSigningService） + ↓ +CLOB API (POST /orders) + ↓ +订单跟踪表（copy_order_tracking） +``` + +## 五、注意事项 + +1. **去重机制**： + - 使用 `leaderId + tradeId` 作为唯一标识 + - 在 `processed_trade` 表中记录已处理的交易 + - 避免重复处理同一笔交易 + +2. **价格调整**： + - 买入时应用价格容忍度（提高买入价格） + - 卖出时优先使用订单簿 bestBid，失败则使用 Leader 价格的 90% + +3. **订单簿检查**： + - 买入前检查订单簿中是否有可匹配的卖单 + - 确保调整后的买入价格不低于最佳卖单价格 + +4. **FIFO 匹配**： + - 卖出时按买入时间顺序匹配（先进先出） + - 支持部分匹配 + +5. **错误处理**： + - 订单创建失败时记录到 `failed_trade` 表 + - 支持重试机制（最多 2 次） + - 发送失败通知（如果配置了 `pushFailedOrders`） + diff --git a/docs/zh/copy-trading-monitor-strategy.md b/docs/zh/copy-trading-monitor-strategy.md new file mode 100644 index 0000000..ac00d25 --- /dev/null +++ b/docs/zh/copy-trading-monitor-strategy.md @@ -0,0 +1,653 @@ +# 跟单监听策略：链上 WebSocket + 轮询并行方案 + +## 一、方案概述 + +系统**同时运行**两种监听方式，并行处理，哪个数据先返回就用哪个： + +1. **链上 WebSocket 监听**：通过 Polygon RPC 的 `eth_subscribe` 实时监听链上交易 +2. **轮询监听**：通过 Polymarket Data API 定期轮询交易记录 + +**核心特点**： +- 两种方式**并行运行**，不互相排斥 +- WS 断开时**不断重试连接**，不停止轮询 +- 哪个数据先返回就用哪个，确保最快响应 +- 通过去重机制确保同一笔交易只处理一次 + +## 二、关键流程图 + +``` +系统启动 + ↓ +同时启动两种监听方式 + ↓ + ├─→ 链上 WS 监听（并行） + │ ↓ + │ 尝试连接 WS RPC + │ ↓ + │ 连接成功？ + │ ├─→ 是 → 订阅 Leader 钱包地址 + │ │ (eth_subscribe: USDC Transfer + ERC1155 Transfer) + │ │ ↓ + │ │ 实时接收交易日志 + │ │ ↓ + │ │ 解析交易 receipt + │ │ ↓ + │ │ 转换为 TradeResponse + │ │ ↓ + │ │ 调用 processTrade()（去重检查） + │ │ ↓ + │ │ 连接断开？ + │ │ ├─→ 是 → 等待重连延迟 → 重试连接（循环） + │ │ └─→ 否 → 继续监听 + │ │ + │ └─→ 否 → 等待重连延迟 → 重试连接（循环） + │ + └─→ 轮询监听（并行） + ↓ + 定期轮询 (每 2 秒) + ↓ + 查询 /activity 接口 + ↓ + diff 检测新增交易 + ↓ + 调用 processTrade()（去重检查） + ↓ + 继续轮询（循环） +``` + +## 三、并行处理流程 + +``` +交易发生 + ↓ + ├─→ WS 监听（实时，秒级） + │ ↓ + │ 收到链上日志 + │ ↓ + │ 解析并转换为 TradeResponse + │ ↓ + │ 调用 processTrade() + │ ↓ + │ 去重检查（processed_trade 表） + │ ├─→ 已处理 → 跳过 + │ └─→ 未处理 → 处理交易 + │ + └─→ 轮询监听（延迟，2秒间隔） + ↓ + 轮询到新交易 + ↓ + 转换为 TradeResponse + ↓ + 调用 processTrade() + ↓ + 去重检查（processed_trade 表） + ├─→ 已处理 → 跳过（WS 已处理） + └─→ 未处理 → 处理交易 +``` + +## 三、实现方案 + +### 3.1 服务架构 + +``` +CopyTradingMonitorService (主服务) + ├─→ OnChainWsService (链上 WS 监听，独立运行) + └─→ CopyTradingPollingService (轮询监听，独立运行) +``` + +**关键点**： +- 两个服务**独立运行**，互不影响 +- 主服务负责启动和协调两个服务 +- 两个服务都调用同一个 `processTrade()` 方法 +- 去重由 `processTrade()` 内部处理 + +### 3.2 WS 重连机制 + +**重连策略**： +1. WS 连接断开时，**不停止轮询服务** +2. 等待重连延迟（如 3 秒） +3. 自动重试连接 +4. 连接成功后重新订阅所有 Leader +5. 如果连接失败，继续重试（无限重试） + +**重连流程**： +``` +WS 连接断开 + ↓ +记录断开日志 + ↓ +等待重连延迟（3秒） + ↓ +尝试重新连接 + ↓ +连接成功？ + ├─→ 是 → 重新订阅所有 Leader → 继续监听 + └─→ 否 → 等待重连延迟 → 继续重试（循环） +``` + +**关键实现**： +- 使用协程或后台线程持续重试 +- 不阻塞主流程 +- 轮询服务继续运行，不受 WS 状态影响 + +### 3.3 去重机制 + +**去重标识**： +- 使用 `leaderId + transactionHash` 作为唯一标识 +- 在 `processed_trade` 表中记录已处理的交易 + +**去重流程**： +``` +收到交易数据（WS 或轮询） + ↓ +调用 processTrade(leaderId, trade, source) + ↓ +检查 processed_trade 表 + ├─→ 已存在 → 跳过处理（返回成功） + └─→ 不存在 → 继续处理 + ↓ + 处理交易（创建订单等） + ↓ + 保存到 processed_trade 表 + ├─→ leaderId + ├─→ leaderTradeId (transactionHash) + ├─→ tradeType (BUY/SELL) + ├─→ source (onchain-ws / polling) + └─→ status (SUCCESS/FAILED) +``` + +**并发安全**： +- 使用数据库唯一约束（`leaderId + leaderTradeId`） +- 处理唯一约束冲突（并发情况下可能多个请求同时处理同一笔交易） +- 如果冲突，再次查询确认状态 + +### 3.4 链上 WS 监听实现要点 + +**订阅参数**： +- 钱包地址：Leader 的 `leaderAddress` +- 订阅类型： + - USDC Transfer（`from` 或 `to` 为钱包地址） + - ERC1155 TransferSingle/Batch（`from` 或 `to` 为钱包地址） + +**消息处理流程**： +1. 接收 `eth_subscription` 消息 +2. 提取 `transactionHash` +3. 调用 RPC 获取交易 receipt +4. 解析 USDC Transfer 和 ERC1155 Transfer 日志 +5. 计算交易方向（BUY/SELL）、数量、价格 +6. 调用 Gamma API 补齐市场元数据（conditionId、outcomeIndex 等） +7. 转换为 `TradeResponse` +8. 调用 `processTrade(leaderId, trade, "onchain-ws")` 处理交易（去重由内部处理） + +**重连实现**： +```kotlin +// 伪代码示例 +while (isActive) { + try { + // 尝试连接 + val ws = connectWebSocket() + + // 订阅所有 Leader + subscribeAllLeaders(ws) + + // 监听消息 + ws.listen { message -> + handleMessage(message) + } + + // 连接断开，等待重连 + waitReconnectDelay() + } catch (e: Exception) { + logger.warn("WS 连接失败，等待重连: ${e.message}") + waitReconnectDelay() + } +} +``` + +### 3.5 轮询监听实现要点 + +**轮询流程**： +1. 定期查询 `/activity` 接口（每 2 秒） +2. 通过 diff 检测新增交易 +3. 转换为 `TradeResponse` +4. 调用 `processTrade(leaderId, trade, "polling")` 处理交易（去重由内部处理） + +**关键点**： +- 轮询服务**独立运行**，不受 WS 状态影响 +- 即使 WS 正常工作，轮询也继续运行（作为备份） +- 去重机制确保不会重复处理同一笔交易 + +## 四、配置参数 + +### 4.1 RPC 配置获取 + +**RPC 配置从后台配置中读取**，通过 `RpcNodeService` 获取： + +```kotlin +// 获取 HTTP RPC URL +val httpUrl = rpcNodeService.getHttpUrl() + +// 获取 WebSocket RPC URL +val wsUrl = rpcNodeService.getWsUrl() + +// 获取可用节点配置（包含完整信息） +val nodeResult = rpcNodeService.getAvailableNode() +if (nodeResult.isSuccess) { + val node = nodeResult.getOrNull() + val httpUrl = node?.httpUrl + val wsUrl = node?.wsUrl +} +``` + +**配置管理**： +- **RPC 节点配置存储在数据库**（`rpc_node_config` 表），不从配置文件读取 +- 支持多个节点配置，按优先级选择 +- 支持健康检查，自动选择可用节点 +- 前端可以通过系统设置页面配置 RPC 节点 +- 配置变更后，WS 重连时会自动使用新配置 + +**配置字段**： +- `httpUrl`: HTTP RPC 地址 +- `wsUrl`: WebSocket RPC 地址（可选） +- `enabled`: 是否启用 +- `priority`: 优先级（数字越小优先级越高） +- `lastCheckStatus`: 最后检查状态（HEALTHY/UNHEALTHY/UNKNOWN） + +### 4.2 WS 连接配置 + +```properties +# WS 连接超时（毫秒） +polygen.ws.connect.timeout=5000 + +# WS 重连延迟（毫秒） +polygen.ws.reconnect.delay=3000 +``` + +### 4.3 WS 重连配置 + +```properties +# WS 重连延迟（毫秒） +polygen.ws.reconnect.delay=3000 + +# WS 重连最大延迟（毫秒，指数退避上限） +polygen.ws.reconnect.max.delay=60000 + +# WS 重连是否启用指数退避 +polygen.ws.reconnect.exponential.backoff=true +``` + +### 4.4 合约地址配置 + +```properties +# USDC 合约地址 +usdc.contract.address=0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174 + +# ERC1155 合约地址（Polymarket） +erc1155.contract.address=0x4d97dcd97ec945f40cf65f87097ace5ea0476045 +``` + +## 五、并行方案优势 + +### 5.1 双重保障 + +- **实时性**：WS 提供秒级实时通知 +- **可靠性**：轮询作为备份，确保不遗漏交易 +- **容错性**：WS 断开时轮询继续工作，WS 恢复后自动继续 + +### 5.2 性能优化 + +- **最快响应**：哪个数据先返回就用哪个 +- **负载均衡**：两种方式并行，减少单点压力 +- **去重保护**：确保同一笔交易只处理一次 + +### 5.3 实现简单 + +- **无需切换逻辑**：两种方式始终运行 +- **独立管理**：WS 和轮询各自管理自己的状态 +- **易于维护**：逻辑清晰，易于调试 + +## 六、注意事项 + +1. **WS 重连策略**： + - WS 断开时**不断重试**，不停止轮询 + - 使用指数退避策略，避免频繁重连 + - 记录重连日志，便于监控 + +2. **去重机制**： + - 使用 `leaderId + transactionHash` 作为唯一标识 + - 在 `processTrade()` 方法内部统一处理去重 + - 使用数据库唯一约束确保并发安全 + +3. **性能考虑**： + - WS 模式下需要为每个 Leader 订阅日志 + - 大量 Leader 时可能需要优化订阅策略（批量订阅） + - 轮询间隔可以适当调整（默认 2 秒） + +4. **错误处理**： + - WS 连接失败不影响轮询服务 + - 记录详细的错误日志，便于排查问题 + - 两种方式的错误独立处理，互不影响 + +5. **元数据补齐**： + - 链上数据不包含市场元数据（conditionId、outcomeIndex 等） + - 需要通过 Gamma API 或内部元数据服务补齐 + - 轮询数据已包含元数据，无需额外补齐 + +6. **数据源标识**： + - WS 数据：`source = "onchain-ws"` + - 轮询数据：`source = "polling"` + - 便于统计和分析不同数据源的处理情况 + +## 七、实现要点 + +### 7.1 主服务启动 + +```kotlin +@Service +class CopyTradingMonitorService( + private val rpcNodeService: RpcNodeService, + private val onChainWsService: OnChainWsService, + private val pollingService: CopyTradingPollingService +) { + @PostConstruct + fun init() { + scope.launch { + // 同时启动两种监听方式 + launch { onChainWsService.start() } // WS 监听（独立协程） + launch { pollingService.start() } // 轮询监听（独立协程） + } + } +} +``` + +### 7.2 WS 服务实现（从后台配置获取 RPC） + +```kotlin +@Service +class OnChainWsService( + private val rpcNodeService: RpcNodeService, + private val copyOrderTrackingService: CopyOrderTrackingService +) { + suspend fun start() { + while (isActive) { + try { + // 从后台配置获取 WS RPC URL + val wsUrl = rpcNodeService.getWsUrl() + val httpUrl = rpcNodeService.getHttpUrl() + + // 连接并订阅 + connectAndSubscribe(wsUrl, httpUrl) + + // 连接成功后持续监听 + waitForDisconnect() + } catch (e: Exception) { + logger.warn("WS 连接失败，等待重连: ${e.message}") + delay(reconnectDelay) + } + } + } + + private suspend fun connectAndSubscribe(wsUrl: String, httpUrl: String) { + // 使用 wsUrl 和 httpUrl 连接和订阅 + // ... + } +} +``` + +**关键点**： +- 每次重连时都从 `RpcNodeService` 获取最新的 RPC 配置 +- 如果配置变更，下次重连时会自动使用新配置 +- 支持多个节点配置，自动选择可用节点 + +### 7.3 RPC 配置变更处理 + +**配置变更时的处理**： +- WS 连接断开时，下次重连会从 `RpcNodeService` 获取最新配置 +- 如果当前使用的节点不可用，`getAvailableNode()` 会自动选择下一个可用节点 +- 支持动态切换节点，无需重启服务 + +**节点选择策略**： +1. 优先使用 `lastCheckStatus = HEALTHY` 的节点 +2. 按 `priority` 排序，数字越小优先级越高 +3. 如果所有节点都不可用，返回失败（不降级到默认节点，因为默认节点可能也不可用） + +### 7.4 线程安全保证（单实例推荐方案） + +`processTrade` 方法需要保证线程安全，因为： +- WS 和轮询可能同时处理同一笔交易 +- 多个协程可能并发调用 `processTrade` +- 需要确保同一笔交易只处理一次 + +**推荐方案：使用 Mutex（应用级锁）** + +对于单实例部署，**最轻量的方案是使用 Kotlin 协程的 Mutex**： +- ✅ 无需额外依赖（不需要 Redis） +- ✅ 性能开销小（内存锁，无网络开销） +- ✅ 实现简单（Kotlin 标准库） +- ✅ 协程友好（支持 suspend 函数） + +**实现代码**： + +```kotlin +@Service +open class CopyOrderTrackingService( + // ... 其他依赖 +) { + // 使用 Mutex 保证线程安全（按交易ID锁定） + private val tradeMutexMap = ConcurrentHashMap() + + /** + * 获取或创建 Mutex（按交易ID） + */ + private fun getMutex(leaderId: Long, tradeId: String): Mutex { + val key = "${leaderId}_${tradeId}" + return tradeMutexMap.getOrPut(key) { Mutex() } + } + + /** + * 清理不再使用的 Mutex（可选，避免内存泄漏） + */ + private fun cleanupMutex(leaderId: Long, tradeId: String) { + val key = "${leaderId}_${tradeId}" + // 延迟清理，避免频繁创建/删除 + // 可以定期清理或使用 WeakReference + } + + /** + * 处理交易事件（WebSocket 或轮询） + * 使用 Mutex 保证线程安全 + */ + @Transactional + suspend fun processTrade(leaderId: Long, trade: TradeResponse, source: String): Result { + // 获取该交易的 Mutex + val mutex = getMutex(leaderId, trade.id) + + return mutex.withLock { + try { + // 1. 检查是否已处理（去重） + val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId( + leaderId, + trade.id + ) + + if (existingProcessed != null) { + if (existingProcessed.status == "FAILED") { + return@withLock Result.success(Unit) + } + return@withLock Result.success(Unit) + } + + // 检查是否已记录为失败交易 + val failedTrade = failedTradeRepository.findByLeaderIdAndLeaderTradeId( + leaderId, + trade.id + ) + if (failedTrade != null) { + return@withLock Result.success(Unit) + } + + // 2. 处理交易逻辑 + val result = when (trade.side.uppercase()) { + "BUY" -> processBuyTrade(leaderId, trade) + "SELL" -> processSellTrade(leaderId, trade) + else -> { + logger.warn("未知的交易方向: ${trade.side}") + Result.failure(IllegalArgumentException("未知的交易方向: ${trade.side}")) + } + } + + if (result.isFailure) { + logger.error( + "处理交易失败: leaderId=$leaderId, tradeId=${trade.id}, side=${trade.side}", + result.exceptionOrNull() + ) + return@withLock result + } + + // 3. 标记为已处理（成功状态） + // 由于使用了 Mutex，这里不会出现并发冲突 + try { + val processed = ProcessedTrade( + leaderId = leaderId, + leaderTradeId = trade.id, + tradeType = trade.side.uppercase(), + source = source, + status = "SUCCESS", + processedAt = System.currentTimeMillis() + ) + processedTradeRepository.save(processed) + } catch (e: Exception) { + // 理论上不会发生，但保留异常处理作为兜底 + if (isUniqueConstraintViolation(e)) { + val existing = processedTradeRepository.findByLeaderIdAndLeaderTradeId( + leaderId, + trade.id + ) + if (existing != null) { + logger.debug("交易已处理（并发检测）: leaderId=$leaderId, tradeId=${trade.id}") + return@withLock Result.success(Unit) + } + } else { + throw e + } + } + + Result.success(Unit) + } catch (e: Exception) { + logger.error("处理交易异常: leaderId=$leaderId, tradeId=${trade.id}", e) + Result.failure(e) + } + } + } +} +``` + +**关键点**： +1. **按交易ID锁定**：每个交易使用独立的 Mutex，不同交易可以并行处理 +2. **Mutex 复用**：使用 `ConcurrentHashMap` 缓存 Mutex，避免频繁创建 +3. **协程友好**：`Mutex.withLock` 是 suspend 函数，不会阻塞线程 +4. **性能优化**：只锁定同一笔交易，不影响其他交易的并发处理 + +**当前实现（基于数据库唯一约束，作为兜底）**： + +```kotlin +@Transactional +suspend fun processTrade(leaderId: Long, trade: TradeResponse, source: String): Result { + // 1. 检查是否已处理（去重） + val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId( + leaderId, + trade.id + ) + + if (existingProcessed != null) { + return Result.success(Unit) // 已处理，跳过 + } + + // 2. 处理交易逻辑 + val result = when (trade.side.uppercase()) { + "BUY" -> processBuyTrade(leaderId, trade) + "SELL" -> processSellTrade(leaderId, trade) + else -> Result.failure(IllegalArgumentException("未知的交易方向")) + } + + if (result.isFailure) { + return result + } + + // 3. 标记为已处理（使用数据库唯一约束保证并发安全） + try { + val processed = ProcessedTrade( + leaderId = leaderId, + leaderTradeId = trade.id, + tradeType = trade.side.uppercase(), + source = source, + status = "SUCCESS", + processedAt = System.currentTimeMillis() + ) + processedTradeRepository.save(processed) + } catch (e: Exception) { + // 处理唯一约束冲突（并发情况下可能发生） + if (isUniqueConstraintViolation(e)) { + // 再次检查确认状态 + val existing = processedTradeRepository.findByLeaderIdAndLeaderTradeId( + leaderId, + trade.id + ) + if (existing != null) { + logger.debug("交易已处理（并发检测）: leaderId=$leaderId, tradeId=${trade.id}") + return Result.success(Unit) + } + } else { + throw e + } + } + + return Result.success(Unit) +} +``` + +**线程安全机制**： + +1. **数据库唯一约束**： + - `UNIQUE KEY uk_leader_trade (leader_id, leader_trade_id)` + - 确保同一笔交易只能插入一次 + - 即使两个线程同时通过检查，也只有一个能成功保存 + +2. **事务隔离**： + - 使用 `@Transactional` 注解 + - 默认隔离级别（通常是 READ_COMMITTED） + - 保证事务内的数据一致性 + +3. **异常处理**： + - 捕获唯一约束冲突异常 + - 重新查询确认状态 + - 避免重复处理 + +**潜在问题**： + +虽然数据库唯一约束可以防止重复插入，但在检查 `existingProcessed` 和保存 `processed` 之间存在时间窗口（TOCTOU），可能导致： +- 两个线程同时通过检查 +- 两个线程都执行 `processBuyTrade` 或 `processSellTrade` +- 虽然只有一个能成功保存 `processed`，但可能创建了重复的订单 + +**其他方案（不推荐，仅作参考）**： + +**方案 1：使用数据库锁（SELECT FOR UPDATE）** +- ❌ 需要修改 Repository 方法 +- ❌ 增加数据库负载 +- ❌ 不适合高并发场景 + +**方案 2：使用分布式锁（Redis，仅适用于多实例）** +- ❌ 需要 Redis 依赖 +- ❌ 有网络开销 +- ❌ 单实例场景不需要 + +**总结**： + +- ✅ **单实例部署推荐**：使用 **Mutex（应用级锁）** + - 最轻量：无需额外依赖，性能开销小 + - 最简单：Kotlin 标准库，实现简单 + - 最高效：内存锁，无网络开销，支持协程并发 +- ❌ **不推荐**：仅依赖数据库唯一约束（存在 TOCTOU 问题，可能创建重复订单） +- ❌ **不推荐**：数据库锁或分布式锁（单实例场景过于复杂） +