diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/PositionCheckService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/PositionCheckService.kt index f351bff..4cfff21 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/PositionCheckService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/PositionCheckService.kt @@ -383,9 +383,9 @@ class PositionCheckService( if (ordersToMarkAsSold.isNotEmpty()) { // 有订单创建时间超过2分钟，认为仓位已被出售 try { - val currentPrice = getCurrentMarketPrice(marketId, outcomeIndex) - updateOrdersAsSold(ordersToMarkAsSold, currentPrice, copyTrading.id, marketId, outcomeIndex) - logger.debug("仓位不存在且订单创建时间超过2分钟，标记为已卖出: marketId=$marketId, outcomeIndex=$outcomeIndex, orderCount=${ordersToMarkAsSold.size}") + val currentPrice = getCurrentMarketPrice(marketId, outcomeIndex) + updateOrdersAsSold(ordersToMarkAsSold, currentPrice, copyTrading.id, marketId, outcomeIndex) + logger.debug("仓位不存在且订单创建时间超过2分钟，标记为已卖出: marketId=$marketId, outcomeIndex=$outcomeIndex, orderCount=${ordersToMarkAsSold.size}") } catch (e: Exception) { logger.warn("无法获取市场价格，跳过标记为已卖出: marketId=$marketId, outcomeIndex=$outcomeIndex, error=${e.message}") // 无法获取价格时，跳过该市场的处理，等待下次检查时再试 @@ -420,9 +420,9 @@ class PositionCheckService( // 如果已成交数量 > 0，按FIFO顺序匹配订单 try { - val currentPrice = getCurrentMarketPrice(marketId, outcomeIndex) - updateOrdersAsSoldByFIFO(orders, soldQuantity, currentPrice, - copyTrading.id, marketId, outcomeIndex) + val currentPrice = getCurrentMarketPrice(marketId, outcomeIndex) + updateOrdersAsSoldByFIFO(orders, soldQuantity, currentPrice, + copyTrading.id, marketId, outcomeIndex) } catch (e: Exception) { logger.warn("无法获取市场价格，跳过FIFO匹配: marketId=$marketId, outcomeIndex=$outcomeIndex, error=${e.message}") // 无法获取价格时，跳过该市场的处理，等待下次检查时再试 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 2fd679e..3276a5d 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 @@ -715,7 +715,7 @@ class BlockchainService( /** * 从链上查询市场条件（Condition）的结算结果 - * 通过调用 ConditionalTokens 合约的 getCondition 函数获取 payouts + * 通过调用 ConditionalTokens 合约的 conditions mapping 和 payoutNumerators mapping * * @param conditionId 市场条件ID（bytes32，必须是 0x 开头的 66 位十六进制字符串） * @return Result> @@ -734,44 +734,101 @@ class BlockchainService( val rpcApi = polygonRpcApi - // 构建 getCondition(bytes32) 函数调用 - // 函数签名: getCondition(bytes32) - val functionSelector = EthereumUtils.getFunctionSelector("getCondition(bytes32)") + // 1. 调用 conditions(bytes32) 获取 outcomeSlotCount 和 payoutDenominator + // 注意：这是一个公开的 mapping，Solidity 自动生成的 getter + // 函数签名: conditions(bytes32) returns (uint outcomeSlotCount, uint payoutDenominator) + val conditionsFunctionSelector = EthereumUtils.getFunctionSelector("conditions(bytes32)") val encodedConditionId = EthereumUtils.encodeBytes32(conditionId) - val data = functionSelector + encodedConditionId + val conditionsData = conditionsFunctionSelector + encodedConditionId // 构建 JSON-RPC 请求 - val rpcRequest = JsonRpcRequest( + val conditionsRequest = JsonRpcRequest( method = "eth_call", params = listOf( mapOf( "to" to conditionalTokensAddress, - "data" to data + "data" to conditionsData ), "latest" ) ) // 发送 RPC 请求 - val response = rpcApi.call(rpcRequest) + val conditionsResponse = rpcApi.call(conditionsRequest) - if (!response.isSuccessful || response.body() == null) { - return Result.failure(Exception("RPC 请求失败: ${response.code()} ${response.message()}")) + if (!conditionsResponse.isSuccessful || conditionsResponse.body() == null) { + return Result.failure(Exception("RPC 请求失败 (conditions): ${conditionsResponse.code()} ${conditionsResponse.message()}")) } - val rpcResponse = response.body()!! + val conditionsRpcResponse = conditionsResponse.body()!! // 检查错误 - if (rpcResponse.error != null) { - return Result.failure(Exception("RPC 错误: ${rpcResponse.error.message}")) + if (conditionsRpcResponse.error != null) { + // 记录完整的错误信息，包括 code 和 data + val errorMsg = "RPC 错误 (code=${conditionsRpcResponse.error.code}): ${conditionsRpcResponse.error.message}, data=${conditionsRpcResponse.error.data}" + logger.error("查询市场条件(conditions)出现RPC错误: conditionId=$conditionId, $errorMsg") + logger.debug("RPC 请求详情: to=$conditionalTokensAddress, data=$conditionsData") + return Result.failure(Exception(errorMsg)) } // 使用 Gson 解析 result（JsonElement） - val hexResult = rpcResponse.result?.asString + val hexResult = conditionsRpcResponse.result?.asString ?: return Result.failure(Exception("RPC 响应格式错误: result 为空")) - // 解析 ABI 编码的返回结果 - val (payoutDenominator, payouts) = EthereumUtils.decodeConditionResult(hexResult) + // 解析返回的 (outcomeSlotCount, payoutDenominator) + val cleanHex = hexResult.removePrefix("0x") + val outcomeSlotCountHex = cleanHex.substring(0, 64) + val payoutDenominatorHex = cleanHex.substring(64, 128) + + val outcomeSlotCount = BigInteger(outcomeSlotCountHex, 16).toInt() + val payoutDenominator = BigInteger(payoutDenominatorHex, 16) + + // 如果 outcomeSlotCount 为 0，说明市场尚未创建或不存在 + if (outcomeSlotCount <= 0) { + logger.debug("市场尚未创建或不存在: conditionId=$conditionId, outcomeSlotCount=$outcomeSlotCount") + return Result.success(Pair(BigInteger.ZERO, emptyList())) + } + + // 如果 payoutDenominator 为 0，说明市场尚未结算 + if (payoutDenominator == BigInteger.ZERO) { + logger.debug("市场尚未结算: conditionId=$conditionId, payoutDenominator=$payoutDenominator") + return Result.success(Pair(BigInteger.ZERO, emptyList())) + } + + // 2. 查询每个 outcome 的 payoutNumerators + val payouts = mutableListOf() + for (i in 0 until outcomeSlotCount) { + val payoutNumeratorsFunctionSelector = EthereumUtils.getFunctionSelector("payoutNumerators(bytes32,uint256)") + val encodedIndex = EthereumUtils.encodeUint256(BigInteger.valueOf(i.toLong())) + val payoutNumeratorsData = payoutNumeratorsFunctionSelector + encodedConditionId + encodedIndex + + val payoutRequest = JsonRpcRequest( + method = "eth_call", + params = listOf( + mapOf( + "to" to conditionalTokensAddress, + "data" to payoutNumeratorsData + ), + "latest" + ) + ) + + val payoutResponse = rpcApi.call(payoutRequest) + if (!payoutResponse.isSuccessful || payoutResponse.body() == null) { + logger.warn("查询 payoutNumerators 失败: index=$i") + continue + } + + val payoutRpcResponse = payoutResponse.body()!! + if (payoutRpcResponse.error != null) { + logger.warn("查询 payoutNumerators 错误: index=$i, error=${payoutRpcResponse.error.message}") + continue + } + + val payoutHex = payoutRpcResponse.result?.asString ?: "0x0" + val payout = EthereumUtils.decodeUint256(payoutHex) + payouts.add(payout) + } Result.success(Pair(payoutDenominator, payouts)) } catch (e: Exception) { diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/MarketPriceService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/MarketPriceService.kt index 9c94641..7a49a97 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/MarketPriceService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/MarketPriceService.kt @@ -31,7 +31,8 @@ class MarketPriceService( * 获取当前市场最新价 * 优先级： * 1. 链上查询市场结算结果（如果已结算，返回 1.0 或 0.0） - * 2. CLOB API 查询订单簿价格（最准确，使用 bestBid） + * 2. CLOB API 查询订单簿价格（最准确，优先使用，使用 bestBid） + * 3. Gamma Market API 查询市场价格（快速，作为备选） * * 价格会被截位到 4 位小数（向下截断，不四舍五入），用于显示和后续计算 * @@ -48,15 +49,22 @@ class MarketPriceService( return chainPrice.setScale(4, java.math.RoundingMode.DOWN) } - // 2. 从 CLOB API 查询订单簿价格（最准确） + // 2. 从 CLOB API 查询订单簿价格（最准确，优先使用） val orderbookPrice = getPriceFromClobOrderbook(marketId, outcomeIndex) if (orderbookPrice != null) { // 截位到 4 位小数（向下截断，不四舍五入） return orderbookPrice.setScale(4, java.math.RoundingMode.DOWN) } + // 3. 从 Gamma Market API 查询市场价格（作为备选） + val marketPrice = getPriceFromGammaMarket(marketId, outcomeIndex) + if (marketPrice != null) { + // 截位到 4 位小数（向下截断，不四舍五入） + return marketPrice.setScale(4, java.math.RoundingMode.DOWN) + } + // 如果所有数据源都失败，抛出异常 - val errorMsg = "无法获取市场价格: marketId=$marketId, outcomeIndex=$outcomeIndex (链上查询和订单簿查询均失败)" + val errorMsg = "无法获取市场价格: marketId=$marketId, outcomeIndex=$outcomeIndex (链上查询、订单簿查询和 Market API 均失败)" logger.error(errorMsg) throw IllegalStateException(errorMsg) } @@ -106,6 +114,61 @@ class MarketPriceService( } } + /** + * 从 Gamma Market API 获取价格 + * 使用 outcomePrices 字段，格式通常为 JSON 字符串 "[\"0.5\", \"0.5\"]" + * 如果查询失败或 outcomePrices 为空，返回 null + */ + private suspend fun getPriceFromGammaMarket(marketId: String, outcomeIndex: Int): BigDecimal? { + return try { + val gammaApi = retrofitFactory.createGammaApi() + val marketResponse = gammaApi.listMarkets(conditionIds = listOf(marketId)) + + if (!marketResponse.isSuccessful || marketResponse.body() == null) { + logger.debug("Gamma Market API 查询失败: marketId=$marketId, code=${marketResponse.code()}") + return null + } + + val markets = marketResponse.body()!! + if (markets.isEmpty()) { + logger.debug("Gamma Market API 未找到市场: marketId=$marketId") + return null + } + + val market = markets.first() + + // 尝试从 outcomePrices 字段获取价格 + val outcomePricesStr = market.outcomePrices + if (outcomePricesStr.isNullOrBlank()) { + logger.debug("Market outcomePrices 为空: marketId=$marketId") + return null + } + + // 解析 outcomePrices（通常是 JSON 数组字符串） + val outcomePrices = try { + // 移除首尾的方括号和引号，按逗号分割 + val cleanStr = outcomePricesStr.trim().removeSurrounding("[", "]") + cleanStr.split(",").map { + it.trim().removeSurrounding("\"").toSafeBigDecimal() + } + } catch (e: Exception) { + logger.warn("解析 outcomePrices 失败: marketId=$marketId, outcomePrices=$outcomePricesStr, error=${e.message}") + null + } + + if (outcomePrices != null && outcomeIndex < outcomePrices.size) { + val price = outcomePrices[outcomeIndex] + logger.debug("从 Gamma Market API 获取价格: marketId=$marketId, outcomeIndex=$outcomeIndex, price=$price") + return price + } + + null + } catch (e: Exception) { + logger.debug("Gamma Market API 查询异常: marketId=$marketId, outcomeIndex=$outcomeIndex, error=${e.message}") + null + } + } + /** * 从 CLOB API 查询订单簿价格 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 index 3ee5e7c..328ffb4 100644 --- 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 @@ -89,6 +89,8 @@ class OnChainWsService( private suspend fun handleLeaderTransaction(leaderId: Long, txHash: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) { val leader = monitoredLeaders[leaderId] ?: return + logger.debug("开始处理 Leader 交易: leaderId=$leaderId, txHash=$txHash, leaderAddress=${leader.leaderAddress}") + try { // 获取交易 receipt val receiptRequest = JsonRpcRequest( @@ -98,11 +100,13 @@ class OnChainWsService( val receiptResponse = rpcApi.call(receiptRequest) if (!receiptResponse.isSuccessful || receiptResponse.body() == null) { + logger.warn("获取交易 receipt 失败: leaderId=$leaderId, txHash=$txHash, code=${receiptResponse.code()}") return } val receiptRpcResponse = receiptResponse.body()!! if (receiptRpcResponse.error != null || receiptRpcResponse.result == null) { + logger.warn("交易 receipt 错误: leaderId=$leaderId, txHash=$txHash, error=${receiptRpcResponse.error}") return } @@ -118,8 +122,12 @@ class OnChainWsService( } // 解析 receipt 中的 Transfer 日志 - val logs = receiptJson.getAsJsonArray("logs") ?: return + val logs = receiptJson.getAsJsonArray("logs") ?: run { + logger.warn("交易 receipt 中没有日志: leaderId=$leaderId, txHash=$txHash") + return + } val (erc20Transfers, erc1155Transfers) = OnChainWsUtils.parseReceiptTransfers(logs) + logger.debug("解析交易日志: leaderId=$leaderId, txHash=$txHash, erc20Transfers=${erc20Transfers.size}, erc1155Transfers=${erc1155Transfers.size}") // 解析交易信息 val trade = OnChainWsUtils.parseTradeFromTransfers( @@ -132,12 +140,15 @@ class OnChainWsService( ) if (trade != null) { + logger.info("成功解析交易: leaderId=$leaderId, txHash=$txHash, side=${trade.side}, market=${trade.market}, size=${trade.size}") // 调用 processTrade 处理交易 copyOrderTrackingService.processTrade( leaderId = leaderId, trade = trade, source = "onchain-ws" ) + } else { + logger.warn("无法解析交易（返回 null）: leaderId=$leaderId, txHash=$txHash, erc20Transfers=${erc20Transfers.size}, erc1155Transfers=${erc1155Transfers.size}") } } catch (e: Exception) { logger.error("处理 Leader 交易失败: leaderId=$leaderId, txHash=$txHash, ${e.message}", e) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/OnChainWsUtils.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/OnChainWsUtils.kt index 7ba35c6..f5f1a0e 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/OnChainWsUtils.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/OnChainWsUtils.kt @@ -207,6 +207,7 @@ object OnChainWsUtils { usdcRaw = usdcIn } else { // 无法判断交易方向 + logger.debug("无法判断交易方向: txHash=$txHash, bestInId=$bestInId, bestInVal=$bestInVal, bestOutId=$bestOutId, bestOutVal=$bestOutVal, usdcOut=$usdcOut, usdcIn=$usdcIn") return null } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/UnifiedOnChainWsService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/UnifiedOnChainWsService.kt index 22d02ab..94d45f2 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/UnifiedOnChainWsService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/UnifiedOnChainWsService.kt @@ -1,6 +1,7 @@ package com.wrbug.polymarketbot.service.copytrading.monitor import com.google.gson.Gson +import com.google.gson.JsonArray import com.google.gson.JsonObject import com.wrbug.polymarketbot.api.* import com.wrbug.polymarketbot.service.system.RpcNodeService @@ -19,6 +20,7 @@ import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Value import org.springframework.stereotype.Service import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicInteger /** * 统一的链上 WebSocket 服务 @@ -38,27 +40,8 @@ class UnifiedOnChainWsService( private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) - // WebSocket 连接（唯一） - private var webSocket: WebSocket? = null - @Volatile - private var isConnected = false - - // 订阅ID计数器（用于请求 ID） - private var requestIdCounter = 0 - - // 连接任务（确保只有一个连接任务在运行） - private var connectionJob: Job? = null - - // 存储所有订阅：subscriptionId -> 订阅信息 - private val subscriptions = ConcurrentHashMap() - - // 存储请求 ID 到订阅 ID 的映射：requestId -> subscriptionId - // 用于在收到订阅响应时，将 subscription ID 关联到对应的订阅 - private val requestIdToSubscriptionId = ConcurrentHashMap() - - // 存储 RPC subscriptionId 到订阅 ID 的映射：rpcSubscriptionId -> subscriptionId - // 用于在收到日志通知时，知道是哪个订阅 - private val rpcSubscriptionIdToSubscriptionId = ConcurrentHashMap() + // 存储所有地址的连接：address -> AddressWsConnection + private val addressConnections = ConcurrentHashMap() /** * 订阅信息 @@ -88,31 +71,24 @@ class UnifiedOnChainWsService( callback: suspend (String, OkHttpClient, EthereumRpcApi) -> Unit ): Boolean { try { - // 如果已经订阅，先取消 - if (subscriptions.containsKey(subscriptionId)) { - unsubscribe(subscriptionId) + val lowerAddress = address.lowercase() + + // 找到或创建该地址的连接 + val connection = addressConnections.computeIfAbsent(lowerAddress) { + AddressWsConnection(it).apply { start() } } // 创建订阅信息 val subscription = SubscriptionInfo( subscriptionId = subscriptionId, - address = address.lowercase(), + address = lowerAddress, entityType = entityType, entityId = entityId, callback = callback ) - subscriptions[subscriptionId] = subscription - - // 如果已连接，立即订阅 - if (isConnected) { - scope.launch { - subscribeAddress(subscription) - } - } else { - // 如果未连接，启动连接 - startConnection() - } + // 添加订阅 + connection.addSubscription(subscription) logger.info("订阅地址监听: subscriptionId=$subscriptionId, address=$address, entityType=$entityType, entityId=$entityId") return true @@ -126,434 +102,37 @@ class UnifiedOnChainWsService( * 取消订阅 */ fun unsubscribe(subscriptionId: String) { - val subscription = subscriptions.remove(subscriptionId) - - if (subscription != null && isConnected) { - // 取消该订阅的所有 RPC 订阅 - scope.launch { - // 查找该订阅的所有 RPC subscriptionId - val rpcSubscriptionIds = rpcSubscriptionIdToSubscriptionId.entries - .filter { it.value == subscriptionId } - .map { it.key } + // 遍历所有连接找到含有该订阅的连接 + for (connection in addressConnections.values) { + if (connection.hasSubscription(subscriptionId)) { + connection.removeSubscription(subscriptionId) - for (rpcSubId in rpcSubscriptionIds) { - unsubscribeRpc(rpcSubId) - rpcSubscriptionIdToSubscriptionId.remove(rpcSubId) - } - } - - logger.info("取消订阅: subscriptionId=$subscriptionId") - } - - // 如果没有订阅了，停止连接 - if (subscriptions.isEmpty()) { - stop() - } - } - - /** - * 启动连接（如果还没有连接） - */ - private fun startConnection() { - // 如果没有订阅，不启动连接 - if (subscriptions.isEmpty()) { - return - } - - // 如果连接任务已经在运行，不重复启动 - if (connectionJob != null && connectionJob!!.isActive) { - return - } - - // 启动连接任务 - connectionJob = scope.launch { - startConnectionLoop() - } - } - - /** - * 启动连接循环 - */ - private suspend fun startConnectionLoop() { - while (scope.isActive) { - try { - // 如果没有订阅，停止连接 - if (subscriptions.isEmpty()) { - logger.info("没有订阅，停止连接") - stop() - break + // 如果该连接没有订阅了，停止并移除 + if (connection.isSubscriptionsEmpty()) { + connection.stop() + addressConnections.remove(connection.address) + logger.info("连接已无订阅，关闭连接: address=${connection.address}") } - // 如果已经连接，等待断开 - if (isConnected && webSocket != null) { - waitForDisconnect() - continue - } - - // 获取可用的 RPC 节点 - val wsUrl = rpcNodeService.getWsUrl() - val httpUrl = rpcNodeService.getHttpUrl() - - if (wsUrl.isBlank() || httpUrl.isBlank()) { - logger.warn("没有可用的 RPC 节点，等待重试...") - delay(reconnectDelay) - continue - } - - logger.info("连接链上 WebSocket: $wsUrl (${subscriptions.size} 个订阅)") - - // 创建 HTTP 客户端（用于 RPC 调用） - val httpClient = createHttpClient() - - // 创建 RPC API 客户端 - val rpcApi = retrofitFactory.createEthereumRpcApi(httpUrl) - - // 连接 WebSocket - connectWebSocket(wsUrl, httpClient, rpcApi) - - // 等待连接建立 - waitForConnect() - - // 如果连接成功，订阅所有地址 - if (isConnected) { - logger.info("WebSocket 连接已建立，开始订阅") - for (subscription in subscriptions.values) { - subscribeAddress(subscription) - } - - // 等待连接断开 - waitForDisconnect() - } - - // 连接断开后，如果没有订阅了，不再重连 - if (subscriptions.isEmpty()) { - logger.info("没有订阅，停止重连") - break - } - - // 等待后重连 - logger.info("WebSocket 连接断开，等待 ${reconnectDelay}ms 后重连") - delay(reconnectDelay) - - } catch (e: Exception) { - logger.error("连接异常: ${e.message}", e) - 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) { - // 先关闭旧连接 - webSocket?.close(1000, "重新连接") - 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 连接成功") - } - - 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 - } - }) - } - - /** - * 等待连接建立 - */ - private suspend fun waitForConnect() { - var waited = 0L - val timeout = 15000L // 15秒超时 - - while (!isConnected && waited < timeout) { - delay(100) - waited += 100 - } - - if (!isConnected) { - logger.warn("WebSocket 连接超时，等待重连") - } - } - - /** - * 等待连接断开 - */ - private suspend fun waitForDisconnect() { - while (isConnected && scope.isActive) { - delay(1000) - } - } - - /** - * 订阅地址（为每个地址订阅 6 个事件） - */ - private suspend fun subscribeAddress(subscription: SubscriptionInfo) { - if (webSocket == null || !isConnected) { - return - } - - val address = subscription.address - val walletTopic = OnChainWsUtils.addressToTopic32(address) - val subscriptionId = subscription.subscriptionId - - try { - // 订阅 USDC Transfer (from wallet) - subscribeLogs(OnChainWsUtils.USDC_CONTRACT, listOf(OnChainWsUtils.ERC20_TRANSFER_TOPIC, walletTopic), subscriptionId) - - // 订阅 USDC Transfer (to wallet) - subscribeLogs(OnChainWsUtils.USDC_CONTRACT, listOf(OnChainWsUtils.ERC20_TRANSFER_TOPIC, null, walletTopic), subscriptionId) - - // 订阅 ERC1155 TransferSingle (from wallet) - subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_SINGLE_TOPIC, null, walletTopic), subscriptionId) - - // 订阅 ERC1155 TransferSingle (to wallet) - subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_SINGLE_TOPIC, null, null, walletTopic), subscriptionId) - - // 订阅 ERC1155 TransferBatch (from wallet) - subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_BATCH_TOPIC, null, walletTopic), subscriptionId) - - // 订阅 ERC1155 TransferBatch (to wallet) - subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_BATCH_TOPIC, null, null, walletTopic), subscriptionId) - - logger.debug("已订阅地址: subscriptionId=$subscriptionId, address=$address") - } catch (e: Exception) { - logger.error("订阅地址失败: subscriptionId=$subscriptionId, address=$address, error=${e.message}", e) - } - } - - /** - * 订阅日志 - */ - private fun subscribeLogs(address: String, topics: List, subscriptionId: String) { - val ws = webSocket ?: return - - val params = mapOf( - "address" to address.lowercase(), - "topics" to topics.filterNotNull() - ) - - val requestId = ++requestIdCounter - requestIdToSubscriptionId[requestId] = subscriptionId - - val request = mapOf( - "jsonrpc" to "2.0", - "id" to requestId, - "method" to "eth_subscribe", - "params" to listOf("logs", params) - ) - - val message = gson.toJson(request) - ws.send(message) - } - - /** - * 取消 RPC 订阅 - */ - private fun unsubscribeRpc(rpcSubscriptionId: String) { - val ws = webSocket ?: return - - val requestId = ++requestIdCounter - val request = mapOf( - "jsonrpc" to "2.0", - "id" to requestId, - "method" to "eth_unsubscribe", - "params" to listOf(rpcSubscriptionId) - ) - - val message = gson.toJson(request) - ws.send(message) - } - - /** - * 处理 WebSocket 消息 - */ - private suspend fun handleMessage(text: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) { - try { - val message = gson.fromJson(text, JsonObject::class.java) - - // 处理订阅响应 - if (message.has("result") && message.has("id")) { - val requestId = message.get("id")?.asInt - val rpcSubscriptionId = message.get("result")?.asString - - if (requestId != null && rpcSubscriptionId != null) { - val subscriptionId = requestIdToSubscriptionId.remove(requestId) - if (subscriptionId != null) { - // 保存 RPC subscriptionId 到订阅的映射 - rpcSubscriptionIdToSubscriptionId[rpcSubscriptionId] = subscriptionId - logger.debug("订阅成功: subscriptionId=$subscriptionId, rpcSubscriptionId=$rpcSubscriptionId") - } - } + logger.info("取消订阅: subscriptionId=$subscriptionId") return } - - // 处理日志通知 - if (message.has("params")) { - val params = message.getAsJsonObject("params") - val subscriptionIdParam = params.get("subscription")?.asString - val result = params.getAsJsonObject("result") - - if (result != null) { - val txHash = result.get("transactionHash")?.asString - if (txHash != null && subscriptionIdParam != null) { - // 根据 RPC subscriptionId 找到对应的订阅 - val subscriptionId = rpcSubscriptionIdToSubscriptionId[subscriptionIdParam] - if (subscriptionId != null) { - // 处理交易，分发给对应的订阅者 - processTransactionForSubscription(txHash, subscriptionId, httpClient, rpcApi) - } else { - // 如果没有找到订阅，可能是新订阅还未建立映射，尝试处理所有订阅 - processTransaction(txHash, httpClient, rpcApi) - } - } - } - } - } catch (e: Exception) { - logger.error("处理 WebSocket 消息失败: ${e.message}", e) } } /** - * 处理交易（为特定订阅） - * 直接调用订阅的回调 - */ - private suspend fun processTransactionForSubscription( - txHash: String, - subscriptionId: String, - httpClient: OkHttpClient, - rpcApi: EthereumRpcApi - ) { - val subscription = subscriptions[subscriptionId] ?: return - - try { - subscription.callback(txHash, httpClient, rpcApi) - } catch (e: Exception) { - logger.error("调用订阅回调失败: subscriptionId=$subscriptionId, txHash=$txHash, error=${e.message}", e) - } - } - - /** - * 处理交易（为所有订阅，用于兼容） - * 解析交易中的 Transfer 事件，分发给所有订阅者 - */ - 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 - val receiptJson = receiptRpcResponse.result.asJsonObject - - // 解析 receipt 中的 Transfer 日志 - val logs = receiptJson.getAsJsonArray("logs") ?: return - val (erc20Transfers, erc1155Transfers) = OnChainWsUtils.parseReceiptTransfers(logs) - - // 为每个订阅检查是否匹配，如果匹配则调用回调 - for (subscription in subscriptions.values) { - val address = subscription.address - - // 检查该地址是否参与了交易（通过检查 Transfer 日志） - val isInvolved = erc20Transfers.any { - it.from.lowercase() == address || it.to.lowercase() == address - } || erc1155Transfers.any { - it.from.lowercase() == address || it.to.lowercase() == address - } - - if (isInvolved) { - // 该地址参与了交易，调用回调 - try { - subscription.callback(txHash, httpClient, rpcApi) - } catch (e: Exception) { - logger.error("调用订阅回调失败: subscriptionId=${subscription.subscriptionId}, txHash=$txHash, error=${e.message}", e) - } - } - } - } catch (e: Exception) { - logger.error("处理交易失败: txHash=$txHash, ${e.message}", e) - } - } - - /** - * 停止连接 + * 停止所有服务 */ fun stop() { - connectionJob?.cancel() - connectionJob = null - - // 关闭 WebSocket 连接 - webSocket?.close(1000, "停止监听") - webSocket = null - isConnected = false - - // 清空订阅信息 - subscriptions.clear() - requestIdToSubscriptionId.clear() - rpcSubscriptionIdToSubscriptionId.clear() + for (connection in addressConnections.values) { + connection.stop() + } + addressConnections.clear() } @PostConstruct fun init() { - // 服务启动时不自动连接，等待有订阅时再连接 - logger.info("统一链上 WebSocket 服务已初始化") + logger.info("统一链上 WebSocket 服务已初始化 (独立连接模式)") } @PreDestroy @@ -561,5 +140,295 @@ class UnifiedOnChainWsService( stop() scope.cancel() } + + /** + * 单个地址的 WebSocket 连接管理 + */ + inner class AddressWsConnection(val address: String) { + private var webSocket: WebSocket? = null + @Volatile + private var isConnected = false + + // 订阅ID计数器（用于请求 ID） + private var requestIdCounter = AtomicInteger(0) + + // 连接任务 + private var connectionJob: Job? = null + + // 该连接下的所有订阅：subscriptionId -> SubscriptionInfo + // 理论上一个地址可能被多个业务订阅（如：既是被跟单者又是普通监控），虽然业务上通常只有一个 + private val subscriptions = ConcurrentHashMap() + + // 存储请求 ID 到订阅 ID 的映射：requestId -> subscriptionId + private val requestIdToSubscriptionId = ConcurrentHashMap() + + // 存储 RPC subscriptionId 到订阅 ID 的映射：rpcSubscriptionId -> subscriptionId + private val rpcSubscriptionIdToSubscriptionId = ConcurrentHashMap() + + fun start() { + if (connectionJob != null && connectionJob!!.isActive) return + connectionJob = scope.launch { + startConnectionLoop() + } + } + + fun stop() { + connectionJob?.cancel() + connectionJob = null + webSocket?.close(1000, "停止监听") + webSocket = null + isConnected = false + subscriptions.clear() + requestIdToSubscriptionId.clear() + rpcSubscriptionIdToSubscriptionId.clear() + } + + fun addSubscription(subscription: SubscriptionInfo) { + // 如果已经存在，先移除旧的 + removeSubscription(subscription.subscriptionId) + subscriptions[subscription.subscriptionId] = subscription + + // 如果已经连接，立即发送链上订阅请求 + if (isConnected) { + scope.launch { + subscribeAddressOnChain(subscription) + } + } + } + + fun removeSubscription(subscriptionId: String) { + subscriptions.remove(subscriptionId) + // 不需要显式发送 eth_unsubscribe，因为连接是 per-address 的， + // 只要只要连接还在，就保持该地址相关的所有 logs 订阅。 + // 只有当所有 subscription 都移除了，连接才会关闭。 + } + + fun hasSubscription(subscriptionId: String): Boolean { + return subscriptions.containsKey(subscriptionId) + } + + fun isSubscriptionsEmpty(): Boolean { + return subscriptions.isEmpty() + } + + private suspend fun startConnectionLoop() { + while (scope.isActive) { + try { + if (subscriptions.isEmpty()) { + // 如果启动循环时还没订阅（不太可能，通常是先 addSubscription 再 start，或者是 start 后 addSubscription） + // 或者订阅被清空了，外部应当掉 stop，但这里作为防守 + delay(1000) + continue + } + + if (isConnected && webSocket != null) { + waitForDisconnect() + continue + } + + // 获取可用的 RPC 节点 + val wsUrl = rpcNodeService.getWsUrl() + val httpUrl = rpcNodeService.getHttpUrl() + + logger.info("[$address] 连接链上 WebSocket: $wsUrl") + + val httpClient = createHttpClient() + val rpcApi = retrofitFactory.createEthereumRpcApi(httpUrl) + + connectWebSocket(wsUrl, httpClient, rpcApi) + waitForConnect() + + if (isConnected) { + logger.info("[$address] WebSocket 连接已建立，开始注册订阅") + // 重新为所有订阅注册链上监听 + for (subscription in subscriptions.values) { + subscribeAddressOnChain(subscription) + } + waitForDisconnect() + } + + logger.info("[$address] WebSocket 连接断开，等待 ${reconnectDelay}ms 后重连") + delay(reconnectDelay) + + } catch (e: Exception) { + logger.error("[$address] 连接异常: ${e.message}", e) + delay(reconnectDelay) + } + } + } + + private fun connectWebSocket(wsUrl: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) { + webSocket?.close(1000, "重新连接") + 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("[$address] 链上 WebSocket 连接成功") + } + + 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("[$address] 链上 WebSocket 连接关闭: code=$code, reason=$reason") + } + + override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { + isConnected = false + logger.warn("[$address] 链上 WebSocket 连接已关闭: code=$code, reason=$reason") + } + + override fun onFailure(webSocket: WebSocket, t: Throwable, response: okhttp3.Response?) { + logger.error("[$address] 链上 WebSocket 连接失败: ${t.message}", t) + isConnected = false + } + }) + } + + private suspend fun subscribeAddressOnChain(subscription: SubscriptionInfo) { + if (webSocket == null || !isConnected) return + + val walletTopic = OnChainWsUtils.addressToTopic32(address) + val subId = subscription.subscriptionId + + try { + // 订阅该地址相关的所有事件 + // USDC Transfer (from/to) + subscribeLogs(OnChainWsUtils.USDC_CONTRACT, listOf(OnChainWsUtils.ERC20_TRANSFER_TOPIC, walletTopic), subId) + subscribeLogs(OnChainWsUtils.USDC_CONTRACT, listOf(OnChainWsUtils.ERC20_TRANSFER_TOPIC, null, walletTopic), subId) + + // ERC1155 TransferSingle (from/to) + subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_SINGLE_TOPIC, null, walletTopic), subId) + subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_SINGLE_TOPIC, null, null, walletTopic), subId) + + // ERC1155 TransferBatch (from/to) + subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_BATCH_TOPIC, null, walletTopic), subId) + subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_BATCH_TOPIC, null, null, walletTopic), subId) + + logger.debug("[$address] 已发送链上订阅请求: subscriptionId=$subId") + } catch (e: Exception) { + logger.error("[$address] 发送链上订阅请求失败: error=${e.message}", e) + } + } + + private fun subscribeLogs(contractAddress: String, topics: List, subscriptionId: String) { + val ws = webSocket ?: return + + val topicsArray = gson.toJsonTree(topics).asJsonArray + val logParams = JsonObject() + logParams.addProperty("address", contractAddress.lowercase()) + logParams.add("topics", topicsArray) + + val requestId = requestIdCounter.incrementAndGet() + requestIdToSubscriptionId[requestId] = subscriptionId + + val request = JsonObject() + request.addProperty("jsonrpc", "2.0") + request.addProperty("id", requestId) + request.addProperty("method", "eth_subscribe") + val paramsArray = JsonArray() + paramsArray.add("logs") + paramsArray.add(logParams) + request.add("params", paramsArray) + + ws.send(gson.toJson(request)) + } + + private suspend fun handleMessage(text: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) { + try { + val message = gson.fromJson(text, JsonObject::class.java) + + // 1. 处理订阅响应 (eth_subscribe response) + if (message.has("result") && message.has("id")) { + val requestId = message.get("id")?.asInt + val rpcSubscriptionId = message.get("result")?.asString + + if (requestId != null && rpcSubscriptionId != null) { + val subId = requestIdToSubscriptionId.remove(requestId) + if (subId != null) { + rpcSubscriptionIdToSubscriptionId[rpcSubscriptionId] = subId + logger.debug("[$address] 链上订阅成功: mapped connection rpcSubId=$rpcSubscriptionId to localSubId=$subId") + } + } + return + } + + // 2. 处理日志通知 (eth_subscription) + val method = message.get("method")?.asString + if (method == "eth_subscription") { + val params = message.getAsJsonObject("params") ?: return + val rpcSubParam = params.get("subscription")?.asString + val result = params.getAsJsonObject("result") ?: return + val txHash = result.get("transactionHash")?.asString + + if (txHash != null && rpcSubParam != null) { + // 找到触发此通知的本地订阅 ID + // 因为我们在这个连接里只订阅了 this.address，所以理论上所有通知都跟这个 address 有关 + // 但我们需要找到对应的 callback + val localSubId = rpcSubscriptionIdToSubscriptionId[rpcSubParam] + + if (localSubId != null) { + val subscription = subscriptions[localSubId] + if (subscription != null) { + logger.info("[$address] 收到交易通知: txHash=$txHash, subId=$localSubId") + runCatching { + subscription.callback(txHash, httpClient, rpcApi) + }.onFailure { e -> + logger.error("[$address] 回调执行失败: ${e.message}", e) + } + } + } else { + // 找不到具体是哪个订阅请求触发的（可能是重启后之前的订阅残留？或者映射丢失？） + // 在单地址单连接模式下，只要是这个 connection 收到的，肯定是关于这个 address 的 + // 我们可以尝试通知所有订阅者（通常一个地址只有一个订阅者，除非此地址既是Leader又是User） + logger.warn("[$address] 未找到映射的订阅ID: rpcSubId=$rpcSubParam. 广播给所有订阅者.") + subscriptions.values.forEach { sub -> + runCatching { + sub.callback(txHash, httpClient, rpcApi) + }.onFailure { e -> + logger.error("[$address] 广播回调执行失败: ${e.message}", e) + } + } + } + } + } + } catch (e: Exception) { + logger.error("[$address] 处理消息失败: ${e.message}", e) + } + } + + private suspend fun waitForConnect() { + var waited = 0L + val timeout = 15000L + while (!isConnected && waited < timeout) { + delay(100) + waited += 100 + } + if (!isConnected) logger.warn("[$address] WebSocket 连接超时") + } + + private suspend fun waitForDisconnect() { + while (isConnected && scope.isActive) { + delay(1000) + } + } + + private fun createHttpClient(): OkHttpClient { + val proxy = getProxyConfig() + val builder = createClient() + if (proxy != null) builder.proxy(proxy) + return builder.build() + } + } } 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 913ad9f..2bc31e4 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 @@ -253,7 +253,7 @@ open class CopyOrderTrackingService( // 先计算跟单金额（用于仓位检查） // 注意：这里先计算金额，即使后续被过滤也会记录 val tradePrice = trade.price.toSafeBigDecimal() - val buyQuantity = try { + var buyQuantity = try { calculateBuyQuantity(trade, copyTrading) } catch (e: Exception) { logger.warn("计算买入数量失败: ${e.message}", e) @@ -369,10 +369,14 @@ open class CopyOrderTrackingService( // 买入数量已在过滤检查前计算，这里直接使用 // 如果数量为0或负数，跳过 if (buyQuantity.lte(BigDecimal.ZERO)) { - logger.warn("计算出的买入数量为0或负数，跳过: copyTradingId=${copyTrading.id}, tradeId=${trade.id}") + logger.warn("计算得到的买入数量为0，跳过跟单: copyTradingId=${copyTrading.id}, tradeId=${trade.id}") continue } + if (buyQuantity.lt(BigDecimal.ONE)) { + logger.warn("计算得到的买入数量小于1，自动调整为1 (Polymarket 最小下单数量): copyTradingId=${copyTrading.id}, tradeId=${trade.id}, originalQuantity=$buyQuantity") + buyQuantity = BigDecimal.ONE + } // 验证订单数量限制（仅比例模式） var finalBuyQuantity = buyQuantity if (copyTrading.copyMode == "RATIO") { @@ -656,8 +660,8 @@ open class CopyOrderTrackingService( private fun calculateBuyQuantity(trade: TradeResponse, copyTrading: CopyTrading): BigDecimal { return when (copyTrading.copyMode) { "RATIO" -> { - // 比例模式：Leader 数量 × 比例 - trade.size.toSafeBigDecimal().multi(copyTrading.copyRatio) + // 比例模式：Leader 数量 × (比例 / 100) + trade.size.toSafeBigDecimal().multi(copyTrading.copyRatio.div(100)) } "FIXED" -> { @@ -689,7 +693,7 @@ open class CopyOrderTrackingService( val leader = leaderRepository.findById(copyTrading.leaderId).orElse(null) ?: run { logger.warn("Leader 不存在，使用默认比例: leaderId=${copyTrading.leaderId}") - return leaderSellQuantity.multi(copyTrading.copyRatio) + return leaderSellQuantity.multi(copyTrading.copyRatio.div(100)) } // 创建不需要认证的 CLOB API 客户端（用于查询公开的交易数据） @@ -760,7 +764,7 @@ open class CopyOrderTrackingService( // 如果无法计算总比例（查询失败），使用默认比例 if (totalLeaderQuantity.lte(BigDecimal.ZERO)) { logger.warn("无法计算总比例（Leader 买入数量为 0），使用默认比例: copyTradingId=${copyTrading.id}") - return leaderSellQuantity.multi(copyTrading.copyRatio) + return leaderSellQuantity.multi(copyTrading.copyRatio.div(100)) } // 计算实际比例：跟单买入数量 / Leader 买入数量 @@ -835,15 +839,23 @@ open class CopyOrderTrackingService( ) } "RATIO" -> { - // 比例模式：直接使用配置的 copyRatio - leaderSellTrade.size.toSafeBigDecimal().multi(copyTrading.copyRatio) + // 比例模式：直接使用配置的 copyRatio (需要除以100) + leaderSellTrade.size.toSafeBigDecimal().multi(copyTrading.copyRatio.div(100)) } else -> { logger.warn("不支持的 copyMode: ${copyTrading.copyMode}，使用默认比例模式") - leaderSellTrade.size.toSafeBigDecimal().multi(copyTrading.copyRatio) + leaderSellTrade.size.toSafeBigDecimal().multi(copyTrading.copyRatio.div(100)) } } + // 如果需要卖出的数量小于1（但大于0），自动调整为1（Polymarket 最小下单数量） + // 注意：如果实际持有数量不足1，后续的 totalMatched 检查会拦截 + var finalNeedMatch = needMatch + if (finalNeedMatch.gt(BigDecimal.ZERO) && finalNeedMatch.lt(BigDecimal.ONE)) { + logger.warn("计算得到的卖出数量小于1，自动调整为1: copyTradingId=${copyTrading.id}, original=$needMatch") + finalNeedMatch = BigDecimal.ONE + } + // 4. 获取tokenId（直接使用outcomeIndex，支持多元市场） val tokenIdResult = blockchainService.getTokenId(leaderSellTrade.market, leaderSellTrade.outcomeIndex) if (tokenIdResult.isFailure) { @@ -867,7 +879,7 @@ open class CopyOrderTrackingService( // 6. 按FIFO顺序匹配，计算实际可以卖出的数量 // 使用计算出的实际卖出价格（而不是 Leader 价格）来创建匹配明细 var totalMatched = BigDecimal.ZERO - var remaining = needMatch + var remaining = finalNeedMatch val matchDetails = mutableListOf() for (order in unmatchedOrders) { @@ -904,6 +916,11 @@ open class CopyOrderTrackingService( return } + if (totalMatched.lt(BigDecimal.ONE)) { + logger.warn("卖出数量小于1，跳过卖出 (Polymarket 最小下单数量为 1): copyTradingId=${copyTrading.id}, tradeId=${leaderSellTrade.id}, quantity=$totalMatched") + return + } + // 7. 解密 API 凭证 val apiSecret = try { decryptApiSecret(account) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/OrderStatusUpdateService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/OrderStatusUpdateService.kt index ad6fa12..e9a1fcc 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/OrderStatusUpdateService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/OrderStatusUpdateService.kt @@ -143,14 +143,16 @@ class OrderStatusUpdateService( // 计算30秒前的时间戳 val thirtySecondsAgo = System.currentTimeMillis() - 30000 - // 查询30秒前创建的订单 - val ordersToCheck = copyOrderTrackingRepository.findByCreatedAtBefore(thirtySecondsAgo) + // 查询30秒前创建的订单,并过滤掉已经完全匹配的订单 + // 已经完全匹配的订单(status = "fully_matched")不需要再检查 + val allOrdersToCheck = copyOrderTrackingRepository.findByCreatedAtBefore(thirtySecondsAgo) + val ordersToCheck = allOrdersToCheck.filter { it.status != "fully_matched" } if (ordersToCheck.isEmpty()) { return } - logger.debug("检查 ${ordersToCheck.size} 个30秒前创建的订单是否成交") + logger.debug("检查 ${ordersToCheck.size} 个30秒前创建的订单是否成交 (已过滤 ${allOrdersToCheck.size - ordersToCheck.size} 个已完全匹配的订单)") // 按账户分组，避免重复创建 API 客户端 val ordersByAccount = ordersToCheck.groupBy { it.accountId } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/TelegramNotificationService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/TelegramNotificationService.kt index 24860ae..11a9992 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/TelegramNotificationService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/TelegramNotificationService.kt @@ -686,6 +686,13 @@ class TelegramNotificationService( else -> side } + // 获取图标 + val icon = when (side.uppercase()) { + "BUY" -> "🚀" + "SELL" -> "💰" + else -> "📣" + } + // 构建账户信息（格式：账户名(钱包地址)） val accountInfo = buildAccountInfo(accountName, walletAddress, unknownAccount) @@ -761,7 +768,7 @@ class TelegramNotificationService( val priceDisplay = formatPrice(price) val sizeDisplay = formatQuantity(size) - return """✅ $orderCreatedSuccess + return """$icon $orderCreatedSuccess 📊 $orderInfo： • $orderIdLabel: ${orderId ?: unknown} @@ -989,7 +996,7 @@ class TelegramNotificationService( " • ${position.marketId.substring(0, 8)}... (${position.side}): $quantityDisplay shares = $valueDisplay USDC" } - return """✅ $redeemSuccess + return """💸 $redeemSuccess 📊 $redeemInfo： • $accountLabel: $escapedAccountInfo diff --git a/docs/zh/smart-money-analysis.md b/docs/zh/smart-money-analysis.md index 0b69605..2d78f51 100644 --- a/docs/zh/smart-money-analysis.md +++ b/docs/zh/smart-money-analysis.md @@ -833,3 +833,4 @@ fun analyzeTrader(@RequestBody request: SmartMoneyAnalyzeRequest): ResponseEntit 4. **风险预警**：监控聪明钱交易者的风险指标，及时预警 5. **多维度分析**：增加更多分析维度（如市场类型、时间分布等） + diff --git a/frontend/src/locales/zh-CN/common.json b/frontend/src/locales/zh-CN/common.json index ca201a1..8b7015c 100644 --- a/frontend/src/locales/zh-CN/common.json +++ b/frontend/src/locales/zh-CN/common.json @@ -1026,7 +1026,7 @@ "remainingQuantity": "剩余", "sellStatus": "卖出状态", "status": "状态", - "statusFilled": "已完成", + "statusFilled": "未成交", "statusPartiallySold": "部分成交", "statusFullySold": "全部成交", "realizedPnl": "已实现盈亏", diff --git a/frontend/src/pages/CopyTradingOrders/BuyOrdersTab.tsx b/frontend/src/pages/CopyTradingOrders/BuyOrdersTab.tsx index de66e87..d011f61 100644 --- a/frontend/src/pages/CopyTradingOrders/BuyOrdersTab.tsx +++ b/frontend/src/pages/CopyTradingOrders/BuyOrdersTab.tsx @@ -8,11 +8,14 @@ import type { BuyOrderInfo, OrderTrackingRequest, OrderTrackingListResponse } fr const { Option } = Select +import { ReloadOutlined } from '@ant-design/icons' + interface BuyOrdersTabProps { copyTradingId: string + active?: boolean } -const BuyOrdersTab: React.FC = ({ copyTradingId }) => { +const BuyOrdersTab: React.FC = ({ copyTradingId, active = false }) => { const { t } = useTranslation() const isMobile = useMediaQuery({ maxWidth: 768 }) const [loading, setLoading] = useState(false) @@ -25,16 +28,16 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { side?: string status?: string }>({}) - + useEffect(() => { - if (copyTradingId) { + if (copyTradingId && active) { fetchOrders() } - }, [copyTradingId, page, limit, filters]) - + }, [copyTradingId, active, page, limit, filters]) + const fetchOrders = async () => { if (!copyTradingId) return - + setLoading(true) try { const request: OrderTrackingRequest = { @@ -44,7 +47,7 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { limit, ...filters } - + const response = await apiService.orderTracking.list(request) if (response.data.code === 0 && response.data.data) { const data = response.data.data as OrderTrackingListResponse @@ -57,17 +60,17 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { setLoading(false) } } - + const getStatusTag = (status: string) => { const statusMap: Record = { - filled: { color: 'processing', text: t('copyTradingOrders.statusFilled') || '已完成' }, + filled: { color: 'processing', text: t('copyTradingOrders.statusFilled') || '未成交' }, partially_matched: { color: 'warning', text: t('copyTradingOrders.statusPartiallySold') || '部分成交' }, fully_matched: { color: 'success', text: t('copyTradingOrders.statusFullySold') || '全部成交' } } const config = statusMap[status] || { color: 'default', text: status } return {config.text} } - + const columns = [ { title: t('copyTradingOrders.orderId') || '订单ID', @@ -76,7 +79,7 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { width: isMobile ? 100 : 150, render: (text: string) => ( - {isMobile + {isMobile ? `${text.slice(0, 6)}...${text.slice(-4)}` : `${text.slice(0, 8)}...${text.slice(-6)}` } @@ -90,7 +93,7 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { width: isMobile ? 100 : 150, render: (text: string) => ( - {isMobile + {isMobile ? `${text.slice(0, 6)}...${text.slice(-4)}` : `${text.slice(0, 8)}...${text.slice(-6)}` } @@ -104,7 +107,7 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { width: isMobile ? 100 : 150, render: (text: string) => ( - {isMobile + {isMobile ? `${text.slice(0, 6)}...${text.slice(-4)}` : `${text.slice(0, 8)}...${text.slice(-6)}` } @@ -184,7 +187,7 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { width: isMobile ? 120 : 160, render: (timestamp: number) => ( - {isMobile + {isMobile ? new Date(timestamp).toLocaleDateString('zh-CN') : new Date(timestamp).toLocaleString('zh-CN') } @@ -192,7 +195,7 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { ) } ] - + return (
@@ -203,7 +206,7 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { value={filters.marketId} onChange={(e) => setFilters({ ...filters, marketId: e.target.value || undefined })} /> - + - + - - + +
- + {isMobile ? (
{loading ? ( @@ -255,7 +256,7 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { }) const amount = (parseFloat(order.quantity) * parseFloat(order.price)).toString() const displaySide = order.side === '0' ? 'YES' : order.side === '1' ? 'NO' : order.side - + return ( = ({ copyTradingId }) => { bodyStyle={{ padding: '16px' }} >
-
@@ -280,9 +281,9 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { {getStatusTag(order.status)}
- + - +
{t('copyTradingOrders.buyInfo') || '买入信息'}
@@ -292,28 +293,28 @@ const BuyOrdersTab: React.FC = ({ copyTradingId }) => { {t('copyTradingOrders.amount') || '金额'}: {formatUSDC(amount)} USDC
- +
{t('copyTradingOrders.matchInfo') || '匹配信息'}
{t('copyTradingOrders.matched') || '已匹配'}: {formatUSDC(order.matchedQuantity)} | {t('copyTradingOrders.remaining') || '剩余'}: {formatUSDC(order.remainingQuantity)}
- +
{t('copyTradingOrders.leaderTradeId') || 'Leader 交易ID'}
{order.leaderTradeId.slice(0, 8)}...{order.leaderTradeId.slice(-6)}
- +
{t('copyTradingOrders.marketId') || '市场ID'}
{order.marketId.slice(0, 8)}...{order.marketId.slice(-6)}
- +
{t('copyTradingOrders.createdAt') || '创建时间'}: {formattedDate} diff --git a/frontend/src/pages/CopyTradingOrders/MatchedOrdersTab.tsx b/frontend/src/pages/CopyTradingOrders/MatchedOrdersTab.tsx index bcd27be..9a2da18 100644 --- a/frontend/src/pages/CopyTradingOrders/MatchedOrdersTab.tsx +++ b/frontend/src/pages/CopyTradingOrders/MatchedOrdersTab.tsx @@ -6,11 +6,14 @@ import { useMediaQuery } from 'react-responsive' import { useTranslation } from 'react-i18next' import type { MatchedOrderInfo, OrderTrackingRequest, OrderTrackingListResponse } from '../../types' +import { ReloadOutlined } from '@ant-design/icons' + interface MatchedOrdersTabProps { copyTradingId: string + active?: boolean } -const MatchedOrdersTab: React.FC = ({ copyTradingId }) => { +const MatchedOrdersTab: React.FC = ({ copyTradingId, active = false }) => { const { t } = useTranslation() const isMobile = useMediaQuery({ maxWidth: 768 }) const [loading, setLoading] = useState(false) @@ -22,16 +25,16 @@ const MatchedOrdersTab: React.FC = ({ copyTradingId }) => sellOrderId?: string buyOrderId?: string }>({}) - + useEffect(() => { - if (copyTradingId) { + if (copyTradingId && active) { fetchOrders() } - }, [copyTradingId, page, limit, filters]) - + }, [copyTradingId, active, page, limit, filters]) + const fetchOrders = async () => { if (!copyTradingId) return - + setLoading(true) try { const request: OrderTrackingRequest = { @@ -41,7 +44,7 @@ const MatchedOrdersTab: React.FC = ({ copyTradingId }) => limit, ...filters } - + const response = await apiService.orderTracking.list(request) if (response.data.code === 0 && response.data.data) { const data = response.data.data as OrderTrackingListResponse @@ -54,13 +57,13 @@ const MatchedOrdersTab: React.FC = ({ copyTradingId }) => setLoading(false) } } - + const getPnlColor = (value: string): string => { const num = parseFloat(value) if (isNaN(num)) return '#666' return num >= 0 ? '#3f8600' : '#cf1322' } - + const columns = [ { title: t('copyTradingOrders.sellOrderId') || '卖出订单ID', @@ -69,7 +72,7 @@ const MatchedOrdersTab: React.FC = ({ copyTradingId }) => width: isMobile ? 100 : 150, render: (text: string) => ( - {isMobile + {isMobile ? `${text.slice(0, 6)}...${text.slice(-4)}` : `${text.slice(0, 8)}...${text.slice(-6)}` } @@ -83,7 +86,7 @@ const MatchedOrdersTab: React.FC = ({ copyTradingId }) => width: isMobile ? 100 : 150, render: (text: string) => ( - {isMobile + {isMobile ? `${text.slice(0, 6)}...${text.slice(-4)}` : `${text.slice(0, 8)}...${text.slice(-6)}` } @@ -123,8 +126,8 @@ const MatchedOrdersTab: React.FC = ({ copyTradingId }) => key: 'realizedPnl', width: isMobile ? 100 : 120, render: (value: string) => ( - @@ -139,7 +142,7 @@ const MatchedOrdersTab: React.FC = ({ copyTradingId }) => width: isMobile ? 120 : 160, render: (timestamp: number) => ( - {isMobile + {isMobile ? new Date(timestamp).toLocaleDateString('zh-CN') : new Date(timestamp).toLocaleString('zh-CN') } @@ -147,7 +150,7 @@ const MatchedOrdersTab: React.FC = ({ copyTradingId }) => ) } ] - + return (
@@ -158,7 +161,7 @@ const MatchedOrdersTab: React.FC = ({ copyTradingId }) => value={filters.sellOrderId} onChange={(e) => setFilters({ ...filters, sellOrderId: e.target.value || undefined })} /> - + = ({ copyTradingId }) => value={filters.buyOrderId} onChange={(e) => setFilters({ ...filters, buyOrderId: e.target.value || undefined })} /> - - + +
- + {isMobile ? (
{loading ? ( @@ -191,7 +194,7 @@ const MatchedOrdersTab: React.FC = ({ copyTradingId }) => hour: '2-digit', minute: '2-digit' }) - + return ( = ({ copyTradingId }) => >
{t('copyTradingOrders.sellOrderId') || '卖出订单ID'}
-
= ({ copyTradingId }) => {order.sellOrderId.slice(0, 8)}...{order.sellOrderId.slice(-6)}
{t('copyTradingOrders.buyOrderId') || '买入订单ID'}
-
{order.buyOrderId.slice(0, 8)}...{order.buyOrderId.slice(-6)}
- + - +
{t('copyTradingOrders.matchedQuantity') || '匹配数量'}
{formatUSDC(order.matchedQuantity)}
- +
{t('copyTradingOrders.priceInfo') || '价格信息'}
{t('copyTradingOrders.buy') || '买入'}: {formatUSDC(order.buyPrice)} | {t('copyTradingOrders.sell') || '卖出'}: {formatUSDC(order.sellPrice)}
- +
{t('copyTradingOrders.realizedPnl') || '盈亏'}
-
{formatUSDC(order.realizedPnl)} USDC
- +
{t('copyTradingOrders.matchedAt') || '匹配时间'}: {formattedDate} diff --git a/frontend/src/pages/CopyTradingOrders/SellOrdersTab.tsx b/frontend/src/pages/CopyTradingOrders/SellOrdersTab.tsx index d00ec53..a99dc2a 100644 --- a/frontend/src/pages/CopyTradingOrders/SellOrdersTab.tsx +++ b/frontend/src/pages/CopyTradingOrders/SellOrdersTab.tsx @@ -8,11 +8,14 @@ import type { SellOrderInfo, OrderTrackingRequest, OrderTrackingListResponse } f const { Option } = Select +import { ReloadOutlined } from '@ant-design/icons' + interface SellOrdersTabProps { copyTradingId: string + active?: boolean } -const SellOrdersTab: React.FC = ({ copyTradingId }) => { +const SellOrdersTab: React.FC = ({ copyTradingId, active = false }) => { const { t } = useTranslation() const isMobile = useMediaQuery({ maxWidth: 768 }) const [loading, setLoading] = useState(false) @@ -24,16 +27,16 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { marketId?: string side?: string }>({}) - + useEffect(() => { - if (copyTradingId) { + if (copyTradingId && active) { fetchOrders() } - }, [copyTradingId, page, limit, filters]) - + }, [copyTradingId, active, page, limit, filters]) + const fetchOrders = async () => { if (!copyTradingId) return - + setLoading(true) try { const request: OrderTrackingRequest = { @@ -43,7 +46,7 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { limit, ...filters } - + const response = await apiService.orderTracking.list(request) if (response.data.code === 0 && response.data.data) { const data = response.data.data as OrderTrackingListResponse @@ -56,13 +59,13 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { setLoading(false) } } - + const getPnlColor = (value: string): string => { const num = parseFloat(value) if (isNaN(num)) return '#666' return num >= 0 ? '#3f8600' : '#cf1322' } - + const columns = [ { title: t('copyTradingOrders.orderId') || '订单ID', @@ -71,7 +74,7 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { width: isMobile ? 100 : 150, render: (text: string) => ( - {isMobile + {isMobile ? `${text.slice(0, 6)}...${text.slice(-4)}` : `${text.slice(0, 8)}...${text.slice(-6)}` } @@ -85,7 +88,7 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { width: isMobile ? 100 : 150, render: (text: string) => ( - {isMobile + {isMobile ? `${text.slice(0, 6)}...${text.slice(-4)}` : `${text.slice(0, 8)}...${text.slice(-6)}` } @@ -99,7 +102,7 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { width: isMobile ? 100 : 150, render: (text: string) => ( - {isMobile + {isMobile ? `${text.slice(0, 6)}...${text.slice(-4)}` : `${text.slice(0, 8)}...${text.slice(-6)}` } @@ -153,8 +156,8 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { key: 'realizedPnl', width: isMobile ? 100 : 120, render: (value: string) => ( - @@ -169,7 +172,7 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { width: isMobile ? 120 : 160, render: (timestamp: number) => ( - {isMobile + {isMobile ? new Date(timestamp).toLocaleDateString('zh-CN') : new Date(timestamp).toLocaleString('zh-CN') } @@ -177,7 +180,7 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { ) } ] - + return (
@@ -188,7 +191,7 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { value={filters.marketId} onChange={(e) => setFilters({ ...filters, marketId: e.target.value || undefined })} /> - + - - + +
- + {isMobile ? (
{loading ? ( @@ -228,7 +231,7 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { }) const amount = (parseFloat(order.quantity) * parseFloat(order.price)).toString() const displaySide = order.side === '0' ? 'YES' : order.side === '1' ? 'NO' : order.side - + return ( = ({ copyTradingId }) => { bodyStyle={{ padding: '16px' }} >
-
@@ -252,9 +255,9 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { {displaySide}
- + - +
{t('copyTradingOrders.sellInfo') || '卖出信息'}
@@ -264,32 +267,32 @@ const SellOrdersTab: React.FC = ({ copyTradingId }) => { {t('copyTradingOrders.amount') || '金额'}: {formatUSDC(amount)} USDC
- +
{t('copyTradingOrders.realizedPnl') || '已实现盈亏'}
-
{formatUSDC(order.realizedPnl)} USDC
- +
{t('copyTradingOrders.leaderTradeId') || 'Leader 交易ID'}
{order.leaderTradeId.slice(0, 8)}...{order.leaderTradeId.slice(-6)}
- +
{t('copyTradingOrders.marketId') || '市场ID'}
{order.marketId.slice(0, 8)}...{order.marketId.slice(-6)}
- +
{t('copyTradingOrders.createdAt') || '创建时间'}: {formattedDate} diff --git a/frontend/src/pages/CopyTradingOrders/index.tsx b/frontend/src/pages/CopyTradingOrders/index.tsx index 0919f75..6b842d7 100644 --- a/frontend/src/pages/CopyTradingOrders/index.tsx +++ b/frontend/src/pages/CopyTradingOrders/index.tsx @@ -22,13 +22,13 @@ const CopyTradingOrdersModal: React.FC = ({ }) => { const { t } = useTranslation() const [activeTab, setActiveTab] = useState(defaultTab) - + useEffect(() => { if (open) { setActiveTab(defaultTab) } }, [open, defaultTab]) - + return ( = ({ width="90%" style={{ top: 20 }} bodyStyle={{ padding: '24px', maxHeight: 'calc(100vh - 100px)', overflow: 'auto' }} + destroyOnClose > - setActiveTab(key as TabType)} items={[ { key: 'buy', label: t('copyTradingOrders.buyOrders') || '买入订单', - children: + children: }, { key: 'sell', label: t('copyTradingOrders.sellOrders') || '卖出订单', - children: + children: }, { key: 'matched', label: t('copyTradingOrders.matchedOrders') || '匹配关系', - children: + children: } ]} />