diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/cryptotail/CryptoTailStrategyController.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/cryptotail/CryptoTailStrategyController.kt index c8fb02c..25217b9 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/cryptotail/CryptoTailStrategyController.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/cryptotail/CryptoTailStrategyController.kt @@ -146,7 +146,13 @@ class CryptoTailStrategyController( return try { val options = listOf( CryptoTailMarketOptionDto(slug = "btc-updown-5m", title = "Bitcoin Up or Down - 5 minute", intervalSeconds = 300, periodStartUnix = 0L, endDate = null), - CryptoTailMarketOptionDto(slug = "btc-updown-15m", title = "Bitcoin Up or Down - 15 minute", intervalSeconds = 900, periodStartUnix = 0L, endDate = null) + CryptoTailMarketOptionDto(slug = "btc-updown-15m", title = "Bitcoin Up or Down - 15 minute", intervalSeconds = 900, periodStartUnix = 0L, endDate = null), + CryptoTailMarketOptionDto(slug = "eth-updown-5m", title = "Ethereum Up or Down - 5 minute", intervalSeconds = 300, periodStartUnix = 0L, endDate = null), + CryptoTailMarketOptionDto(slug = "eth-updown-15m", title = "Ethereum Up or Down - 15 minute", intervalSeconds = 900, periodStartUnix = 0L, endDate = null), + CryptoTailMarketOptionDto(slug = "sol-updown-5m", title = "Solana Up or Down - 5 minute", intervalSeconds = 300, periodStartUnix = 0L, endDate = null), + CryptoTailMarketOptionDto(slug = "sol-updown-15m", title = "Solana Up or Down - 15 minute", intervalSeconds = 900, periodStartUnix = 0L, endDate = null), + CryptoTailMarketOptionDto(slug = "xrp-updown-5m", title = "XRP Up or Down - 5 minute", intervalSeconds = 300, periodStartUnix = 0L, endDate = null), + CryptoTailMarketOptionDto(slug = "xrp-updown-15m", title = "XRP Up or Down - 15 minute", intervalSeconds = 900, periodStartUnix = 0L, endDate = null) ) ResponseEntity.ok(ApiResponse.success(options)) } catch (e: Exception) { @@ -168,7 +174,9 @@ class CryptoTailStrategyController( } val periodStartUnix = (request["periodStartUnix"] as? Number)?.toLong() ?: (System.currentTimeMillis() / 1000 / intervalSeconds) * intervalSeconds - val pair = binanceKlineAutoSpreadService.computeAndCache(intervalSeconds, periodStartUnix) + // 默认使用 BTC 市场（向后兼容） + val marketSlugPrefix = (request["marketSlugPrefix"] as? String) ?: "btc-updown" + val pair = binanceKlineAutoSpreadService.computeAndCache(marketSlugPrefix, intervalSeconds, periodStartUnix) ?: return ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_ERROR, "fetch_failed", messageSource)) val body = CryptoTailAutoMinSpreadResponse( minSpreadUp = pair.first.toPlainString(), diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineAutoSpreadService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineAutoSpreadService.kt index 23e6a2b..dd80681 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineAutoSpreadService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineAutoSpreadService.kt @@ -9,7 +9,7 @@ import java.math.RoundingMode import java.util.concurrent.ConcurrentHashMap /** - * 自动最小价差：按周期计算。每个周期首次需要时，拉取该周期前的 20 根已收盘 K 线，按方向筛选、IQR 剔除后求平均，缓存 100% 基准值 (interval, period)。 + * 自动最小价差：按周期计算。每个周期首次需要时，拉取该周期前的 20 根已收盘 K 线，按方向筛选、IQR 剔除后求平均，缓存 100% 基准值 (marketSlugPrefix, interval, period)。 * 触发时由调用方按窗口进度计算动态系数（100%→50%）后得到有效最小价差。不在保存策略时计算。 */ @Service @@ -19,29 +19,68 @@ class BinanceKlineAutoSpreadService( private val logger = LoggerFactory.getLogger(BinanceKlineAutoSpreadService::class.java) - private val symbol = "BTCUSDC" + /** 市场 slug 前缀 -> Binance 交易对映射 */ + private val marketToSymbol = mapOf( + "btc-updown" to "BTCUSDC", + "eth-updown" to "ETHUSDC", + "sol-updown" to "SOLUSDC", + "xrp-updown" to "XRPUSDC" + ) + private val historyLimit = 20 private val minSamplesAfterIqr = 3 - /** (intervalSeconds, periodStartUnix) -> (baseSpreadUp, baseSpreadDown)，100% 基准价差 */ + /** (marketSlugPrefix, intervalSeconds, periodStartUnix) -> (baseSpreadUp, baseSpreadDown)，100% 基准价差 */ private val cache = ConcurrentHashMap>() - private fun cacheKey(intervalSeconds: Int, periodStartUnix: Long): String = "$intervalSeconds-$periodStartUnix" + /** 缓存保留时间（秒），超过则清理，防止无界增长 */ + private val cacheExpireSeconds = 3600L + + /** 从市场 slug 前缀获取 Binance 交易对；支持完整 slug（如 eth-updown-5m）或前缀（如 eth-updown） */ + private fun getSymbol(marketSlugPrefix: String): String? { + val base = marketSlugPrefix.lowercase().removeSuffix("-15m").removeSuffix("-5m") + return marketToSymbol[base] + } + + private fun cacheKey(marketSlugPrefix: String, intervalSeconds: Int, periodStartUnix: Long): String { + return "$marketSlugPrefix-$intervalSeconds-$periodStartUnix" + } + + /** 清理已过期的价差缓存，避免内存泄漏 */ + private fun cleanExpiredCache() { + val nowSeconds = System.currentTimeMillis() / 1000 + val expireThreshold = nowSeconds - cacheExpireSeconds + val keysToRemove = cache.keys.filter { key -> + // key 格式: marketSlugPrefix-intervalSeconds-periodStartUnix + val parts = key.split('-') + if (parts.size >= 3) { + parts.last().toLongOrNull()?.let { it < expireThreshold } ?: false + } else { + false + } + } + keysToRemove.forEach { cache.remove(it) } + } /** 返回该周期、该方向的 100% 基准价差，供调用方按窗口进度应用动态系数。 */ - fun getAutoMinSpreadBase(intervalSeconds: Int, periodStartUnix: Long, outcomeIndex: Int): BigDecimal? { - val key = cacheKey(intervalSeconds, periodStartUnix) + fun getAutoMinSpreadBase(marketSlugPrefix: String, intervalSeconds: Int, periodStartUnix: Long, outcomeIndex: Int): BigDecimal? { + val key = cacheKey(marketSlugPrefix, intervalSeconds, periodStartUnix) val (up, down) = cache[key] ?: run { - computeAndCache(intervalSeconds, periodStartUnix) ?: return null + computeAndCache(marketSlugPrefix, intervalSeconds, periodStartUnix) ?: return null } return if (outcomeIndex == 0) up else down } /** 计算并缓存 100% 基准价差（IQR 平均，不乘系数）。预加载与触发时共用此缓存。 */ - fun computeAndCache(intervalSeconds: Int, periodStartUnix: Long): Pair? { + fun computeAndCache(marketSlugPrefix: String, intervalSeconds: Int, periodStartUnix: Long): Pair? { + cleanExpiredCache() + val symbol = getSymbol(marketSlugPrefix) ?: run { + logger.warn("不支持的市场 slug 前缀: $marketSlugPrefix") + return null + } val intervalStr = if (intervalSeconds == 300) "5m" else "15m" val endTimeMs = periodStartUnix * 1000L - val klines = fetchKlines(intervalStr, historyLimit, endTime = endTimeMs) ?: return null + val klines = fetchKlines(symbol, intervalStr, historyLimit, endTime = endTimeMs) ?: return null val spreadsUp = mutableListOf() val spreadsDown = mutableListOf() for (k in klines) { @@ -53,16 +92,16 @@ class BinanceKlineAutoSpreadService( } val baseUp = averageAfterIqr(spreadsUp).setScale(8, RoundingMode.HALF_UP) val baseDown = averageAfterIqr(spreadsDown).setScale(8, RoundingMode.HALF_UP) - cache[cacheKey(intervalSeconds, periodStartUnix)] = baseUp to baseDown + cache[cacheKey(marketSlugPrefix, intervalSeconds, periodStartUnix)] = baseUp to baseDown logger.info( - "尾盘自动价差已计算并缓存(100%基准): interval=${intervalSeconds}s periodStartUnix=$periodStartUnix | " + + "尾盘自动价差已计算并缓存(100%基准): market=$marketSlugPrefix symbol=$symbol interval=${intervalSeconds}s periodStartUnix=$periodStartUnix | " + "Up方向: 样本数=${spreadsUp.size}, baseSpreadUp=${baseUp.toPlainString()} | " + "Down方向: 样本数=${spreadsDown.size}, baseSpreadDown=${baseDown.toPlainString()}" ) return baseUp to baseDown } - private fun fetchKlines(interval: String, limit: Int, endTime: Long? = null): List>? { + private fun fetchKlines(symbol: String, interval: String, limit: Int, endTime: Long? = null): List>? { return try { val api = retrofitFactory.createBinanceApi() val call = api.getKlines(symbol = symbol, interval = interval, limit = limit, endTime = endTime) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineService.kt index a72c585..a469948 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineService.kt @@ -16,10 +16,11 @@ import org.springframework.stereotype.Service import java.math.BigDecimal import jakarta.annotation.PreDestroy import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicReference /** - * 币安 K 线 WebSocket：订阅 BTCUSDC 5m/15m，维护当前周期 (open, close)，供尾盘策略价差校验使用。 + * 币安 K 线 WebSocket：按需订阅尾盘策略使用的币种 5m/15m，维护当前周期 (open, close)，供价差校验使用。 + * 仅当存在启用策略且策略使用到某市场时才订阅对应币种，无策略时不建立连接。 */ @Service class BinanceKlineService { @@ -30,86 +31,125 @@ class BinanceKlineService { private val wsBase = "wss://stream.binance.com:9443" private val client = createClient().build() - /** (intervalSeconds, periodStartUnix) -> (open, close) */ + /** (marketSlugPrefix, intervalSeconds, periodStartUnix) -> (open, close) */ private val openCloseByPeriod = ConcurrentHashMap>() - private var ws5m: WebSocket? = null - private var ws15m: WebSocket? = null - private var reconnectJob: Job? = null - private val connected5m = AtomicBoolean(false) - private val connected15m = AtomicBoolean(false) - - init { - connectAll() - } - - private fun key(intervalSeconds: Int, periodStartUnix: Long): String = "$intervalSeconds-$periodStartUnix" - - fun getCurrentOpenClose(intervalSeconds: Int, periodStartUnix: Long): Pair? { - return openCloseByPeriod[key(intervalSeconds, periodStartUnix)] - } - - /** 供 API 健康检查使用：5m / 15m 连接是否正常 */ - fun getConnectionStatuses(): Map = mapOf( - "5m" to connected5m.get(), - "15m" to connected15m.get() + + /** 市场 slug 前缀（如 btc-updown）-> Binance 交易对映射 */ + private val marketToSymbol = mapOf( + "btc-updown" to "BTCUSDC", + "eth-updown" to "ETHUSDC", + "sol-updown" to "SOLUSDC", + "xrp-updown" to "XRPUSDC" ) + + /** 已连接的 WebSocket: wsKey (symbol-interval) -> WebSocket */ + private val connectedWebSockets = ConcurrentHashMap() + /** 当前需要订阅的完整市场集合（如 btc-updown-5m、btc-updown-15m），由尾盘策略刷新时更新 */ + private val requiredMarketPrefixes = AtomicReference>(emptySet()) + private val subscriptionLock = Any() + private var reconnectJob: Job? = null - private fun connectAll() { - if (ws5m != null && ws15m != null) return - connectStream("btcusdc@kline_5m") { intervalSec, tMs, openP, closeP -> - val periodSec = tMs / 1000 - openCloseByPeriod[key(intervalSec, periodSec)] = openP to closeP - }.also { ws5m = it } - connectStream("btcusdc@kline_15m") { intervalSec, tMs, openP, closeP -> - val periodSec = tMs / 1000 - openCloseByPeriod[key(intervalSec, periodSec)] = openP to closeP - }.also { ws15m = it } + /** 解析完整市场 slug（如 btc-updown-5m）为 (basePrefix, interval)，不支持则返回 null */ + private fun parseMarketSlug(full: String): Pair? { + val lower = full.lowercase() + return when { + lower.endsWith("-5m") -> Pair(lower.removeSuffix("-5m"), "5m") + lower.endsWith("-15m") -> Pair(lower.removeSuffix("-15m"), "15m") + else -> null + } + } + + /** 从市场 base 前缀（如 btc-updown）获取 Binance 交易对 */ + private fun getSymbol(basePrefix: String): String? = marketToSymbol[basePrefix] + + private fun key(marketSlugPrefix: String, intervalSeconds: Int, periodStartUnix: Long): String { + return "$marketSlugPrefix-$intervalSeconds-$periodStartUnix" + } + + fun getCurrentOpenClose(marketSlugPrefix: String, intervalSeconds: Int, periodStartUnix: Long): Pair? { + return openCloseByPeriod[key(marketSlugPrefix, intervalSeconds, periodStartUnix)] + } + + /** 供 API 健康检查使用：各币种各周期的连接状态 */ + fun getConnectionStatuses(): Map { + return connectedWebSockets.keys.associateWith { connectedWebSockets[it] != null } + } + + /** + * 按需更新订阅：仅订阅策略用到的 (币种, 周期)，例如只开 btc 5min 则只建 btc 5min K 线连接。 + * 由 CryptoTailOrderbookWsService 在刷新订阅时根据启用策略的 marketSlugPrefix 调用。 + * @param marketPrefixes 当前启用策略用到的完整市场集合，如 ["btc-updown-5m"] 或 ["btc-updown-5m", "eth-updown-15m"]；空集合时关闭所有连接 + */ + fun updateSubscriptions(marketPrefixes: Set) { + val normalized = marketPrefixes.map { it.lowercase() }.toSet() + val parsed = normalized.mapNotNull { full -> + parseMarketSlug(full)?.let { (base, interval) -> + getSymbol(base)?.let { symbol -> Triple(full, symbol, interval) } + } + }.toSet() + val wsKeysNeeded = parsed.map { (_, symbol, interval) -> "$symbol-$interval" }.toSet() + if (normalized == requiredMarketPrefixes.get()) return + requiredMarketPrefixes.set(normalized) + synchronized(subscriptionLock) { + connectedWebSockets.keys.toList().forEach { wsKey -> + if (wsKey !in wsKeysNeeded) { + connectedWebSockets.remove(wsKey)?.close(1000, "subscription_update") + logger.info("币安 K 线 WS 已关闭（无策略使用）: $wsKey") + } + } + parsed.forEach { (fullPrefix, symbol, interval) -> + connectStream(symbol, interval, fullPrefix) { marketPrefixParam, intervalSec, tMs, openP, closeP -> + val periodSec = tMs / 1000 + openCloseByPeriod[key(marketPrefixParam, intervalSec, periodSec)] = openP to closeP + } + } + } } private fun connectStream( - streamName: String, - onKline: (intervalSeconds: Int, openTimeMs: Long, open: BigDecimal, close: BigDecimal) -> Unit - ): WebSocket { + symbol: String, + interval: String, + marketPrefix: String, + onKline: (marketPrefix: String, intervalSeconds: Int, openTimeMs: Long, open: BigDecimal, close: BigDecimal) -> Unit + ) { + val streamName = "${symbol.lowercase()}@kline_$interval" + val wsKey = "$symbol-$interval" + if (connectedWebSockets[wsKey] != null) return + val url = "$wsBase/ws/$streamName" - val intervalSeconds = when { - streamName.contains("kline_5m") -> 300 - streamName.contains("kline_15m") -> 900 + val intervalSeconds = when (interval) { + "5m" -> 300 + "15m" -> 900 else -> 300 } val request = Request.Builder().url(url).build() - val connectedFlag = when { - streamName.contains("kline_5m") -> connected5m - streamName.contains("kline_15m") -> connected15m - else -> null - } val ws = client.newWebSocket(request, object : WebSocketListener() { override fun onOpen(webSocket: WebSocket, response: okhttp3.Response) { - connectedFlag?.set(true) + connectedWebSockets[wsKey] = webSocket + logger.info("币安 K 线 WS 已连接: $streamName") } override fun onMessage(webSocket: WebSocket, text: String) { parseKlineMessage(text, intervalSeconds)?.let { (tMs, o, c) -> - onKline(intervalSeconds, tMs, o, c) + onKline(marketPrefix, intervalSeconds, tMs, o, c) } } override fun onFailure(webSocket: WebSocket, t: Throwable, response: okhttp3.Response?) { - connectedFlag?.set(false) + connectedWebSockets.remove(wsKey) logger.warn("币安 K 线 WS 异常 $streamName: ${t.message}") scheduleReconnect() } override fun onClosing(webSocket: WebSocket, code: Int, reason: String) { - connectedFlag?.set(false) + connectedWebSockets.remove(wsKey) if (code != 1000) scheduleReconnect() } override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { - connectedFlag?.set(false) + connectedWebSockets.remove(wsKey) } }) - logger.info("币安 K 线 WS 已连接: $streamName") - return ws } private fun parseKlineMessage(text: String, intervalSeconds: Int): Triple? { @@ -132,23 +172,18 @@ class BinanceKlineService { reconnectJob = scope.launch { delay(3_000) reconnectJob = null - ws5m?.close(1000, "reconnect") - ws15m?.close(1000, "reconnect") - ws5m = null - ws15m = null - connected5m.set(false) - connected15m.set(false) + val current = requiredMarketPrefixes.get() + connectedWebSockets.values.forEach { it.close(1000, "reconnect") } + connectedWebSockets.clear() logger.info("币安 K 线 WS 尝试重连") - connectAll() + updateSubscriptions(current) } } @PreDestroy fun destroy() { reconnectJob?.cancel() - ws5m?.close(1000, "shutdown") - ws15m?.close(1000, "shutdown") - ws5m = null - ws15m = null + connectedWebSockets.values.forEach { it.close(1000, "shutdown") } + connectedWebSockets.clear() } } 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 index f0dfece..93d3d26 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationPollingService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationPollingService.kt @@ -19,6 +19,7 @@ import org.springframework.context.ApplicationContextAware import org.springframework.scheduling.annotation.Scheduled import org.springframework.stereotype.Service import org.springframework.transaction.annotation.Transactional +import jakarta.annotation.PreDestroy /** * 尾盘策略订单 TG 通知轮询服务（与跟单一致） @@ -36,7 +37,8 @@ class CryptoTailOrderNotificationPollingService( ) : ApplicationContextAware { private val logger = LoggerFactory.getLogger(CryptoTailOrderNotificationPollingService::class.java) - private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + private val scopeJob = SupervisorJob() + private val scope = CoroutineScope(Dispatchers.IO + scopeJob) private var applicationContext: ApplicationContext? = null @@ -143,4 +145,11 @@ class CryptoTailOrderNotificationPollingService( logger.info("尾盘订单 TG 通知已发送: orderId=$orderId, strategyId=${strategy.id}, triggerId=${trigger.id}") return true } + + @PreDestroy + fun destroy() { + notificationJob?.cancel() + notificationJob = null + scopeJob.cancel() + } } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderbookWsService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderbookWsService.kt index 8797df3..84db01c 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderbookWsService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderbookWsService.kt @@ -7,6 +7,7 @@ import com.wrbug.polymarketbot.enums.SpreadMode import com.wrbug.polymarketbot.event.CryptoTailStrategyChangedEvent import com.wrbug.polymarketbot.repository.CryptoTailStrategyRepository import com.wrbug.polymarketbot.service.binance.BinanceKlineAutoSpreadService +import com.wrbug.polymarketbot.service.binance.BinanceKlineService import com.wrbug.polymarketbot.util.RetrofitFactory import com.wrbug.polymarketbot.util.createClient import com.wrbug.polymarketbot.util.fromJson @@ -28,6 +29,7 @@ import org.slf4j.LoggerFactory import org.springframework.context.event.EventListener import org.springframework.stereotype.Service import jakarta.annotation.PostConstruct +import jakarta.annotation.PreDestroy import java.math.BigDecimal import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicReference @@ -40,12 +42,14 @@ class CryptoTailOrderbookWsService( private val strategyRepository: CryptoTailStrategyRepository, private val executionService: CryptoTailStrategyExecutionService, private val retrofitFactory: RetrofitFactory, - private val binanceKlineAutoSpreadService: BinanceKlineAutoSpreadService + private val binanceKlineAutoSpreadService: BinanceKlineAutoSpreadService, + private val binanceKlineService: BinanceKlineService ) { private val logger = LoggerFactory.getLogger(CryptoTailOrderbookWsService::class.java) - private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + private val scopeJob = SupervisorJob() + private val scope = CoroutineScope(Dispatchers.Default + scopeJob) /** tokenId -> list of (strategy, periodStartUnix, marketTitle, tokenIds, outcomeIndex) */ private val tokenToEntries = AtomicReference>>(emptyMap()) @@ -66,6 +70,12 @@ class CryptoTailOrderbookWsService( /** 保护 connect() 的互斥锁，避免多线程并发创建连接 */ private val connectLock = Any() + /** 保护 refreshAndSubscribe() 的互斥锁，避免多线程并发刷新订阅 */ + private val refreshLock = Any() + + /** 标记是否正在刷新订阅，避免重复调用 */ + private val isRefreshing = AtomicBoolean(false) + data class WsBookEntry( val strategy: CryptoTailStrategy, val periodStartUnix: Long, @@ -79,6 +89,26 @@ class CryptoTailOrderbookWsService( if (strategyRepository.findAllByEnabledTrue().isNotEmpty()) connect() } + @PreDestroy + fun destroy() { + periodEndCountdownJob?.cancel() + periodEndCountdownJob = null + reconnectJob?.cancel() + reconnectJob = null + synchronized(precomputeJobs) { + precomputeJobs.forEach { it.cancel() } + precomputeJobs.clear() + } + closedForNoStrategies.set(true) + try { + webSocket?.close(1000, "shutdown") + } catch (e: Exception) { + logger.debug("关闭尾盘策略 WebSocket 时异常: ${e.message}") + } + webSocket = null + scopeJob.cancel() + } + private fun connect() { synchronized(connectLock) { if (webSocket != null) return @@ -171,16 +201,14 @@ class CryptoTailOrderbookWsService( if (nowSeconds < windowStart || nowSeconds >= windowEnd) continue scope.launch { try { - runBlocking { - executionService.tryTriggerWithPriceFromWs( - strategy = e.strategy, - periodStartUnix = e.periodStartUnix, - marketTitle = e.marketTitle, - tokenIds = e.tokenIds, - outcomeIndex = e.outcomeIndex, - bestBid = bestBid - ) - } + executionService.tryTriggerWithPriceFromWs( + strategy = e.strategy, + periodStartUnix = e.periodStartUnix, + marketTitle = e.marketTitle, + tokenIds = e.tokenIds, + outcomeIndex = e.outcomeIndex, + bestBid = bestBid + ) } catch (ex: Exception) { logger.error("WS 触发下单异常: strategyId=${e.strategy.id}, ${ex.message}", ex) } @@ -213,42 +241,56 @@ class CryptoTailOrderbookWsService( } private fun refreshAndSubscribe(fromConnect: Boolean = false) { - periodEndCountdownJob?.cancel() - periodEndCountdownJob = null - val oldTokenIds = tokenToEntries.get().keys.toSet() - val (tokenIds, newMap) = buildSubscriptionMap() - tokenToEntries.set(newMap) - if (tokenIds.isEmpty()) { - closeWebSocketForNoStrategies() - return - } - if (!fromConnect) { - if (webSocket == null) { - connect() + synchronized(refreshLock) { + // 如果正在刷新，直接返回，避免重复调用 + if (isRefreshing.get()) { + logger.debug("尾盘策略订阅刷新已在进行中，跳过本次调用") return } - if (oldTokenIds == tokenIds.toSet()) { - scheduleRefreshAtPeriodEnd(newMap) - precomputeAutoSpreadForCurrentPeriods(newMap) - return - } - closeWebSocketAndReconnect() - return + isRefreshing.set(true) } - val marketSlugs = newMap.values.asSequence().flatten() - .distinctBy { "${it.strategy.marketSlugPrefix}-${it.periodStartUnix}" } - .map { "${it.strategy.marketSlugPrefix}-${it.periodStartUnix}" } - .toList() - val msg = """{"type":"MARKET","assets_ids":${tokenIds.toJson()}}""" try { - webSocket?.send(msg) - logger.info("尾盘策略订单簿订阅: ${tokenIds.size} 个 token, 市场: $marketSlugs") - } catch (e: Exception) { - logger.warn("发送订阅失败: ${e.message}") - return + val strategies = strategyRepository.findAllByEnabledTrue() + binanceKlineService.updateSubscriptions(strategies.map { it.marketSlugPrefix }.toSet()) + periodEndCountdownJob?.cancel() + periodEndCountdownJob = null + val oldTokenIds = tokenToEntries.get().keys.toSet() + val (tokenIds, newMap) = buildSubscriptionMap() + tokenToEntries.set(newMap) + if (tokenIds.isEmpty()) { + closeWebSocketForNoStrategies() + return + } + if (!fromConnect) { + if (webSocket == null) { + connect() + return + } + if (oldTokenIds == tokenIds.toSet()) { + scheduleRefreshAtPeriodEnd(newMap) + precomputeAutoSpreadForCurrentPeriods(newMap) + return + } + closeWebSocketAndReconnect() + return + } + val marketSlugs = newMap.values.asSequence().flatten() + .distinctBy { "${it.strategy.marketSlugPrefix}-${it.periodStartUnix}" } + .map { "${it.strategy.marketSlugPrefix}-${it.periodStartUnix}" } + .toList() + val msg = """{"type":"MARKET","assets_ids":${tokenIds.toJson()}}""" + try { + webSocket?.send(msg) + logger.info("尾盘策略订单簿订阅: ${tokenIds.size} 个 token, 市场: $marketSlugs") + } catch (e: Exception) { + logger.warn("发送订阅失败: ${e.message}") + return + } + scheduleRefreshAtPeriodEnd(newMap) + precomputeAutoSpreadForCurrentPeriods(newMap) + } finally { + isRefreshing.set(false) } - scheduleRefreshAtPeriodEnd(newMap) - precomputeAutoSpreadForCurrentPeriods(newMap) } /** @@ -267,31 +309,39 @@ class CryptoTailOrderbookWsService( } } + /** 跟踪预计算价差的协程 Job，用于在关闭时取消 */ + private val precomputeJobs = mutableSetOf() + /** * AUTO 模式：在周期开始（刷新订阅）时预拉历史 30 根 K 线并计算该周期价差，触发时直接用缓存。 */ private fun precomputeAutoSpreadForCurrentPeriods(newMap: Map>) { val autoPeriods = newMap.values.asSequence().flatten() .filter { it.strategy.spreadMode == SpreadMode.AUTO } - .distinctBy { "${it.strategy.intervalSeconds}-${it.periodStartUnix}" } - .map { it.strategy.intervalSeconds to it.periodStartUnix } + .distinctBy { "${it.strategy.marketSlugPrefix}-${it.strategy.intervalSeconds}-${it.periodStartUnix}" } + .map { Triple(it.strategy.marketSlugPrefix, it.strategy.intervalSeconds, it.periodStartUnix) } .toList() if (autoPeriods.isEmpty()) return - scope.launch { - for ((intervalSeconds, periodStartUnix) in autoPeriods) { + val job = scope.launch { + for ((marketPrefix, intervalSeconds, periodStartUnix) in autoPeriods) { try { - val pair = binanceKlineAutoSpreadService.computeAndCache(intervalSeconds, periodStartUnix) + val pair = binanceKlineAutoSpreadService.computeAndCache(marketPrefix, intervalSeconds, periodStartUnix) if (pair != null) { logger.info( - "周期开始初始价差: interval=${intervalSeconds}s periodStartUnix=$periodStartUnix " + + "周期开始初始价差: market=$marketPrefix interval=${intervalSeconds}s periodStartUnix=$periodStartUnix " + "baseSpreadUp=${pair.first.toPlainString()} baseSpreadDown=${pair.second.toPlainString()}" ) } } catch (e: Exception) { - logger.warn("周期开始预计算 AUTO 价差失败: interval=$intervalSeconds periodStartUnix=$periodStartUnix ${e.message}") + logger.warn("周期开始预计算 AUTO 价差失败: market=$marketPrefix interval=$intervalSeconds periodStartUnix=$periodStartUnix ${e.message}") } } } + synchronized(precomputeJobs) { + precomputeJobs.add(job) + // 清理已完成的 Job，避免集合无限增长 + precomputeJobs.removeIf { !it.isActive } + } } /** @@ -345,7 +395,7 @@ class CryptoTailOrderbookWsService( continue } val slug = "${strategy.marketSlugPrefix}-$periodStartUnix" - val event = fetchEventBySlugWithRetry(slug).getOrNull() + val event = runBlocking { fetchEventBySlugWithRetry(slug).getOrNull() } if (event == null) { logger.warn("尾盘策略跳过（拉取事件失败）: strategyId=${strategy.id}, slug=$slug，请确认 Gamma 是否存在该 slug 或稍后重试") continue @@ -372,21 +422,21 @@ class CryptoTailOrderbookWsService( } /** 拉取事件，失败时重试最多 2 次（间隔 1s），避免瞬时失败导致多策略只订阅到其中一个 */ - private fun fetchEventBySlugWithRetry(slug: String, maxAttempts: Int = 3): Result { + private suspend fun fetchEventBySlugWithRetry(slug: String, maxAttempts: Int = 3): Result { var lastFailure: Exception? = null repeat(maxAttempts) { attempt -> val result = fetchEventBySlug(slug) if (result.isSuccess) return result lastFailure = result.exceptionOrNull() as? Exception - if (attempt < maxAttempts - 1) runBlocking { delay(1000L) } + if (attempt < maxAttempts - 1) delay(1000L) } return Result.failure(lastFailure ?: Exception("fetchEventBySlug failed")) } - private fun fetchEventBySlug(slug: String): Result { + private suspend fun fetchEventBySlug(slug: String): Result { return try { val api = retrofitFactory.createGammaApi() - val response = runBlocking { api.getEventBySlug(slug) } + val response = api.getEventBySlug(slug) if (response.isSuccessful && response.body() != null) { Result.success(response.body()!!) } else { diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailSettlementService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailSettlementService.kt index 241daae..068ae91 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailSettlementService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailSettlementService.kt @@ -22,6 +22,7 @@ import org.slf4j.LoggerFactory import org.springframework.scheduling.annotation.Scheduled import org.springframework.stereotype.Service import org.springframework.transaction.annotation.Transactional +import jakarta.annotation.PreDestroy import java.math.BigDecimal import java.math.RoundingMode @@ -44,7 +45,8 @@ class CryptoTailSettlementService( private val triggerFixedPrice = BigDecimal("0.99") private val pnlScale = 8 - private val settlementScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + private val settlementScopeJob = SupervisorJob() + private val settlementScope = CoroutineScope(Dispatchers.IO + settlementScopeJob) /** 跟踪上一轮结算任务的 Job，防止并发执行（与 OrderStatusUpdateService 一致） */ @Volatile @@ -273,4 +275,11 @@ class CryptoTailSettlementService( amountUsdc.negate() } } + + @PreDestroy + fun destroy() { + settlementJob?.cancel() + settlementJob = null + settlementScopeJob.cancel() + } } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailStrategyExecutionService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailStrategyExecutionService.kt index 3537851..32c0543 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailStrategyExecutionService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailStrategyExecutionService.kt @@ -28,6 +28,7 @@ import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import org.slf4j.LoggerFactory import org.springframework.stereotype.Service +import jakarta.annotation.PreDestroy import java.math.BigDecimal import java.math.RoundingMode import java.util.concurrent.ConcurrentHashMap @@ -83,10 +84,25 @@ class CryptoTailStrategyExecutionService( /** 按 (strategyId, periodStartUnix) 加锁，避免同一周期被调度器与 WebSocket 等多路并发重复下单 */ private val triggerMutexMap = ConcurrentHashMap() + /** 过期锁 key 保留时间（秒），超过则清理，防止 map 无界增长 */ + private val triggerMutexExpireSeconds = 3600L + private fun triggerLockKey(strategyId: Long, periodStartUnix: Long): String = "$strategyId-$periodStartUnix" - private fun getTriggerMutex(strategyId: Long, periodStartUnix: Long): Mutex = - triggerMutexMap.getOrPut(triggerLockKey(strategyId, periodStartUnix)) { Mutex() } + private fun getTriggerMutex(strategyId: Long, periodStartUnix: Long): Mutex { + cleanExpiredTriggerMutexKeys() + return triggerMutexMap.getOrPut(triggerLockKey(strategyId, periodStartUnix)) { Mutex() } + } + + /** 清理已过期的 (strategyId, periodStartUnix) 锁，避免内存泄漏 */ + private fun cleanExpiredTriggerMutexKeys() { + val nowSeconds = System.currentTimeMillis() / 1000 + val expireThreshold = nowSeconds - triggerMutexExpireSeconds + val keysToRemove = triggerMutexMap.keys.filter { key -> + key.substringAfterLast('-').toLongOrNull()?.let { it < expireThreshold } ?: false + } + keysToRemove.forEach { triggerMutexMap.remove(it) } + } /** 周期预置上下文缓存：(strategyId-periodStartUnix) -> PeriodContext，过期周期在读取时剔除 */ private val periodContextCache = ConcurrentHashMap() @@ -165,11 +181,20 @@ class CryptoTailStrategyExecutionService( val ctx = periodContextCache[key] ?: return null if (periodStartUnix + strategy.intervalSeconds <= nowSeconds) { periodContextCache.remove(key) + cleanExpiredPeriodContextCache(nowSeconds) return null } return ctx } + /** 清理已过期的周期上下文缓存，避免内存泄漏 */ + private fun cleanExpiredPeriodContextCache(nowSeconds: Long) { + val keysToRemove = periodContextCache.entries + .filter { (_, ctx) -> ctx.periodStartUnix + ctx.strategy.intervalSeconds <= nowSeconds } + .map { it.key } + keysToRemove.forEach { periodContextCache.remove(it) } + } + /** * 由订单簿 WebSocket 触发：当收到某 token 的 bestBid 且满足区间时调用，若本周期未触发则下单。 */ @@ -190,7 +215,7 @@ class CryptoTailStrategyExecutionService( val logKey = triggerLockKey(strategy.id!!, periodStartUnix) if (conditionLoggedCache.getIfPresent(logKey) == null) { conditionLoggedCache.put(logKey, periodStartUnix + strategy.intervalSeconds) - val oc = binanceKlineService.getCurrentOpenClose(strategy.intervalSeconds, periodStartUnix) + val oc = binanceKlineService.getCurrentOpenClose(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix) val openPrice = oc?.first?.toPlainString() ?: "-" val closePrice = oc?.second?.toPlainString() ?: "-" val strategyName = strategy.name?.takeIf { it.isNotBlank() } ?: "尾盘策略-${strategy.marketSlugPrefix}" @@ -210,7 +235,7 @@ class CryptoTailStrategyExecutionService( private fun passSpreadCheck(strategy: CryptoTailStrategy, periodStartUnix: Long, outcomeIndex: Int): Boolean { if (strategy.spreadMode == SpreadMode.NONE) return true - val oc = binanceKlineService.getCurrentOpenClose(strategy.intervalSeconds, periodStartUnix) + val oc = binanceKlineService.getCurrentOpenClose(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix) ?: return false val (openP, closeP) = oc val spreadAbs = closeP.subtract(openP).abs() @@ -247,8 +272,8 @@ class CryptoTailStrategyExecutionService( ) private fun computeAutoEffectiveSpread(strategy: CryptoTailStrategy, periodStartUnix: Long, outcomeIndex: Int): AutoSpreadResult? { - val baseSpread = binanceKlineAutoSpreadService.getAutoMinSpreadBase(strategy.intervalSeconds, periodStartUnix, outcomeIndex) - ?: binanceKlineAutoSpreadService.computeAndCache(strategy.intervalSeconds, periodStartUnix)?.let { if (outcomeIndex == 0) it.first else it.second } + val baseSpread = binanceKlineAutoSpreadService.getAutoMinSpreadBase(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix, outcomeIndex) + ?: binanceKlineAutoSpreadService.computeAndCache(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix)?.let { if (outcomeIndex == 0) it.first else it.second } ?: return null if (baseSpread <= BigDecimal.ZERO) return null val windowStartMs = (periodStartUnix + strategy.windowStartSeconds) * 1000L @@ -495,4 +520,13 @@ class CryptoTailStrategyExecutionService( ) triggerRepository.save(record) } + + @PreDestroy + fun destroy() { + // 清理所有周期上下文缓存，避免敏感信息（明文私钥、API Secret）在内存中保留 + periodContextCache.clear() + // 清理所有锁，避免内存泄漏 + triggerMutexMap.clear() + logger.debug("尾盘策略执行服务已清理缓存和锁") + } } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/ApiHealthCheckService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/ApiHealthCheckService.kt index 748c9f5..426c8b2 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/ApiHealthCheckService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/system/ApiHealthCheckService.kt @@ -243,7 +243,14 @@ class ApiHealthCheckService( name = "币安 WebSocket", url = binanceWsUrl, status = "success", - message = "连接正常 (5m、15m)" + message = "连接正常 (按策略订阅)" + ) + } else if (total == 0) { + ApiHealthCheckDto( + name = "币安 WebSocket", + url = binanceWsUrl, + status = "success", + message = "无尾盘策略，未订阅" ) } else if (connected > 0) { val which = statuses.filter { it.value }.keys.joinToString("、") diff --git a/frontend/src/pages/CryptoTailStrategyList.tsx b/frontend/src/pages/CryptoTailStrategyList.tsx index 5d03d9a..bbd8f90 100644 --- a/frontend/src/pages/CryptoTailStrategyList.tsx +++ b/frontend/src/pages/CryptoTailStrategyList.tsx @@ -26,7 +26,7 @@ import { } from 'antd' import type { Dayjs } from 'dayjs' import dayjs from 'dayjs' -import { PlusOutlined, EditOutlined, UnorderedListOutlined, InfoCircleOutlined, WarningOutlined, CalendarOutlined, ArrowUpOutlined, ArrowDownOutlined, FileTextOutlined } from '@ant-design/icons' +import { PlusOutlined, EditOutlined, UnorderedListOutlined, InfoCircleOutlined, WarningOutlined, CalendarOutlined, FileTextOutlined } from '@ant-design/icons' import { useTranslation } from 'react-i18next' import { useMediaQuery } from 'react-responsive' import { apiService } from '../services/api' @@ -555,7 +555,7 @@ const CryptoTailStrategyList: React.FC = () => { {t('cryptoTailStrategy.list.configGuide')} - {binanceUnhealthy.length > 0 && ( + {binanceUnhealthy.length > 0 && list.some((s) => s.enabled) && ( { align: 'center', render: (i: number) => i === 0 ? ( - } color="green">{t('cryptoTailStrategy.triggerRecords.up')} + {t('cryptoTailStrategy.triggerRecords.up')} ) : ( - } color="volcano">{t('cryptoTailStrategy.triggerRecords.down')} + {t('cryptoTailStrategy.triggerRecords.down')} ) }, { @@ -1035,9 +1035,9 @@ const CryptoTailStrategyList: React.FC = () => { align: 'center', render: (i: number) => i === 0 ? ( - } color="green">{t('cryptoTailStrategy.triggerRecords.up')} + {t('cryptoTailStrategy.triggerRecords.up')} ) : ( - } color="volcano">{t('cryptoTailStrategy.triggerRecords.down')} + {t('cryptoTailStrategy.triggerRecords.down')} ) }, {