From b1e69135b8913d9b6af51a0948e4bd6245759bff Mon Sep 17 00:00:00 2001 From: WrBug Date: Thu, 15 Jan 2026 04:33:55 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E8=B7=9F=E5=8D=95?= =?UTF-8?q?=E8=AE=A2=E5=8D=95=E6=9D=A5=E6=BA=90=E5=AD=97=E6=AE=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在 CopyOrderTracking 实体添加 source 字段，记录订单来源 - 支持 activity-ws（Polymarket WebSocket）和 onchain-ws（OnChain WebSocket）两种来源 - 更新 processBuyTrade 方法，接收并保存 source 参数 - 更新 OrderStatusUpdateService，保留原始 source 信息 - 添加数据库迁移脚本 V25 - 统一 CopyTradingWebSocketService 使用 activity-ws 作为来源 --- .../polymarketbot/entity/CopyOrderTracking.kt | 5 +- .../monitor/CopyTradingWebSocketService.kt | 2 +- .../statistics/CopyOrderTrackingService.kt | 7 +- .../statistics/OrderStatusUpdateService.kt | 393 ++++++++++-------- ...V25__add_source_to_copy_order_tracking.sql | 12 + 5 files changed, 230 insertions(+), 189 deletions(-) create mode 100644 backend/src/main/resources/db/migration/V25__add_source_to_copy_order_tracking.sql diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/entity/CopyOrderTracking.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/entity/CopyOrderTracking.kt index be0224b..fbfd97e 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/entity/CopyOrderTracking.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/entity/CopyOrderTracking.kt @@ -58,7 +58,10 @@ data class CopyOrderTracking( @Column(name = "notification_sent", nullable = false) var notificationSent: Boolean = false, // 是否已发送通知（从订单详情获取实际数据后发送） - + + @Column(name = "source", nullable = false, length = 20) + val source: String, // 订单来源：activity-ws（Polymarket WebSocket）、onchain-ws（OnChain WebSocket） + @Column(name = "created_at", nullable = false) val createdAt: Long = System.currentTimeMillis(), diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingWebSocketService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingWebSocketService.kt index 23b2974..8583d45 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingWebSocketService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingWebSocketService.kt @@ -213,7 +213,7 @@ class CopyTradingWebSocketService( // 处理交易 scope.launch { try { - copyOrderTrackingService.processTrade(leaderId, trade, "websocket") + copyOrderTrackingService.processTrade(leaderId, trade, "activity-ws") } catch (e: Exception) { logger.error("处理交易失败: leaderId=$leaderId, tradeId=${trade.id}", e) } 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 83590f1..5f300bc 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 @@ -140,7 +140,7 @@ open class CopyOrderTrackingService( // 2. 处理交易逻辑 val result = when (trade.side.uppercase()) { - "BUY" -> processBuyTrade(leaderId, trade) + "BUY" -> processBuyTrade(leaderId, trade, source) "SELL" -> processSellTrade(leaderId, trade) else -> { logger.warn("未知的交易方向: ${trade.side}") @@ -213,7 +213,7 @@ open class CopyOrderTrackingService( * 创建跟单买入订单并记录到跟踪表 */ @Transactional - suspend fun processBuyTrade(leaderId: Long, trade: TradeResponse): Result { + suspend fun processBuyTrade(leaderId: Long, trade: TradeResponse, source: String): Result { return try { // 1. 查找所有启用且支持该Leader的跟单关系 val copyTradings = copyTradingRepository.findByLeaderIdAndEnabledTrue(leaderId) @@ -622,7 +622,8 @@ open class CopyOrderTrackingService( price = buyPrice, // 使用下单价格，临时值 remainingQuantity = finalBuyQuantity, status = "filled", - notificationSent = false // 标记为未发送通知，等待轮询任务获取实际数据后发送 + notificationSent = false, // 标记为未发送通知，等待轮询任务获取实际数据后发送 + source = source // 订单来源 ) copyOrderTrackingRepository.save(tracking) 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 d55f43d..bdca416 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 @@ -38,25 +38,25 @@ class OrderStatusUpdateService( private val marketService: MarketService, // 市场信息服务 private val telegramNotificationService: TelegramNotificationService? ) { - + private val logger = LoggerFactory.getLogger(OrderStatusUpdateService::class.java) - + private val updateScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) - + // 缓存首次检测到订单详情为 null 的时间戳（订单ID -> 首次检测时间） private val orderNullDetectionTime = ConcurrentHashMap() - + // 订单详情为 null 的重试时间窗口（1分钟） private val ORDER_NULL_RETRY_WINDOW_MS = 60000L - + // 订单详情为 null 但已部分卖出的清理时间窗口（1小时） private val PARTIAL_SOLD_CLEANUP_WINDOW_MS = 3600000L - + @EventListener(ApplicationReadyEvent::class) fun onApplicationReady() { logger.info("订单状态更新服务已启动，将每5秒轮询一次") } - + /** * 定时更新订单状态 * 每5秒执行一次 @@ -67,13 +67,13 @@ class OrderStatusUpdateService( try { // 1. 清理已删除账户的订单 cleanupDeletedAccountOrders() - + // 2. 检查30秒前创建的订单，如果未成交则删除 checkAndDeleteUnfilledOrders() - + // 3. 更新卖出订单的实际成交价并发送通知（priceUpdated 共用字段） updatePendingSellOrderPrices() - + // 4. 更新买入订单的实际数据并发送通知 updatePendingBuyOrders() } catch (e: Exception) { @@ -81,11 +81,11 @@ class OrderStatusUpdateService( } } } - + /** * 验证订单ID格式 * 订单ID必须以 0x 开头，且是有效的 16 进制字符串 - * + * * @param orderId 订单ID * @return 如果格式有效返回 true，否则返回 false */ @@ -101,7 +101,7 @@ class OrderStatusUpdateService( // 检查是否只包含 0-9, a-f, A-F return hexPart.all { it in '0'..'9' || it in 'a'..'f' || it in 'A'..'F' } } - + /** * 清理已删除账户的订单 */ @@ -110,41 +110,41 @@ class OrderStatusUpdateService( try { // 查询所有卖出记录 val allRecords = sellMatchRecordRepository.findAll() - + // 查询所有有效的账户ID val validAccountIds = accountRepository.findAll().mapNotNull { it.id }.toSet() - + // 查询所有有效的跟单关系 val validCopyTradingIds = copyTradingRepository.findAll() .filter { it.accountId in validAccountIds } .mapNotNull { it.id } .toSet() - + // 找出需要删除的记录（关联的跟单关系已不存在或账户已删除） val recordsToDelete = allRecords.filter { record -> val copyTrading = copyTradingRepository.findById(record.copyTradingId).orElse(null) copyTrading == null || copyTrading.accountId !in validAccountIds } - + if (recordsToDelete.isNotEmpty()) { logger.info("清理已删除账户的订单: ${recordsToDelete.size} 条记录") - + // 删除匹配明细 for (record in recordsToDelete) { val details = sellMatchDetailRepository.findByMatchRecordId(record.id!!) sellMatchDetailRepository.deleteAll(details) } - + // 删除卖出记录 sellMatchRecordRepository.deleteAll(recordsToDelete) - + logger.info("已清理 ${recordsToDelete.size} 条已删除账户的订单记录") } } catch (e: Exception) { logger.error("清理已删除账户订单异常: ${e.message}", e) } } - + /** * 检查30秒前创建的订单，如果未成交则删除 * 首次检测但加入缓存中30s后还没有成交，则删除 @@ -154,24 +154,24 @@ class OrderStatusUpdateService( try { // 计算30秒前的时间戳 val thirtySecondsAgo = System.currentTimeMillis() - 30000 - + // 查询30秒前创建的订单,并过滤掉已经完全匹配的订单 // 已经完全匹配的订单(status = "fully_matched")不需要再检查 // 使用数据库查询过滤，避免加载过多数据 val ordersToCheck = copyOrderTrackingRepository.findByCreatedAtBeforeAndStatusNot( - thirtySecondsAgo, + thirtySecondsAgo, "fully_matched" ) - + if (ordersToCheck.isEmpty()) { return } - + logger.debug("检查 ${ordersToCheck.size} 个30秒前创建的订单是否成交") - + // 按账户分组，避免重复创建 API 客户端 val ordersByAccount = ordersToCheck.groupBy { it.accountId } - + for ((accountId, orders) in ordersByAccount) { try { // 获取账户 @@ -180,13 +180,13 @@ class OrderStatusUpdateService( logger.warn("账户不存在，跳过检查: accountId=$accountId") continue } - + // 检查账户是否配置了 API 凭证 if (account.apiKey == null || account.apiSecret == null || account.apiPassphrase == null) { logger.debug("账户未配置 API 凭证，跳过检查: accountId=${account.id}") continue } - + // 解密 API 凭证 val apiSecret = try { cryptoUtils.decrypt(account.apiSecret!!) @@ -194,14 +194,14 @@ class OrderStatusUpdateService( logger.warn("解密 API Secret 失败: accountId=${account.id}, error=${e.message}") continue } - + val apiPassphrase = try { cryptoUtils.decrypt(account.apiPassphrase!!) } catch (e: Exception) { logger.warn("解密 API Passphrase 失败: accountId=${account.id}, error=${e.message}") continue } - + // 创建带认证的 CLOB API 客户端 val clobApi = retrofitFactory.createClobApi( account.apiKey!!, @@ -209,13 +209,13 @@ class OrderStatusUpdateService( apiPassphrase, account.walletAddress ) - + // 检查每个订单 for (order in orders) { try { // 查询订单详情 val orderResponse = clobApi.getOrder(order.buyOrderId) - + // 先检查 HTTP 状态码，非 200 的都跳过 if (orderResponse.code() != 200) { // HTTP 非 200，记录日志并跳过，等待下次轮询 @@ -224,18 +224,20 @@ class OrderStatusUpdateService( logger.debug("订单查询失败（HTTP非200），等待下次轮询: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, code=${orderResponse.code()}, errorBody=$errorBody") continue } - + // HTTP 200，检查响应体 // 响应体也可能返回字符串 "null"，Gson 解析时会返回 null val orderDetail = orderResponse.body() if (orderDetail == null) { // HTTP 200 且响应体为 null（或字符串 "null"），可能是网络异常或 API 暂时不可用 // 使用兜底逻辑：首次检测不删除，1分钟后仍为 null 才删除 - val firstDetectionTime = orderNullDetectionTime.getOrPut(order.buyOrderId) { System.currentTimeMillis() } + val firstDetectionTime = + orderNullDetectionTime.getOrPut(order.buyOrderId) { System.currentTimeMillis() } val currentTime = System.currentTimeMillis() - + // 检查订单是否已部分卖出，如果已部分卖出则保留订单用于统计 - val hasMatchedDetails = sellMatchDetailRepository.findByTrackingId(order.id!!).isNotEmpty() + val hasMatchedDetails = + sellMatchDetailRepository.findByTrackingId(order.id!!).isNotEmpty() if (hasMatchedDetails || order.matchedQuantity > BigDecimal.ZERO) { // 检查是否超过清理时间窗口（1小时） val orderAge = currentTime - order.createdAt @@ -247,7 +249,10 @@ class OrderStatusUpdateService( // 清除缓存 orderNullDetectionTime.remove(order.buyOrderId) } catch (e: Exception) { - logger.error("删除本地订单失败: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, error=${e.message}", e) + logger.error( + "删除本地订单失败: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, error=${e.message}", + e + ) } continue } else { @@ -257,7 +262,7 @@ class OrderStatusUpdateService( continue } } - + // 检查是否超过重试时间窗口 if (currentTime - firstDetectionTime < ORDER_NULL_RETRY_WINDOW_MS) { // 未超过重试窗口，记录日志并等待下次轮询 @@ -265,7 +270,7 @@ class OrderStatusUpdateService( logger.debug("订单详情为 null（可能是网络异常），等待重试: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, 已等待=${elapsedSeconds}s, 重试窗口=${ORDER_NULL_RETRY_WINDOW_MS / 1000}s") continue } - + // 超过重试窗口，删除本地订单 logger.warn("订单详情为 null 超过重试窗口，删除本地订单: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, 已等待=$((currentTime - firstDetectionTime) / 1000}s") try { @@ -274,14 +279,17 @@ class OrderStatusUpdateService( // 清除缓存 orderNullDetectionTime.remove(order.buyOrderId) } catch (e: Exception) { - logger.error("删除本地订单失败: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, error=${e.message}", e) + logger.error( + "删除本地订单失败: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, error=${e.message}", + e + ) } continue } - + // 订单详情不为 null，清除缓存 orderNullDetectionTime.remove(order.buyOrderId) - + // 检查订单是否成交 // 如果订单状态不是 FILLED 且已成交数量为0，说明未成交，删除 val sizeMatched = orderDetail.sizeMatched?.toSafeBigDecimal() ?: BigDecimal.ZERO @@ -291,10 +299,11 @@ class OrderStatusUpdateService( copyOrderTrackingRepository.deleteById(order.id!!) logger.info("已删除未成交订单: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}") } catch (e: Exception) { - logger.error("删除未成交订单失败: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, error=${e.message}", e) + logger.error( + "删除未成交订单失败: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, error=${e.message}", + e + ) } - } else { - logger.debug("订单已成交或部分成交，保留: orderId=${order.buyOrderId}, status=${orderDetail.status}, sizeMatched=$sizeMatched") } } catch (e: Exception) { logger.error("检查订单失败: orderId=${order.buyOrderId}, error=${e.message}", e) @@ -308,7 +317,7 @@ class OrderStatusUpdateService( logger.error("检查未成交订单异常: ${e.message}", e) } } - + /** * 更新待更新的卖出订单价格 * 注意：priceUpdated 现在同时表示价格已更新和通知已发送（共用字段） @@ -318,13 +327,13 @@ class OrderStatusUpdateService( try { // 查询所有价格未更新的卖出记录（priceUpdated = false 表示未处理） val pendingRecords = sellMatchRecordRepository.findByPriceUpdatedFalse() - + if (pendingRecords.isEmpty()) { return } - + logger.debug("找到 ${pendingRecords.size} 条待更新价格的卖出订单") - + for (record in pendingRecords) { try { // 获取跟单关系 @@ -333,20 +342,20 @@ class OrderStatusUpdateService( logger.warn("跟单关系不存在，跳过更新: copyTradingId=${record.copyTradingId}") continue } - + // 获取账户 val account = accountRepository.findById(copyTrading.accountId).orElse(null) if (account == null) { logger.warn("账户不存在，跳过更新: accountId=${copyTrading.accountId}") continue } - + // 检查账户是否配置了 API 凭证 if (account.apiKey == null || account.apiSecret == null || account.apiPassphrase == null) { logger.debug("账户未配置 API 凭证，跳过更新: accountId=${account.id}") continue } - + // 解密 API 凭证 val apiSecret = try { cryptoUtils.decrypt(account.apiSecret!!) @@ -354,14 +363,14 @@ class OrderStatusUpdateService( logger.warn("解密 API Secret 失败: accountId=${account.id}, error=${e.message}") continue } - + val apiPassphrase = try { cryptoUtils.decrypt(account.apiPassphrase!!) } catch (e: Exception) { logger.warn("解密 API Passphrase 失败: accountId=${account.id}, error=${e.message}") continue } - + // 创建带认证的 CLOB API 客户端 val clobApi = retrofitFactory.createClobApi( account.apiKey!!, @@ -369,16 +378,16 @@ class OrderStatusUpdateService( apiPassphrase, account.walletAddress ) - + // 如果 orderId 不是 0x 开头，直接标记为已处理（priceUpdated = true 表示已处理，包括价格更新和通知发送） if (!record.sellOrderId.startsWith("0x", ignoreCase = true)) { logger.debug("卖出订单ID非0x开头，直接标记为已处理: orderId=${record.sellOrderId}") - + // 检查是否为自动生成的订单（AUTO_ 或 AUTO_FIFO_ 开头），如果是则不发送通知 - val isAutoOrder = record.sellOrderId.startsWith("AUTO_", ignoreCase = true) || - record.sellOrderId.startsWith("AUTO_FIFO_", ignoreCase = true) || - record.sellOrderId.startsWith("AUTO_WS_", ignoreCase = true) - + val isAutoOrder = record.sellOrderId.startsWith("AUTO_", ignoreCase = true) || + record.sellOrderId.startsWith("AUTO_FIFO_", ignoreCase = true) || + record.sellOrderId.startsWith("AUTO_WS_", ignoreCase = true) + if (!isAutoOrder) { // 非自动订单，发送通知（使用临时数据） sendSellOrderNotification( @@ -394,7 +403,7 @@ class OrderStatusUpdateService( } else { logger.debug("自动生成的订单，跳过发送通知: orderId=${record.sellOrderId}") } - + // 标记为已处理（priceUpdated = true 同时表示价格已更新和通知已发送） val updatedRecord = SellMatchRecord( id = record.id, @@ -413,12 +422,12 @@ class OrderStatusUpdateService( sellMatchRecordRepository.save(updatedRecord) continue } - + // 检查是否为自动生成的订单（AUTO_ 或 AUTO_FIFO_ 开头），如果是则跳过发送通知 - val isAutoOrder = record.sellOrderId.startsWith("AUTO_", ignoreCase = true) || - record.sellOrderId.startsWith("AUTO_FIFO_", ignoreCase = true) || - record.sellOrderId.startsWith("AUTO_WS_", ignoreCase = true) - + val isAutoOrder = record.sellOrderId.startsWith("AUTO_", ignoreCase = true) || + record.sellOrderId.startsWith("AUTO_FIFO_", ignoreCase = true) || + record.sellOrderId.startsWith("AUTO_WS_", ignoreCase = true) + if (isAutoOrder) { logger.debug("自动生成的订单，跳过发送通知并直接标记为已处理: orderId=${record.sellOrderId}") // 直接标记为已处理，不发送通知 @@ -439,23 +448,24 @@ class OrderStatusUpdateService( sellMatchRecordRepository.save(updatedRecord) continue } - + // 查询订单详情，获取实际成交价 val actualSellPrice = trackingService.getActualExecutionPrice( orderId = record.sellOrderId, clobApi = clobApi, fallbackPrice = record.sellPrice ) - + // 如果价格已更新（与当前价格不同），更新数据库 if (actualSellPrice != record.sellPrice) { // 重新计算盈亏 val details = sellMatchDetailRepository.findByMatchRecordId(record.id!!) var totalRealizedPnl = BigDecimal.ZERO - + for (detail in details) { - val updatedRealizedPnl = actualSellPrice.subtract(detail.buyPrice).multi(detail.matchedQuantity) - + val updatedRealizedPnl = + actualSellPrice.subtract(detail.buyPrice).multi(detail.matchedQuantity) + // 更新明细的卖出价格和盈亏 // 注意：SellMatchDetail 的字段都是 val，需要创建新对象 val updatedDetail = SellMatchDetail( @@ -470,10 +480,10 @@ class OrderStatusUpdateService( createdAt = detail.createdAt ) sellMatchDetailRepository.save(updatedDetail) - + totalRealizedPnl = totalRealizedPnl.add(updatedRealizedPnl) } - + // 先更新卖出记录，标记 priceUpdated = true（在发送通知之前更新） // 注意：SellMatchRecord 的字段都是 val，需要创建新对象 val updatedRecord = SellMatchRecord( @@ -491,9 +501,9 @@ class OrderStatusUpdateService( createdAt = record.createdAt ) sellMatchRecordRepository.save(updatedRecord) - + logger.info("更新卖出订单价格成功: orderId=${record.sellOrderId}, 原价格=${record.sellPrice}, 新价格=$actualSellPrice") - + // 发送通知（使用实际价格） sendSellOrderNotification( record = updatedRecord, @@ -524,9 +534,9 @@ class OrderStatusUpdateService( createdAt = record.createdAt ) sellMatchRecordRepository.save(updatedRecord) - + logger.debug("卖出订单价格无需更新: orderId=${record.sellOrderId}, price=$actualSellPrice") - + // 发送通知 sendSellOrderNotification( record = updatedRecord, @@ -550,7 +560,7 @@ class OrderStatusUpdateService( logger.error("更新待更新卖出订单价格异常: ${e.message}", e) } } - + /** * 更新待发送通知的买入订单 * 查询订单详情获取实际价格和数量，然后发送通知并更新数据库 @@ -560,13 +570,13 @@ class OrderStatusUpdateService( try { // 查询所有未发送通知的买入订单 val pendingOrders = copyOrderTrackingRepository.findByNotificationSentFalse() - + if (pendingOrders.isEmpty()) { return } - + logger.debug("找到 ${pendingOrders.size} 条待发送通知的买入订单") - + for (order in pendingOrders) { try { // 验证 orderId 格式（必须以 0x 开头的 16 进制） @@ -589,35 +599,40 @@ class OrderStatusUpdateService( remainingQuantity = order.remainingQuantity, status = order.status, notificationSent = true, // 标记为已发送通知 + source = order.source, // 保留原始订单来源 createdAt = order.createdAt, updatedAt = System.currentTimeMillis() ) copyOrderTrackingRepository.save(updatedOrder) - - sendBuyOrderNotification(updatedOrder, useTemporaryData = true, orderCreatedAt = order.createdAt) + + sendBuyOrderNotification( + updatedOrder, + useTemporaryData = true, + orderCreatedAt = order.createdAt + ) continue } - + // 获取跟单关系 val copyTrading = copyTradingRepository.findById(order.copyTradingId).orElse(null) if (copyTrading == null) { logger.warn("跟单关系不存在，跳过更新: copyTradingId=${order.copyTradingId}") continue } - + // 获取账户 val account = accountRepository.findById(order.accountId).orElse(null) if (account == null) { logger.warn("账户不存在，跳过更新: accountId=${order.accountId}") continue } - + // 检查账户是否配置了 API 凭证 if (account.apiKey == null || account.apiSecret == null || account.apiPassphrase == null) { logger.debug("账户未配置 API 凭证，跳过更新: accountId=${account.id}") continue } - + // 解密 API 凭证 val apiSecret = try { cryptoUtils.decrypt(account.apiSecret!!) @@ -625,14 +640,14 @@ class OrderStatusUpdateService( logger.warn("解密 API Secret 失败: accountId=${account.id}, error=${e.message}") continue } - + val apiPassphrase = try { cryptoUtils.decrypt(account.apiPassphrase!!) } catch (e: Exception) { logger.warn("解密 API Passphrase 失败: accountId=${account.id}, error=${e.message}") continue } - + // 创建带认证的 CLOB API 客户端 val clobApi = retrofitFactory.createClobApi( account.apiKey!!, @@ -640,26 +655,27 @@ class OrderStatusUpdateService( apiPassphrase, account.walletAddress ) - + // 查询订单详情 val orderResponse = clobApi.getOrder(order.buyOrderId) - + // 先检查 HTTP 状态码，非 200 的都跳过 if (orderResponse.code() != 200) { val errorBody = orderResponse.errorBody()?.string()?.take(200) ?: "无错误详情" logger.debug("查询订单详情失败（HTTP非200），等待下次轮询: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, code=${orderResponse.code()}, errorBody=$errorBody") continue } - + // HTTP 200，检查响应体 // 响应体也可能返回字符串 "null"，Gson 解析时会返回 null val orderDetail = orderResponse.body() if (orderDetail == null) { // HTTP 200 且响应体为 null（或字符串 "null"），可能是网络异常或 API 暂时不可用 // 使用兜底逻辑：首次检测不删除，1分钟后仍为 null 才删除 - val firstDetectionTime = orderNullDetectionTime.getOrPut(order.buyOrderId) { System.currentTimeMillis() } + val firstDetectionTime = + orderNullDetectionTime.getOrPut(order.buyOrderId) { System.currentTimeMillis() } val currentTime = System.currentTimeMillis() - + // 检查订单是否已部分卖出，如果已部分卖出则保留订单用于统计 val hasMatchedDetails = sellMatchDetailRepository.findByTrackingId(order.id!!).isNotEmpty() if (hasMatchedDetails || order.matchedQuantity > BigDecimal.ZERO) { @@ -673,7 +689,10 @@ class OrderStatusUpdateService( // 清除缓存 orderNullDetectionTime.remove(order.buyOrderId) } catch (e: Exception) { - logger.error("删除本地订单失败: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, error=${e.message}", e) + logger.error( + "删除本地订单失败: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, error=${e.message}", + e + ) } continue } else { @@ -683,7 +702,7 @@ class OrderStatusUpdateService( continue } } - + // 检查是否超过重试时间窗口 if (currentTime - firstDetectionTime < ORDER_NULL_RETRY_WINDOW_MS) { // 未超过重试窗口，记录日志并等待下次轮询 @@ -691,7 +710,7 @@ class OrderStatusUpdateService( logger.debug("订单详情为 null（可能是网络异常），等待重试: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, 已等待=${elapsedSeconds}s, 重试窗口=${ORDER_NULL_RETRY_WINDOW_MS / 1000}s") continue } - + // 超过重试窗口，删除本地订单 logger.warn("订单详情为 null 超过重试窗口，删除本地订单: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, 已等待=$((currentTime - firstDetectionTime) / 1000)s") try { @@ -700,66 +719,70 @@ class OrderStatusUpdateService( // 清除缓存 orderNullDetectionTime.remove(order.buyOrderId) } catch (e: Exception) { - logger.error("删除本地订单失败: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, error=${e.message}", e) + logger.error( + "删除本地订单失败: orderId=${order.buyOrderId}, copyOrderTrackingId=${order.id}, error=${e.message}", + e + ) } continue } - + // 订单详情不为 null，清除缓存 orderNullDetectionTime.remove(order.buyOrderId) - + // 获取实际价格和数量 val actualPrice = orderDetail.price?.toSafeBigDecimal() ?: order.price val actualSize = orderDetail.originalSize?.toSafeBigDecimal() ?: order.quantity val actualOutcome = orderDetail.outcome - + // 更新订单数据（如果实际数据与临时数据不同） val needUpdate = actualPrice != order.price || actualSize != order.quantity - - // 先保存更新后的订单，标记 notificationSent = true - // 这样可以防止其他并发任务重复发送通知 - val updatedOrder = CopyOrderTracking( - id = order.id, - copyTradingId = order.copyTradingId, - accountId = order.accountId, - leaderId = order.leaderId, - marketId = order.marketId, - side = order.side, - outcomeIndex = order.outcomeIndex, - buyOrderId = order.buyOrderId, - leaderBuyTradeId = order.leaderBuyTradeId, - quantity = actualSize, // 使用实际数量 - price = actualPrice, // 使用实际价格 - matchedQuantity = order.matchedQuantity, - remainingQuantity = order.remainingQuantity, - status = order.status, - notificationSent = true, // 标记为已发送通知 - createdAt = order.createdAt, - updatedAt = System.currentTimeMillis() - ) - - // 保存更新后的订单（在发送通知之前保存） - copyOrderTrackingRepository.save(updatedOrder) - - if (needUpdate) { - logger.info("更新买入订单数据成功: orderId=${order.buyOrderId}, 原价格=${order.price}, 新价格=$actualPrice, 原数量=${order.quantity}, 新数量=$actualSize") - } else { - logger.debug("买入订单数据无需更新: orderId=${order.buyOrderId}") - } - - // 发送通知（使用实际数据） - sendBuyOrderNotification( - order = updatedOrder, - actualPrice = actualPrice.toString(), - actualSize = actualSize.toString(), - actualOutcome = actualOutcome, - account = account, - copyTrading = copyTrading, - clobApi = clobApi, - apiSecret = apiSecret, - apiPassphrase = apiPassphrase, - orderCreatedAt = order.createdAt - ) + + // 先保存更新后的订单，标记 notificationSent = true + // 这样可以防止其他并发任务重复发送通知 + val updatedOrder = CopyOrderTracking( + id = order.id, + copyTradingId = order.copyTradingId, + accountId = order.accountId, + leaderId = order.leaderId, + marketId = order.marketId, + side = order.side, + outcomeIndex = order.outcomeIndex, + buyOrderId = order.buyOrderId, + leaderBuyTradeId = order.leaderBuyTradeId, + quantity = actualSize, // 使用实际数量 + price = actualPrice, // 使用实际价格 + matchedQuantity = order.matchedQuantity, + remainingQuantity = order.remainingQuantity, + status = order.status, + notificationSent = true, // 标记为已发送通知 + source = order.source, // 保留原始订单来源 + createdAt = order.createdAt, + updatedAt = System.currentTimeMillis() + ) + + // 保存更新后的订单（在发送通知之前保存） + copyOrderTrackingRepository.save(updatedOrder) + + if (needUpdate) { + logger.info("更新买入订单数据成功: orderId=${order.buyOrderId}, 原价格=${order.price}, 新价格=$actualPrice, 原数量=${order.quantity}, 新数量=$actualSize") + } else { + logger.debug("买入订单数据无需更新: orderId=${order.buyOrderId}") + } + + // 发送通知（使用实际数据） + sendBuyOrderNotification( + order = updatedOrder, + actualPrice = actualPrice.toString(), + actualSize = actualSize.toString(), + actualOutcome = actualOutcome, + account = account, + copyTrading = copyTrading, + clobApi = clobApi, + apiSecret = apiSecret, + apiPassphrase = apiPassphrase, + orderCreatedAt = order.createdAt + ) } catch (e: Exception) { logger.warn("更新买入订单失败: orderId=${order.buyOrderId}, error=${e.message}", e) // 继续处理下一条记录 @@ -769,7 +792,7 @@ class OrderStatusUpdateService( logger.error("更新待发送通知买入订单异常: ${e.message}", e) } } - + /** * 发送买入订单通知 */ @@ -789,7 +812,7 @@ class OrderStatusUpdateService( if (telegramNotificationService == null) { return } - + try { // 获取跟单关系和账户信息（如果未提供） val finalCopyTrading = copyTrading ?: copyTradingRepository.findById(order.copyTradingId).orElse(null) @@ -797,7 +820,7 @@ class OrderStatusUpdateService( logger.warn("跟单关系不存在，跳过发送通知: copyTradingId=${order.copyTradingId}") return } - + val finalAccount = account ?: accountRepository.findById(order.accountId).orElse(null) if (finalAccount == null) { logger.warn("账户不存在，跳过发送通知: accountId=${order.accountId}") @@ -807,31 +830,32 @@ class OrderStatusUpdateService( // 获取市场信息 val market = marketService.getMarket(order.marketId) val marketTitle = market?.title ?: order.marketId - + // 获取 Leader 和跟单配置信息 val leader = leaderRepository.findById(order.leaderId).orElse(null) val leaderName = leader?.leaderName val configName = finalCopyTrading.configName - + // 获取当前语言设置 val locale = try { LocaleContextHolder.getLocale() } catch (e: Exception) { java.util.Locale("zh", "CN") // 默认简体中文 } - + // 创建 CLOB API 客户端（如果未提供） - val finalClobApi = clobApi ?: if (finalAccount.apiKey != null && apiSecret != null && apiPassphrase != null) { - retrofitFactory.createClobApi( - finalAccount.apiKey!!, - apiSecret, - apiPassphrase, - finalAccount.walletAddress - ) - } else { - null - } - + val finalClobApi = + clobApi ?: if (finalAccount.apiKey != null && apiSecret != null && apiPassphrase != null) { + retrofitFactory.createClobApi( + finalAccount.apiKey!!, + apiSecret, + apiPassphrase, + finalAccount.walletAddress + ) + } else { + null + } + // 发送通知 telegramNotificationService.sendOrderSuccessNotification( orderId = order.buyOrderId, @@ -854,14 +878,14 @@ class OrderStatusUpdateService( configName = configName, orderTime = orderCreatedAt // 使用订单创建时间 ) - + logger.info("买入订单通知已发送: orderId=${order.buyOrderId}, copyTradingId=${order.copyTradingId}") } catch (e: Exception) { logger.warn("发送买入订单通知失败: orderId=${order.buyOrderId}, error=${e.message}", e) } } - - + + /** * 发送卖出订单通知 */ @@ -881,7 +905,7 @@ class OrderStatusUpdateService( if (telegramNotificationService == null) { return } - + try { // 获取跟单关系和账户信息（如果未提供） val finalCopyTrading = copyTrading ?: copyTradingRepository.findById(record.copyTradingId).orElse(null) @@ -889,7 +913,7 @@ class OrderStatusUpdateService( logger.warn("跟单关系不存在，跳过发送通知: copyTradingId=${record.copyTradingId}") return } - + val finalAccount = account ?: accountRepository.findById(finalCopyTrading.accountId).orElse(null) if (finalAccount == null) { logger.warn("账户不存在，跳过发送通知: accountId=${finalCopyTrading.accountId}") @@ -899,31 +923,32 @@ class OrderStatusUpdateService( // 获取市场信息 val market = marketService.getMarket(record.marketId) val marketTitle = market?.title ?: record.marketId - + // 获取 Leader 和跟单配置信息 val leader = leaderRepository.findById(finalCopyTrading.leaderId).orElse(null) val leaderName = leader?.leaderName val configName = finalCopyTrading.configName - + // 获取当前语言设置 val locale = try { LocaleContextHolder.getLocale() } catch (e: Exception) { java.util.Locale("zh", "CN") // 默认简体中文 } - + // 创建 CLOB API 客户端（如果未提供） - val finalClobApi = clobApi ?: if (finalAccount.apiKey != null && apiSecret != null && apiPassphrase != null) { - retrofitFactory.createClobApi( - finalAccount.apiKey!!, - apiSecret, - apiPassphrase, - finalAccount.walletAddress - ) - } else { - null - } - + val finalClobApi = + clobApi ?: if (finalAccount.apiKey != null && apiSecret != null && apiPassphrase != null) { + retrofitFactory.createClobApi( + finalAccount.apiKey!!, + apiSecret, + apiPassphrase, + finalAccount.walletAddress + ) + } else { + null + } + // 发送通知 telegramNotificationService.sendOrderSuccessNotification( orderId = record.sellOrderId, @@ -946,7 +971,7 @@ class OrderStatusUpdateService( configName = configName, orderTime = orderCreatedAt // 使用订单创建时间 ) - + logger.info("卖出订单通知已发送: orderId=${record.sellOrderId}, copyTradingId=${record.copyTradingId}") } catch (e: Exception) { logger.warn("发送卖出订单通知失败: orderId=${record.sellOrderId}, error=${e.message}", e) diff --git a/backend/src/main/resources/db/migration/V25__add_source_to_copy_order_tracking.sql b/backend/src/main/resources/db/migration/V25__add_source_to_copy_order_tracking.sql new file mode 100644 index 0000000..5a4dc8c --- /dev/null +++ b/backend/src/main/resources/db/migration/V25__add_source_to_copy_order_tracking.sql @@ -0,0 +1,12 @@ +-- ============================================ +-- V25: 添加订单来源字段到跟单订单跟踪表 +-- 用于记录订单是从哪个数据源接收到的（activity-ws 或 onchain-ws） +-- ============================================ + +-- 添加订单来源字段 +ALTER TABLE copy_order_tracking +ADD COLUMN source VARCHAR(20) NOT NULL DEFAULT 'unknown' COMMENT '订单来源：activity-ws（Polymarket WebSocket）、onchain-ws（OnChain WebSocket）'; + +-- 对于已有数据，设置为默认值 unknown（不影响现有功能） +-- 新创建的记录会在创建时自动填充此字段 +