diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/entity/CryptoTailStrategyTrigger.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/entity/CryptoTailStrategyTrigger.kt index 030e474..9444770 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/entity/CryptoTailStrategyTrigger.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/entity/CryptoTailStrategyTrigger.kt @@ -57,5 +57,8 @@ data class CryptoTailStrategyTrigger( val failReason: String? = null, @Column(name = "created_at", nullable = false) - val createdAt: Long = System.currentTimeMillis() + val createdAt: Long = System.currentTimeMillis(), + + @Column(name = "notification_sent", nullable = false) + var notificationSent: Boolean = false ) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CryptoTailStrategyTriggerRepository.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CryptoTailStrategyTriggerRepository.kt index 3423cd7..f8e1fd5 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CryptoTailStrategyTriggerRepository.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/repository/CryptoTailStrategyTriggerRepository.kt @@ -18,9 +18,12 @@ interface CryptoTailStrategyTriggerRepository : JpaRepository - /** 根据订单 ID 查询尾盘触发记录（用于 WS 推送时匹配并发送 TG 通知） */ + /** 根据订单 ID 查询尾盘触发记录 */ fun findByOrderId(orderId: String): CryptoTailStrategyTrigger? + /** 轮询发 TG：status=success、orderId 非空、未发过通知，按创建时间正序 */ + fun findByStatusAndOrderIdIsNotNullAndNotificationSentFalseOrderByCreatedAtAsc(status: String): List + /** 策略已结算订单的总已实现盈亏（用于收益统计） */ @Query("SELECT COALESCE(SUM(t.realizedPnl), 0) FROM CryptoTailStrategyTrigger t WHERE t.strategyId = :strategyId AND t.resolved = true") fun sumRealizedPnlByStrategyId(@Param("strategyId") strategyId: Long): BigDecimal? diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationPollingService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationPollingService.kt new file mode 100644 index 0000000..44e8594 --- /dev/null +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationPollingService.kt @@ -0,0 +1,140 @@ +package com.wrbug.polymarketbot.service.cryptotail + +import com.wrbug.polymarketbot.entity.CryptoTailStrategyTrigger +import com.wrbug.polymarketbot.repository.AccountRepository +import com.wrbug.polymarketbot.repository.CryptoTailStrategyRepository +import com.wrbug.polymarketbot.repository.CryptoTailStrategyTriggerRepository +import com.wrbug.polymarketbot.service.common.MarketService +import com.wrbug.polymarketbot.service.system.TelegramNotificationService +import com.wrbug.polymarketbot.util.CryptoUtils +import com.wrbug.polymarketbot.util.RetrofitFactory +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.launch +import org.slf4j.LoggerFactory +import org.springframework.context.ApplicationContext +import org.springframework.context.ApplicationContextAware +import org.springframework.scheduling.annotation.Scheduled +import org.springframework.stereotype.Service +import org.springframework.transaction.annotation.Transactional + +/** + * 尾盘策略订单 TG 通知轮询服务（与跟单一致） + * 定时查询「下单成功且未发 TG」的触发记录，通过 CLOB getOrder 获取订单详情后发送 TG 并标记已发。 + */ +@Service +class CryptoTailOrderNotificationPollingService( + private val triggerRepository: CryptoTailStrategyTriggerRepository, + private val strategyRepository: CryptoTailStrategyRepository, + private val accountRepository: AccountRepository, + private val retrofitFactory: RetrofitFactory, + private val cryptoUtils: CryptoUtils, + private val marketService: MarketService, + private val telegramNotificationService: TelegramNotificationService +) : ApplicationContextAware { + + private val logger = LoggerFactory.getLogger(CryptoTailOrderNotificationPollingService::class.java) + private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + + private var applicationContext: ApplicationContext? = null + + override fun setApplicationContext(applicationContext: ApplicationContext) { + this.applicationContext = applicationContext + } + + private fun getSelf(): CryptoTailOrderNotificationPollingService { + return applicationContext?.getBean(CryptoTailOrderNotificationPollingService::class.java) + ?: throw IllegalStateException("ApplicationContext not initialized") + } + + @Volatile + private var notificationJob: Job? = null + + @Scheduled(fixedDelay = 5000) + fun scheduledSendPendingNotifications() { + if (notificationJob != null && notificationJob!!.isActive) { + logger.debug("上一轮尾盘 TG 通知任务仍在执行，跳过本次") + return + } + notificationJob = scope.launch { + try { + getSelf().sendPendingNotifications() + } catch (e: Exception) { + logger.error("尾盘 TG 通知轮询异常: ${e.message}", e) + } finally { + notificationJob = null + } + } + } + + @Transactional + suspend fun sendPendingNotifications() { + val pending = triggerRepository.findByStatusAndOrderIdIsNotNullAndNotificationSentFalseOrderByCreatedAtAsc("success") + if (pending.isEmpty()) return + for (trigger in pending) { + try { + if (sendNotificationForTrigger(trigger)) { + trigger.notificationSent = true + triggerRepository.save(trigger) + } + } catch (e: Exception) { + logger.warn("尾盘 TG 通知单条失败: triggerId=${trigger.id}, orderId=${trigger.orderId}, ${e.message}", e) + } + } + } + + private suspend fun sendNotificationForTrigger(trigger: CryptoTailStrategyTrigger): Boolean { + val strategy = strategyRepository.findById(trigger.strategyId).orElse(null) ?: return false + val account = accountRepository.findById(strategy.accountId).orElse(null) ?: return false + val orderId = trigger.orderId ?: return false + if (account.apiKey == null || account.apiSecret == null || account.apiPassphrase == null) { + logger.debug("账户未配置 API 凭证，跳过 TG: accountId=${account.id}") + return false + } + val apiSecret = try { + cryptoUtils.decrypt(account.apiSecret) ?: return false + } catch (e: Exception) { + logger.warn("解密 API Secret 失败: accountId=${account.id}", e) + return false + } + val apiPassphrase = try { + cryptoUtils.decrypt(account.apiPassphrase) ?: "" + } catch (e: Exception) { "" } + val clobApi = retrofitFactory.createClobApi( + account.apiKey!!, + apiSecret, + apiPassphrase, + account.walletAddress + ) + val orderResponse = clobApi.getOrder(orderId) + if (!orderResponse.isSuccessful) { + logger.debug("查询订单详情失败，等待下次轮询: orderId=$orderId, code=${orderResponse.code()}") + return false + } + val order = orderResponse.body() ?: run { + logger.debug("订单详情为空，等待下次轮询: orderId=$orderId") + return false + } + val market = marketService.getMarket(order.market) + val marketTitle = trigger.marketTitle?.takeIf { it.isNotBlank() } ?: market?.title ?: order.market + val orderTimeMs = if (order.createdAt < 1_000_000_000_000L) order.createdAt * 1000 else order.createdAt + telegramNotificationService.sendCryptoTailOrderSuccessNotification( + orderId = orderId, + marketTitle = marketTitle, + marketId = order.market, + marketSlug = market?.eventSlug ?: market?.slug, + side = order.side, + outcome = order.outcome, + price = order.price, + size = order.originalSize, + strategyName = strategy.name, + accountName = account.accountName, + walletAddress = account.walletAddress, + orderTime = orderTimeMs + ) + logger.info("尾盘订单 TG 通知已发送: orderId=$orderId, strategyId=${strategy.id}, triggerId=${trigger.id}") + return true + } +} diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationSubscriber.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationSubscriber.kt deleted file mode 100644 index cd99b4a..0000000 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationSubscriber.kt +++ /dev/null @@ -1,80 +0,0 @@ -package com.wrbug.polymarketbot.service.cryptotail - -import com.wrbug.polymarketbot.dto.OrderPushMessage -import com.wrbug.polymarketbot.repository.AccountRepository -import com.wrbug.polymarketbot.repository.CryptoTailStrategyRepository -import com.wrbug.polymarketbot.repository.CryptoTailStrategyTriggerRepository -import com.wrbug.polymarketbot.service.copytrading.orders.OrderPushService -import com.wrbug.polymarketbot.service.system.TelegramNotificationService -import jakarta.annotation.PostConstruct -import jakarta.annotation.PreDestroy -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.launch -import org.slf4j.LoggerFactory -import org.springframework.stereotype.Service - -/** - * 尾盘策略订单 TG 通知订阅者 - * 与跟单订单广播方式一致：通过 OrderPushService.subscribeAllEnabled 订阅订单推送， - * 收到广播后匹配是否为尾盘订单，若是则发送 TG 通知。 - */ -@Service -class CryptoTailOrderNotificationSubscriber( - private val orderPushService: OrderPushService, - private val triggerRepository: CryptoTailStrategyTriggerRepository, - private val strategyRepository: CryptoTailStrategyRepository, - private val accountRepository: AccountRepository, - private val telegramNotificationService: TelegramNotificationService -) { - - private val logger = LoggerFactory.getLogger(CryptoTailOrderNotificationSubscriber::class.java) - private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob()) - - private var orderPushCallback: ((OrderPushMessage) -> Unit)? = null - - @PostConstruct - fun subscribe() { - val callback: (OrderPushMessage) -> Unit = { message -> onOrderPush(message) } - orderPushCallback = callback - orderPushService.subscribeAllEnabled(callback) - logger.info("尾盘订单 TG 通知已订阅 OrderPushService 广播") - } - - @PreDestroy - fun unsubscribe() { - orderPushCallback?.let { orderPushService.unsubscribeAll(it) } - orderPushCallback = null - logger.info("尾盘订单 TG 通知已取消订阅") - } - - private fun onOrderPush(message: OrderPushMessage) { - val trigger = triggerRepository.findByOrderId(message.order.id) ?: return - val strategy = strategyRepository.findById(trigger.strategyId).orElse(null) ?: return - val account = accountRepository.findById(strategy.accountId).orElse(null) ?: return - val orderTimeMs = message.order.timestamp.toLongOrNull()?.let { ts -> - if (ts < 1_000_000_000_000L) ts * 1000 else ts - } - scope.launch { - try { - telegramNotificationService.sendCryptoTailOrderSuccessNotification( - orderId = message.order.id, - marketTitle = trigger.marketTitle ?: message.orderDetail?.marketName ?: "", - marketId = message.order.market, - marketSlug = message.orderDetail?.marketSlug, - side = message.order.side, - outcome = message.order.outcome, - price = message.order.price, - size = message.order.originalSize, - strategyName = strategy.name, - accountName = account.accountName, - walletAddress = account.walletAddress, - orderTime = orderTimeMs - ) - } catch (e: Exception) { - logger.warn("尾盘订单 TG 通知失败: orderId=${message.order.id}, ${e.message}", e) - } - } - } -} diff --git a/backend/src/main/resources/db/migration/V36__add_crypto_tail_trigger_notification_sent.sql b/backend/src/main/resources/db/migration/V36__add_crypto_tail_trigger_notification_sent.sql new file mode 100644 index 0000000..af5910c --- /dev/null +++ b/backend/src/main/resources/db/migration/V36__add_crypto_tail_trigger_notification_sent.sql @@ -0,0 +1,8 @@ +-- ============================================ +-- V36: 尾盘策略触发记录 - TG 通知已发标记（与跟单轮询发 TG 一致） +-- ============================================ + +ALTER TABLE crypto_tail_strategy_trigger + ADD COLUMN notification_sent TINYINT(1) NOT NULL DEFAULT 0 COMMENT '是否已发送 TG 通知: 0=未发送, 1=已发送'; + +CREATE INDEX idx_trigger_notification ON crypto_tail_strategy_trigger (status, notification_sent);