From 2bb8cbc564209018468812c8102ef7c7694c14cf Mon Sep 17 00:00:00 2001 From: WrBug Date: Sun, 4 Jan 2026 15:50:23 +0800 Subject: [PATCH] =?UTF-8?q?=E7=A7=BB=E9=99=A4=E8=B7=9F=E5=8D=95=E4=BF=A1?= =?UTF-8?q?=E5=8F=B7=E4=B8=AD=E7=9A=84=E8=BD=AE=E8=AF=A2=20activity=20?= =?UTF-8?q?=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) - } - } -} -