Merge pull request #32 from WrBug/dev

Release v2.3.2: 尾盘策略多市场与币安按需订阅
2026-02-20 23:50:10 +08:00
parent 7c1f8df590 d3196a783f
commit a0c2d7995b
9 changed files with 337 additions and 146 deletions
@@ -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(),
@@ -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<String, Pair<BigDecimal, BigDecimal>>()

-    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<BigDecimal, BigDecimal>? {
+    fun computeAndCache(marketSlugPrefix: String, intervalSeconds: Int, periodStartUnix: Long): Pair<BigDecimal, BigDecimal>? {
+        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<BigDecimal>()
        val spreadsDown = mutableListOf<BigDecimal>()
        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<List<Any>>? {
+    private fun fetchKlines(symbol: String, interval: String, limit: Int, endTime: Long? = null): List<List<Any>>? {
        return try {
            val api = retrofitFactory.createBinanceApi()
            val call = api.getKlines(symbol = symbol, interval = interval, limit = limit, endTime = endTime)
@@ -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<String, Pair<BigDecimal, BigDecimal>>()
-    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<BigDecimal, BigDecimal>? {
-        return openCloseByPeriod[key(intervalSeconds, periodStartUnix)]
-    }
-
-    /** 供 API 健康检查使用：5m / 15m 连接是否正常 */
-    fun getConnectionStatuses(): Map<String, Boolean> = 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<String, WebSocket>()
+    /** 当前需要订阅的完整市场集合（如 btc-updown-5m、btc-updown-15m），由尾盘策略刷新时更新 */
+    private val requiredMarketPrefixes = AtomicReference<Set<String>>(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<String, String>? {
+        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<BigDecimal, BigDecimal>? {
+        return openCloseByPeriod[key(marketSlugPrefix, intervalSeconds, periodStartUnix)]
+    }
+
+    /** 供 API 健康检查使用：各币种各周期的连接状态 */
+    fun getConnectionStatuses(): Map<String, Boolean> {
+        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<String>) {
+        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<Long, BigDecimal, BigDecimal>? {
@@ -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()
    }
 }
@@ -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()
+    }
 }
@@ -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<Map<String, List<WsBookEntry>>>(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<Job>()
+
    /**
     * AUTO 模式：在周期开始（刷新订阅）时预拉历史 30 根 K 线并计算该周期价差，触发时直接用缓存。
     */
    private fun precomputeAutoSpreadForCurrentPeriods(newMap: Map<String, List<WsBookEntry>>) {
        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<GammaEventBySlugResponse> {
+    private suspend fun fetchEventBySlugWithRetry(slug: String, maxAttempts: Int = 3): Result<GammaEventBySlugResponse> {
        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<GammaEventBySlugResponse> {
+    private suspend fun fetchEventBySlug(slug: String): Result<GammaEventBySlugResponse> {
        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 {
@@ -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()
+    }
 }
@@ -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<String, Mutex>()

+    /** 过期锁 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<String, PeriodContext>()
@@ -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("尾盘策略执行服务已清理缓存和锁")
+    }
 }
@@ -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("、")
@@ -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')}
        </Button>
      </div>
-      {binanceUnhealthy.length > 0 && (
+      {binanceUnhealthy.length > 0 && list.some((s) => s.enabled) && (
        <Alert
          type="error"
          showIcon
@@ -960,9 +960,9 @@ const CryptoTailStrategyList: React.FC = () => {
                        align: 'center',
                        render: (i: number) =>
                          i === 0 ? (
-                            <Tag icon={<ArrowUpOutlined />} color="green">{t('cryptoTailStrategy.triggerRecords.up')}</Tag>
+                            <Tag color="green">{t('cryptoTailStrategy.triggerRecords.up')}</Tag>
                          ) : (
-                            <Tag icon={<ArrowDownOutlined />} color="volcano">{t('cryptoTailStrategy.triggerRecords.down')}</Tag>
+                            <Tag color="volcano">{t('cryptoTailStrategy.triggerRecords.down')}</Tag>
                          )
                      },
                      {
@@ -1035,9 +1035,9 @@ const CryptoTailStrategyList: React.FC = () => {
                        align: 'center',
                        render: (i: number) =>
                          i === 0 ? (
-                            <Tag icon={<ArrowUpOutlined />} color="green">{t('cryptoTailStrategy.triggerRecords.up')}</Tag>
+                            <Tag color="green">{t('cryptoTailStrategy.triggerRecords.up')}</Tag>
                          ) : (
-                            <Tag icon={<ArrowDownOutlined />} color="volcano">{t('cryptoTailStrategy.triggerRecords.down')}</Tag>
+                            <Tag color="volcano">{t('cryptoTailStrategy.triggerRecords.down')}</Tag>
                          )
                      },
                      {