From 9f0b22fab53313ac7fb658abc0c687e406af177d Mon Sep 17 00:00:00 2001 From: WrBug Date: Sun, 28 Dec 2025 00:02:29 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BC=98=E5=8C=96=E5=B8=82=E5=9C=BA?= =?UTF-8?q?=E7=8A=B6=E6=80=81=E6=9F=A5=E8=AF=A2=E5=92=8C=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E8=AE=A2=E5=8D=95=E9=80=9A=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 使用 Retrofit 替代直接 HTTP 请求查询市场信息（通过 clob_token_ids） - 实现链上查询市场结算结果（方案1）：通过 ConditionalTokens 合约的 getCondition 方法实时获取市场结算状态 - 自动生成的订单（AUTO_、AUTO_FIFO_、AUTO_WS_）不发送 Telegram 通知 - 添加 ABI 编码解析工具方法（decodeUint256Array、decodeConditionResult） - 优化 PositionCheckService：优先使用链上查询，失败时降级到 API 查询 --- .../polymarketbot/api/PolymarketGammaApi.kt | 7 +- .../service/accounts/PositionCheckService.kt | 125 ++- .../service/common/BlockchainService.kt | 67 ++ .../copytrading/configs/CopyTradingService.kt | 16 +- .../monitor/AccountOnChainMonitorService.kt | 341 +++++++ .../monitor/CopyTradingMonitorService.kt | 51 +- .../copytrading/monitor/OnChainWsService.kt | 862 ++---------------- .../copytrading/monitor/OnChainWsUtils.kt | 359 ++++++++ .../monitor/UnifiedOnChainWsService.kt | 567 ++++++++++++ .../statistics/OrderStatusUpdateService.kt | 57 +- .../wrbug/polymarketbot/util/EthereumUtils.kt | 61 ++ 11 files changed, 1668 insertions(+), 845 deletions(-) create mode 100644 backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/AccountOnChainMonitorService.kt create mode 100644 backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/OnChainWsUtils.kt create mode 100644 backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/UnifiedOnChainWsService.kt diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/api/PolymarketGammaApi.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/api/PolymarketGammaApi.kt index fa87527..bec3b68 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/api/PolymarketGammaApi.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/api/PolymarketGammaApi.kt @@ -16,12 +16,14 @@ interface PolymarketGammaApi { * 根据 condition ID 列表获取市场信息 * 文档: https://docs.polymarket.com/api-reference/markets/list-markets * @param conditionIds condition ID 数组（16 进制字符串，如 "0x..."） + * @param clobTokenIds CLOB token ID 数组（用于通过 tokenId 查询市场） * @param includeTag 是否包含标签信息 * @return 市场信息数组 */ @GET("/markets") suspend fun listMarkets( @Query("condition_ids") conditionIds: List? = null, + @Query("clob_token_ids") clobTokenIds: List? = null, @Query("include_tag") includeTag: Boolean? = null ): Response> } @@ -51,6 +53,9 @@ data class MarketResponse( val liquidityNum: Double? = null, val lastTradePrice: Double? = null, val bestBid: Double? = null, - val bestAsk: Double? = null + val bestAsk: Double? = null, + // 以下字段可能存在于响应中，但不在标准文档中 + val clobTokenIds: String? = null, // CLOB token IDs（可能是 JSON 字符串或数组） + val clob_token_ids: String? = null // 下划线格式（兼容不同 API 版本） ) 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 6fc85ac..c34d951 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 @@ -23,8 +23,10 @@ import com.wrbug.polymarketbot.service.system.RelayClientService import com.wrbug.polymarketbot.service.system.TelegramNotificationService import com.wrbug.polymarketbot.util.RetrofitFactory import com.wrbug.polymarketbot.util.JsonUtils +import com.wrbug.polymarketbot.service.common.BlockchainService import org.springframework.stereotype.Service import java.math.BigDecimal +import java.math.BigInteger import java.util.concurrent.ConcurrentHashMap /** @@ -45,7 +47,8 @@ class PositionCheckService( private val telegramNotificationService: TelegramNotificationService?, private val accountRepository: AccountRepository, private val messageSource: MessageSource, - private val retrofitFactory: RetrofitFactory + private val retrofitFactory: RetrofitFactory, + private val blockchainService: BlockchainService ) { private val logger = LoggerFactory.getLogger(PositionCheckService::class.java) @@ -427,14 +430,45 @@ class PositionCheckService( /** * 获取当前市场最新价（用于更新订单卖出价） - * 优先使用 bestBid（最优买价），如果没有则使用 midpoint（中间价） + * 优先使用链上查询获取市场结算结果，如果未结算则使用 API 查询 * 如果市场已关闭： * - 该 outcome 赢了，返回 1 * - 该 outcome 输了，返回 0 */ private suspend fun getCurrentMarketPrice(marketId: String, outcomeIndex: Int): BigDecimal { return try { - // 先获取市场信息，检查市场是否已关闭 + // 优先从链上查询市场结算结果（实时性高） + val chainResult = blockchainService.getCondition(marketId) + chainResult.fold( + onSuccess = { (payoutDenominator, payouts) -> + // 如果 payouts 不为空，说明市场已结算 + if (payouts.isNotEmpty() && outcomeIndex < payouts.size) { + val payout = payouts[outcomeIndex] + when { + payout > BigInteger.ZERO -> { + // payout > 0 表示赢了 + logger.info("从链上查询到市场已结算，该 outcome 赢了: marketId=$marketId, outcomeIndex=$outcomeIndex, payout=$payout") + return BigDecimal.ONE + } + payout == BigInteger.ZERO -> { + // payout == 0 表示输了 + logger.info("从链上查询到市场已结算，该 outcome 输了: marketId=$marketId, outcomeIndex=$outcomeIndex, payout=$payout") + return BigDecimal.ZERO + } + else -> { + logger.warn("从链上查询到异常的 payout 值: marketId=$marketId, outcomeIndex=$outcomeIndex, payout=$payout") + } + } + } else { + logger.debug("从链上查询到市场尚未结算: marketId=$marketId, payouts=${payouts.size}") + } + }, + onFailure = { e -> + logger.debug("链上查询市场条件失败，降级到 API 查询: marketId=$marketId, error=${e.message}") + } + ) + + // 链上查询失败或市场未结算，降级到 API 查询 val gammaApi = retrofitFactory.createGammaApi() val marketResponse = gammaApi.listMarkets(conditionIds = listOf(marketId)) @@ -442,20 +476,27 @@ class PositionCheckService( val markets = marketResponse.body()!! val market = markets.firstOrNull() - if (market != null && market.closed == true) { - // 市场已关闭，检查该 outcome 是赢了还是输了 - val outcomeResult = checkOutcomeResult(market, outcomeIndex) - when (outcomeResult) { - OutcomeResult.WON -> { - logger.info("市场已关闭且该 outcome 赢了，返回价格为 1: marketId=$marketId, outcomeIndex=$outcomeIndex") - return BigDecimal.ONE - } - OutcomeResult.LOST -> { - logger.info("市场已关闭且该 outcome 输了，返回价格为 0: marketId=$marketId, outcomeIndex=$outcomeIndex") - return BigDecimal.ZERO - } - OutcomeResult.UNKNOWN -> { - // 无法判断，继续使用正常价格逻辑 + if (market != null) { + // 检查市场是否已结束：1) closed == true 或 2) endDate 已过 + val isMarketEnded = checkIfMarketEnded(market) + + if (isMarketEnded) { + logger.debug("市场已结束: marketId=$marketId, closed=${market.closed}, endDate=${market.endDate}") + // 市场已结束，检查该 outcome 是赢了还是输了 + val outcomeResult = checkOutcomeResult(market, outcomeIndex) + when (outcomeResult) { + OutcomeResult.WON -> { + logger.info("市场已结束且该 outcome 赢了，返回价格为 1: marketId=$marketId, outcomeIndex=$outcomeIndex") + return BigDecimal.ONE + } + OutcomeResult.LOST -> { + logger.info("市场已结束且该 outcome 输了，返回价格为 0: marketId=$marketId, outcomeIndex=$outcomeIndex") + return BigDecimal.ZERO + } + OutcomeResult.UNKNOWN -> { + // 无法判断，记录警告并继续使用正常价格逻辑 + logger.warn("市场已结束但无法判断 outcome 结果，使用正常价格: marketId=$marketId, outcomeIndex=$outcomeIndex, closed=${market.closed}, endDate=${market.endDate}, outcomePrices=${market.outcomePrices}, bestBid=${market.bestBid}, bestAsk=${market.bestAsk}") + } } } } @@ -477,6 +518,50 @@ class PositionCheckService( } } + /** + * 检查市场是否已结束 + * 判断条件： + * 1. closed == true + * 2. 或 endDate 已过（如果 endDate 不为空） + */ + private fun checkIfMarketEnded(market: com.wrbug.polymarketbot.api.MarketResponse): Boolean { + // 1. 检查 closed 字段 + if (market.closed == true) { + return true + } + + // 2. 检查 endDate 是否已过 + val endDateStr = market.endDate + if (endDateStr != null && endDateStr.isNotBlank()) { + try { + // endDate 可能是 ISO 8601 格式字符串或时间戳 + val endDate = if (endDateStr.matches(Regex("^\\d+$"))) { + // 时间戳（秒或毫秒） + val timestamp = endDateStr.toLong() + // 判断是秒还是毫秒（如果小于 10^10，认为是秒） + if (timestamp < 10000000000L) { + timestamp * 1000 // 转换为毫秒 + } else { + timestamp + } + } else { + // ISO 8601 格式，尝试解析 + java.time.Instant.parse(endDateStr).toEpochMilli() + } + + val now = System.currentTimeMillis() + if (now >= endDate) { + logger.debug("市场 endDate 已过: marketId=${market.conditionId}, endDate=$endDateStr, now=$now") + return true + } + } catch (e: Exception) { + logger.warn("解析 endDate 失败: marketId=${market.conditionId}, endDate=$endDateStr, error=${e.message}") + } + } + + return false + } + /** * Outcome 结果枚举 */ @@ -628,7 +713,8 @@ class PositionCheckService( outcomeIndex = outcomeIndex, totalMatchedQuantity = totalMatchedQuantity, sellPrice = sellPrice, - totalRealizedPnl = totalRealizedPnl + totalRealizedPnl = totalRealizedPnl, + priceUpdated = true // 自动生成的订单，直接标记为已处理，不发送通知 ) val savedRecord = sellMatchRecordRepository.save(matchRecord) @@ -742,7 +828,8 @@ class PositionCheckService( outcomeIndex = outcomeIndex, totalMatchedQuantity = totalMatchedQuantity, sellPrice = sellPrice, - totalRealizedPnl = totalRealizedPnl + totalRealizedPnl = totalRealizedPnl, + priceUpdated = true // 自动生成的订单，直接标记为已处理，不发送通知 ) val savedRecord = sellMatchRecordRepository.save(matchRecord) 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 58312ae..e400398 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 @@ -602,6 +602,73 @@ class BlockchainService( return Result.success(txHash) } + /** + * 从链上查询市场条件（Condition）的结算结果 + * 通过调用 ConditionalTokens 合约的 getCondition 函数获取 payouts + * + * @param conditionId 市场条件ID（bytes32，必须是 0x 开头的 66 位十六进制字符串） + * @return Result> + * - payoutDenominator: 支付分母（通常为 1） + * - payouts: 每个 outcome 的支付金额数组（0 或 1） + * - 如果 payouts[outcomeIndex] == 1，表示该 outcome 赢了 + * - 如果 payouts[outcomeIndex] == 0，表示该 outcome 输了 + * - 如果 payouts 为空，表示市场尚未结算 + */ + suspend fun getCondition(conditionId: String): Result>> { + return try { + // 验证 conditionId 格式 + if (conditionId.isBlank() || !conditionId.startsWith("0x") || conditionId.length != 66) { + return Result.failure(IllegalArgumentException("conditionId 格式错误，必须是 0x 开头的 66 位十六进制字符串")) + } + + val rpcApi = polygonRpcApi + + // 构建 getCondition(bytes32) 函数调用 + // 函数签名: getCondition(bytes32) + val functionSelector = EthereumUtils.getFunctionSelector("getCondition(bytes32)") + val encodedConditionId = EthereumUtils.encodeBytes32(conditionId) + val data = functionSelector + encodedConditionId + + // 构建 JSON-RPC 请求 + val rpcRequest = JsonRpcRequest( + method = "eth_call", + params = listOf( + mapOf( + "to" to conditionalTokensAddress, + "data" to data + ), + "latest" + ) + ) + + // 发送 RPC 请求 + val response = rpcApi.call(rpcRequest) + + if (!response.isSuccessful || response.body() == null) { + return Result.failure(Exception("RPC 请求失败: ${response.code()} ${response.message()}")) + } + + val rpcResponse = response.body()!! + + // 检查错误 + if (rpcResponse.error != null) { + return Result.failure(Exception("RPC 错误: ${rpcResponse.error.message}")) + } + + // 使用 Gson 解析 result（JsonElement） + val hexResult = rpcResponse.result?.asString + ?: return Result.failure(Exception("RPC 响应格式错误: result 为空")) + + // 解析 ABI 编码的返回结果 + val (payoutDenominator, payouts) = EthereumUtils.decodeConditionResult(hexResult) + + Result.success(Pair(payoutDenominator, payouts)) + } catch (e: Exception) { + logger.error("查询市场条件失败: conditionId=$conditionId, ${e.message}", e) + Result.failure(e) + } + } + /** * 查询交易详情（用于调试和分析） * @param txHash 交易哈希 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 9fecc17..af9f4a1 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,14 @@ class CopyTradingService( val saved = copyTradingRepository.save(copyTrading) - // 如果跟单已启用，更新 Leader 监听（增量更新，不重启所有监听） + // 如果跟单已启用，更新 Leader 监听和账户监听（增量更新，不重启所有监听） if (saved.enabled) { kotlinx.coroutines.runBlocking { try { monitorService.updateLeaderMonitoring(saved.leaderId) + monitorService.updateAccountMonitoring(saved.accountId) } catch (e: Exception) { - logger.error("更新 Leader 监听失败", e) + logger.error("更新监听失败", e) } } } @@ -219,12 +220,13 @@ class CopyTradingService( val saved = copyTradingRepository.save(updated) - // 更新 Leader 监听（增量更新，根据 enabled 状态决定添加或移除） + // 更新 Leader 监听和账户监听（增量更新，根据 enabled 状态决定添加或移除） kotlinx.coroutines.runBlocking { try { monitorService.updateLeaderMonitoring(saved.leaderId) + monitorService.updateAccountMonitoring(saved.accountId) } catch (e: Exception) { - logger.error("更新 Leader 监听失败", e) + logger.error("更新监听失败", e) } } @@ -323,14 +325,16 @@ class CopyTradingService( ?: return Result.failure(IllegalArgumentException("跟单配置不存在")) val leaderId = copyTrading.leaderId + val accountId = copyTrading.accountId copyTradingRepository.delete(copyTrading) - // 更新 Leader 监听（检查该 Leader 是否还有其他启用的跟单配置） + // 更新 Leader 监听和账户监听（检查是否还有其他启用的跟单配置） kotlinx.coroutines.runBlocking { try { monitorService.removeLeaderMonitoring(leaderId) + monitorService.updateAccountMonitoring(accountId) } catch (e: Exception) { - logger.error("更新 Leader 监听失败", e) + logger.error("更新监听失败", e) } } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/AccountOnChainMonitorService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/AccountOnChainMonitorService.kt new file mode 100644 index 0000000..6209925 --- /dev/null +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/AccountOnChainMonitorService.kt @@ -0,0 +1,341 @@ +package com.wrbug.polymarketbot.service.copytrading.monitor + +import com.wrbug.polymarketbot.api.* +import com.wrbug.polymarketbot.entity.Account +import com.wrbug.polymarketbot.entity.CopyOrderTracking +import com.wrbug.polymarketbot.entity.SellMatchDetail +import com.wrbug.polymarketbot.entity.SellMatchRecord +import com.wrbug.polymarketbot.repository.AccountRepository +import com.wrbug.polymarketbot.repository.CopyOrderTrackingRepository +import com.wrbug.polymarketbot.repository.CopyTradingRepository +import com.wrbug.polymarketbot.repository.SellMatchDetailRepository +import com.wrbug.polymarketbot.repository.SellMatchRecordRepository +import com.wrbug.polymarketbot.util.RetrofitFactory +import com.wrbug.polymarketbot.util.multi +import com.wrbug.polymarketbot.util.toSafeBigDecimal +import jakarta.annotation.PreDestroy +import kotlinx.coroutines.* +import okhttp3.OkHttpClient +import org.slf4j.LoggerFactory +import org.springframework.stereotype.Service +import java.math.BigDecimal +import java.util.concurrent.ConcurrentHashMap + +/** + * 跟单账户链上 WebSocket 监听服务 + * 通过统一服务订阅跟单账户的卖出和赎回事件 + * 用于更新订单状态，不再依赖轮询 + */ +@Service +class AccountOnChainMonitorService( + private val unifiedOnChainWsService: UnifiedOnChainWsService, + private val retrofitFactory: RetrofitFactory, + private val accountRepository: AccountRepository, + private val copyTradingRepository: CopyTradingRepository, + private val copyOrderTrackingRepository: CopyOrderTrackingRepository, + private val sellMatchRecordRepository: SellMatchRecordRepository, + private val sellMatchDetailRepository: SellMatchDetailRepository +) { + + private val logger = LoggerFactory.getLogger(AccountOnChainMonitorService::class.java) + + // 存储需要监听的账户：accountId -> Account + private val monitoredAccounts = ConcurrentHashMap() + + /** + * 启动链上 WebSocket 监听 + * 通过统一服务订阅所有跟单账户 + */ + fun start(accounts: List) { + // 如果没有账户，取消所有订阅 + if (accounts.isEmpty()) { + logger.info("没有需要监听的跟单账户，取消所有订阅") + stop() + return + } + + // 更新账户列表 + monitoredAccounts.clear() + accounts.forEach { account -> + addAccount(account) + } + } + + /** + * 添加账户监听 + * 通过统一服务订阅该账户的地址 + */ + fun addAccount(account: Account) { + if (account.id == null) { + logger.warn("账户 ID 为空，跳过: ${account.proxyAddress}") + return + } + + val accountId = account.id!! + + // 如果已经在监听列表中，不重复添加 + if (monitoredAccounts.containsKey(accountId)) { + return + } + + monitoredAccounts[accountId] = account + + // 通过统一服务订阅 + val subscriptionId = "ACCOUNT_$accountId" + unifiedOnChainWsService.subscribe( + subscriptionId = subscriptionId, + address = account.proxyAddress, + entityType = "ACCOUNT", + entityId = accountId, + callback = { txHash, httpClient, rpcApi -> + handleAccountTransaction(accountId, txHash, httpClient, rpcApi) + } + ) + + logger.info("已添加跟单账户进行链上监听: accountId=${accountId}, address=${account.proxyAddress}") + } + + /** + * 处理账户的交易 + */ + private suspend fun handleAccountTransaction(accountId: Long, txHash: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) { + val account = monitoredAccounts[accountId] ?: return + + 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 + + // 获取区块号和时间戳 + val blockNumber = receiptJson.get("blockNumber")?.asString + val blockTimestamp = if (blockNumber != null) { + OnChainWsUtils.getBlockTimestamp(blockNumber, rpcApi) + } else { + null + } + + // 解析 receipt 中的 Transfer 日志 + val logs = receiptJson.getAsJsonArray("logs") ?: return + val (erc20Transfers, erc1155Transfers) = OnChainWsUtils.parseReceiptTransfers(logs) + + // 解析交易信息 + val trade = OnChainWsUtils.parseTradeFromTransfers( + txHash = txHash, + timestamp = blockTimestamp, + walletAddress = account.proxyAddress, + erc20Transfers = erc20Transfers, + erc1155Transfers = erc1155Transfers, + retrofitFactory = retrofitFactory + ) + + if (trade != null && trade.side == "SELL") { + // 检测到卖出或赎回事件，更新订单状态 + handleAccountSellOrRedeem(account, trade) + } + } catch (e: Exception) { + logger.error("处理账户交易失败: accountId=$accountId, txHash=$txHash, ${e.message}", e) + } + } + + /** + * 处理账户的卖出或赎回事件 + * 更新对应的订单状态 + */ + private suspend fun handleAccountSellOrRedeem(account: Account, trade: TradeResponse) { + try { + // 获取该账户的所有启用的跟单配置 + val copyTradings = copyTradingRepository.findByAccountId(account.id!!) + .filter { it.enabled } + + if (copyTradings.isEmpty()) { + return + } + + // 使用 trade 中已有的市场信息 + val marketId = trade.market // conditionId + val outcomeIndex = trade.outcomeIndex ?: 0 + + // 计算卖出价格 + val sellPrice = trade.price.toSafeBigDecimal() + + // 为每个跟单配置更新订单状态 + for (copyTrading in copyTradings) { + // 查找该跟单配置下所有未卖出的订单（remaining_quantity > 0） + val unmatchedOrders = copyOrderTrackingRepository.findByCopyTradingId(copyTrading.id!!) + .filter { + it.remainingQuantity > BigDecimal.ZERO && + it.marketId == marketId && + it.outcomeIndex == outcomeIndex + } + .sortedBy { it.createdAt } // 按创建时间排序（FIFO） + + if (unmatchedOrders.isEmpty()) { + continue + } + + // 卖出数量就是交易的 size + val soldQuantity = trade.size.toSafeBigDecimal() + + // 更新订单状态为已卖出 + updateOrdersAsSoldByFIFO( + unmatchedOrders, + soldQuantity, + sellPrice, + copyTrading.id!!, + marketId, + outcomeIndex + ) + + logger.info("跟单账户卖出/赎回事件处理完成: accountId=${account.id}, copyTradingId=${copyTrading.id}, txHash=${trade.id}, soldQuantity=$soldQuantity, sellPrice=$sellPrice") + } + } catch (e: Exception) { + logger.error("处理账户卖出/赎回事件失败: accountId=${account.id}, txHash=${trade.id}, error=${e.message}", e) + } + } + + /** + * 按 FIFO 顺序更新订单为已卖出 + */ + private suspend fun updateOrdersAsSoldByFIFO( + orders: List, + soldQuantity: BigDecimal, + sellPrice: BigDecimal, + copyTradingId: Long, + marketId: String, + outcomeIndex: Int + ) { + var remainingSoldQuantity = soldQuantity + val matchDetails = mutableListOf() + var totalMatchedQuantity = BigDecimal.ZERO + var totalRealizedPnl = BigDecimal.ZERO + + for (order in orders) { + if (remainingSoldQuantity <= BigDecimal.ZERO) { + break + } + + val currentOrderRemaining = order.remainingQuantity.toSafeBigDecimal() + val matchedQty = minOf(currentOrderRemaining, remainingSoldQuantity) + + if (matchedQty <= BigDecimal.ZERO) { + continue + } + + // 计算盈亏 + val buyPrice = order.price.toSafeBigDecimal() + val realizedPnl = sellPrice.subtract(buyPrice).multi(matchedQty) + + // 创建匹配明细 + val detail = SellMatchDetail( + matchRecordId = 0, // 稍后设置 + trackingId = order.id!!, + buyOrderId = order.buyOrderId, + matchedQuantity = matchedQty, + buyPrice = buyPrice, + sellPrice = sellPrice, + realizedPnl = realizedPnl + ) + matchDetails.add(detail) + + totalMatchedQuantity = totalMatchedQuantity.add(matchedQty) + totalRealizedPnl = totalRealizedPnl.add(realizedPnl) + + // 更新订单状态 + order.matchedQuantity = order.matchedQuantity.add(matchedQty) + order.remainingQuantity = currentOrderRemaining.subtract(matchedQty) + order.status = if (order.remainingQuantity <= BigDecimal.ZERO) "fully_matched" else "partially_matched" + order.updatedAt = System.currentTimeMillis() + copyOrderTrackingRepository.save(order) + + remainingSoldQuantity = remainingSoldQuantity.subtract(matchedQty) + } + + // 如果有匹配的订单，创建卖出记录 + if (totalMatchedQuantity > BigDecimal.ZERO && matchDetails.isNotEmpty()) { + val timestamp = System.currentTimeMillis() + val sellOrderId = "AUTO_WS_${timestamp}_${copyTradingId}" // 区分 WS 自动卖出 + val leaderSellTradeId = "AUTO_WS_${timestamp}" + + val matchRecord = SellMatchRecord( + copyTradingId = copyTradingId, + sellOrderId = sellOrderId, + leaderSellTradeId = leaderSellTradeId, + marketId = marketId, + side = outcomeIndex.toString(), + outcomeIndex = outcomeIndex, + totalMatchedQuantity = totalMatchedQuantity, + sellPrice = sellPrice, + totalRealizedPnl = totalRealizedPnl, + priceUpdated = true // WS 实时获取，直接标记为已更新 + ) + + val savedRecord = sellMatchRecordRepository.save(matchRecord) + + // 保存匹配明细 + for (detail in matchDetails) { + val savedDetail = detail.copy(matchRecordId = savedRecord.id!!) + sellMatchDetailRepository.save(savedDetail) + } + + logger.info("创建跟单账户链上自动卖出记录: copyTradingId=$copyTradingId, marketId=$marketId, totalMatched=$totalMatchedQuantity, totalPnl=$totalRealizedPnl") + } + } + + /** + * 移除账户监听 + * 取消该账户的订阅 + */ + fun removeAccount(accountId: Long) { + monitoredAccounts.remove(accountId) + + // 通过统一服务取消订阅 + val subscriptionId = "ACCOUNT_$accountId" + unifiedOnChainWsService.unsubscribe(subscriptionId) + + logger.info("已移除跟单账户的链上监听: accountId=$accountId") + } + + /** + * 更新账户监听状态 + */ + fun updateAccountMonitoring(accountId: Long) { + val account = accountRepository.findById(accountId).orElse(null) + if (account != null && account.isEnabled) { + addAccount(account) + } else { + removeAccount(accountId) + } + } + + /** + * 停止监听 + */ + fun stop() { + // 取消所有账户的订阅 + val accountIds = monitoredAccounts.keys.toList() + for (accountId in accountIds) { + removeAccount(accountId) + } + monitoredAccounts.clear() + } + + @PreDestroy + fun destroy() { + stop() + } +} 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 937fef0..d4b5c89 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 @@ -2,6 +2,7 @@ package com.wrbug.polymarketbot.service.copytrading.monitor import com.wrbug.polymarketbot.entity.CopyTrading import com.wrbug.polymarketbot.entity.Leader +import com.wrbug.polymarketbot.repository.AccountRepository import com.wrbug.polymarketbot.repository.CopyTradingRepository import com.wrbug.polymarketbot.repository.LeaderRepository import jakarta.annotation.PostConstruct @@ -14,13 +15,16 @@ import org.springframework.stereotype.Service * 跟单监听服务（主服务） * 管理所有Leader的交易监听 * 同时运行链上 WebSocket 监听和轮询监听（并行处理） + * 同时监听跟单账户的卖出/赎回事件（通过链上 WebSocket） */ @Service class CopyTradingMonitorService( private val copyTradingRepository: CopyTradingRepository, private val leaderRepository: LeaderRepository, + private val accountRepository: AccountRepository, private val pollingService: CopyTradingPollingService, - private val onChainWsService: OnChainWsService + private val onChainWsService: OnChainWsService, + private val accountOnChainMonitorService: AccountOnChainMonitorService ) { private val logger = LoggerFactory.getLogger(CopyTradingMonitorService::class.java) @@ -50,11 +54,13 @@ class CopyTradingMonitorService( // 停止轮询和链上 WS 监听 pollingService.stop() onChainWsService.stop() + accountOnChainMonitorService.stop() } /** * 启动监听 * 同时启动链上 WebSocket 监听和轮询监听（并行运行） + * 同时启动跟单账户的链上 WebSocket 监听（用于检测卖出/赎回事件） */ suspend fun startMonitoring() { // 1. 获取所有启用的跟单关系 @@ -70,12 +76,21 @@ class CopyTradingMonitorService( leaderRepository.findById(leaderId).orElse(null) } - // 3. 同时启动链上 WebSocket 监听和轮询监听（并行运行） - // 链上 WS 监听（实时，秒级延迟） + // 3. 获取所有需要监听的跟单账户（去重） + val accountIds = enabledCopyTradings.map { it.accountId }.distinct() + val accounts = accountIds.mapNotNull { accountId -> + accountRepository.findById(accountId).orElse(null) + } + + // 4. 同时启动链上 WebSocket 监听和轮询监听（并行运行） + // 链上 WS 监听 Leader 的交易（实时，秒级延迟） onChainWsService.start(leaders) - // 轮询监听（延迟，2秒间隔，作为备份） + // 轮询监听 Leader 的交易（延迟，2秒间隔，作为备份） pollingService.start(leaders) + + // 5. 启动跟单账户的链上 WebSocket 监听（用于检测卖出/赎回事件） + accountOnChainMonitorService.start(accounts) } /** @@ -125,6 +140,15 @@ class CopyTradingMonitorService( // 有启用的跟单配置，确保在监听列表中 onChainWsService.addLeader(leader) pollingService.addLeader(leader) + + // 更新账户监听（添加该配置关联的账户） + val accountIds = copyTradings.map { it.accountId }.distinct() + accountIds.forEach { accountId -> + val account = accountRepository.findById(accountId).orElse(null) + if (account != null) { + accountOnChainMonitorService.addAccount(account) + } + } } else { // 没有启用的跟单配置，移除监听 onChainWsService.removeLeader(leaderId) @@ -132,6 +156,25 @@ class CopyTradingMonitorService( } } + /** + * 更新账户监听（当跟单配置状态改变时调用） + * 根据当前状态决定添加或移除账户监听 + */ + suspend fun updateAccountMonitoring(accountId: Long) { + val copyTradings = copyTradingRepository.findByAccountId(accountId) + .filter { it.enabled } + val account = accountRepository.findById(accountId).orElse(null) + ?: return + + if (copyTradings.isNotEmpty()) { + // 有启用的跟单配置，确保账户在监听列表中 + accountOnChainMonitorService.addAccount(account) + } else { + // 没有启用的跟单配置，移除账户监听 + accountOnChainMonitorService.removeAccount(accountId) + } + } + /** * 重新启动监听（当跟单关系状态改变时调用） * 注意：这个方法会停止所有监听并重新启动，建议使用 updateLeaderMonitoring 进行增量更新 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 18145cd..3ee5e7c 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 @@ -1,37 +1,24 @@ 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 实时监听链上交易 + * 通过统一服务订阅 Leader 的链上交易 */ @Service class OnChainWsService( - private val rpcNodeService: RpcNodeService, + private val unifiedOnChainWsService: UnifiedOnChainWsService, private val retrofitFactory: RetrofitFactory, private val copyOrderTrackingService: CopyOrderTrackingService, private val leaderRepository: LeaderRepository @@ -39,84 +26,31 @@ class OnChainWsService( 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 订阅 + * 通过统一服务订阅所有 Leader */ fun start(leaders: List) { - // 如果没有 Leader，不启动连接 + // 如果没有 Leader，取消所有订阅 if (leaders.isEmpty()) { - logger.info("没有需要监听的 Leader，不启动链上 WebSocket 连接") + logger.info("没有需要监听的 Leader，取消所有订阅") 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 已经在监听列表中，不重复添加 - * 如果已连接，立即订阅 + * 通过统一服务订阅该 Leader 的地址 */ fun addLeader(leader: Leader) { if (leader.id == null) { @@ -133,406 +67,28 @@ class OnChainWsService( } monitoredLeaders[leaderId] = leader + + // 通过统一服务订阅 + val subscriptionId = "LEADER_$leaderId" + unifiedOnChainWsService.subscribe( + subscriptionId = subscriptionId, + address = leader.leaderAddress, + entityType = "LEADER", + entityId = leaderId, + callback = { txHash, httpClient, rpcApi -> + handleLeaderTransaction(leaderId, txHash, httpClient, rpcApi) + } + ) + 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 连接 + * 处理 Leader 的交易 */ - fun removeLeader(leaderId: Long) { - val leader = monitoredLeaders.remove(leaderId) - if (leader != null) { - logger.info("移除 Leader 监听: ${leader.leaderName} (${leader.leaderAddress})") - } + private suspend fun handleLeaderTransaction(leaderId: Long, txHash: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) { + val leader = monitoredLeaders[leaderId] ?: return - // 取消该 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( @@ -550,376 +106,72 @@ class OnChainWsService( return } - // 使用 Gson 解析 receipt JSON（result 是 JsonElement） + // 使用 Gson 解析 receipt JSON val receiptJson = receiptRpcResponse.result.asJsonObject // 获取区块号和时间戳 val blockNumber = receiptJson.get("blockNumber")?.asString val blockTimestamp = if (blockNumber != null) { - getBlockTimestamp(blockNumber, rpcApi) + OnChainWsUtils.getBlockTimestamp(blockNumber, rpcApi) } else { null } // 解析 receipt 中的 Transfer 日志 val logs = receiptJson.getAsJsonArray("logs") ?: return - val (erc20Transfers, erc1155Transfers) = parseReceiptTransfers(logs) + val (erc20Transfers, erc1155Transfers) = OnChainWsUtils.parseReceiptTransfers(logs) - // 为每个 Leader 处理交易 - for (leader in monitoredLeaders.values) { - val trade = parseTradeFromTransfers( - txHash = txHash, - timestamp = blockTimestamp, - walletAddress = leader.leaderAddress, - erc20Transfers = erc20Transfers, - erc1155Transfers = erc1155Transfers + // 解析交易信息 + val trade = OnChainWsUtils.parseTradeFromTransfers( + txHash = txHash, + timestamp = blockTimestamp, + walletAddress = leader.leaderAddress, + erc20Transfers = erc20Transfers, + erc1155Transfers = erc1155Transfers, + retrofitFactory = retrofitFactory + ) + + if (trade != null) { + // 调用 processTrade 处理交易 + copyOrderTrackingService.processTrade( + leaderId = leaderId, + trade = trade, + source = "onchain-ws" ) - - 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) + logger.error("处理 Leader 交易失败: leaderId=$leaderId, txHash=$txHash, ${e.message}", e) } } /** - * 解析 receipt 中的 Transfer 日志 + * 移除Leader监听 + * 取消该 Leader 的订阅 */ - private fun parseReceiptTransfers(logs: com.google.gson.JsonArray): Pair, List> { - val erc20 = mutableListOf() - val erc1155 = mutableListOf() + fun removeLeader(leaderId: Long) { + monitoredLeaders.remove(leaderId) - 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)) - } - } - } + // 通过统一服务取消订阅 + val subscriptionId = "LEADER_$leaderId" + unifiedOnChainWsService.unsubscribe(subscriptionId) - return Pair(erc20, erc1155) + logger.info("移除 Leader 监听: leaderId=$leaderId") } /** - * 从 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 - } + fun stop() { + // 取消所有 Leader 的订阅 + val leaderIds = monitoredLeaders.keys.toList() + for (leaderId in leaderIds) { + removeLeader(leaderId) } - - // 找到最大的流入和流出 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 + monitoredLeaders.clear() } @PreDestroy fun destroy() { stop() - scope.cancel() } } - 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 new file mode 100644 index 0000000..a45221b --- /dev/null +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/OnChainWsUtils.kt @@ -0,0 +1,359 @@ +package com.wrbug.polymarketbot.service.copytrading.monitor + +import com.google.gson.JsonArray +import com.google.gson.reflect.TypeToken +import com.wrbug.polymarketbot.api.* +import com.wrbug.polymarketbot.service.system.RpcNodeService +import com.wrbug.polymarketbot.util.RetrofitFactory +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import okhttp3.OkHttpClient +import org.slf4j.LoggerFactory +import java.math.BigInteger + +/** + * 链上 WebSocket 工具类 + * 提取公共的交易解析、工具函数等逻辑 + */ +object OnChainWsUtils { + + private val logger = LoggerFactory.getLogger(OnChainWsUtils::class.java) + + // 合约地址 + const val USDC_CONTRACT = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174" + const val ERC1155_CONTRACT = "0x4d97dcd97ec945f40cf65f87097ace5ea0476045" + const val ERC20_TRANSFER_TOPIC = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" + const val ERC1155_TRANSFER_SINGLE_TOPIC = "0xc3d58168c5ae7397731d063d5bbf3d657854427343f4c083240f7aacaa2d0f62" + const val ERC1155_TRANSFER_BATCH_TOPIC = "0x4a39dc06d4c0dbc64b70af90fd698a233a518aa5d07e595d983b8c0526c8f7fb" + + /** + * ERC20 Transfer 数据类 + */ + data class Erc20Transfer( + val from: String, + val to: String, + val value: BigInteger + ) + + /** + * ERC1155 Transfer 数据类 + */ + data class Erc1155Transfer( + val from: String, + val to: String, + val tokenId: BigInteger, + val value: BigInteger + ) + + /** + * 市场信息数据类 + */ + data class MarketInfo( + val conditionId: String, + val outcomeIndex: Int?, // 可空，因为可能找不到对应的 tokenId + val outcome: String? + ) + + /** + * 解析 receipt 中的 Transfer 日志 + */ + fun parseReceiptTransfers(logs: 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 日志解析交易信息 + */ + suspend fun parseTradeFromTransfers( + txHash: String, + timestamp: Long?, + walletAddress: String, + erc20Transfers: List, + erc1155Transfers: List, + retrofitFactory: RetrofitFactory + ): 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(), retrofitFactory) + + // 创建 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） + * 使用 Retrofit 接口，支持 clob_token_ids 参数 + */ + suspend fun fetchMarketByTokenId(tokenId: String, retrofitFactory: RetrofitFactory): MarketInfo? { + return try { + val gammaApi = retrofitFactory.createGammaApi() + val marketsResponse = gammaApi.listMarkets( + conditionIds = null, + clobTokenIds = listOf(tokenId), + includeTag = null + ) + + if (!marketsResponse.isSuccessful || marketsResponse.body() == null) { + return null + } + + val markets = marketsResponse.body()!! + val market = markets.firstOrNull() + + if (market == null) { + return null + } + + // 解析 clobTokenIds（可能是 JSON 字符串或数组） + val clobTokenIdsRaw = market.clobTokenIds ?: market.clob_token_ids + val clobTokenIds = when { + clobTokenIdsRaw == null -> null + else -> { + try { + // 尝试解析 JSON 字符串 + val gson = com.google.gson.Gson() + val listType = object : com.google.gson.reflect.TypeToken>() {}.type + gson.fromJson>(clobTokenIdsRaw, listType) + } catch (e: Exception) { + // 如果不是 JSON 字符串，可能是其他格式，返回 null + null + } + } + } + + // 解析 outcomes（可能是 JSON 字符串或数组） + val outcomes = com.wrbug.polymarketbot.util.JsonUtils.parseStringArray(market.outcomes) + + // 查找 tokenId 在 clobTokenIds 中的索引 + val outcomeIndex = clobTokenIds?.indexOfFirst { + it.equals(tokenId, ignoreCase = true) + }?.takeIf { it >= 0 } + + // 获取 outcome 名称 + val outcome = if (outcomeIndex != null && outcomes.isNotEmpty() && outcomeIndex < outcomes.size) { + outcomes[outcomeIndex] + } else { + null + } + + val conditionId = market.conditionId ?: return null + + MarketInfo( + conditionId = conditionId, + outcomeIndex = outcomeIndex, + outcome = outcome + ) + } catch (e: Exception) { + logger.warn("查询市场信息失败: tokenId=$tokenId, error=${e.message}") + null + } + } + + /** + * 获取区块时间戳 + */ + 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) { + val blockRpcResponse = blockResponse.body()!! + if (blockRpcResponse.error == null && blockRpcResponse.result != null) { + val blockJson = blockRpcResponse.result.asJsonObject + val timestampHex = blockJson.get("timestamp")?.asString + if (timestampHex != null) { + BigInteger(timestampHex.removePrefix("0x"), 16).toLong() * 1000 // 转换为毫秒 + } else { + null + } + } else { + null + } + } else { + null + } + } catch (e: Exception) { + logger.warn("获取区块时间戳失败: blockNumber=$blockNumber, error=${e.message}") + null + } + } + + /** + * 工具函数：地址转 topic（32字节，左对齐） + */ + fun addressToTopic32(address: String): String { + val clean = address.removePrefix("0x").lowercase() + return "0x" + clean.padStart(64, '0') + } + + /** + * 工具函数：topic 转地址 + */ + fun topicToAddress(topic: String): String { + val clean = topic.removePrefix("0x").lowercase() + return "0x" + clean.takeLast(40) + } + + /** + * 工具函数：十六进制转 BigInteger + */ + fun hexToBigInt(hex: String): BigInteger { + val clean = hex.removePrefix("0x") + return if (clean.isBlank()) BigInteger.ZERO else BigInteger(clean, 16) + } + + /** + * 工具函数：十六进制转字节数组 + */ + fun bytesFromHex(hex: String): ByteArray { + val clean = hex.removePrefix("0x") + return ByteArray(clean.length / 2) { i -> + clean.substring(i * 2, i * 2 + 2).toInt(16).toByte() + } + } + + /** + * 工具函数：从字节数组切片 BigInteger（32字节） + */ + fun sliceBigInt32(bytes: ByteArray, offset: Int): BigInteger { + if (offset + 32 > bytes.size) return BigInteger.ZERO + val slice = bytes.sliceArray(offset until offset + 32) + return BigInteger(1, slice) + } +} + 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 new file mode 100644 index 0000000..c9f91ce --- /dev/null +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/UnifiedOnChainWsService.kt @@ -0,0 +1,567 @@ +package com.wrbug.polymarketbot.service.copytrading.monitor + +import com.google.gson.Gson +import com.google.gson.JsonObject +import com.wrbug.polymarketbot.api.* +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.PostConstruct +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.util.concurrent.ConcurrentHashMap + +/** + * 统一的链上 WebSocket 服务 + * 管理唯一的 WebSocket 连接，其他服务通过订阅的方式接收链上事件 + */ +@Service +class UnifiedOnChainWsService( + private val rpcNodeService: RpcNodeService, + private val retrofitFactory: RetrofitFactory +) { + + private val logger = LoggerFactory.getLogger(UnifiedOnChainWsService::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()) + + // 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() + + /** + * 订阅信息 + */ + data class SubscriptionInfo( + val subscriptionId: String, // 订阅的唯一标识 + val address: String, // 要监听的地址（Leader 地址或账户代理地址） + val entityType: String, // 实体类型：LEADER 或 ACCOUNT + val entityId: Long, // 实体 ID（Leader ID 或 Account ID） + val callback: suspend (String, OkHttpClient, EthereumRpcApi) -> Unit // 回调函数 + ) + + /** + * 订阅地址监听 + * @param subscriptionId 订阅的唯一标识（建议格式："{entityType}_{entityId}"） + * @param address 要监听的地址（Leader 地址或账户代理地址） + * @param entityType 实体类型：LEADER 或 ACCOUNT + * @param entityId 实体 ID（Leader ID 或 Account ID） + * @param callback 回调函数，当检测到该地址的交易时调用 + * @return 是否订阅成功 + */ + fun subscribe( + subscriptionId: String, + address: String, + entityType: String, + entityId: Long, + callback: suspend (String, OkHttpClient, EthereumRpcApi) -> Unit + ): Boolean { + try { + // 如果已经订阅，先取消 + if (subscriptions.containsKey(subscriptionId)) { + unsubscribe(subscriptionId) + } + + // 创建订阅信息 + val subscription = SubscriptionInfo( + subscriptionId = subscriptionId, + address = address.lowercase(), + entityType = entityType, + entityId = entityId, + callback = callback + ) + + subscriptions[subscriptionId] = subscription + + // 如果已连接，立即订阅 + if (isConnected) { + scope.launch { + subscribeAddress(subscription) + } + } else { + // 如果未连接，启动连接 + startConnection() + } + + logger.info("订阅地址监听: subscriptionId=$subscriptionId, address=$address, entityType=$entityType, entityId=$entityId") + return true + } catch (e: Exception) { + logger.error("订阅地址监听失败: subscriptionId=$subscriptionId, address=$address, error=${e.message}", e) + return false + } + } + + /** + * 取消订阅 + */ + 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 (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 (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") + } + } + 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() + } + + @PostConstruct + fun init() { + // 服务启动时不自动连接，等待有订阅时再连接 + logger.info("统一链上 WebSocket 服务已初始化") + } + + @PreDestroy + fun destroy() { + stop() + scope.cancel() + } +} + 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 adba5c9..0f6281c 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 @@ -194,16 +194,27 @@ class OrderStatusUpdateService( // 如果 orderId 不是 0x 开头，直接标记为已处理（priceUpdated = true 表示已处理，包括价格更新和通知发送） if (!record.sellOrderId.startsWith("0x", ignoreCase = true)) { logger.debug("卖出订单ID非0x开头，直接标记为已处理: orderId=${record.sellOrderId}") - // 发送通知（使用临时数据） - sendSellOrderNotification( - record = record, - useTemporaryData = true, - account = account, - copyTrading = copyTrading, - clobApi = clobApi, - apiSecret = apiSecret, - apiPassphrase = apiPassphrase - ) + + // 检查是否为自动生成的订单（AUTO_ 或 AUTO_FIFO_ 开头），如果是则不发送通知 + val isAutoOrder = record.sellOrderId.startsWith("AUTO_", ignoreCase = true) || + record.sellOrderId.startsWith("AUTO_FIFO_", ignoreCase = true) || + record.sellOrderId.startsWith("AUTO_WS_", ignoreCase = true) + + if (!isAutoOrder) { + // 非自动订单，发送通知（使用临时数据） + sendSellOrderNotification( + record = record, + useTemporaryData = true, + account = account, + copyTrading = copyTrading, + clobApi = clobApi, + apiSecret = apiSecret, + apiPassphrase = apiPassphrase + ) + } else { + logger.debug("自动生成的订单，跳过发送通知: orderId=${record.sellOrderId}") + } + // 标记为已处理（priceUpdated = true 同时表示价格已更新和通知已发送） val updatedRecord = SellMatchRecord( id = record.id, @@ -223,6 +234,32 @@ class OrderStatusUpdateService( continue } + // 检查是否为自动生成的订单（AUTO_ 或 AUTO_FIFO_ 开头），如果是则跳过发送通知 + val isAutoOrder = record.sellOrderId.startsWith("AUTO_", ignoreCase = true) || + record.sellOrderId.startsWith("AUTO_FIFO_", ignoreCase = true) || + record.sellOrderId.startsWith("AUTO_WS_", ignoreCase = true) + + if (isAutoOrder) { + logger.debug("自动生成的订单，跳过发送通知并直接标记为已处理: orderId=${record.sellOrderId}") + // 直接标记为已处理，不发送通知 + val updatedRecord = SellMatchRecord( + id = record.id, + copyTradingId = record.copyTradingId, + sellOrderId = record.sellOrderId, + leaderSellTradeId = record.leaderSellTradeId, + marketId = record.marketId, + side = record.side, + outcomeIndex = record.outcomeIndex, + totalMatchedQuantity = record.totalMatchedQuantity, + sellPrice = record.sellPrice, + totalRealizedPnl = record.totalRealizedPnl, + priceUpdated = true, // 标记为已处理 + createdAt = record.createdAt + ) + sellMatchRecordRepository.save(updatedRecord) + continue + } + // 查询订单详情，获取实际成交价 val actualSellPrice = trackingService.getActualExecutionPrice( orderId = record.sellOrderId, diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/util/EthereumUtils.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/util/EthereumUtils.kt index 51f3d35..fff918d 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/util/EthereumUtils.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/util/EthereumUtils.kt @@ -77,6 +77,67 @@ object EthereumUtils { return BigInteger(cleanHex, 16) } + /** + * 从 ABI 编码的响应中解析 uint256 数组 + * ABI 编码格式： + * - offset (32 bytes): 数组数据的位置偏移量 + * - length (32 bytes): 数组长度 + * - data: 每个元素 32 字节 + * @param hexResult 十六进制结果（完整的 ABI 编码响应） + * @param offset 数组数据的偏移位置（字节数，从 offset 位置开始读取） + * @return BigInteger 数组 + */ + fun decodeUint256Array(hexResult: String, offset: Int = 0): List { + val cleanHex = hexResult.removePrefix("0x") + if (cleanHex.length < (offset + 1) * 64) { + return emptyList() + } + + // 从 offset 位置开始读取 + val startPos = offset * 64 // 每个 uint256 是 64 个十六进制字符 + val lengthHex = cleanHex.substring(startPos, startPos + 64) + val length = BigInteger(lengthHex, 16).toInt() + + if (length <= 0 || length > 100) { // 防止异常数据 + return emptyList() + } + + val result = mutableListOf() + for (i in 0 until length) { + val elementStart = startPos + 64 + (i * 64) // 跳过长度字段 + if (elementStart + 64 > cleanHex.length) { + break + } + val elementHex = cleanHex.substring(elementStart, elementStart + 64) + result.add(BigInteger(elementHex, 16)) + } + + return result + } + + /** + * 从 ABI 编码的元组响应中解析数据 + * 用于解析 getCondition 返回的 (uint256 payoutDenominator, uint256[] payouts) + * @param hexResult 十六进制结果 + * @return Pair + */ + fun decodeConditionResult(hexResult: String): Pair> { + val cleanHex = hexResult.removePrefix("0x") + + // 第一个 32 字节：payoutDenominator + val payoutDenominatorHex = cleanHex.substring(0, 64) + val payoutDenominator = BigInteger(payoutDenominatorHex, 16) + + // 第二个 32 字节：payouts 数组的偏移量（通常是 0x40 = 64 字节） + val offsetHex = cleanHex.substring(64, 128) + val offset = BigInteger(offsetHex, 16).toInt() / 32 // 转换为 32 字节单位 + + // 从 offset 位置解析数组 + val payouts = decodeUint256Array(hexResult, offset) + + return Pair(payoutDenominator, payouts) + } + /** * 计算 Keccak-256 哈希（Ethereum 标准） * 使用 BouncyCastle 库实现真正的 Keccak-256