From 2bb8cbc564209018468812c8102ef7c7694c14cf Mon Sep 17 00:00:00 2001 From: WrBug Date: Sun, 4 Jan 2026 15:50:23 +0800 Subject: [PATCH 1/5] =?UTF-8?q?=E7=A7=BB=E9=99=A4=E8=B7=9F=E5=8D=95?= =?UTF-8?q?=E4=BF=A1=E5=8F=B7=E4=B8=AD=E7=9A=84=E8=BD=AE=E8=AF=A2=20activi?= =?UTF-8?q?ty=20=E6=8E=A5=E5=8F=A3=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 移除 CopyTradingMonitorService 中的 pollingService 相关调用 - 删除不再使用的 CopyTradingPollingService 类 - 跟单信号现在完全依赖链上 WebSocket 监听，不再使用轮询接口 - 优化账户列表接口性能（仅返回基本信息，不查询统计数据） - 改进 PositionCheckService 的错误处理（市场价格获取失败时跳过处理） --- .../service/accounts/AccountService.kt | 28 +- .../service/accounts/PositionCheckService.kt | 24 +- .../monitor/CopyTradingMonitorService.kt | 21 +- .../monitor/CopyTradingPollingService.kt | 279 ------------------ 4 files changed, 49 insertions(+), 303 deletions(-) delete mode 100644 backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingPollingService.kt diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/AccountService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/AccountService.kt index f9f7048..a031cb8 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/AccountService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/AccountService.kt @@ -228,11 +228,12 @@ class AccountService( /** * 查询账户列表 + * 列表接口只返回基本信息，不查询统计信息（统计信息只在详情接口中查询） */ fun getAccountList(): Result { return try { val accounts = accountRepository.findAllByOrderByCreatedAtAsc() - val accountDtos = accounts.map { toDto(it) } + val accountDtos = accounts.map { toBasicDto(it) } Result.success( AccountListResponse( @@ -353,7 +354,30 @@ class AccountService( } /** - * 转换为 DTO + * 转换为基础 DTO（列表使用，不包含统计信息） + * 列表接口只返回基本信息，不查询统计信息，以提高性能 + */ + private fun toBasicDto(account: Account): AccountDto { + return AccountDto( + id = account.id!!, + walletAddress = account.walletAddress, + proxyAddress = account.proxyAddress, + accountName = account.accountName, + isEnabled = account.isEnabled, + walletType = account.walletType, + apiKeyConfigured = account.apiKey != null, + apiSecretConfigured = account.apiSecret != null, + apiPassphraseConfigured = account.apiPassphrase != null, + totalOrders = null, + totalPnl = null, + activeOrders = null, + completedOrders = null, + positionCount = null + ) + } + + /** + * 转换为完整 DTO（详情使用，包含交易统计数据） * 包含交易统计数据（总订单数、总盈亏、活跃订单数、已完成订单数、持仓数量） */ private fun toDto(account: Account): AccountDto { 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 b33c395..f351bff 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 @@ -382,9 +382,15 @@ class PositionCheckService( if (ordersToMarkAsSold.isNotEmpty()) { // 有订单创建时间超过2分钟，认为仓位已被出售 - val currentPrice = getCurrentMarketPrice(marketId, outcomeIndex) - updateOrdersAsSold(ordersToMarkAsSold, currentPrice, copyTrading.id, marketId, outcomeIndex) - logger.debug("仓位不存在且订单创建时间超过2分钟，标记为已卖出: marketId=$marketId, outcomeIndex=$outcomeIndex, orderCount=${ordersToMarkAsSold.size}") + try { + 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}") + // 无法获取价格时，跳过该市场的处理，等待下次检查时再试 + continue + } } else { // 订单创建时间不足2分钟，可能是刚创建的订单，暂时不处理 logger.debug("仓位不存在但订单创建时间不足2分钟，暂不标记为已卖出: marketId=$marketId, outcomeIndex=$outcomeIndex, orderCount=${orders.size}, oldestOrderAge=${orders.minOfOrNull { now - it.createdAt }?.let { "${it}ms" } ?: "N/A"}") @@ -413,9 +419,15 @@ class PositionCheckService( } // 如果已成交数量 > 0，按FIFO顺序匹配订单 - val currentPrice = getCurrentMarketPrice(marketId, outcomeIndex) - updateOrdersAsSoldByFIFO(orders, soldQuantity, currentPrice, - copyTrading.id, marketId, outcomeIndex) + try { + 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}") + // 无法获取价格时，跳过该市场的处理，等待下次检查时再试 + continue + } } } } 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 d4b5c89..29a60bb 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 @@ -14,7 +14,7 @@ import org.springframework.stereotype.Service /** * 跟单监听服务（主服务） * 管理所有Leader的交易监听 - * 同时运行链上 WebSocket 监听和轮询监听（并行处理） + * 使用链上 WebSocket 监听 Leader 的交易（实时，秒级延迟） * 同时监听跟单账户的卖出/赎回事件（通过链上 WebSocket） */ @Service @@ -22,7 +22,6 @@ class CopyTradingMonitorService( private val copyTradingRepository: CopyTradingRepository, private val leaderRepository: LeaderRepository, private val accountRepository: AccountRepository, - private val pollingService: CopyTradingPollingService, private val onChainWsService: OnChainWsService, private val accountOnChainMonitorService: AccountOnChainMonitorService ) { @@ -51,15 +50,14 @@ class CopyTradingMonitorService( @PreDestroy fun destroy() { scope.cancel() - // 停止轮询和链上 WS 监听 - pollingService.stop() + // 停止链上 WS 监听 onChainWsService.stop() accountOnChainMonitorService.stop() } /** * 启动监听 - * 同时启动链上 WebSocket 监听和轮询监听（并行运行） + * 启动链上 WebSocket 监听 Leader 的交易（实时，秒级延迟） * 同时启动跟单账户的链上 WebSocket 监听（用于检测卖出/赎回事件） */ suspend fun startMonitoring() { @@ -82,13 +80,9 @@ class CopyTradingMonitorService( accountRepository.findById(accountId).orElse(null) } - // 4. 同时启动链上 WebSocket 监听和轮询监听（并行运行） - // 链上 WS 监听 Leader 的交易（实时，秒级延迟） + // 4. 启动链上 WebSocket 监听 Leader 的交易（实时，秒级延迟） onChainWsService.start(leaders) - // 轮询监听 Leader 的交易（延迟，2秒间隔，作为备份） - pollingService.start(leaders) - // 5. 启动跟单账户的链上 WebSocket 监听（用于检测卖出/赎回事件） accountOnChainMonitorService.start(accounts) } @@ -106,9 +100,8 @@ class CopyTradingMonitorService( return } - // 同时添加到链上 WS 监听和轮询监听（如果不在列表中才添加） + // 添加到链上 WS 监听（如果不在列表中才添加） onChainWsService.addLeader(leader) - pollingService.addLeader(leader) } /** @@ -124,7 +117,6 @@ class CopyTradingMonitorService( // 没有启用的跟单配置了，移除监听 onChainWsService.removeLeader(leaderId) - pollingService.removeLeader(leaderId) } /** @@ -139,7 +131,6 @@ class CopyTradingMonitorService( if (copyTradings.isNotEmpty()) { // 有启用的跟单配置，确保在监听列表中 onChainWsService.addLeader(leader) - pollingService.addLeader(leader) // 更新账户监听（添加该配置关联的账户） val accountIds = copyTradings.map { it.accountId }.distinct() @@ -152,7 +143,6 @@ class CopyTradingMonitorService( } else { // 没有启用的跟单配置，移除监听 onChainWsService.removeLeader(leaderId) - pollingService.removeLeader(leaderId) } } @@ -182,7 +172,6 @@ class CopyTradingMonitorService( suspend fun restartMonitoring() { // 停止所有监听 onChainWsService.stop() - pollingService.stop() delay(1000) // 等待1秒 startMonitoring() } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingPollingService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingPollingService.kt deleted file mode 100644 index e5cbef1..0000000 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingPollingService.kt +++ /dev/null @@ -1,279 +0,0 @@ -package com.wrbug.polymarketbot.service.copytrading.monitor - -import com.wrbug.polymarketbot.api.TradeResponse -import com.wrbug.polymarketbot.api.UserActivityResponse -import com.wrbug.polymarketbot.entity.Leader -import com.wrbug.polymarketbot.repository.CopyTradingTemplateRepository -import com.wrbug.polymarketbot.util.RetrofitFactory -import jakarta.annotation.PreDestroy -import kotlinx.coroutines.* -import org.slf4j.LoggerFactory -import org.springframework.beans.factory.annotation.Value -import com.wrbug.polymarketbot.service.copytrading.statistics.CopyOrderTrackingService -import org.springframework.stereotype.Service -import retrofit2.Response -import java.util.concurrent.ConcurrentHashMap - -/** - * 跟单轮询监听服务 - * 通过定期轮询 Polymarket Data API 的 /activity 接口获取Leader的交易记录 - * 使用 /activity 接口可以查询用户的链上活动，包括交易 - */ -@Service -class CopyTradingPollingService( - private val copyOrderTrackingService: CopyOrderTrackingService, - private val retrofitFactory: RetrofitFactory, - private val templateRepository: CopyTradingTemplateRepository -) { - - private val logger = LoggerFactory.getLogger(CopyTradingPollingService::class.java) - - @Value("\${copy.trading.polling.interval:2000}") - private var pollingInterval: Long = 2000 // 轮询间隔（毫秒），默认2秒 - - @Value("\${copy.trading.polling.enabled:true}") - private var pollingEnabled: Boolean = true // 是否启用轮询 - - private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) - - // 存储需要监听的Leader：leaderId -> Leader - private val monitoredLeaders = ConcurrentHashMap() - - // 存储每个Leader已缓存的交易ID集合：leaderId -> Set - private val cachedTradeIds = ConcurrentHashMap>() - - // 存储每个Leader是否首次轮询：leaderId -> isFirstPoll - private val isFirstPoll = ConcurrentHashMap() - - // 轮询任务 - private var pollingJob: Job? = null - - /** - * 启动轮询监听 - */ - fun start(leaders: List) { - if (!pollingEnabled) { - return - } - - leaders.forEach { leader -> - addLeader(leader) - } - - // 启动轮询任务 - startPolling() - } - - /** - * 添加Leader监听 - */ - fun addLeader(leader: Leader) { - if (leader.id == null) { - logger.warn("Leader ID为空，跳过: ${leader.leaderAddress}") - return - } - - val leaderId = leader.id - monitoredLeaders[leaderId] = leader - // 初始化缓存的交易ID集合 - cachedTradeIds[leaderId] = mutableSetOf() - // 首次轮询标志，用于缓存数据而不处理 - isFirstPoll[leaderId] = true - - // 如果轮询任务没有运行，启动它 - startPolling() - } - - /** - * 移除Leader监听 - */ - fun removeLeader(leaderId: Long) { - monitoredLeaders.remove(leaderId) - cachedTradeIds.remove(leaderId) - isFirstPoll.remove(leaderId) - - // 如果没有需要监听的Leader了，停止轮询任务 - if (monitoredLeaders.isEmpty()) { - stopPolling() - } - } - - /** - * 停止所有监听 - */ - fun stop() { - stopPolling() - monitoredLeaders.clear() - cachedTradeIds.clear() - isFirstPoll.clear() - } - - /** - * 启动轮询任务 - */ - private fun startPolling() { - if (pollingJob != null && pollingJob!!.isActive) { - return - } - - if (monitoredLeaders.isEmpty()) { - return - } - - pollingJob = scope.launch { - - while (isActive) { - try { - // 轮询所有Leader的交易 - pollAllLeaders() - - // 等待下一次轮询 - delay(pollingInterval) - } catch (e: Exception) { - logger.error("轮询任务异常", e) - delay(pollingInterval) // 异常后继续等待 - } - } - } - } - - /** - * 停止轮询任务 - */ - private fun stopPolling() { - pollingJob?.cancel() - pollingJob = null - } - - /** - * 轮询所有Leader的交易 - */ - private suspend fun pollAllLeaders() { - val leaders = monitoredLeaders.values.toList() - - if (leaders.isEmpty()) { - return - } - - // 并发轮询所有Leader（限制并发数） - leaders.chunked(10).forEach { chunk -> - chunk.forEach { leader -> - try { - pollLeaderTrades(leader) - } catch (e: Exception) { - logger.error("轮询Leader交易失败: leaderId=${leader.id}, address=${leader.leaderAddress}", e) - } - } - // 每个chunk之间稍作延迟，避免API限流 - delay(100) - } - } - - /** - * 轮询单个Leader的交易 - * 使用 Polymarket Data API 的 /activity 接口 - * 通过 diff 分析增量数据，不使用 start 字段 - */ - private suspend fun pollLeaderTrades(leader: Leader) { - if (leader.id == null) { - return - } - - val leaderId = leader.id - val leaderAddress = leader.leaderAddress - - try { - val firstPoll = isFirstPoll[leaderId] == true - val cachedIds = cachedTradeIds[leaderId] ?: mutableSetOf() - - // 创建 Data API 客户端（不需要认证） - val dataApi = retrofitFactory.createDataApi() - - // 查询用户活动（只查询交易类型，不使用 start 字段） - // 查询最近的数据（limit=100），通过 diff 找出增量 - val response: Response> = dataApi.getUserActivity( - user = leaderAddress, - limit = 100, // 每次最多查询100条 - offset = 0, - type = listOf("TRADE"), // 只查询交易类型 - start = null, // 不使用 start 字段 - sortBy = "TIMESTAMP", - sortDirection = "DESC" // 按时间戳降序，最新的在前 - ) - - if (!response.isSuccessful || response.body() == null) { - logger.warn("获取Leader活动失败: leaderId=$leaderId, address=$leaderAddress, code=${response.code()}, message=${response.message()}") - return - } - - val activities = response.body()!! - - // 将 UserActivityResponse 转换为 TradeResponse - val allTrades = activities.mapNotNull { activity -> - // 只处理交易类型 - if (activity.type != "TRADE" || activity.side == null || activity.price == null || activity.size == null) { - return@mapNotNull null - } - - // 转换为 TradeResponse - TradeResponse( - id = activity.transactionHash ?: "${activity.timestamp}_${activity.conditionId}", - market = activity.conditionId, - side = activity.side, // BUY 或 SELL - price = activity.price.toString(), - size = activity.size.toString(), - timestamp = activity.timestamp.toString(), // 时间戳（秒） - user = activity.proxyWallet, - outcomeIndex = activity.outcomeIndex, // 结果索引（0=YES, 1=NO） - outcome = activity.outcome // 结果名称 - ) - } - - if (firstPoll) { - // 首次轮询：缓存所有查询到的交易ID，不处理 - val tradeIds = allTrades.map { it.id }.toSet() - cachedIds.addAll(tradeIds) - cachedTradeIds[leaderId] = cachedIds - - // 标记首次轮询完成 - isFirstPoll[leaderId] = false - } else { - // 后续轮询：通过 diff 找出新增的交易 - val newTradeIds = allTrades.map { it.id }.toSet() - val incrementalTradeIds = newTradeIds - cachedIds - - if (incrementalTradeIds.isNotEmpty()) { - // 找出新增的交易 - val incrementalTrades = allTrades.filter { it.id in incrementalTradeIds } - - - // 处理新增的交易 - incrementalTrades.forEach { trade -> - try { - // 检查是否已处理（去重由processTrade内部处理） - copyOrderTrackingService.processTrade(leaderId, trade, "polling") - } catch (e: Exception) { - logger.error("处理交易失败: leaderId=$leaderId, tradeId=${trade.id}", e) - } - } - - // 更新缓存：添加新增的交易ID - cachedIds.addAll(incrementalTradeIds) - cachedTradeIds[leaderId] = cachedIds - - } else { - } - - // 限制缓存大小，避免内存溢出（只保留最近1000条） - if (cachedIds.size > 1000) { - // 保留最新的1000条（由于查询是按时间戳降序，保留前1000条即可） - val sortedTradeIds = allTrades.map { it.id }.take(1000).toSet() - cachedTradeIds[leaderId] = sortedTradeIds.toMutableSet() - } - } - } catch (e: Exception) { - logger.error("轮询Leader交易异常: leaderId=$leaderId, address=$leaderAddress", e) - } - } -} - From 3e667b70fddf38c46dd5d935e9f33640d00b0cea Mon Sep 17 00:00:00 2001 From: WrBug Date: Sun, 4 Jan 2026 20:36:41 +0800 Subject: [PATCH 2/5] feat: enhance copy trading logic and UI - fix: correct copyRatio scaling (divide by 100) - fix: enforce minimum order size with round-up logic - feat: customize Telegram notification icons - feat: optimize frontend order list (active loading, refresh buttons) - refactor: optimize OnChain WebSocket connection management --- .../service/accounts/PositionCheckService.kt | 12 +- .../service/common/BlockchainService.kt | 89 +- .../service/common/MarketPriceService.kt | 69 +- .../copytrading/monitor/OnChainWsService.kt | 13 +- .../copytrading/monitor/OnChainWsUtils.kt | 1 + .../monitor/UnifiedOnChainWsService.kt | 767 ++++++++---------- .../statistics/CopyOrderTrackingService.kt | 37 +- .../statistics/OrderStatusUpdateService.kt | 8 +- .../system/TelegramNotificationService.kt | 11 +- docs/zh/smart-money-analysis.md | 1 + frontend/src/locales/zh-CN/common.json | 2 +- .../pages/CopyTradingOrders/BuyOrdersTab.tsx | 67 +- .../CopyTradingOrders/MatchedOrdersTab.tsx | 65 +- .../pages/CopyTradingOrders/SellOrdersTab.tsx | 67 +- .../src/pages/CopyTradingOrders/index.tsx | 15 +- 15 files changed, 630 insertions(+), 594 deletions(-) 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: } ]} /> From 6b8c11f1718bbef9a804052882d98d69d3b10599 Mon Sep 17 00:00:00 2001 From: WrBug Date: Sun, 4 Jan 2026 21:30:46 +0800 Subject: [PATCH 3/5] fix: refine price rounding and optimize order polling - fix: round buy prices UP and sell prices DOWN to 2 decimal places to match USDC precision - perf: optimize order status polling by filtering fully_matched orders in SQL - chore: add debug logging for order creation and FAK error investigation --- .../repository/CopyOrderTrackingRepository.kt | 5 + .../statistics/CopyOrderTrackingService.kt | 267 ++++++++++-------- .../statistics/OrderStatusUpdateService.kt | 9 +- 3 files changed, 158 insertions(+), 123 deletions(-) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CopyOrderTrackingRepository.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CopyOrderTrackingRepository.kt index 84d1bae..9697f57 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CopyOrderTrackingRepository.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CopyOrderTrackingRepository.kt @@ -61,5 +61,10 @@ interface CopyOrderTrackingRepository : JpaRepository { */ @Query("SELECT t FROM CopyOrderTracking t WHERE t.createdAt <= :beforeTime") fun findByCreatedAtBefore(beforeTime: Long): List + + /** + * 查询指定时间之前创建且状态不为指定状态的订单 + */ + fun findByCreatedAtBeforeAndStatusNot(beforeTime: Long, status: String): List } 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 2bc31e4..88fa255 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 @@ -25,6 +25,7 @@ import com.wrbug.polymarketbot.util.CryptoUtils import org.springframework.stereotype.Service import org.springframework.transaction.annotation.Transactional import java.math.BigDecimal +import kotlin.math.max /** * 订单跟踪服务 @@ -54,16 +55,16 @@ open class CopyOrderTrackingService( // 协程作用域（用于异步发送通知） private val notificationScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) - + // 使用 Mutex 保证线程安全（按交易ID锁定） private val tradeMutexMap = ConcurrentHashMap() - + // 订单创建重试配置 companion object { private const val MAX_RETRY_ATTEMPTS = 2 // 最多重试次数（首次 + 1次重试） private const val RETRY_DELAY_MS = 3000L // 重试前等待时间（毫秒，3秒） } - + /** * 获取或创建 Mutex（按交易ID） */ @@ -121,85 +122,85 @@ open class CopyOrderTrackingService( suspend fun processTrade(leaderId: Long, trade: TradeResponse, source: String): Result { // 获取该交易的 Mutex（按交易ID锁定，不同交易可以并行处理） val mutex = getMutex(leaderId, trade.id) - + return mutex.withLock { try { - // 1. 检查是否已处理（去重，包括失败状态） - val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) + // 1. 检查是否已处理（去重，包括失败状态） + val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) - if (existingProcessed != null) { - if (existingProcessed.status == "FAILED") { + if (existingProcessed != null) { + if (existingProcessed.status == "FAILED") { return@withLock Result.success(Unit) - } - return@withLock Result.success(Unit) - } - - // 2. 处理交易逻辑 - val result = when (trade.side.uppercase()) { - "BUY" -> processBuyTrade(leaderId, trade) - "SELL" -> processSellTrade(leaderId, trade) - else -> { - logger.warn("未知的交易方向: ${trade.side}") - Result.failure(IllegalArgumentException("未知的交易方向: ${trade.side}")) - } - } - - if (result.isFailure) { - logger.error( - "处理交易失败: leaderId=$leaderId, tradeId=${trade.id}, side=${trade.side}", - result.exceptionOrNull() - ) - return@withLock result - } - - // 3. 标记为已处理（成功状态） - // 由于使用了 Mutex，这里理论上不会出现并发冲突，但保留异常处理作为兜底 - try { - val processed = ProcessedTrade( - leaderId = leaderId, - leaderTradeId = trade.id, - tradeType = trade.side.uppercase(), - source = source, - status = "SUCCESS", - processedAt = System.currentTimeMillis() - ) - processedTradeRepository.save(processed) - } catch (e: Exception) { - // 检查是否是唯一键冲突异常（理论上不会发生，但保留作为兜底） - if (isUniqueConstraintViolation(e)) { - val existing = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) - if (existing != null) { - if (existing.status == "FAILED") { - logger.debug("交易已标记为失败，跳过处理: leaderId=$leaderId, tradeId=${trade.id}") - return@withLock Result.success(Unit) - } - logger.debug("交易已处理（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existing.status}") - return@withLock Result.success(Unit) - } else { - // 如果检查不到，可能是事务隔离级别问题，等待一下再查询 - delay(100) - val existingAfterDelay = - processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) - if (existingAfterDelay != null) { - logger.debug("延迟查询到记录（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existingAfterDelay.status}") - return@withLock Result.success(Unit) - } - logger.warn( - "保存ProcessedTrade时发生唯一约束冲突，但查询不到记录: leaderId=$leaderId, tradeId=${trade.id}", - e - ) - return@withLock Result.success(Unit) } - } else { - // 其他类型的异常，重新抛出 - throw e + return@withLock Result.success(Unit) } - } - Result.success(Unit) - } catch (e: Exception) { - logger.error("处理交易异常: leaderId=$leaderId, tradeId=${trade.id}", e) - Result.failure(e) + // 2. 处理交易逻辑 + val result = when (trade.side.uppercase()) { + "BUY" -> processBuyTrade(leaderId, trade) + "SELL" -> processSellTrade(leaderId, trade) + else -> { + logger.warn("未知的交易方向: ${trade.side}") + Result.failure(IllegalArgumentException("未知的交易方向: ${trade.side}")) + } + } + + if (result.isFailure) { + logger.error( + "处理交易失败: leaderId=$leaderId, tradeId=${trade.id}, side=${trade.side}", + result.exceptionOrNull() + ) + return@withLock result + } + + // 3. 标记为已处理（成功状态） + // 由于使用了 Mutex，这里理论上不会出现并发冲突，但保留异常处理作为兜底 + try { + val processed = ProcessedTrade( + leaderId = leaderId, + leaderTradeId = trade.id, + tradeType = trade.side.uppercase(), + source = source, + status = "SUCCESS", + processedAt = System.currentTimeMillis() + ) + processedTradeRepository.save(processed) + } catch (e: Exception) { + // 检查是否是唯一键冲突异常（理论上不会发生，但保留作为兜底） + if (isUniqueConstraintViolation(e)) { + val existing = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) + if (existing != null) { + if (existing.status == "FAILED") { + logger.debug("交易已标记为失败，跳过处理: leaderId=$leaderId, tradeId=${trade.id}") + return@withLock Result.success(Unit) + } + logger.debug("交易已处理（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existing.status}") + return@withLock Result.success(Unit) + } else { + // 如果检查不到，可能是事务隔离级别问题，等待一下再查询 + delay(100) + val existingAfterDelay = + processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) + if (existingAfterDelay != null) { + logger.debug("延迟查询到记录（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existingAfterDelay.status}") + return@withLock Result.success(Unit) + } + logger.warn( + "保存ProcessedTrade时发生唯一约束冲突，但查询不到记录: leaderId=$leaderId, tradeId=${trade.id}", + e + ) + return@withLock Result.success(Unit) + } + } else { + // 其他类型的异常，重新抛出 + throw e + } + } + + Result.success(Unit) + } catch (e: Exception) { + logger.error("处理交易异常: leaderId=$leaderId, tradeId=${trade.id}", e) + Result.failure(e) } } } @@ -259,7 +260,7 @@ open class CopyOrderTrackingService( logger.warn("计算买入数量失败: ${e.message}", e) continue } - + // 计算跟单金额（USDC）= 买入数量 × 价格 val copyOrderAmount = buyQuantity.multi(tradePrice) @@ -268,8 +269,8 @@ open class CopyOrderTrackingService( // 传入跟单金额和市场ID，用于仓位检查（按市场检查仓位） // 订单簿只请求一次，返回给后续逻辑使用 val filterResult = filterService.checkFilters( - copyTrading, - tokenId, + copyTrading, + tokenId, tradePrice = tradePrice, copyOrderAmount = copyOrderAmount, marketId = trade.market @@ -382,44 +383,59 @@ open class CopyOrderTrackingService( if (copyTrading.copyMode == "RATIO") { val tradePrice = trade.price.toSafeBigDecimal() val rawOrderAmount = buyQuantity.multi(tradePrice) - + // 对按比例计算的金额进行向上取整处理（确保满足最小限制） // 向上取整到 2 位小数（USDC 精度） val roundedOrderAmount = rawOrderAmount.setScale(2, java.math.RoundingMode.CEILING) - + // 如果原始金额或向上取整后的金额小于最小限制，调整 buyQuantity 以满足最小限制 // 这样可以避免精度问题导致订单被错误地跳过 if (roundedOrderAmount.lt(copyTrading.minOrderSize)) { logger.debug("订单金额（向上取整后）低于最小限制，调整数量以满足最小限制: copyTradingId=${copyTrading.id}, rawAmount=$rawOrderAmount, roundedAmount=$roundedOrderAmount, min=${copyTrading.minOrderSize}") // 计算满足最小限制所需的数量（向上取整） - val minQuantity = copyTrading.minOrderSize.div(tradePrice, 8, java.math.RoundingMode.CEILING) + val minQuantity = + copyTrading.minOrderSize.div(tradePrice, 8, java.math.RoundingMode.CEILING) if (minQuantity.lte(BigDecimal.ZERO)) { logger.warn("计算出的最小数量为0或负数，跳过: copyTradingId=${copyTrading.id}") continue } // 使用调整后的数量 finalBuyQuantity = minQuantity - logger.debug("已调整数量以满足最小限制: copyTradingId=${copyTrading.id}, originalQuantity=$buyQuantity, adjustedQuantity=$finalBuyQuantity, adjustedAmount=${finalBuyQuantity.multi(tradePrice)}") + logger.debug( + "已调整数量以满足最小限制: copyTradingId=${copyTrading.id}, originalQuantity=$buyQuantity, adjustedQuantity=$finalBuyQuantity, adjustedAmount=${ + finalBuyQuantity.multi( + tradePrice + ) + }" + ) } else if (rawOrderAmount.lt(copyTrading.minOrderSize)) { // 原始金额小于最小限制，但向上取整后满足，调整数量以满足最小限制 logger.debug("订单金额（精度处理后）低于最小限制，调整数量以满足最小限制: copyTradingId=${copyTrading.id}, rawAmount=$rawOrderAmount, min=${copyTrading.minOrderSize}") // 计算满足最小限制所需的数量（向上取整） - val minQuantity = copyTrading.minOrderSize.div(tradePrice, 8, java.math.RoundingMode.CEILING) + val minQuantity = + copyTrading.minOrderSize.div(tradePrice, 8, java.math.RoundingMode.CEILING) if (minQuantity.lte(BigDecimal.ZERO)) { logger.warn("计算出的最小数量为0或负数，跳过: copyTradingId=${copyTrading.id}") continue } // 使用调整后的数量 finalBuyQuantity = minQuantity - logger.debug("已调整数量以满足最小限制: copyTradingId=${copyTrading.id}, originalQuantity=$buyQuantity, adjustedQuantity=$finalBuyQuantity, adjustedAmount=${finalBuyQuantity.multi(tradePrice)}") + logger.debug( + "已调整数量以满足最小限制: copyTradingId=${copyTrading.id}, originalQuantity=$buyQuantity, adjustedQuantity=$finalBuyQuantity, adjustedAmount=${ + finalBuyQuantity.multi( + tradePrice + ) + }" + ) } - + // 检查最大限制（使用调整后的数量） val finalOrderAmount = finalBuyQuantity.multi(tradePrice) if (finalOrderAmount.gt(copyTrading.maxOrderSize)) { logger.warn("订单金额超过最大限制，调整数量: copyTradingId=${copyTrading.id}, amount=$finalOrderAmount, max=${copyTrading.maxOrderSize}") // 调整数量到最大值 - val adjustedQuantity = copyTrading.maxOrderSize.div(tradePrice, 8, java.math.RoundingMode.DOWN) + val adjustedQuantity = + copyTrading.maxOrderSize.div(tradePrice, 8, java.math.RoundingMode.DOWN) if (adjustedQuantity.lte(BigDecimal.ZERO)) { logger.warn("调整后的数量为0或负数，跳过: copyTradingId=${copyTrading.id}") continue @@ -444,7 +460,7 @@ open class CopyOrderTrackingService( // 计算价格（应用价格容忍度） val buyPrice = calculateAdjustedPrice(trade.price.toSafeBigDecimal(), copyTrading, isBuy = true) - + logger.debug("计算价格结果：$buyPrice") // 在创建订单前，检查订单簿中是否有可匹配的订单（避免 FAK 订单失败） // 如果过滤检查时已经获取了订单簿，直接使用；否则重新获取 val orderbookForCheck = orderbook ?: run { @@ -455,21 +471,21 @@ open class CopyOrderTrackingService( null } } - + // 检查是否有可匹配的卖单（asks） if (orderbookForCheck != null) { val bestAsk = orderbookForCheck.asks .mapNotNull { it.price.toSafeBigDecimal() } .minOrNull() - + if (bestAsk == null) { logger.warn("订单簿中没有卖单，跳过创建订单: copyTradingId=${copyTrading.id}, tradeId=${trade.id}") continue } - + // 如果调整后的买入价格低于最佳卖单价格，无法匹配 if (buyPrice.lt(bestAsk)) { - logger.warn("调整后的买入价格 ($buyPrice) 低于最佳卖单价格 ($bestAsk)，无法匹配，跳过创建订单: copyTradingId=${copyTrading.id}, tradeId=${trade.id}, leaderPrice=${trade.price}, tolerance=${copyTrading.priceTolerance}") + logger.info("调整后的买入价格 ($buyPrice) 低于最佳卖单价格 ($bestAsk)，无法匹配，跳过创建订单: copyTradingId=${copyTrading.id}, tradeId=${trade.id}, leaderPrice=${trade.price}, tolerance=${copyTrading.priceTolerance}") continue } } @@ -499,6 +515,8 @@ open class CopyOrderTrackingService( // 解密私钥 val decryptedPrivateKey = decryptPrivateKey(account) + logger.info("准备创建买入订单: copyTradingId=${copyTrading.id}, tradeId=${trade.id}, leaderPrice=${trade.price}, tolerance=${copyTrading.priceTolerance}, calculatedPrice=$buyPrice, quantity=$finalBuyQuantity") + // 调用API创建订单（带重试机制） // 重试策略：最多重试 MAX_RETRY_ATTEMPTS 次，每次重试前等待 RETRY_DELAY_MS 毫秒 // 每次重试都会重新生成salt并重新签名，确保签名唯一性 @@ -519,6 +537,7 @@ open class CopyOrderTrackingService( if (createOrderResult.isFailure) { // 提取错误信息（只保留 code 和 errorBody） val exception = createOrderResult.exceptionOrNull() + logger.error("创建买入订单失败: copyTradingId=${copyTrading.id}, tradeId=${trade.id}, leaderPrice=${trade.price}, myPrice=$buyPrice, error=${exception?.message}") // 发送订单失败通知（异步，不阻塞，仅在 pushFailedOrders 为 true 时发送） if (copyTrading.pushFailedOrders) { @@ -574,7 +593,7 @@ open class CopyOrderTrackingService( } val realOrderId = createOrderResult.getOrNull() ?: continue - + // 验证 orderId 格式（必须以 0x 开头的 16 进制） if (!isValidOrderId(realOrderId)) { logger.warn("买入订单ID格式无效，跳过保存: orderId=$realOrderId") @@ -712,7 +731,7 @@ open class CopyOrderTrackingService( for (order in unmatchedOrders) { val copyQty = order.quantity.toSafeBigDecimal() var leaderQty: BigDecimal? = null - + // 优先使用存储的 leaderBuyQuantity if (order.leaderBuyQuantity != null) { leaderQty = order.leaderBuyQuantity.toSafeBigDecimal() @@ -723,7 +742,7 @@ open class CopyOrderTrackingService( logger.debug("Leader 买入数量未存储，尝试查询 API: leaderBuyTradeId=${order.leaderBuyTradeId}, copyOrderId=${order.buyOrderId}") try { val tradesResponse = clobApi.getTrades(id = order.leaderBuyTradeId) - + if (tradesResponse.isSuccessful && tradesResponse.body() != null) { val tradesData = tradesResponse.body()!!.data if (tradesData.isNotEmpty()) { @@ -749,7 +768,7 @@ open class CopyOrderTrackingService( failCount++ } } - + // 如果成功获取到 Leader 买入数量，累加 if (leaderQty != null && leaderQty.gt(BigDecimal.ZERO)) { totalCopyQuantity = totalCopyQuantity.add(copyQty) @@ -758,7 +777,7 @@ open class CopyOrderTrackingService( logger.warn("无法获取 Leader 买入数量，跳过该订单: copyOrderId=${order.buyOrderId}, leaderBuyTradeId=${order.leaderBuyTradeId}") } } - + logger.info("固定金额模式计算结果汇总: copyTradingId=${copyTrading.id}, successCount=$successCount, failCount=$failCount, totalCopyQuantity=$totalCopyQuantity, totalLeaderQuantity=$totalLeaderQuantity") // 如果无法计算总比例（查询失败），使用默认比例 @@ -769,12 +788,12 @@ open class CopyOrderTrackingService( // 计算实际比例：跟单买入数量 / Leader 买入数量 val actualRatio = totalCopyQuantity.div(totalLeaderQuantity) - + // 计算需要卖出的数量：Leader 卖出数量 × 实际比例 val needMatch = leaderSellQuantity.multi(actualRatio) - + logger.debug("固定金额模式卖出数量计算: copyTradingId=${copyTrading.id}, leaderSellQuantity=$leaderSellQuantity, totalCopyQuantity=$totalCopyQuantity, totalLeaderQuantity=$totalLeaderQuantity, actualRatio=$actualRatio, needMatch=$needMatch") - + return needMatch } @@ -838,10 +857,12 @@ open class CopyOrderTrackingService( copyTrading = copyTrading ) } + "RATIO" -> { // 比例模式：直接使用配置的 copyRatio (需要除以100) leaderSellTrade.size.toSafeBigDecimal().multi(copyTrading.copyRatio.div(100)) } + else -> { logger.warn("不支持的 copyMode: ${copyTrading.copyMode}，使用默认比例模式") leaderSellTrade.size.toSafeBigDecimal().multi(copyTrading.copyRatio.div(100)) @@ -937,7 +958,7 @@ open class CopyOrderTrackingService( // 8. 解密私钥（在方法开始时解密一次，后续复用） val decryptedPrivateKey = decryptPrivateKey(account) - + // 9. 创建并签名卖出订单 val signedOrder = try { orderSigningService.createAndSignOrder( @@ -1007,7 +1028,7 @@ open class CopyOrderTrackingService( } else { logger.debug("卖出订单ID为0x开头，等待定时任务更新价格: orderId=$realSellOrderId") } - + // 使用下单价格，等待定时任务更新实际成交价 val actualSellPrice = sellPrice @@ -1056,19 +1077,19 @@ open class CopyOrderTrackingService( val savedDetail = detail.copy(matchRecordId = savedRecord.id!!) sellMatchDetailRepository.save(savedDetail) } - + logger.info("卖出订单已保存，等待轮询任务获取实际数据后发送通知: orderId=$realSellOrderId, copyTradingId=${copyTrading.id}") } /** * 创建订单（带重试机制） - * + * * 重试策略： * - 最多重试 MAX_RETRY_ATTEMPTS 次（首次尝试 + 重试） * - 每次重试前等待 RETRY_DELAY_MS 毫秒 * - 每次重试都重新生成salt并重新签名，确保签名唯一性 - * + * * @param clobApi CLOB API 客户端 * @param privateKey 私钥（用于签名） * @param makerAddress 代理钱包地址 @@ -1134,10 +1155,10 @@ open class CopyOrderTrackingService( } val errorMsg = "code=${orderResponse.code()}, errorBody=${errorBody ?: "null"}" lastError = Exception(errorMsg) - + // 记录错误日志 logger.error("创建订单失败 (尝试 $attempt/$MAX_RETRY_ATTEMPTS): copyTradingId=$copyTradingId, tradeId=$tradeId, $errorMsg") - + // 如果不是最后一次尝试，等待后重试 if (attempt < MAX_RETRY_ATTEMPTS) { delay(RETRY_DELAY_MS) @@ -1151,10 +1172,10 @@ open class CopyOrderTrackingService( if (!response.success || response.orderId == null) { val errorMsg = "errorMsg=${response.errorMsg}" lastError = Exception(errorMsg) - + // 记录错误日志 logger.error("创建订单失败 (尝试 $attempt/$MAX_RETRY_ATTEMPTS): copyTradingId=$copyTradingId, tradeId=$tradeId, $errorMsg") - + // 如果不是最后一次尝试，等待后重试 if (attempt < MAX_RETRY_ATTEMPTS) { delay(RETRY_DELAY_MS) @@ -1166,14 +1187,17 @@ open class CopyOrderTrackingService( // 创建订单成功 logger.info("创建订单成功: copyTradingId=$copyTradingId, tradeId=$tradeId, orderId=${response.orderId}, attempt=$attempt") return Result.success(response.orderId) - + } catch (e: Exception) { val errorMsg = "error=${e.message}" lastError = Exception(errorMsg, e) - + // 记录错误日志（包含堆栈） - logger.error("创建订单异常 (尝试 $attempt/$MAX_RETRY_ATTEMPTS): copyTradingId=$copyTradingId, tradeId=$tradeId, $errorMsg", e) - + logger.error( + "创建订单异常 (尝试 $attempt/$MAX_RETRY_ATTEMPTS): copyTradingId=$copyTradingId, tradeId=$tradeId, $errorMsg", + e + ) + // 如果不是最后一次尝试，等待后重试 if (attempt < MAX_RETRY_ATTEMPTS) { delay(RETRY_DELAY_MS) @@ -1185,7 +1209,10 @@ open class CopyOrderTrackingService( // 所有重试都失败 val finalError = lastError ?: Exception("error=未知错误") - logger.error("创建订单失败（所有重试都失败）: copyTradingId=$copyTradingId, tradeId=$tradeId, side=$side, price=$price, size=$size", finalError) + logger.error( + "创建订单失败（所有重试都失败）: copyTradingId=$copyTradingId, tradeId=$tradeId, side=$side, price=$price, size=$size", + finalError + ) return Result.failure(finalError) } @@ -1320,7 +1347,7 @@ open class CopyOrderTrackingService( // 计算价格调整范围（百分比） val tolerancePercent = tolerance.div(100) - val adjustment = originalPrice.multi(tolerancePercent) + val adjustment = originalPrice.multi(tolerancePercent).max(0.01.toSafeBigDecimal()) return if (isBuy) { // 买入：可以稍微加价以确保成交（在原价格基础上加容忍度） @@ -1373,7 +1400,7 @@ open class CopyOrderTrackingService( /** * 验证订单ID格式 * 订单ID必须以 0x 开头，且是有效的 16 进制字符串 - * + * * @param orderId 订单ID * @return 如果格式有效返回 true，否则返回 false */ @@ -1389,11 +1416,11 @@ open class CopyOrderTrackingService( // 检查是否只包含 0-9, a-f, A-F return hexPart.all { it in '0'..'9' || it in 'a'..'f' || it in 'A'..'F' } } - + /** * 获取订单的实际成交价 * 通过查询订单详情和关联的交易记录，计算加权平均成交价 - * + * * @param orderId 订单ID * @param clobApi CLOB API 客户端（已认证） * @param fallbackPrice 如果查询失败，使用此价格作为默认值 @@ -1412,14 +1439,14 @@ open class CopyOrderTrackingService( logger.warn("查询订单详情失败: orderId=$orderId, code=${orderResponse.code()}, errorBody=$errorBody") return fallbackPrice } - + val order = orderResponse.body() if (order == null) { // 响应体为空，可能是订单不存在或已过期 logger.warn("查询订单详情失败: 响应体为空, orderId=$orderId, code=${orderResponse.code()}") return fallbackPrice } - + // 2. 如果订单未成交，使用下单价格 if (order.status != "FILLED" && order.sizeMatched.toSafeBigDecimal() <= BigDecimal.ZERO) { logger.debug("订单未成交，使用下单价格: orderId=$orderId, status=${order.status}") @@ -1460,7 +1487,7 @@ open class CopyOrderTrackingService( for (trade in trades) { val tradePrice = trade.price.toSafeBigDecimal() val tradeSize = trade.size.toSafeBigDecimal() - + if (tradeSize > BigDecimal.ZERO) { totalAmount = totalAmount.add(tradePrice.multiply(tradeSize)) totalSize = totalSize.add(tradeSize) 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 e9a1fcc..d34cc71 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 @@ -145,14 +145,17 @@ class OrderStatusUpdateService( // 查询30秒前创建的订单,并过滤掉已经完全匹配的订单 // 已经完全匹配的订单(status = "fully_matched")不需要再检查 - val allOrdersToCheck = copyOrderTrackingRepository.findByCreatedAtBefore(thirtySecondsAgo) - val ordersToCheck = allOrdersToCheck.filter { it.status != "fully_matched" } + // 使用数据库查询过滤，避免加载过多数据 + val ordersToCheck = copyOrderTrackingRepository.findByCreatedAtBeforeAndStatusNot( + thirtySecondsAgo, + "fully_matched" + ) if (ordersToCheck.isEmpty()) { return } - logger.debug("检查 ${ordersToCheck.size} 个30秒前创建的订单是否成交 (已过滤 ${allOrdersToCheck.size - ordersToCheck.size} 个已完全匹配的订单)") + logger.debug("检查 ${ordersToCheck.size} 个30秒前创建的订单是否成交") // 按账户分组，避免重复创建 API 客户端 val ordersByAccount = ordersToCheck.groupBy { it.accountId } From c0e0942404223415ac4815f8425cc859bdeb548a Mon Sep 17 00:00:00 2001 From: WrBug Date: Sun, 4 Jan 2026 21:36:02 +0800 Subject: [PATCH 4/5] fix: prevent duplicate telegram notifications for same orderId - feat: add deduplication cache in TelegramNotificationService - fix: skip sending notification if orderId was sent within last 5 minutes --- .../system/TelegramNotificationService.kt | 39 +++++++++++++++++++ 1 file changed, 39 insertions(+) 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 11a9992..625c244 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 @@ -43,6 +43,25 @@ class TelegramNotificationService( // 协程作用域 private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + /** + * 发送订单成功通知 + * @param orderId 订单ID（用于查询订单详情获取实际价格和数量） + * @param marketTitle 市场标题 + * @param marketId 市场ID（conditionId），用于生成链接 + * @param marketSlug 市场slug，用于生成链接 + * @param side 订单方向（BUY/SELL），用于多语言显示 + * @param accountName 账户名称 + * @param walletAddress 钱包地址 + * @param clobApi CLOB API 客户端（可选，如果提供则查询订单详情获取实际价格和数量） + * @param apiKey API Key（可选，用于查询订单详情） + * @param apiSecret API Secret（可选，用于查询订单详情） + * @param apiPassphrase API Passphrase（可选，用于查询订单详情） + * @param walletAddressForApi 钱包地址（可选，用于查询订单详情） + * @param locale 语言设置（可选，如果提供则使用，否则使用 LocaleContextHolder 获取） + */ + // 已发送通知的订单ID缓存（key: orderId, value: timestamp） + private val sentOrderIds = java.util.concurrent.ConcurrentHashMap() + /** * 发送订单成功通知 * @param orderId 订单ID（用于查询订单详情获取实际价格和数量） @@ -79,6 +98,26 @@ class TelegramNotificationService( leaderName: String? = null, // Leader 名称（备注） configName: String? = null // 跟单配置名 ) { + // 1. 如果提供了 orderId，检查是否已发送过通知（去重） + if (orderId != null) { + val lastSentTime = sentOrderIds[orderId] + if (lastSentTime != null) { + // 如果5分钟内已发送过，跳过 + if (System.currentTimeMillis() - lastSentTime < 5 * 60 * 1000) { + logger.info("订单通知已发送过（5分钟内），跳过: orderId=$orderId") + return + } + } + // 记录发送时间 + sentOrderIds[orderId] = System.currentTimeMillis() + + // 简单的清理逻辑：如果缓存过大，清理过期的 + if (sentOrderIds.size > 1000) { + val expiryTime = System.currentTimeMillis() - 5 * 60 * 1000 + sentOrderIds.entries.removeIf { it.value < expiryTime } + } + } + // 获取语言设置（优先使用传入的 locale，否则从 LocaleContextHolder 获取） val currentLocale = locale ?: try { LocaleContextHolder.getLocale() From e48149a19b657c2dae3fa575dd5985a70fcfca8c Mon Sep 17 00:00:00 2001 From: WrBug Date: Mon, 5 Jan 2026 10:49:59 +0800 Subject: [PATCH 5/5] =?UTF-8?q?feat:=20=E4=BC=98=E5=8C=96=E8=B7=9F?= =?UTF-8?q?=E5=8D=95=E4=BB=93=E4=BD=8D=E6=A3=80=E6=9F=A5=E9=80=BB=E8=BE=91?= =?UTF-8?q?=EF=BC=8C=E4=BD=BF=E7=94=A8=E6=9C=AC=E5=9C=B0=E8=AE=A2=E5=8D=95?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=BA=93=E8=A7=A3=E5=86=B3=E6=8E=A8=E9=80=81?= =?UTF-8?q?=E5=BB=B6=E8=BF=9F=E9=97=AE=E9=A2=98=EF=BC=8C=E5=B9=B6=E5=BC=95?= =?UTF-8?q?=E5=85=A5=E6=B7=B7=E5=90=88=E6=A3=80=E6=9F=A5=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../repository/CopyOrderTrackingRepository.kt | 18 ++++++++ .../configs/CopyTradingFilterService.kt | 42 +++++++++++++------ 2 files changed, 47 insertions(+), 13 deletions(-) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CopyOrderTrackingRepository.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CopyOrderTrackingRepository.kt index 9697f57..018190f 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CopyOrderTrackingRepository.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CopyOrderTrackingRepository.kt @@ -66,5 +66,23 @@ interface CopyOrderTrackingRepository : JpaRepository { * 查询指定时间之前创建且状态不为指定状态的订单 */ fun findByCreatedAtBeforeAndStatusNot(beforeTime: Long, status: String): List + + /** + * 查询指定跟单配置下的活跃仓位数量 + * 活跃仓位定义为 remainingQuantity > 0 的不同 (marketId, outcomeIndex) 组合 + */ + @Query("SELECT COUNT(DISTINCT CONCAT(t.marketId, '_', COALESCE(t.outcomeIndex, -1))) FROM CopyOrderTracking t WHERE t.copyTradingId = :copyTradingId AND t.remainingQuantity > 0") + fun countActivePositions(copyTradingId: Long): Int + + /** + * 检查指定市场是否存在活跃仓位 + */ + fun existsByCopyTradingIdAndMarketIdAndRemainingQuantityGreaterThan(copyTradingId: Long, marketId: String, remainingQuantity: BigDecimal): Boolean + + /** + * 计算指定跟单配置和市场下的当前持仓总价值 (成本价计算) + */ + @Query("SELECT SUM(t.remainingQuantity * t.price) FROM CopyOrderTracking t WHERE t.copyTradingId = :copyTradingId AND t.marketId = :marketId AND t.remainingQuantity > 0") + fun sumCurrentPositionValueByMarket(copyTradingId: Long, marketId: String): BigDecimal? } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/configs/CopyTradingFilterService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/configs/CopyTradingFilterService.kt index 3a73ff0..9ca5413 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/configs/CopyTradingFilterService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/configs/CopyTradingFilterService.kt @@ -9,6 +9,7 @@ import com.wrbug.polymarketbot.util.toSafeBigDecimal import org.slf4j.LoggerFactory import com.wrbug.polymarketbot.service.common.PolymarketClobService import com.wrbug.polymarketbot.service.accounts.AccountService +import com.wrbug.polymarketbot.repository.CopyOrderTrackingRepository import org.springframework.stereotype.Service import java.math.BigDecimal @@ -18,7 +19,8 @@ import java.math.BigDecimal @Service class CopyTradingFilterService( private val clobService: PolymarketClobService, - private val accountService: AccountService + private val accountService: AccountService, + private val copyOrderTrackingRepository: CopyOrderTrackingRepository ) { private val logger = LoggerFactory.getLogger(CopyTradingFilterService::class.java) @@ -234,32 +236,46 @@ class CopyTradingFilterService( // 检查最大仓位金额（如果配置了） if (copyTrading.maxPositionValue != null) { - // 计算该市场的当前仓位总价值（累加该市场所有仓位的 currentValue） - val currentPositionValue = marketPositions.sumOf { position -> - position.currentValue.toSafeBigDecimal() - } + // 比较数据库成本价（本地订单记录）和外部持仓市值（可能来自其他终端的操作），取最大值 + val dbValue = copyOrderTrackingRepository.sumCurrentPositionValueByMarket(copyTrading.id!!, marketId) ?: BigDecimal.ZERO + val extValue = marketPositions.sumOf { it.currentValue.toSafeBigDecimal() } + val currentPositionValue = dbValue.max(extValue) // 检查：该市场的当前仓位 + 跟单金额 <= 最大仓位金额 val totalValueAfterOrder = currentPositionValue.add(copyOrderAmount) if (totalValueAfterOrder.gt(copyTrading.maxPositionValue)) { return FilterResult.maxPositionValueFailed( - "超过最大仓位金额限制: 当前该市场仓位=${currentPositionValue} USDC, 跟单金额=${copyOrderAmount} USDC, 总计=${totalValueAfterOrder} USDC > 最大限制=${copyTrading.maxPositionValue} USDC" + "超过最大仓位金额限制: 当前该市场仓位(取最大值)=${currentPositionValue} USDC (DB=${dbValue}, Ext=${extValue}), 跟单金额=${copyOrderAmount} USDC, 总计=${totalValueAfterOrder} USDC > 最大限制=${copyTrading.maxPositionValue} USDC" ) } } // 检查最大仓位数量（如果配置了） if (copyTrading.maxPositionCount != null) { - // 计算该市场的当前仓位数量（该市场不同方向的仓位算不同仓位） - val currentPositionCount = marketPositions.size + // 使用数据库中的订单记录计算活跃仓位数量（解决延迟问题） + val dbCount = copyOrderTrackingRepository.countActivePositions(copyTrading.id!!) - // 检查：该市场的当前仓位数量 <= 最大仓位数量 - // 注意：如果该市场已有仓位，跟单可能会增加新的仓位（不同方向）或增加现有仓位 - // 为了简化，我们检查当前该市场的仓位数量是否已经达到或超过限制 - if (currentPositionCount >= copyTrading.maxPositionCount) { + // 计算外部持仓中的唯一市场数量（防止遗漏非本项目创建的仓位） + val extCount = positions.currentPositions + .filter { it.accountId == copyTrading.accountId } + .map { it.marketId } + .distinct() + .size + + val currentPositionCount = maxOf(dbCount, extCount) + + // 检查：如果当前没有该市场的活跃仓位，且总仓位数量已达到限制，则不允许开新仓 + // 判断当前市场是否已有活跃仓位（数据库或外部持仓） + val hasDbPosition = copyOrderTrackingRepository.existsByCopyTradingIdAndMarketIdAndRemainingQuantityGreaterThan( + copyTrading.id, marketId, BigDecimal.ZERO + ) + val hasExtPosition = marketPositions.isNotEmpty() + val hasCurrentMarketPosition = hasDbPosition || hasExtPosition + + if (!hasCurrentMarketPosition && currentPositionCount >= copyTrading.maxPositionCount) { return FilterResult.maxPositionCountFailed( - "超过最大仓位数量限制: 当前该市场仓位数量=${currentPositionCount} >= 最大限制=${copyTrading.maxPositionCount}" + "超过最大仓位数量限制: 当前活跃仓位总数(取最大值)=${currentPositionCount} (DB=${dbCount}, Ext=${extCount}) >= 最大限制=${copyTrading.maxPositionCount}" ) } }