feat(cryptotail): 尾盘 TG 改为轮询实现并修复自调用
- 新增 notification_sent 字段与 V36 迁移,轮询未发 TG 的 trigger - 新增 CryptoTailOrderNotificationPollingService:每 5 秒轮询,CLOB getOrder 后发 TG,与跟单一致 - 通过 ApplicationContextAware + getSelf() 解决 @Transactional 自调用问题 - 删除 CryptoTailOrderNotificationSubscriber,移除 WS 推送方式 - Repository 新增 findByStatusAndOrderIdIsNotNullAndNotificationSentFalseOrderByCreatedAtAsc Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -57,5 +57,8 @@ data class CryptoTailStrategyTrigger(
|
|||||||
val failReason: String? = null,
|
val failReason: String? = null,
|
||||||
|
|
||||||
@Column(name = "created_at", nullable = false)
|
@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
|
||||||
)
|
)
|
||||||
|
|||||||
+4
-1
@@ -18,9 +18,12 @@ interface CryptoTailStrategyTriggerRepository : JpaRepository<CryptoTailStrategy
|
|||||||
/** 轮询结算:仅处理下单成功的订单(status=success 且 orderId 非空)、且未结算的触发记录 */
|
/** 轮询结算:仅处理下单成功的订单(status=success 且 orderId 非空)、且未结算的触发记录 */
|
||||||
fun findByStatusAndResolvedAndOrderIdIsNotNullOrderByCreatedAtAsc(status: String, resolved: Boolean): List<CryptoTailStrategyTrigger>
|
fun findByStatusAndResolvedAndOrderIdIsNotNullOrderByCreatedAtAsc(status: String, resolved: Boolean): List<CryptoTailStrategyTrigger>
|
||||||
|
|
||||||
/** 根据订单 ID 查询尾盘触发记录(用于 WS 推送时匹配并发送 TG 通知) */
|
/** 根据订单 ID 查询尾盘触发记录 */
|
||||||
fun findByOrderId(orderId: String): CryptoTailStrategyTrigger?
|
fun findByOrderId(orderId: String): CryptoTailStrategyTrigger?
|
||||||
|
|
||||||
|
/** 轮询发 TG:status=success、orderId 非空、未发过通知,按创建时间正序 */
|
||||||
|
fun findByStatusAndOrderIdIsNotNullAndNotificationSentFalseOrderByCreatedAtAsc(status: String): List<CryptoTailStrategyTrigger>
|
||||||
|
|
||||||
/** 策略已结算订单的总已实现盈亏(用于收益统计) */
|
/** 策略已结算订单的总已实现盈亏(用于收益统计) */
|
||||||
@Query("SELECT COALESCE(SUM(t.realizedPnl), 0) FROM CryptoTailStrategyTrigger t WHERE t.strategyId = :strategyId AND t.resolved = true")
|
@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?
|
fun sumRealizedPnlByStrategyId(@Param("strategyId") strategyId: Long): BigDecimal?
|
||||||
|
|||||||
+140
@@ -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
|
||||||
|
}
|
||||||
|
}
|
||||||
-80
@@ -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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
+8
@@ -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);
|
||||||
Reference in New Issue
Block a user