移除跟单信号中的轮询 activity 接口逻辑

- 移除 CopyTradingMonitorService 中的 pollingService 相关调用 - 删除不再使用的 CopyTradingPollingService 类 - 跟单信号现在完全依赖链上 WebSocket 监听，不再使用轮询接口 - 优化账户列表接口性能（仅返回基本信息，不查询统计数据） - 改进 PositionCheckService 的错误处理（市场价格获取失败时跳过处理）
2026-01-04 15:50:23 +08:00
parent 8b73121c5d
commit 2bb8cbc564
4 changed files with 49 additions and 303 deletions
@@ -228,11 +228,12 @@ class AccountService(

    /**
     * 查询账户列表
+     * 列表接口只返回基本信息，不查询统计信息（统计信息只在详情接口中查询）
     */
    fun getAccountList(): Result<AccountListResponse> {
        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 {
@@ -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
+                        }
                    }
                }
            }
@@ -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()
    }
@@ -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<Long, Leader>()
-    
-    // 存储每个Leader已缓存的交易ID集合：leaderId -> Set<tradeId>
-    private val cachedTradeIds = ConcurrentHashMap<Long, MutableSet<String>>()
-    
-    // 存储每个Leader是否首次轮询：leaderId -> isFirstPoll
-    private val isFirstPoll = ConcurrentHashMap<Long, Boolean>()
-    
-    // 轮询任务
-    private var pollingJob: Job? = null
-    
-    /**
-     * 启动轮询监听
-     */
-    fun start(leaders: List<Leader>) {
-        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<List<UserActivityResponse>> = 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)
-        }
-    }
-}
-