gavindiaz/PolyHermes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加跟单订单来源字段

Browse Source 
- 在 CopyOrderTracking 实体添加 source 字段，记录订单来源
- 支持 activity-ws（Polymarket WebSocket）和 onchain-ws（OnChain WebSocket）两种来源
- 更新 processBuyTrade 方法，接收并保存 source 参数
- 更新 OrderStatusUpdateService，保留原始 source 信息
- 添加数据库迁移脚本 V25
- 统一 CopyTradingWebSocketService 使用 activity-ws 作为来源
This commit is contained in:
WrBug
2026-01-15 04:33:55 +08:00
parent 0c7f34a28a
commit b1e69135b8
5 changed files with 230 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/main/kotlin/com/wrbug/polymarketbot/entity/CopyOrderTracking.kt
+4 -1
View File
@@ -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(),
backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingWebSocketService.kt
+1 -1
View File
@@ -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)
}
backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/CopyOrderTrackingService.kt
+4 -3
View File
@@ -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<Unit> {
suspend fun processBuyTrade(leaderId: Long, trade: TradeResponse, source: String): Result<Unit> {
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)
backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/OrderStatusUpdateService.kt
+209 -184
View File
@@ -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<String, Long>()
// 订单详情为 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)
backend/src/main/resources/db/migration/V25__add_source_to_copy_order_tracking.sql
+12
View File
@@ -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（不影响现有功能）
-- 新创建的记录会在创建时自动填充此字段