feat(cryptotail): 尾盘监控双连接与分时图优化

- 监控 WebSocket 拆分为当前周期连接与下一周期连接，周期切换时关闭过期连接并新建下一周期 - 下一周期市场未创建时也建立第二条空连接，保证始终两条连接 - refreshSubscription 增加 Mutex 防重入，避免周期结束时定时器与消息同时触发导致重复执行 - 修复 initMonitor/buildPushData 中 getCurrentOpenClose、spreadMode/spreadValue 等 API 与实体字段引用 - 移除重复的 buildSubscriptionMap、buildPushData 等方法，修复 StrategyPriceData.periodStartUnix - 前端分时图：市场价折线增加 connectNulls，新周期默认 0.5 价格展示 - 多语言与监控页入口、API 类型与 WebSocket 订阅集成 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-02-25 17:00:41 +08:00
parent d3196a783f
commit 84c79d8812
18 changed files with 2162 additions and 59 deletions
@@ -0,0 +1,23 @@
+package com.wrbug.polymarketbot.config
+
+import com.wrbug.polymarketbot.service.common.WebSocketSubscriptionService
+import com.wrbug.polymarketbot.service.cryptotail.CryptoTailMonitorService
+import jakarta.annotation.PostConstruct
+import org.springframework.context.annotation.Configuration
+
+/**
+ * 尾盘监控服务配置
+ * 处理 WebSocketSubscriptionService 和 CryptoTailMonitorService 之间的循环依赖
+ */
+@Configuration
+class MonitorServiceConfig(
+    private val webSocketSubscriptionService: WebSocketSubscriptionService,
+    private val cryptoTailMonitorService: CryptoTailMonitorService
+) {
+    
+    @PostConstruct
+    fun init() {
+        // 在所有 Bean 初始化后设置引用
+        webSocketSubscriptionService.setCryptoTailMonitorService(cryptoTailMonitorService)
+    }
+}
@@ -11,9 +11,12 @@ import com.wrbug.polymarketbot.dto.CryptoTailStrategyTriggerListResponse
 import com.wrbug.polymarketbot.dto.CryptoTailStrategyUpdateRequest
 import com.wrbug.polymarketbot.dto.CryptoTailMarketOptionDto
 import com.wrbug.polymarketbot.dto.CryptoTailAutoMinSpreadResponse
+import com.wrbug.polymarketbot.dto.CryptoTailMonitorInitRequest
+import com.wrbug.polymarketbot.dto.CryptoTailMonitorInitResponse
 import com.wrbug.polymarketbot.enums.ErrorCode
 import com.wrbug.polymarketbot.service.binance.BinanceKlineAutoSpreadService
 import com.wrbug.polymarketbot.service.cryptotail.CryptoTailStrategyService
+import com.wrbug.polymarketbot.service.cryptotail.CryptoTailMonitorService
 import org.slf4j.LoggerFactory
 import org.springframework.context.MessageSource
 import org.springframework.http.ResponseEntity
@@ -26,6 +29,7 @@ import org.springframework.web.bind.annotation.RestController
@RequestMapping("/api/crypto-tail-strategy")
 class CryptoTailStrategyController(
    private val cryptoTailStrategyService: CryptoTailStrategyService,
+    private val cryptoTailMonitorService: CryptoTailMonitorService,
    private val binanceKlineAutoSpreadService: BinanceKlineAutoSpreadService,
    private val messageSource: MessageSource
 ) {
@@ -173,7 +177,7 @@ class CryptoTailStrategyController(
                return ResponseEntity.ok(ApiResponse.error(ErrorCode.PARAM_ERROR, messageSource = messageSource))
            }
            val periodStartUnix = (request["periodStartUnix"] as? Number)?.toLong()
-                ?: (System.currentTimeMillis() / 1000 / intervalSeconds) * intervalSeconds
+                ?: ((System.currentTimeMillis() / 1000 / intervalSeconds) * intervalSeconds)
            // 默认使用 BTC 市场（向后兼容）
            val marketSlugPrefix = (request["marketSlugPrefix"] as? String) ?: "btc-updown"
            val pair = binanceKlineAutoSpreadService.computeAndCache(marketSlugPrefix, intervalSeconds, periodStartUnix)
@@ -188,4 +192,28 @@ class CryptoTailStrategyController(
            ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_ERROR, e.message, messageSource))
        }
    }
+
+    /**
+     * 初始化尾盘策略监控
+     * 返回策略信息、开盘价、tokenIds等初始化数据
+     */
+    @PostMapping("/monitor/init")
+    fun initMonitor(@RequestBody request: CryptoTailMonitorInitRequest): ResponseEntity<ApiResponse<CryptoTailMonitorInitResponse>> {
+        return try {
+            if (request.strategyId <= 0) {
+                return ResponseEntity.ok(ApiResponse.error(ErrorCode.CRYPTO_TAIL_STRATEGY_NOT_FOUND, messageSource = messageSource))
+            }
+            val result = cryptoTailMonitorService.initMonitor(request)
+            result.fold(
+                onSuccess = { ResponseEntity.ok(ApiResponse.success(it)) },
+                onFailure = { e ->
+                    logger.error("初始化尾盘监控失败: ${e.message}", e)
+                    ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_ERROR, e.message, messageSource))
+                }
+            )
+        } catch (e: Exception) {
+            logger.error("初始化尾盘监控异常: ${e.message}", e)
+            ResponseEntity.ok(ApiResponse.error(ErrorCode.SERVER_ERROR, e.message, messageSource))
+        }
+    }
 }
@@ -0,0 +1,103 @@
+package com.wrbug.polymarketbot.dto
+
+/**
+ * 尾盘策略监控初始化请求
+ */
+data class CryptoTailMonitorInitRequest(
+    /** 策略ID */
+    val strategyId: Long = 0L
+)
+
+/**
+ * 尾盘策略监控初始化响应
+ */
+data class CryptoTailMonitorInitResponse(
+    /** 策略ID */
+    val strategyId: Long = 0L,
+    /** 策略名称 */
+    val name: String = "",
+    /** 账户ID */
+    val accountId: Long = 0L,
+    /** 账户名称 */
+    val accountName: String = "",
+    /** 市场 slug 前缀 */
+    val marketSlugPrefix: String = "",
+    /** 市场标题 */
+    val marketTitle: String = "",
+    /** 周期秒数 (300=5m, 900=15m) */
+    val intervalSeconds: Int = 300,
+    /** 当前周期开始时间 (Unix 秒) */
+    val periodStartUnix: Long = 0L,
+    /** 时间窗口开始秒数 */
+    val windowStartSeconds: Int = 0,
+    /** 时间窗口结束秒数 */
+    val windowEndSeconds: Int = 0,
+    /** 最低价格 */
+    val minPrice: String = "0",
+    /** 最高价格 */
+    val maxPrice: String = "1",
+    /** 最小价差模式: NONE, FIXED, AUTO */
+    val minSpreadMode: String = "NONE",
+    /** 价差方向: MIN（显示周期内最小价差）, MAX（显示周期内最大价差） */
+    val spreadDirection: String = "MIN",
+    /** 最小价差数值 (FIXED 时有值) */
+    val minSpreadValue: String? = null,
+    /** 自动计算的最小价差 (Up方向) */
+    val autoMinSpreadUp: String? = null,
+    /** 自动计算的最小价差 (Down方向) */
+    val autoMinSpreadDown: String? = null,
+    /** BTC 开盘价 USDC（来自币安 K 线 open） */
+    val openPriceBtc: String? = null,
+    /** Up tokenId */
+    val tokenIdUp: String? = null,
+    /** Down tokenId */
+    val tokenIdDown: String? = null,
+    /** 当前时间 (毫秒时间戳) */
+    val currentTimestamp: Long = System.currentTimeMillis(),
+    /** 是否启用 */
+    val enabled: Boolean = true
+)
+
+/**
+ * 尾盘策略监控实时推送数据
+ */
+data class CryptoTailMonitorPushData(
+    /** 策略ID */
+    val strategyId: Long = 0L,
+    /** 推送时间 (毫秒时间戳) */
+    val timestamp: Long = System.currentTimeMillis(),
+    /** 当前周期开始时间 (Unix 秒) */
+    val periodStartUnix: Long = 0L,
+    /** 当前价格 (Up方向，来自订单簿) */
+    val currentPriceUp: String? = null,
+    /** 当前价格 (Down方向，来自订单簿) */
+    val currentPriceDown: String? = null,
+    /** 当前价差 (Up方向: 1 - currentPriceUp) */
+    val spreadUp: String? = null,
+    /** 当前价差 (Down方向: currentPriceUp) */
+    val spreadDown: String? = null,
+    /** 最小价差线 (Up方向) */
+    val minSpreadLineUp: String? = null,
+    /** 最小价差线 (Down方向，USDC 价差) */
+    val minSpreadLineDown: String? = null,
+    /** BTC 开盘价 USDC（币安 K 线 open） */
+    val openPriceBtc: String? = null,
+    /** BTC 最新价 USDC（币安 K 线 close，当前周期实时） */
+    val currentPriceBtc: String? = null,
+    /** BTC 价差 USDC（currentPriceBtc - openPriceBtc） */
+    val spreadBtc: String? = null,
+    /** 周期剩余秒数 */
+    val remainingSeconds: Int = 0,
+    /** 是否在时间窗口内 */
+    val inTimeWindow: Boolean = false,
+    /** 是否在价格区间内 (Up方向) */
+    val inPriceRangeUp: Boolean = false,
+    /** 是否在价格区间内 (Down方向) */
+    val inPriceRangeDown: Boolean = false,
+    /** 是否已触发 */
+    val triggered: Boolean = false,
+    /** 触发方向: UP, DOWN, null */
+    val triggerDirection: String? = null,
+    /** 周期是否已结束 */
+    val periodEnded: Boolean = false
+)
@@ -125,8 +125,8 @@ class AccountService(

            // 7. 加密敏感信息
            val encryptedPrivateKey = cryptoUtils.encrypt(request.privateKey)
-            val encryptedApiSecret = apiKeyCreds.secret?.let { cryptoUtils.encrypt(it) }
-            val encryptedApiPassphrase = apiKeyCreds.passphrase?.let { cryptoUtils.encrypt(it) }
+            val encryptedApiSecret = apiKeyCreds.secret.let { cryptoUtils.encrypt(it) }
+            val encryptedApiPassphrase = apiKeyCreds.passphrase.let { cryptoUtils.encrypt(it) }

            // 8. 生成账户名称（如果未提供，使用 SAFE/MAGIC-代理地址后4位）
            val accountName = if (request.accountName.isNullOrBlank()) {
@@ -518,8 +518,8 @@ class AccountService(
                    }
                    val creds = result.getOrNull()
                        ?: return Result.failure(IllegalStateException("API Key 返回为空"))
-                    val encryptedSecret = creds.secret?.let { cryptoUtils.encrypt(it) }
-                    val encryptedPassphrase = creds.passphrase?.let { cryptoUtils.encrypt(it) }
+                    val encryptedSecret = creds.secret.let { cryptoUtils.encrypt(it) }
+                    val encryptedPassphrase = creds.passphrase.let { cryptoUtils.encrypt(it) }
                    val updated = account.copy(
                        apiKey = creds.apiKey,
                        apiSecret = encryptedSecret,
@@ -1128,7 +1128,7 @@ class AccountService(

            // 3. 验证仓位是否存在并获取原始数量
            val positionsResult = getAllPositions()
-            val (position, originalQuantity) = positionsResult.fold(
+            val (_, originalQuantity) = positionsResult.fold(
                onSuccess = { positionListResponse ->
                    val position = positionListResponse.currentPositions.find {
                        it.accountId == request.accountId &&
@@ -1161,7 +1161,7 @@ class AccountService(
                onFailure = { e ->
                    return Result.failure(Exception("查询仓位失败: ${e.message}"))
                }
-            ) ?: return Result.failure(IllegalArgumentException("仓位不存在"))
+            )

            // 4. 计算实际卖出数量
            val sellQuantity = if (percentDecimal != null) {
@@ -1280,7 +1280,7 @@ class AccountService(

            val newOrderRequest = com.wrbug.polymarketbot.api.NewOrderRequest(
                order = signedOrder,
-                owner = account.apiKey!!,  // API Key
+                owner = account.apiKey,  // API Key
                orderType = orderType,
                deferExec = false
            )
@@ -1300,7 +1300,7 @@ class AccountService(
            }

            val clobApi = retrofitFactory.createClobApi(
-                account.apiKey!!,
+                account.apiKey,
                apiSecret,
                apiPassphrase,
                account.walletAddress
@@ -29,7 +29,9 @@ class BinanceKlineService {
    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())

    private val wsBase = "wss://stream.binance.com:9443"
-    private val client = createClient().build()
+    private val client by lazy {
+        createClient().build()
+    }

    /** (marketSlugPrefix, intervalSeconds, periodStartUnix) -> (open, close) */
    private val openCloseByPeriod = ConcurrentHashMap<String, Pair<BigDecimal, BigDecimal>>()
@@ -82,6 +84,7 @@ class BinanceKlineService {
     */
    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) }
@@ -123,14 +126,14 @@ class BinanceKlineService {
            else -> 300
        }
        val request = Request.Builder().url(url).build()
-        val ws = client.newWebSocket(request, object : WebSocketListener() {
+        client.newWebSocket(request, object : WebSocketListener() {
            override fun onOpen(webSocket: WebSocket, response: okhttp3.Response) {
                connectedWebSockets[wsKey] = webSocket
                logger.info("币安 K 线 WS 已连接: $streamName")
            }

            override fun onMessage(webSocket: WebSocket, text: String) {
-                parseKlineMessage(text, intervalSeconds)?.let { (tMs, o, c) ->
+                parseKlineMessage(text)?.let { (tMs, o, c) ->
                    onKline(marketPrefix, intervalSeconds, tMs, o, c)
                }
            }
@@ -152,7 +155,7 @@ class BinanceKlineService {
        })
    }

-    private fun parseKlineMessage(text: String, intervalSeconds: Int): Triple<Long, BigDecimal, BigDecimal>? {
+    private fun parseKlineMessage(text: String): Triple<Long, BigDecimal, BigDecimal>? {
        return try {
            val json = com.google.gson.JsonParser.parseString(text).asJsonObject
            if (json.get("e")?.asString != "kline") return null
@@ -176,6 +179,8 @@ class BinanceKlineService {
            connectedWebSockets.values.forEach { it.close(1000, "reconnect") }
            connectedWebSockets.clear()
            logger.info("币安 K 线 WS 尝试重连")
+            // 清空 requiredMarketPrefixes，否则 updateSubscriptions(current) 内会因 normalized == requiredMarketPrefixes.get() 直接 return，不会重新 connectStream
+            requiredMarketPrefixes.set(emptySet())
            updateSubscriptions(current)
        }
    }
@@ -1,11 +1,13 @@
 package com.wrbug.polymarketbot.service.common

+import com.wrbug.polymarketbot.dto.CryptoTailMonitorPushData
 import com.wrbug.polymarketbot.dto.OrderPushMessage
 import com.wrbug.polymarketbot.dto.PositionPushMessage
 import com.wrbug.polymarketbot.dto.WebSocketMessage as WsMessage
 import com.wrbug.polymarketbot.dto.WebSocketMessageType
 import com.wrbug.polymarketbot.service.accounts.PositionPushService
 import com.wrbug.polymarketbot.service.copytrading.orders.OrderPushService
+import com.wrbug.polymarketbot.service.cryptotail.CryptoTailMonitorService
 import kotlinx.coroutines.*
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Service
@@ -38,12 +40,26 @@ class WebSocketSubscriptionService(
    // 存储 order 频道的订阅回调：sessionId -> callback（用于取消订阅）
    private val orderChannelCallbacks = ConcurrentHashMap<String, (OrderPushMessage) -> Unit>()
    
+    // 存储尾盘监控频道的订阅回调：sessionId -> (strategyId -> callback)
+    private val monitorChannelCallbacks = ConcurrentHashMap<String, MutableMap<Long, (CryptoTailMonitorPushData) -> Unit>>()
+    
+    // 尾盘监控服务（延迟注入，避免循环依赖）
+    private var cryptoTailMonitorService: CryptoTailMonitorService? = null
+    
+    /**
+     * 设置尾盘监控服务（由 Spring 在初始化后调用）
+     */
+    fun setCryptoTailMonitorService(service: CryptoTailMonitorService) {
+        cryptoTailMonitorService = service
+    }
+    
    /**
     * 注册会话
     */
    fun registerSession(sessionId: String, callback: (WsMessage) -> Unit) {
        sessionCallbacks[sessionId] = callback
        sessionSubscriptions[sessionId] = mutableSetOf()
+        monitorChannelCallbacks[sessionId] = mutableMapOf()
    }
    
    /**
@@ -60,6 +76,12 @@ class WebSocketSubscriptionService(
        // 清理 order 频道的回调
        orderChannelCallbacks.remove(sessionId)
        
+        // 清理尾盘监控频道的回调
+        val monitorCallbacks = monitorChannelCallbacks.remove(sessionId)
+        monitorCallbacks?.keys?.forEach { strategyId ->
+            cryptoTailMonitorService?.unsubscribe(sessionId, strategyId)
+        }
+        
        sessionCallbacks.remove(sessionId)
    }
    
@@ -83,8 +105,8 @@ class WebSocketSubscriptionService(
        sendSubscribeAck(sessionId, channel, true)
        
        // 根据频道类型启动推送服务
-        when (channel) {
-            "position" -> {
+        when {
+            channel == "position" -> {
                positionPushService.subscribe(sessionId) { message ->
                    pushData(sessionId, channel, message)
                }
@@ -97,7 +119,7 @@ class WebSocketSubscriptionService(
                    }
                }
            }
-            "order" -> {
+            channel == "order" -> {
                // 订单推送：自动订阅所有启用的账户
                val callback: (OrderPushMessage) -> Unit = { message ->
                    pushData(sessionId, channel, message)
@@ -105,6 +127,20 @@ class WebSocketSubscriptionService(
                orderChannelCallbacks[sessionId] = callback
                orderPushService.subscribeAllEnabled(callback)
            }
+            channel.startsWith("crypto_tail_monitor_") -> {
+                // 尾盘策略监控频道
+                val strategyId = channel.removePrefix("crypto_tail_monitor_").toLongOrNull()
+                if (strategyId != null && cryptoTailMonitorService != null) {
+                    val callback: (CryptoTailMonitorPushData) -> Unit = { message ->
+                        pushData(sessionId, channel, message)
+                    }
+                    monitorChannelCallbacks.getOrPut(sessionId) { mutableMapOf() }[strategyId] = callback
+                    cryptoTailMonitorService!!.subscribe(sessionId, strategyId, callback)
+                } else {
+                    logger.warn("无效的尾盘监控频道或服务未初始化: $channel")
+                    sendSubscribeAck(sessionId, channel, false, "无效的策略ID")
+                }
+            }
            else -> {
                logger.warn("未知的频道: $channel")
                sendSubscribeAck(sessionId, channel, false, "未知的频道")
@@ -122,15 +158,58 @@ class WebSocketSubscriptionService(
        channelSubscriptions[channel]?.remove(sessionId)
        
        // 取消推送服务的订阅（推送服务内部会处理是否停止轮询）
-        when (channel) {
-            "position" -> positionPushService.unsubscribe(sessionId)
-            "order" -> {
+        when {
+            channel == "position" -> positionPushService.unsubscribe(sessionId)
+            channel == "order" -> {
                // 取消订阅所有账户的订单推送
                val callback = orderChannelCallbacks.remove(sessionId)
                if (callback != null) {
                    orderPushService.unsubscribeAll(callback)
                }
            }
+            channel.startsWith("crypto_tail_monitor_") -> {
+                // 取消尾盘监控订阅
+                val strategyId = channel.removePrefix("crypto_tail_monitor_").toLongOrNull()
+                if (strategyId != null) {
+                    monitorChannelCallbacks[sessionId]?.remove(strategyId)
+                    cryptoTailMonitorService?.unsubscribe(sessionId, strategyId)
+                }
+            }
+        }
+    }
+    
+    /**
+     * 注册尾盘监控回调（由 CryptoTailMonitorService 调用）
+     */
+    fun registerMonitorCallback(sessionId: String, strategyId: Long, callback: (CryptoTailMonitorPushData) -> Unit) {
+        monitorChannelCallbacks.getOrPut(sessionId) { mutableMapOf() }[strategyId] = callback
+    }
+    
+    /**
+     * 注销尾盘监控回调（由 CryptoTailMonitorService 调用）
+     */
+    fun unregisterMonitorCallback(sessionId: String, strategyId: Long) {
+        monitorChannelCallbacks[sessionId]?.remove(strategyId)
+    }
+    
+    /**
+     * 推送尾盘监控数据（由 CryptoTailMonitorService 调用）
+     */
+    fun pushMonitorData(strategyId: Long, data: CryptoTailMonitorPushData) {
+        val channel = "crypto_tail_monitor_$strategyId"
+        val sessionIds = channelSubscriptions[channel] ?: return
+        
+        for (sessionId in sessionIds) {
+            val callback = sessionCallbacks[sessionId]
+            if (callback != null) {
+                val message = WsMessage(
+                    type = WebSocketMessageType.DATA.value,
+                    channel = channel,
+                    payload = data,
+                    timestamp = System.currentTimeMillis()
+                )
+                callback(message)
+            }
        }
    }
    
@@ -168,4 +247,3 @@ class WebSocketSubscriptionService(
        }
    }
 }
-
@@ -0,0 +1,781 @@
+package com.wrbug.polymarketbot.service.cryptotail
+
+import com.wrbug.polymarketbot.api.GammaEventBySlugResponse
+import com.wrbug.polymarketbot.constants.PolymarketConstants
+import com.wrbug.polymarketbot.dto.CryptoTailMonitorInitRequest
+import com.wrbug.polymarketbot.dto.CryptoTailMonitorInitResponse
+import com.wrbug.polymarketbot.dto.CryptoTailMonitorPushData
+import com.wrbug.polymarketbot.entity.CryptoTailStrategy
+import com.wrbug.polymarketbot.repository.AccountRepository
+import com.wrbug.polymarketbot.repository.CryptoTailStrategyRepository
+import com.wrbug.polymarketbot.service.binance.BinanceKlineAutoSpreadService
+import com.wrbug.polymarketbot.service.binance.BinanceKlineService
+import com.wrbug.polymarketbot.service.common.WebSocketSubscriptionService
+import com.wrbug.polymarketbot.util.RetrofitFactory
+import com.wrbug.polymarketbot.util.fromJson
+import com.wrbug.polymarketbot.util.toJson
+import com.wrbug.polymarketbot.util.toSafeBigDecimal
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.sync.Mutex
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.WebSocket
+import okhttp3.WebSocketListener
+import org.slf4j.LoggerFactory
+import org.springframework.context.event.EventListener
+import org.springframework.stereotype.Service
+import jakarta.annotation.PostConstruct
+import jakarta.annotation.PreDestroy
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import java.math.BigDecimal
+import java.math.RoundingMode
+import java.util.Collections
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicReference
+
+/**
+ * 尾盘策略监控服务
+ * 负责实时推送监控数据到前端
+ */
+@Service
+class CryptoTailMonitorService(
+    private val strategyRepository: CryptoTailStrategyRepository,
+    private val accountRepository: AccountRepository,
+    private val retrofitFactory: RetrofitFactory,
+    private val binanceKlineService: BinanceKlineService,
+    private val binanceKlineAutoSpreadService: BinanceKlineAutoSpreadService,
+    private val webSocketSubscriptionService: WebSocketSubscriptionService
+) {
+
+    private val logger = LoggerFactory.getLogger(CryptoTailMonitorService::class.java)
+    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+
+    /** 当前周期 token 映射 */
+    private val currentPeriodTokenToStrategy = AtomicReference<Map<String, List<MonitorEntry>>>(emptyMap())
+    /** 下一周期 token 映射 */
+    private val nextPeriodTokenToStrategy = AtomicReference<Map<String, List<MonitorEntry>>>(emptyMap())
+
+    /** strategyId -> 当前价格数据 */
+    private val strategyPriceData = ConcurrentHashMap<Long, StrategyPriceData>()
+
+    /** strategyId -> 订阅者数量 */
+    private val strategySubscribers = ConcurrentHashMap<Long, Int>()
+
+    private var currentPeriodWebSocket: WebSocket? = null
+    private var nextPeriodWebSocket: WebSocket? = null
+    private val wsUrl = PolymarketConstants.RTDS_WS_URL + "/ws/market"
+    private val client = OkHttpClient.Builder().build()
+
+    private val reconnectDelayMs = 3_000L
+    private var reconnectJob: Job? = null
+    private val closedForNoSubscribers = AtomicBoolean(false)
+    private val connectLock = Any()
+
+    /** 防止 refreshSubscription 并发执行（周期结束时定时器与消息可能同时触发） */
+    private val refreshSubscriptionMutex = Mutex()
+
+    /** 周期结束倒计时 Job */
+    private var periodEndCountdownJob: Job? = null
+
+    /** 定时推送 Job（每 1.5 秒推送一次，保证 BTC 价格和分时图持续更新） */
+    private var periodicPushJob: Job? = null
+    private val pushIntervalMs = 1_500L
+
+    /** 策略推送历史（用于中途进入时补全分时图，最多保留 300 条） */
+    private val strategyPushHistory = ConcurrentHashMap<Long, MutableList<CryptoTailMonitorPushData>>()
+    private val strategyHistoryPeriod = ConcurrentHashMap<Long, Long>()
+    private val maxHistorySize = 300
+
+    data class MonitorEntry(
+        val strategyId: Long,
+        val strategy: CryptoTailStrategy,
+        val periodStartUnix: Long,
+        val outcomeIndex: Int,
+        val tokenId: String,
+        /** 是否为下一个周期（用于预先订阅） */
+        val isNextPeriod: Boolean = false
+    )
+
+    data class StrategyPriceData(
+        val currentPriceUp: BigDecimal? = null,
+        val currentPriceDown: BigDecimal? = null,
+        /** BTC 开盘价 USDC（币安 K 线 open） */
+        val openPriceBtc: BigDecimal? = null,
+        val spreadUp: BigDecimal? = null,
+        val spreadDown: BigDecimal? = null,
+        val minSpreadLineUp: BigDecimal? = null,
+        val minSpreadLineDown: BigDecimal? = null,
+        val triggered: Boolean = false,
+        val triggerDirection: String? = null,
+        val lastUpdateTime: Long = System.currentTimeMillis(),
+        /** 当前周期开始时间（用于双连接周期切换） */
+        val periodStartUnix: Long? = null
+    )
+
+    @PostConstruct
+    fun init() {
+        // 服务启动时不主动连接，等待前端订阅
+    }
+
+    /**
+     * 初始化监控数据
+     */
+    fun initMonitor(request: CryptoTailMonitorInitRequest): Result<CryptoTailMonitorInitResponse> {
+        return try {
+            val strategy = strategyRepository.findById(request.strategyId).orElse(null)
+            if (strategy == null) {
+                return Result.failure(IllegalArgumentException("策略不存在"))
+            }
+
+            val account = accountRepository.findById(strategy.accountId).orElse(null)
+            val nowSeconds = System.currentTimeMillis() / 1000
+            val periodStartUnix = (nowSeconds / strategy.intervalSeconds) * strategy.intervalSeconds
+
+            // 获取市场信息
+            val slug = "${strategy.marketSlugPrefix}-$periodStartUnix"
+            val event = fetchEventBySlug(slug).getOrNull()
+            val market = event?.markets?.firstOrNull()
+            val tokenIds = parseClobTokenIds(market?.clobTokenIds)
+
+            // 获取开盘价（币安 K 线 open = BTC 价格 USDC）
+            val openClose = binanceKlineService.getCurrentOpenClose(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix)
+            val openPriceBtc = openClose?.first
+
+            // 获取自动计算的最小价差
+            var autoMinSpreadUp: BigDecimal? = null
+            var autoMinSpreadDown: BigDecimal? = null
+            if (strategy.spreadMode.name.uppercase() == "AUTO") {
+                val autoSpreads = binanceKlineAutoSpreadService.computeAndCache(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix)
+                autoMinSpreadUp = autoSpreads?.first
+                autoMinSpreadDown = autoSpreads?.second
+            }
+
+            // 保存价格数据到缓存
+            val priceData = StrategyPriceData(
+                openPriceBtc = openPriceBtc,
+                minSpreadLineUp = autoMinSpreadUp ?: strategy.spreadValue?.toSafeBigDecimal(),
+                minSpreadLineDown = autoMinSpreadDown ?: strategy.spreadValue?.toSafeBigDecimal(),
+                periodStartUnix = periodStartUnix
+            )
+            strategyPriceData[strategy.id!!] = priceData
+
+            val response = CryptoTailMonitorInitResponse(
+                strategyId = strategy.id!!,
+                name = strategy.name ?: "",
+                accountId = strategy.accountId,
+                accountName = account?.accountName ?: "",
+                marketSlugPrefix = strategy.marketSlugPrefix,
+                marketTitle = event?.title ?: strategy.marketSlugPrefix,
+                intervalSeconds = strategy.intervalSeconds,
+                periodStartUnix = periodStartUnix,
+                windowStartSeconds = strategy.windowStartSeconds,
+                windowEndSeconds = strategy.windowEndSeconds,
+                minPrice = strategy.minPrice.toPlainString(),
+                maxPrice = strategy.maxPrice.toPlainString(),
+                minSpreadMode = strategy.spreadMode.name,
+                minSpreadValue = strategy.spreadValue?.toPlainString(),
+                autoMinSpreadUp = autoMinSpreadUp?.toPlainString(),
+                autoMinSpreadDown = autoMinSpreadDown?.toPlainString(),
+                openPriceBtc = openPriceBtc?.setScale(2, RoundingMode.HALF_UP)?.toPlainString(),
+                tokenIdUp = tokenIds.getOrNull(0),
+                tokenIdDown = tokenIds.getOrNull(1),
+                currentTimestamp = System.currentTimeMillis(),
+                enabled = strategy.enabled
+            )
+
+            Result.success(response)
+        } catch (e: Exception) {
+            logger.error("初始化监控失败: ${e.message}", e)
+            Result.failure(e)
+        }
+    }
+
+    /**
+     * 订阅策略监控
+     */
+    fun subscribe(sessionId: String, strategyId: Long, callback: (CryptoTailMonitorPushData) -> Unit) {
+        // 增加订阅计数
+        val count = strategySubscribers.merge(strategyId, 1) { old, inc -> old + inc } ?: 1
+        logger.info("策略 $strategyId 订阅数: $count")
+
+        // 注册推送回调
+        webSocketSubscriptionService.registerMonitorCallback(sessionId, strategyId, callback)
+
+        // 如果是第一个订阅者，启动 WebSocket 和定时推送
+        if (count == 1) {
+            scope.launch {
+                refreshSubscription()
+            }
+            startPeriodicPush()
+        }
+
+        // 立即发送当前数据
+        scope.launch {
+            try {
+                sendCurrentData(sessionId, strategyId, callback)
+            } catch (e: Exception) {
+                logger.error("发送当前监控数据失败: $sessionId, ${e.message}")
+            }
+        }
+    }
+
+    /**
+     * 取消订阅策略监控
+     */
+    fun unsubscribe(sessionId: String, strategyId: Long) {
+        // 减少订阅计数
+        val count = strategySubscribers.merge(strategyId, -1) { old, dec -> (old - dec).coerceAtLeast(0) } ?: 0
+        logger.info("策略 $strategyId 订阅数: $count")
+
+        // 移除回调
+        webSocketSubscriptionService.unregisterMonitorCallback(sessionId, strategyId)
+
+        // 如果没有订阅者，关闭 WebSocket 和定时推送
+        if (count == 0) {
+            strategySubscribers.remove(strategyId)
+            scope.launch {
+                refreshSubscription()
+            }
+            stopPeriodicPush()
+        }
+    }
+
+    private fun startPeriodicPush() {
+        if (periodicPushJob?.isActive == true) return
+        periodicPushJob = scope.launch {
+            while (strategySubscribers.isNotEmpty() && strategySubscribers.values.any { (it ?: 0) > 0 }) {
+                delay(pushIntervalMs)
+                if (closedForNoSubscribers.get()) continue
+                val ids = strategySubscribers.filter { (it.value ?: 0) > 0 }.keys.toList()
+                for (strategyId in ids) {
+                    try {
+                        val strategy = strategyRepository.findById(strategyId).orElse(null) ?: continue
+                        val priceData = strategyPriceData[strategyId] ?: continue
+                        val pushData = buildPushData(strategy, priceData)
+                        addToHistoryAndPush(strategyId, pushData)
+                    } catch (e: Exception) {
+                        logger.debug("定时推送失败 strategyId=$strategyId: ${e.message}")
+                    }
+                }
+            }
+        }
+    }
+
+    private fun stopPeriodicPush() {
+        if (strategySubscribers.isEmpty() || strategySubscribers.values.all { (it ?: 0) <= 0 }) {
+            periodicPushJob?.cancel()
+            periodicPushJob = null
+        }
+    }
+
+    /**
+     * 发送当前数据（含历史补全，用于中途进入时填充分时图）
+     */
+    private suspend fun sendCurrentData(sessionId: String, strategyId: Long, callback: (CryptoTailMonitorPushData) -> Unit) {
+        val strategy = strategyRepository.findById(strategyId).orElse(null) ?: return
+        val priceData = strategyPriceData[strategyId] ?: StrategyPriceData()
+
+        val history = strategyPushHistory[strategyId]?.let { list ->
+            synchronized(list) { list.toList() }
+        } ?: emptyList()
+        for (item in history) {
+            callback(item)
+        }
+
+        val pushData = buildPushData(strategy, priceData)
+        callback(pushData)
+    }
+
+    /**
+     * 刷新订阅：双连接模式。当前周期连接 + 下一周期连接；周期切换时关闭过期连接，下一连接晋升为当前，并新建下一周期连接。
+     * 使用 Mutex 防止周期结束时 scheduleRefreshAtPeriodEnd 与 maybeRefreshSubscriptionIfPeriodChanged 同时触发导致重复执行。
+     */
+    private suspend fun refreshSubscription() {
+        if (!refreshSubscriptionMutex.tryLock()) {
+            logger.debug("refreshSubscription 正在执行，跳过本次并发调用")
+            return
+        }
+        try {
+            refreshSubscriptionInternal()
+        } finally {
+            refreshSubscriptionMutex.unlock()
+        }
+    }
+
+    private suspend fun refreshSubscriptionInternal() {
+        periodEndCountdownJob?.cancel()
+        periodEndCountdownJob = null
+
+        val subscribedStrategyIds = strategySubscribers.keys.filter { (strategySubscribers[it] ?: 0) > 0 }
+        if (subscribedStrategyIds.isEmpty()) {
+            closeAllWebSockets()
+            return
+        }
+
+        val strategies = strategyRepository.findAllById(subscribedStrategyIds).filter { it.enabled && it.id != null }
+        if (strategies.isEmpty()) {
+            closeAllWebSockets()
+            return
+        }
+
+        val nowSeconds = System.currentTimeMillis() / 1000
+        val isSwitch = currentPeriodWebSocket != null
+
+        if (isSwitch) {
+            // 周期切换：关闭当前周期连接，下一晋升为当前，新建下一周期连接
+            closeCurrentPeriodWebSocket()
+            currentPeriodWebSocket = nextPeriodWebSocket
+            nextPeriodWebSocket = null
+            val nextMap = nextPeriodTokenToStrategy.get()
+            currentPeriodTokenToStrategy.set(nextMap)
+            val nextPeriodByStrategy = nextMap.values.flatten().distinctBy { it.strategyId }.associate { it.strategyId to it.periodStartUnix }
+            logger.info("周期切换：下一周期连接晋升为当前")
+            for ((strategyId, periodStartUnix) in nextPeriodByStrategy) {
+                updateStrategyPriceDataForPeriod(listOf(strategyId), periodStartUnix, pushDefault = true)
+            }
+            val (newNextTokenIds, newNextMap) = buildSubscriptionMapForNextPeriod(subscribedStrategyIds)
+            nextPeriodTokenToStrategy.set(newNextMap)
+            if (newNextTokenIds.isNotEmpty()) {
+                connectNextPeriod(newNextTokenIds, newNextMap)
+            } else {
+                logger.info("下一周期市场尚未创建，仅建立空连接以便周期切换时复用")
+                connectNextPeriod(emptyList(), emptyMap())
+            }
+            scheduleRefreshAtPeriodEnd(if (newNextMap.isNotEmpty()) newNextMap else nextMap)
+        } else {
+            // 首次：建立当前周期连接 + 下一周期连接
+            val (currentTokenIds, currentMap) = buildSubscriptionMapForCurrentPeriod(subscribedStrategyIds)
+            currentPeriodTokenToStrategy.set(currentMap)
+            for (entry in currentMap.values.flatten().distinctBy { it.strategyId }) {
+                updateStrategyPriceDataForPeriod(listOf(entry.strategyId), entry.periodStartUnix, pushDefault = false)
+            }
+            if (currentTokenIds.isEmpty()) {
+                closeAllWebSockets()
+                return
+            }
+            connectCurrentPeriod(currentTokenIds, currentMap)
+            val (nextTokenIds, nextMap) = buildSubscriptionMapForNextPeriod(subscribedStrategyIds)
+            nextPeriodTokenToStrategy.set(nextMap)
+            if (nextTokenIds.isNotEmpty()) {
+                connectNextPeriod(nextTokenIds, nextMap)
+            } else {
+                logger.info("下一周期市场尚未创建，先建立空连接，周期切换时会重新订阅")
+                connectNextPeriod(emptyList(), emptyMap())
+            }
+            scheduleRefreshAtPeriodEnd(currentMap)
+        }
+    }
+
+    /** 构建当前周期订阅（每个策略按自己的 interval 算当前周期） */
+    private suspend fun buildSubscriptionMapForCurrentPeriod(strategyIds: List<Long>): Pair<List<String>, Map<String, List<MonitorEntry>>> {
+        val strategies = strategyRepository.findAllById(strategyIds)
+        val nowSeconds = System.currentTimeMillis() / 1000
+        val tokenIdSet = mutableSetOf<String>()
+        val map = mutableMapOf<String, MutableList<MonitorEntry>>()
+
+        for (strategy in strategies) {
+            if (!strategy.enabled || strategy.id == null) continue
+            val strategyPeriod = (nowSeconds / strategy.intervalSeconds) * strategy.intervalSeconds
+            val slug = "${strategy.marketSlugPrefix}-$strategyPeriod"
+            val event = fetchEventBySlug(slug).getOrNull() ?: continue
+            val market = event.markets?.firstOrNull() ?: continue
+            val tokenIds = parseClobTokenIds(market.clobTokenIds)
+            if (tokenIds.size < 2) continue
+            for (i in tokenIds.indices) {
+                tokenIdSet.add(tokenIds[i])
+                map.getOrPut(tokenIds[i]) { mutableListOf() }.add(
+                    MonitorEntry(strategy.id!!, strategy, strategyPeriod, i, tokenIds[i], false)
+                )
+            }
+        }
+        return Pair(tokenIdSet.toList(), map)
+    }
+
+    /** 构建下一周期订阅（每个策略按自己的 interval 算下一周期） */
+    private suspend fun buildSubscriptionMapForNextPeriod(strategyIds: List<Long>): Pair<List<String>, Map<String, List<MonitorEntry>>> {
+        val strategies = strategyRepository.findAllById(strategyIds)
+        val nowSeconds = System.currentTimeMillis() / 1000
+        val tokenIdSet = mutableSetOf<String>()
+        val map = mutableMapOf<String, MutableList<MonitorEntry>>()
+
+        for (strategy in strategies) {
+            if (!strategy.enabled || strategy.id == null) continue
+            val currentPeriod = (nowSeconds / strategy.intervalSeconds) * strategy.intervalSeconds
+            val nextPeriod = currentPeriod + strategy.intervalSeconds
+            val slug = "${strategy.marketSlugPrefix}-$nextPeriod"
+            val event = fetchEventBySlug(slug).getOrNull() ?: continue
+            val market = event.markets?.firstOrNull() ?: continue
+            val tokenIds = parseClobTokenIds(market.clobTokenIds)
+            if (tokenIds.size < 2) continue
+            for (i in tokenIds.indices) {
+                tokenIdSet.add(tokenIds[i])
+                map.getOrPut(tokenIds[i]) { mutableListOf() }.add(
+                    MonitorEntry(strategy.id!!, strategy, nextPeriod, i, tokenIds[i], true)
+                )
+            }
+        }
+        return Pair(tokenIdSet.toList(), map)
+    }
+
+    /** 更新策略价格数据为指定周期（开盘价、价差线等），可选是否推送默认 0.5 */
+    private suspend fun updateStrategyPriceDataForPeriod(strategyIds: List<Long>, periodStartUnix: Long, pushDefault: Boolean) {
+        val strategies = strategyRepository.findAllById(strategyIds)
+        for (strategy in strategies) {
+            if (strategy.id == null) continue
+            val openClose = binanceKlineService.getCurrentOpenClose(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix)
+            val openPriceBtc = openClose?.first
+            var minSpreadLineUp: BigDecimal? = null
+            var minSpreadLineDown: BigDecimal? = null
+            when (strategy.spreadMode.name.uppercase()) {
+                "FIXED" -> {
+                    minSpreadLineUp = strategy.spreadValue?.toSafeBigDecimal()
+                    minSpreadLineDown = strategy.spreadValue?.toSafeBigDecimal()
+                }
+                "AUTO" -> {
+                    val autoSpreads = binanceKlineAutoSpreadService.computeAndCache(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix)
+                    minSpreadLineUp = autoSpreads?.first
+                    minSpreadLineDown = autoSpreads?.second
+                }
+            }
+            val existingData = strategyPriceData[strategy.id] ?: StrategyPriceData()
+            val periodChanged = existingData.periodStartUnix != null && existingData.periodStartUnix != periodStartUnix
+            val newData = StrategyPriceData(
+                currentPriceUp = if (periodChanged && pushDefault) BigDecimal("0.5") else existingData.currentPriceUp,
+                currentPriceDown = if (periodChanged && pushDefault) BigDecimal("0.5") else existingData.currentPriceDown,
+                spreadUp = if (periodChanged && pushDefault) BigDecimal("0.5") else existingData.spreadUp,
+                spreadDown = if (periodChanged && pushDefault) BigDecimal("0.5") else existingData.spreadDown,
+                openPriceBtc = openPriceBtc,
+                minSpreadLineUp = minSpreadLineUp,
+                minSpreadLineDown = minSpreadLineDown,
+                periodStartUnix = periodStartUnix
+            )
+            strategyPriceData[strategy.id!!] = newData
+            if (periodChanged && pushDefault) {
+                val pushData = buildPushData(strategy, newData)
+                addToHistoryAndPush(strategy.id!!, pushData)
+            }
+        }
+    }
+
+    private fun connectCurrentPeriod(tokenIds: List<String>, map: Map<String, List<MonitorEntry>>) {
+        if (currentPeriodWebSocket != null) return
+        val request = Request.Builder().url(wsUrl).build()
+        currentPeriodWebSocket = client.newWebSocket(request, object : WebSocketListener() {
+            override fun onOpen(webSocket: WebSocket, response: okhttp3.Response) {
+                closedForNoSubscribers.set(false)
+                val msg = """{"type":"MARKET","assets_ids":${tokenIds.toJson()}}"""
+                try {
+                    webSocket.send(msg)
+                    logger.info("尾盘监控 WebSocket（当前周期）已连接并订阅: ${tokenIds.size} 个 token")
+                } catch (e: Exception) {
+                    logger.warn("发送当前周期订阅失败: ${e.message}")
+                }
+            }
+
+            override fun onMessage(webSocket: WebSocket, text: String) {
+                handleMessage(webSocket, text, isFromCurrentPeriod = true)
+            }
+
+            override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
+                if (this@CryptoTailMonitorService.currentPeriodWebSocket == webSocket) {
+                    this@CryptoTailMonitorService.currentPeriodWebSocket = null
+                    if (!closedForNoSubscribers.get()) scheduleReconnect()
+                }
+            }
+
+            override fun onFailure(webSocket: WebSocket, t: Throwable, response: okhttp3.Response?) {
+                if (this@CryptoTailMonitorService.currentPeriodWebSocket == webSocket) {
+                    this@CryptoTailMonitorService.currentPeriodWebSocket = null
+                    scheduleReconnect()
+                }
+            }
+        })
+    }
+
+    private fun connectNextPeriod(tokenIds: List<String>, map: Map<String, List<MonitorEntry>>) {
+        if (nextPeriodWebSocket != null) return
+        val request = Request.Builder().url(wsUrl).build()
+        nextPeriodWebSocket = client.newWebSocket(request, object : WebSocketListener() {
+            override fun onOpen(webSocket: WebSocket, response: okhttp3.Response) {
+                val msg = """{"type":"MARKET","assets_ids":${tokenIds.toJson()}}"""
+                try {
+                    webSocket.send(msg)
+                    if (tokenIds.isEmpty()) {
+                        logger.info("尾盘监控 WebSocket（下一周期）已连接，暂无 token 订阅，等待周期切换后更新")
+                    } else {
+                        logger.info("尾盘监控 WebSocket（下一周期）已连接并订阅: ${tokenIds.size} 个 token")
+                    }
+                } catch (e: Exception) {
+                    logger.warn("发送下一周期订阅失败: ${e.message}")
+                }
+            }
+
+            override fun onMessage(webSocket: WebSocket, text: String) {
+                handleMessage(webSocket, text, isFromCurrentPeriod = false)
+            }
+
+            override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
+                if (this@CryptoTailMonitorService.nextPeriodWebSocket == webSocket) {
+                    this@CryptoTailMonitorService.nextPeriodWebSocket = null
+                }
+            }
+
+            override fun onFailure(webSocket: WebSocket, t: Throwable, response: okhttp3.Response?) {
+                if (this@CryptoTailMonitorService.nextPeriodWebSocket == webSocket) {
+                    this@CryptoTailMonitorService.nextPeriodWebSocket = null
+                }
+            }
+        })
+    }
+
+    private fun closeCurrentPeriodWebSocket() {
+        currentPeriodWebSocket?.close(1000, "period_ended")
+        currentPeriodWebSocket = null
+        logger.info("尾盘监控 WebSocket（当前周期）已关闭")
+    }
+
+    private fun closeAllWebSockets() {
+        reconnectJob?.cancel()
+        reconnectJob = null
+        closedForNoSubscribers.set(true)
+        currentPeriodWebSocket?.close(1000, "no_subscribers")
+        currentPeriodWebSocket = null
+        nextPeriodWebSocket?.close(1000, "no_subscribers")
+        nextPeriodWebSocket = null
+        logger.info("尾盘监控 WebSocket 已全部关闭（无订阅者）")
+    }
+
+    private fun handleMessage(webSocket: WebSocket, text: String, isFromCurrentPeriod: Boolean) {
+        if (text == "pong" || text.isEmpty()) return
+        if (closedForNoSubscribers.get()) return
+        if (!isFromCurrentPeriod) return
+
+        maybeRefreshSubscriptionIfPeriodChanged()
+
+        val json = text.fromJson<com.google.gson.JsonObject>() ?: return
+        val eventType = (json.get("event_type") as? com.google.gson.JsonPrimitive)?.asString ?: return
+        val map = currentPeriodTokenToStrategy.get()
+
+        when (eventType) {
+            "book" -> {
+                val assetId = (json.get("asset_id") as? com.google.gson.JsonPrimitive)?.asString ?: return
+                val bids = json.get("bids") as? com.google.gson.JsonArray
+                if (bids == null || bids.isEmpty) return
+                val firstBid = bids.get(0) as? com.google.gson.JsonObject
+                val bestBid = (firstBid?.get("price") as? com.google.gson.JsonPrimitive)?.asString?.toSafeBigDecimal()
+                if (bestBid != null) onPriceUpdate(assetId, bestBid, map)
+            }
+            "price_change" -> {
+                val priceChanges = json.get("price_changes") as? com.google.gson.JsonArray ?: return
+                for (i in 0 until priceChanges.size()) {
+                    val pc = priceChanges.get(i) as? com.google.gson.JsonObject ?: continue
+                    val assetId = (pc.get("asset_id") as? com.google.gson.JsonPrimitive)?.asString ?: continue
+                    val bestBidStr = (pc.get("best_bid") as? com.google.gson.JsonPrimitive)?.asString
+                    val bestBid = bestBidStr?.toSafeBigDecimal()
+                    if (bestBid != null) onPriceUpdate(assetId, bestBid, map)
+                }
+            }
+        }
+    }
+
+    private fun onPriceUpdate(tokenId: String, bestBid: BigDecimal, map: Map<String, List<MonitorEntry>>) {
+        if (closedForNoSubscribers.get()) return
+        val entries = map[tokenId] ?: return
+
+        for (entry in entries) {
+            val strategy = entry.strategy
+            val priceData = strategyPriceData[strategy.id!!] ?: StrategyPriceData()
+
+            // 根据方向更新价格
+            val newPriceData = if (entry.outcomeIndex == 0) {
+                // Up 方向
+                priceData.copy(
+                    currentPriceUp = bestBid,
+                    currentPriceDown = BigDecimal.ONE.subtract(bestBid),
+                    spreadUp = BigDecimal.ONE.subtract(bestBid),
+                    spreadDown = bestBid,
+                    lastUpdateTime = System.currentTimeMillis()
+                )
+            } else {
+                // Down 方向
+                priceData.copy(
+                    currentPriceDown = bestBid,
+                    currentPriceUp = BigDecimal.ONE.subtract(bestBid),
+                    spreadUp = bestBid,
+                    spreadDown = BigDecimal.ONE.subtract(bestBid),
+                    lastUpdateTime = System.currentTimeMillis()
+                )
+            }
+
+            strategyPriceData[strategy.id!!] = newPriceData
+
+            val pushData = buildPushData(strategy, newPriceData)
+            addToHistoryAndPush(strategy.id!!, pushData)
+        }
+    }
+
+    private fun addToHistoryAndPush(strategyId: Long, pushData: CryptoTailMonitorPushData) {
+        addToHistory(strategyId, pushData)
+        webSocketSubscriptionService.pushMonitorData(strategyId, pushData)
+    }
+
+    private fun addToHistory(strategyId: Long, pushData: CryptoTailMonitorPushData) {
+        val list = strategyPushHistory.getOrPut(strategyId) {
+            Collections.synchronizedList(mutableListOf<CryptoTailMonitorPushData>())
+        }
+        synchronized(list) {
+            val lastPeriod = strategyHistoryPeriod[strategyId]
+            if (lastPeriod != null && lastPeriod != pushData.periodStartUnix) {
+                list.clear()
+            }
+            strategyHistoryPeriod[strategyId] = pushData.periodStartUnix
+            list.add(pushData)
+            while (list.size > maxHistorySize) {
+                list.removeAt(0)
+            }
+        }
+    }
+
+    /**
+     * 构建推送数据
+     * 最新价、价差使用币安 K 线的 BTC 价格（open/close）
+     */
+    private fun buildPushData(strategy: CryptoTailStrategy, priceData: StrategyPriceData): CryptoTailMonitorPushData {
+        val nowSeconds = System.currentTimeMillis() / 1000
+        val periodStartUnix = (nowSeconds / strategy.intervalSeconds) * strategy.intervalSeconds
+        val periodEndUnix = periodStartUnix + strategy.intervalSeconds
+        val remainingSeconds = (periodEndUnix - nowSeconds).toInt().coerceAtLeast(0)
+
+        val windowStart = periodStartUnix + strategy.windowStartSeconds
+        val windowEnd = periodStartUnix + strategy.windowEndSeconds
+        val inTimeWindow = nowSeconds >= windowStart && nowSeconds < windowEnd
+
+        // 币安 K 线：open = 周期开盘价，close = 当前最新价（实时更新）
+        val openClose = binanceKlineService.getCurrentOpenClose(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix)
+        val openPriceBtc = priceData.openPriceBtc ?: openClose?.first
+        val currentPriceBtc = openClose?.second
+        // K 线数据回来后更新缓存，供后续使用
+        if (openPriceBtc != null && priceData.openPriceBtc == null && strategy.id != null) {
+            strategyPriceData[strategy.id] = priceData.copy(openPriceBtc = openPriceBtc)
+        }
+        val spreadBtc = if (openPriceBtc != null && currentPriceBtc != null) {
+            currentPriceBtc.subtract(openPriceBtc)
+        } else null
+
+        // 判断价格区间（Polymarket 0-1）
+        val currentUp = priceData.currentPriceUp
+        val currentDown = priceData.currentPriceDown
+        val inPriceRangeUp = currentUp != null &&
+            currentUp >= strategy.minPrice && currentUp <= strategy.maxPrice
+        val inPriceRangeDown = currentDown != null &&
+            currentDown >= strategy.minPrice && currentDown <= strategy.maxPrice
+
+        return CryptoTailMonitorPushData(
+            strategyId = strategy.id!!,
+            timestamp = System.currentTimeMillis(),
+            periodStartUnix = periodStartUnix,
+            currentPriceUp = priceData.currentPriceUp?.setScale(4, RoundingMode.HALF_UP)?.toPlainString(),
+            currentPriceDown = priceData.currentPriceDown?.setScale(4, RoundingMode.HALF_UP)?.toPlainString(),
+            spreadUp = priceData.spreadUp?.setScale(4, RoundingMode.HALF_UP)?.toPlainString(),
+            spreadDown = priceData.spreadDown?.setScale(4, RoundingMode.HALF_UP)?.toPlainString(),
+            minSpreadLineUp = priceData.minSpreadLineUp?.setScale(2, RoundingMode.HALF_UP)?.toPlainString(),
+            minSpreadLineDown = priceData.minSpreadLineDown?.setScale(2, RoundingMode.HALF_UP)?.toPlainString(),
+            openPriceBtc = openPriceBtc?.setScale(2, RoundingMode.HALF_UP)?.toPlainString(),
+            currentPriceBtc = currentPriceBtc?.setScale(2, RoundingMode.HALF_UP)?.toPlainString(),
+            spreadBtc = spreadBtc?.setScale(2, RoundingMode.HALF_UP)?.toPlainString(),
+            remainingSeconds = remainingSeconds,
+            inTimeWindow = inTimeWindow,
+            inPriceRangeUp = inPriceRangeUp,
+            inPriceRangeDown = inPriceRangeDown,
+            triggered = priceData.triggered,
+            triggerDirection = priceData.triggerDirection,
+            periodEnded = remainingSeconds <= 0
+        )
+    }
+
+    private fun maybeRefreshSubscriptionIfPeriodChanged() {
+        val subscribed = currentPeriodTokenToStrategy.get().values.flatten().distinctBy { it.strategyId }
+            .associate { it.strategyId to it.periodStartUnix }
+        if (subscribed.isEmpty()) return
+
+        val strategies = strategyRepository.findAllById(subscribed.keys)
+        val nowSeconds = System.currentTimeMillis() / 1000
+
+        for (s in strategies) {
+            if (s.id == null) continue
+            val currentPeriod = (nowSeconds / s.intervalSeconds) * s.intervalSeconds
+            val subPeriod = subscribed[s.id] ?: continue
+            if (currentPeriod != subPeriod) {
+                scope.launch { refreshSubscription() }
+                return
+            }
+        }
+    }
+
+    private fun scheduleRefreshAtPeriodEnd(newMap: Map<String, List<MonitorEntry>>) {
+        val entries = newMap.values.flatten()
+        if (entries.isEmpty()) return
+
+        val nextPeriodEndSeconds = entries.minOf { it.periodStartUnix + it.strategy.intervalSeconds }
+        val delayMs = (nextPeriodEndSeconds * 1000) - System.currentTimeMillis() + 2000
+        if (delayMs <= 0) return
+
+        periodEndCountdownJob = scope.launch {
+            delay(delayMs)
+            periodEndCountdownJob = null
+            refreshSubscription()
+        }
+    }
+
+    private fun closeWebSocketForNoSubscribers() {
+        closeAllWebSockets()
+    }
+
+    private fun scheduleReconnect() {
+        if (reconnectJob?.isActive == true) return
+        reconnectJob = scope.launch {
+            delay(reconnectDelayMs)
+            reconnectJob = null
+            if (strategySubscribers.isNotEmpty()) {
+                logger.info("尾盘监控 WebSocket 尝试重连")
+                refreshSubscription()
+            }
+        }
+    }
+
+    private fun fetchEventBySlug(slug: String): Result<GammaEventBySlugResponse> {
+        return try {
+            val api = retrofitFactory.createGammaApi()
+            val response = runBlocking { api.getEventBySlug(slug) }
+            if (response.isSuccessful && response.body() != null) {
+                Result.success(response.body()!!)
+            } else {
+                Result.failure(Exception("${response.code()}"))
+            }
+        } catch (e: Exception) {
+            Result.failure(e)
+        }
+    }
+
+    private fun parseClobTokenIds(clobTokenIds: String?): List<String> {
+        if (clobTokenIds.isNullOrBlank()) return emptyList()
+        return clobTokenIds.fromJson<List<String>>() ?: emptyList()
+    }
+
+    @PreDestroy
+    fun destroy() {
+        reconnectJob?.cancel()
+        periodEndCountdownJob?.cancel()
+        periodicPushJob?.cancel()
+        currentPeriodWebSocket?.close(1000, "shutdown")
+        currentPeriodWebSocket = null
+        nextPeriodWebSocket?.close(1000, "shutdown")
+        nextPeriodWebSocket = null
+    }
+}
@@ -102,16 +102,16 @@ class CryptoTailOrderNotificationPollingService(
            return false
        }
        val apiSecret = try {
-            cryptoUtils.decrypt(account.apiSecret) ?: return false
+            cryptoUtils.decrypt(account.apiSecret)
        } catch (e: Exception) {
            logger.warn("解密 API Secret 失败: accountId=${account.id}", e)
            return false
        }
        val apiPassphrase = try {
-            cryptoUtils.decrypt(account.apiPassphrase) ?: ""
+            cryptoUtils.decrypt(account.apiPassphrase)
        } catch (e: Exception) { "" }
        val clobApi = retrofitFactory.createClobApi(
-            account.apiKey!!,
+            account.apiKey,
            apiSecret,
            apiPassphrase,
            account.walletAddress
@@ -228,11 +228,11 @@ class CryptoTailSettlementService(
            val match = activities.firstOrNull { a ->
                a.type == "TRADE" &&
                    a.conditionId == conditionId &&
-                    a.outcomeIndex != null && a.outcomeIndex!! in 0..1 &&
+                    a.outcomeIndex != null && a.outcomeIndex in 0..1 &&
                    a.outcomeIndex == trigger.outcomeIndex &&
                    a.side?.uppercase() == "BUY" &&
-                    a.price != null && a.price!! > 0 &&
-                    a.size != null && a.size!! > 0
+                    a.price != null && a.price > 0 &&
+                    a.size != null && a.size > 0
            } ?: run {
                logger.debug("尾盘结算 activity 无匹配成交: triggerId=${trigger.id}, conditionId=$conditionId, outcomeIndex=${trigger.outcomeIndex}, 条数=${activities.size}")
                return null
@@ -135,13 +135,17 @@ class CryptoTailStrategyExecutionService(
            return null
        }
        val apiSecret = try {
-            account.apiSecret?.let { cryptoUtils.decrypt(it) } ?: ""
-        } catch (e: Exception) { "" }
+            account.apiSecret.let { cryptoUtils.decrypt(it) }
+        } catch (e: Exception) {
+            ""
+        }
        val apiPassphrase = try {
-            account.apiPassphrase?.let { cryptoUtils.decrypt(it) } ?: ""
-        } catch (e: Exception) { "" }
+            account.apiPassphrase.let { cryptoUtils.decrypt(it) }
+        } catch (e: Exception) {
+            ""
+        }

-        val clobApi = retrofitFactory.createClobApi(account.apiKey!!, apiSecret, apiPassphrase, account.walletAddress)
+        val clobApi = retrofitFactory.createClobApi(account.apiKey, apiSecret, apiPassphrase, account.walletAddress)
        val feeRateByTokenId = tokenIds.associate { tokenId ->
            tokenId to (clobService.getFeeRate(tokenId).getOrNull()?.toString() ?: "0")
        }
@@ -211,11 +215,19 @@ class CryptoTailStrategyExecutionService(

        val mutex = getTriggerMutex(strategy.id!!, periodStartUnix)
        mutex.withLock {
-            if (triggerRepository.findByStrategyIdAndPeriodStartUnix(strategy.id!!, periodStartUnix) != null) return@withLock
+            if (triggerRepository.findByStrategyIdAndPeriodStartUnix(
+                    strategy.id!!,
+                    periodStartUnix
+                ) != null
+            ) return@withLock
            val logKey = triggerLockKey(strategy.id!!, periodStartUnix)
            if (conditionLoggedCache.getIfPresent(logKey) == null) {
                conditionLoggedCache.put(logKey, periodStartUnix + strategy.intervalSeconds)
-                val oc = binanceKlineService.getCurrentOpenClose(strategy.marketSlugPrefix, 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}"
@@ -223,8 +235,8 @@ class CryptoTailStrategyExecutionService(
                val modeStr = if (strategy.spreadDirection == SpreadDirection.MAX) "最大价差" else "最小价差"
                logger.info(
                    "尾盘策略首次满足条件: strategyName=$strategyName, strategyId=${strategy.id}, " +
-                        "openPrice=$openPrice, closePrice=$closePrice, marketPrice=${bestBid.toPlainString()}, " +
-                        "direction=$direction, outcomeIndex=$outcomeIndex, spreadMode=$modeStr"
+                            "openPrice=$openPrice, closePrice=$closePrice, marketPrice=${bestBid.toPlainString()}, " +
+                            "direction=$direction, outcomeIndex=$outcomeIndex, spreadMode=$modeStr"
                )
            }
            if (!passSpreadCheck(strategy, periodStartUnix, outcomeIndex)) return@withLock
@@ -235,23 +247,29 @@ class CryptoTailStrategyExecutionService(

    private fun passSpreadCheck(strategy: CryptoTailStrategy, periodStartUnix: Long, outcomeIndex: Int): Boolean {
        if (strategy.spreadMode == SpreadMode.NONE) return true
-        val oc = binanceKlineService.getCurrentOpenClose(strategy.marketSlugPrefix, strategy.intervalSeconds, periodStartUnix)
+        val oc = binanceKlineService.getCurrentOpenClose(
+            strategy.marketSlugPrefix,
+            strategy.intervalSeconds,
+            periodStartUnix
+        )
            ?: return false
        val (openP, closeP) = oc
        val spreadAbs = closeP.subtract(openP).abs()
-        
+
        // 获取有效价差
        val effectiveSpread = when (strategy.spreadMode) {
            SpreadMode.FIXED -> {
                strategy.spreadValue?.takeIf { it > BigDecimal.ZERO } ?: return true
            }
+
            SpreadMode.AUTO -> {
                val result = computeAutoEffectiveSpread(strategy, periodStartUnix, outcomeIndex) ?: return true
                result.effectiveSpread.takeIf { it > BigDecimal.ZERO } ?: return true
            }
+
            SpreadMode.NONE -> return true
        }
-        
+
        // 根据价差方向判断
        return if (strategy.spreadDirection == SpreadDirection.MAX) {
            // 最大价差模式：价差 <= 配置值时触发
@@ -271,9 +289,22 @@ class CryptoTailStrategyExecutionService(
        val effectiveSpread: BigDecimal
    )

-    private fun computeAutoEffectiveSpread(strategy: CryptoTailStrategy, periodStartUnix: Long, outcomeIndex: Int): AutoSpreadResult? {
-        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 }
+    private fun computeAutoEffectiveSpread(
+        strategy: CryptoTailStrategy,
+        periodStartUnix: Long,
+        outcomeIndex: Int
+    ): AutoSpreadResult? {
+        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
@@ -306,18 +337,40 @@ class CryptoTailStrategyExecutionService(
            val amountUsdc = when (strategy.amountMode.uppercase()) {
                "RATIO" -> {
                    val balanceResult = accountService.getAccountBalance(ctx.account.id)
-                    val availableBalance = balanceResult.getOrNull()?.availableBalance?.toSafeBigDecimal() ?: BigDecimal.ZERO
+                    val availableBalance =
+                        balanceResult.getOrNull()?.availableBalance?.toSafeBigDecimal() ?: BigDecimal.ZERO
                    availableBalance.multiply(strategy.amountValue).divide(BigDecimal("100"), 18, RoundingMode.DOWN)
                }
+
                else -> strategy.amountValue
            }
            if (amountUsdc < BigDecimal("1")) {
-                saveTriggerRecord(strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, amountUsdc, null, "fail", "投入金额不足")
+                saveTriggerRecord(
+                    strategy,
+                    periodStartUnix,
+                    marketTitle,
+                    outcomeIndex,
+                    triggerPrice,
+                    amountUsdc,
+                    null,
+                    "fail",
+                    "投入金额不足"
+                )
                return
            }

            val tokenId = tokenIds.getOrNull(outcomeIndex) ?: run {
-                saveTriggerRecord(strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, amountUsdc, null, "fail", "tokenIds 越界")
+                saveTriggerRecord(
+                    strategy,
+                    periodStartUnix,
+                    marketTitle,
+                    outcomeIndex,
+                    triggerPrice,
+                    amountUsdc,
+                    null,
+                    "fail",
+                    "tokenIds 越界"
+                )
                return
            }

@@ -350,7 +403,16 @@ class CryptoTailStrategyExecutionService(
                orderType = "FAK",
                deferExec = false
            )
-            submitOrderAndSaveRecord(ctx.clobApi, strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, amountUsdc, orderRequest)
+            submitOrderAndSaveRecord(
+                ctx.clobApi,
+                strategy,
+                periodStartUnix,
+                marketTitle,
+                outcomeIndex,
+                triggerPrice,
+                amountUsdc,
+                orderRequest
+            )
            return
        }

@@ -373,7 +435,17 @@ class CryptoTailStrategyExecutionService(
            if (response.isSuccessful && response.body() != null) {
                val body = response.body()!!
                if (body.success && body.orderId != null) {
-                    saveTriggerRecord(strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, amountUsdc, body.orderId, "success", null)
+                    saveTriggerRecord(
+                        strategy,
+                        periodStartUnix,
+                        marketTitle,
+                        outcomeIndex,
+                        triggerPrice,
+                        amountUsdc,
+                        body.orderId,
+                        "success",
+                        null
+                    )
                    logger.info("尾盘策略下单成功: strategyId=${strategy.id}, periodStartUnix=$periodStartUnix, outcomeIndex=$outcomeIndex, orderId=${body.orderId}")
                    return
                }
@@ -386,7 +458,17 @@ class CryptoTailStrategyExecutionService(
            failReason = e.message ?: e.toString()
            logger.error("尾盘策略下单异常: strategyId=${strategy.id}, periodStartUnix=$periodStartUnix", e)
        }
-        saveTriggerRecord(strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, amountUsdc, null, "fail", failReason)
+        saveTriggerRecord(
+            strategy,
+            periodStartUnix,
+            marketTitle,
+            outcomeIndex,
+            triggerPrice,
+            amountUsdc,
+            null,
+            "fail",
+            failReason
+        )
        logger.error("尾盘策略下单失败: strategyId=${strategy.id}, periodStartUnix=$periodStartUnix, reason=$failReason")
    }

@@ -401,12 +483,32 @@ class CryptoTailStrategyExecutionService(
    ) {
        val account = accountRepository.findById(strategy.accountId).orElse(null) ?: run {
            logger.warn("账户不存在: accountId=${strategy.accountId}")
-            saveTriggerRecord(strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, BigDecimal.ZERO, null, "fail", "账户不存在")
+            saveTriggerRecord(
+                strategy,
+                periodStartUnix,
+                marketTitle,
+                outcomeIndex,
+                triggerPrice,
+                BigDecimal.ZERO,
+                null,
+                "fail",
+                "账户不存在"
+            )
            return
        }
        if (account.apiKey == null || account.apiSecret == null || account.apiPassphrase == null) {
            logger.warn("账户未配置 API 凭证: accountId=${account.id}")
-            saveTriggerRecord(strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, BigDecimal.ZERO, null, "fail", "账户未配置API凭证")
+            saveTriggerRecord(
+                strategy,
+                periodStartUnix,
+                marketTitle,
+                outcomeIndex,
+                triggerPrice,
+                BigDecimal.ZERO,
+                null,
+                "fail",
+                "账户未配置API凭证"
+            )
            return
        }

@@ -417,12 +519,32 @@ class CryptoTailStrategyExecutionService(
            else -> strategy.amountValue
        }
        if (amountUsdc < BigDecimal("1")) {
-            saveTriggerRecord(strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, amountUsdc, null, "fail", "投入金额不足")
+            saveTriggerRecord(
+                strategy,
+                periodStartUnix,
+                marketTitle,
+                outcomeIndex,
+                triggerPrice,
+                amountUsdc,
+                null,
+                "fail",
+                "投入金额不足"
+            )
            return
        }

        val tokenId = tokenIds.getOrNull(outcomeIndex) ?: run {
-            saveTriggerRecord(strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, amountUsdc, null, "fail", "tokenIds 越界")
+            saveTriggerRecord(
+                strategy,
+                periodStartUnix,
+                marketTitle,
+                outcomeIndex,
+                triggerPrice,
+                amountUsdc,
+                null,
+                "fail",
+                "tokenIds 越界"
+            )
            return
        }

@@ -441,16 +563,30 @@ class CryptoTailStrategyExecutionService(
            cryptoUtils.decrypt(account.privateKey) ?: ""
        } catch (e: Exception) {
            logger.error("解密私钥失败: accountId=${account.id}", e)
-            saveTriggerRecord(strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, amountUsdc, null, "fail", "解密私钥失败")
+            saveTriggerRecord(
+                strategy,
+                periodStartUnix,
+                marketTitle,
+                outcomeIndex,
+                triggerPrice,
+                amountUsdc,
+                null,
+                "fail",
+                "解密私钥失败"
+            )
            return
        }
        val apiSecret = try {
-            account.apiSecret?.let { cryptoUtils.decrypt(it) } ?: ""
-        } catch (e: Exception) { "" }
+            account.apiSecret.let { cryptoUtils.decrypt(it) }
+        } catch (e: Exception) {
+            ""
+        }
        val apiPassphrase = try {
-            account.apiPassphrase?.let { cryptoUtils.decrypt(it) } ?: ""
-        } catch (e: Exception) { "" }
-        val clobApi = retrofitFactory.createClobApi(account.apiKey!!, apiSecret, apiPassphrase, account.walletAddress)
+            account.apiPassphrase.let { cryptoUtils.decrypt(it) }
+        } catch (e: Exception) {
+            ""
+        }
+        val clobApi = retrofitFactory.createClobApi(account.apiKey, apiSecret, apiPassphrase, account.walletAddress)
        val feeRateBps = clobService.getFeeRate(tokenId).getOrNull()?.toString() ?: "0"
        val signatureType = orderSigningService.getSignatureTypeForWalletType(account.walletType)

@@ -472,7 +608,16 @@ class CryptoTailStrategyExecutionService(
            orderType = "FAK",
            deferExec = false
        )
-        submitOrderAndSaveRecord(clobApi, strategy, periodStartUnix, marketTitle, outcomeIndex, triggerPrice, amountUsdc, orderRequest)
+        submitOrderAndSaveRecord(
+            clobApi,
+            strategy,
+            periodStartUnix,
+            marketTitle,
+            outcomeIndex,
+            triggerPrice,
+            amountUsdc,
+            orderRequest
+        )
    }

    private suspend fun fetchEventBySlug(slug: String): Result<GammaEventBySlugResponse> {