From 84c79d8812ee7cbf4575fbac99b416e68843399f Mon Sep 17 00:00:00 2001 From: WrBug Date: Wed, 25 Feb 2026 17:00:41 +0800 Subject: [PATCH 1/6] =?UTF-8?q?feat(cryptotail):=20=E5=B0=BE=E7=9B=98?= =?UTF-8?q?=E7=9B=91=E6=8E=A7=E5=8F=8C=E8=BF=9E=E6=8E=A5=E4=B8=8E=E5=88=86?= =?UTF-8?q?=E6=97=B6=E5=9B=BE=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 监控 WebSocket 拆分为当前周期连接与下一周期连接，周期切换时关闭过期连接并新建下一周期 - 下一周期市场未创建时也建立第二条空连接，保证始终两条连接 - refreshSubscription 增加 Mutex 防重入，避免周期结束时定时器与消息同时触发导致重复执行 - 修复 initMonitor/buildPushData 中 getCurrentOpenClose、spreadMode/spreadValue 等 API 与实体字段引用 - 移除重复的 buildSubscriptionMap、buildPushData 等方法，修复 StrategyPriceData.periodStartUnix - 前端分时图：市场价折线增加 connectNulls，新周期默认 0.5 价格展示 - 多语言与监控页入口、API 类型与 WebSocket 订阅集成 Co-authored-by: Cursor --- .../config/MonitorServiceConfig.kt | 23 + .../CryptoTailStrategyController.kt | 30 +- .../polymarketbot/dto/CryptoTailMonitorDto.kt | 103 +++ .../service/accounts/AccountService.kt | 16 +- .../service/binance/BinanceKlineService.kt | 13 +- .../common/WebSocketSubscriptionService.kt | 92 ++- .../cryptotail/CryptoTailMonitorService.kt | 781 ++++++++++++++++++ ...yptoTailOrderNotificationPollingService.kt | 6 +- .../cryptotail/CryptoTailSettlementService.kt | 6 +- .../CryptoTailStrategyExecutionService.kt | 209 ++++- frontend/src/App.tsx | 2 + frontend/src/components/Layout.tsx | 5 + frontend/src/locales/en/common.json | 53 ++ frontend/src/locales/zh-CN/common.json | 53 ++ frontend/src/locales/zh-TW/common.json | 53 ++ frontend/src/pages/CryptoTailMonitor.tsx | 678 +++++++++++++++ frontend/src/services/api.ts | 4 +- frontend/src/types/index.ts | 94 +++ 18 files changed, 2162 insertions(+), 59 deletions(-) create mode 100644 backend/src/main/kotlin/com/wrbug/polymarketbot/config/MonitorServiceConfig.kt create mode 100644 backend/src/main/kotlin/com/wrbug/polymarketbot/dto/CryptoTailMonitorDto.kt create mode 100644 backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailMonitorService.kt create mode 100644 frontend/src/pages/CryptoTailMonitor.tsx diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/config/MonitorServiceConfig.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/config/MonitorServiceConfig.kt new file mode 100644 index 0000000..b2eaa1a --- /dev/null +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/config/MonitorServiceConfig.kt @@ -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) + } +} diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/cryptotail/CryptoTailStrategyController.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/cryptotail/CryptoTailStrategyController.kt index 25217b9..1d0879e 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/cryptotail/CryptoTailStrategyController.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/controller/cryptotail/CryptoTailStrategyController.kt @@ -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> { + 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)) + } + } } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/dto/CryptoTailMonitorDto.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/dto/CryptoTailMonitorDto.kt new file mode 100644 index 0000000..f3eedbe --- /dev/null +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/dto/CryptoTailMonitorDto.kt @@ -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 +) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/AccountService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/AccountService.kt index f2d3219..d0c8cb5 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/AccountService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/accounts/AccountService.kt @@ -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 diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineService.kt index a469948..952970f 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/binance/BinanceKlineService.kt @@ -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>() @@ -82,6 +84,7 @@ class BinanceKlineService { */ fun updateSubscriptions(marketPrefixes: Set) { val normalized = marketPrefixes.map { it.lowercase() }.toSet() + val parsed = normalized.mapNotNull { full -> parseMarketSlug(full)?.let { (base, interval) -> getSymbol(base)?.let { symbol -> Triple(full, symbol, interval) } @@ -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? { + private fun parseKlineMessage(text: String): Triple? { 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) } } diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/WebSocketSubscriptionService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/WebSocketSubscriptionService.kt index 187a91b..8723b71 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/WebSocketSubscriptionService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/common/WebSocketSubscriptionService.kt @@ -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 Unit>() + // 存储尾盘监控频道的订阅回调：sessionId -> (strategyId -> callback) + private val monitorChannelCallbacks = ConcurrentHashMap 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( } } } - diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailMonitorService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailMonitorService.kt new file mode 100644 index 0000000..64fddec --- /dev/null +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailMonitorService.kt @@ -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>>(emptyMap()) + /** 下一周期 token 映射 */ + private val nextPeriodTokenToStrategy = AtomicReference>>(emptyMap()) + + /** strategyId -> 当前价格数据 */ + private val strategyPriceData = ConcurrentHashMap() + + /** strategyId -> 订阅者数量 */ + private val strategySubscribers = ConcurrentHashMap() + + 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>() + private val strategyHistoryPeriod = ConcurrentHashMap() + 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 { + 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): Pair, Map>> { + val strategies = strategyRepository.findAllById(strategyIds) + val nowSeconds = System.currentTimeMillis() / 1000 + val tokenIdSet = mutableSetOf() + val map = mutableMapOf>() + + 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): Pair, Map>> { + val strategies = strategyRepository.findAllById(strategyIds) + val nowSeconds = System.currentTimeMillis() / 1000 + val tokenIdSet = mutableSetOf() + val map = mutableMapOf>() + + 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, 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, map: Map>) { + 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, map: Map>) { + 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() ?: 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>) { + 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()) + } + 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>) { + 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 { + 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 { + if (clobTokenIds.isNullOrBlank()) return emptyList() + return clobTokenIds.fromJson>() ?: emptyList() + } + + @PreDestroy + fun destroy() { + reconnectJob?.cancel() + periodEndCountdownJob?.cancel() + periodicPushJob?.cancel() + currentPeriodWebSocket?.close(1000, "shutdown") + currentPeriodWebSocket = null + nextPeriodWebSocket?.close(1000, "shutdown") + nextPeriodWebSocket = null + } +} diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationPollingService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationPollingService.kt index 93d3d26..47c2593 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationPollingService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailOrderNotificationPollingService.kt @@ -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 diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailSettlementService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailSettlementService.kt index 068ae91..e73326e 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailSettlementService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailSettlementService.kt @@ -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 diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailStrategyExecutionService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailStrategyExecutionService.kt index 32c0543..1a0c4e7 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailStrategyExecutionService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/cryptotail/CryptoTailStrategyExecutionService.kt @@ -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 { diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 0726391..a8c0faa 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -35,6 +35,7 @@ import Announcements from './pages/Announcements' import BacktestList from './pages/BacktestList' import BacktestDetail from './pages/BacktestDetail' import CryptoTailStrategyList from './pages/CryptoTailStrategyList' +import CryptoTailMonitor from './pages/CryptoTailMonitor' import { wsManager } from './services/websocket' import type { OrderPushMessage } from './types' import { apiService } from './services/api' @@ -252,6 +253,7 @@ function App() { } /> } /> } /> + } /> } /> {/* 保留旧路由以保持向后兼容 */} } /> diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index 6c4c8a2..d1617eb 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -162,6 +162,11 @@ const Layout: React.FC = ({ children }) => { icon: , label: t('menu.cryptoTailStrategy') }, + { + key: '/crypto-tail-monitor', + icon: , + label: t('menu.cryptoTailMonitor') + }, { key: '/positions', icon: , diff --git a/frontend/src/locales/en/common.json b/frontend/src/locales/en/common.json index d424ede..bb67515 100644 --- a/frontend/src/locales/en/common.json +++ b/frontend/src/locales/en/common.json @@ -313,6 +313,7 @@ "templates": "Templates", "copyTradingConfig": "Copy Trading Config", "cryptoTailStrategy": "Tail Strategy", + "cryptoTailMonitor": "Tail Monitor", "positions": "Position Management", "backtest": "Backtest", "statistics": "Statistics", @@ -1541,5 +1542,57 @@ "emptyFail": "No failed records", "totalCount": "{count} record(s)" } + }, + "cryptoTailMonitor": { + "title": "Tail Strategy Monitor", + "selectStrategy": "Strategy", + "selectStrategyPlaceholder": "Select a strategy to monitor", + "direction": "Direction", + "directionUp": "Up", + "directionDown": "Down", + "noData": "Select a strategy to start monitoring", + "priceRange": "Price Range", + "timeWindow": "Time Window", + "stat": { + "openPrice": "Open Price", + "currentPrice": "Current Price", + "spread": "Spread", + "remainingTime": "Remaining", + "configuredSpread": "Configured Spread", + "configuredSpreadMin": "Min Spread", + "configuredSpreadMax": "Max Spread", + "status": "Status", + "minSpreadLine": "Min Spread", + "periodSpreadMinMax": "Period Spread", + "periodSpreadMin": "Period Min Spread", + "periodSpreadMax": "Period Max Spread", + "minSpread": "Min", + "maxSpread": "Max" + }, + "status": { + "triggered": "Triggered", + "periodEnded": "Period Ended", + "inCondition": "In Condition", + "waiting": "Waiting" + }, + "chart": { + "title": "Price Chart", + "btcTitle": "BTC Price Chart", + "marketTitle": "Market Price Chart", + "price": "BTC Price", + "openPrice": "Open", + "spread": "Spread", + "minSpreadLine": "Min Spread Line", + "marketUp": "Up", + "marketDown": "Down", + "time": "Time" + }, + "strategyInfo": { + "title": "Strategy Info", + "market": "Market", + "interval": "Interval", + "account": "Account", + "spreadMode": "Spread Mode" + } } } \ No newline at end of file diff --git a/frontend/src/locales/zh-CN/common.json b/frontend/src/locales/zh-CN/common.json index 3b3f561..f431683 100644 --- a/frontend/src/locales/zh-CN/common.json +++ b/frontend/src/locales/zh-CN/common.json @@ -312,6 +312,7 @@ "templates": "跟单模板", "copyTradingConfig": "跟单配置", "cryptoTailStrategy": "尾盘策略", + "cryptoTailMonitor": "尾盘监控", "positions": "仓位管理", "backtest": "回测", "statistics": "统计信息", @@ -1540,5 +1541,57 @@ "emptyFail": "暂无失败记录", "totalCount": "共 {count} 条" } + }, + "cryptoTailMonitor": { + "title": "尾盘策略监控", + "selectStrategy": "选择策略", + "selectStrategyPlaceholder": "请选择要监控的策略", + "direction": "监控方向", + "directionUp": "Up", + "directionDown": "Down", + "noData": "请选择一个策略开始监控", + "priceRange": "价格区间", + "timeWindow": "时间窗口", + "stat": { + "openPrice": "开盘价", + "currentPrice": "最新价", + "spread": "价差", + "remainingTime": "剩余时间", + "configuredSpread": "配置价差", + "configuredSpreadMin": "最小价差", + "configuredSpreadMax": "最大价差", + "status": "状态", + "minSpreadLine": "最小价差线", + "periodSpreadMinMax": "周期内价差", + "periodSpreadMin": "周期内最小价差", + "periodSpreadMax": "周期内最大价差", + "minSpread": "最小", + "maxSpread": "最大" + }, + "status": { + "triggered": "已触发", + "periodEnded": "周期结束", + "inCondition": "满足条件", + "waiting": "等待中" + }, + "chart": { + "title": "分时图", + "btcTitle": "BTC 分时图", + "marketTitle": "市场分时图", + "price": "BTC 价格", + "openPrice": "开盘价", + "spread": "价差", + "minSpreadLine": "最小价差线", + "marketUp": "Up", + "marketDown": "Down", + "time": "时间" + }, + "strategyInfo": { + "title": "策略信息", + "market": "市场", + "interval": "周期", + "account": "账户", + "spreadMode": "价差模式" + } } } \ No newline at end of file diff --git a/frontend/src/locales/zh-TW/common.json b/frontend/src/locales/zh-TW/common.json index ab6a7c5..900a6a3 100644 --- a/frontend/src/locales/zh-TW/common.json +++ b/frontend/src/locales/zh-TW/common.json @@ -313,6 +313,7 @@ "templates": "跟單模板", "copyTradingConfig": "跟單配置", "cryptoTailStrategy": "尾盤策略", + "cryptoTailMonitor": "尾盤監控", "positions": "倉位管理", "backtest": "回測", "statistics": "統計信息", @@ -1541,5 +1542,57 @@ "emptyFail": "暫無失敗記錄", "totalCount": "共 {count} 條" } + }, + "cryptoTailMonitor": { + "title": "尾盤策略監控", + "selectStrategy": "選擇策略", + "selectStrategyPlaceholder": "請選擇要監控的策略", + "direction": "監控方向", + "directionUp": "Up", + "directionDown": "Down", + "noData": "請選擇一個策略開始監控", + "priceRange": "價格區間", + "timeWindow": "時間窗口", + "stat": { + "openPrice": "開盤價", + "currentPrice": "最新價", + "spread": "價差", + "remainingTime": "剩餘時間", + "configuredSpread": "配置價差", + "configuredSpreadMin": "最小價差", + "configuredSpreadMax": "最大價差", + "status": "狀態", + "minSpreadLine": "最小價差線", + "periodSpreadMinMax": "週期內價差", + "periodSpreadMin": "週期內最小價差", + "periodSpreadMax": "週期內最大價差", + "minSpread": "最小", + "maxSpread": "最大" + }, + "status": { + "triggered": "已觸發", + "periodEnded": "週期結束", + "inCondition": "滿足條件", + "waiting": "等待中" + }, + "chart": { + "title": "分時圖", + "btcTitle": "BTC 分時圖", + "marketTitle": "市場分時圖", + "price": "BTC 價格", + "openPrice": "開盤價", + "spread": "價差", + "minSpreadLine": "最小價差線", + "marketUp": "Up", + "marketDown": "Down", + "time": "時間" + }, + "strategyInfo": { + "title": "策略信息", + "market": "市場", + "interval": "週期", + "account": "賬戶", + "spreadMode": "價差模式" + } } } \ No newline at end of file diff --git a/frontend/src/pages/CryptoTailMonitor.tsx b/frontend/src/pages/CryptoTailMonitor.tsx new file mode 100644 index 0000000..ccd5916 --- /dev/null +++ b/frontend/src/pages/CryptoTailMonitor.tsx @@ -0,0 +1,678 @@ +import { useEffect, useState, useRef, useCallback } from 'react' +import { + Card, + Select, + Space, + Statistic, + Row, + Col, + Typography, + Spin, + Empty, + Alert +} from 'antd' +import { ClockCircleOutlined } from '@ant-design/icons' +import { useTranslation } from 'react-i18next' +import { useMediaQuery } from 'react-responsive' +import * as echarts from 'echarts' +import type { EChartsOption } from 'echarts' +import { apiService } from '../services/api' +import { useWebSocketSubscription } from '../hooks/useWebSocket' +import { formatNumber } from '../utils' +import type { + CryptoTailStrategyDto, + CryptoTailMonitorInitResponse, + CryptoTailMonitorPushData +} from '../types' + +const { Title, Text } = Typography + +/** 分时图数据点：时间戳、BTC 价格 USDC、市场 Up/Down 价格 0-1 */ +interface PriceDataPoint { + time: number + btcPrice: number | null + marketPriceUp: number | null + marketPriceDown: number | null +} + +const CryptoTailMonitor: React.FC = () => { + const { t } = useTranslation() + const isMobile = useMediaQuery({ maxWidth: 768 }) + + // 策略列表 + const [strategies, setStrategies] = useState([]) + const [strategiesLoading, setStrategiesLoading] = useState(false) + + // 选中的策略 + const [selectedStrategyId, setSelectedStrategyId] = useState(null) + + // 监控数据 + const [initData, setInitData] = useState(null) + const [pushData, setPushData] = useState(null) + const [initLoading, setInitLoading] = useState(false) + + // 价格历史数据（用于分时图） + const [priceHistory, setPriceHistory] = useState([]) + const chartRef = useRef(null) + const chartInstance = useRef(null) + const marketChartRef = useRef(null) + const marketChartInstance = useRef(null) + const lastPeriodStartRef = useRef(null) + // 记录首次数据进入时间（用于中途进入时的横轴起点） + const [firstDataTime, setFirstDataTime] = useState(null) + // 标记是否已切换过周期（切换后使用完整周期） + const [hasSwitchedPeriod, setHasSwitchedPeriod] = useState(false) + + // 获取策略列表 + useEffect(() => { + const fetchStrategies = async () => { + setStrategiesLoading(true) + try { + const res = await apiService.cryptoTailStrategy.list({ enabled: true }) + if (res.data.code === 0 && res.data.data) { + setStrategies(res.data.data.list ?? []) + // 自动选择第一个策略 + if (res.data.data.list?.length > 0 && !selectedStrategyId) { + setSelectedStrategyId(res.data.data.list[0].id) + } + } + } catch (e) { + console.error('Failed to fetch strategies:', e) + } finally { + setStrategiesLoading(false) + } + } + fetchStrategies() + }, []) + + // 初始化监控数据 + useEffect(() => { + if (!selectedStrategyId) { + setInitData(null) + setPushData(null) + setPriceHistory([]) + setFirstDataTime(null) + setHasSwitchedPeriod(false) + return + } + + const initMonitor = async () => { + setInitLoading(true) + setPriceHistory([]) + setFirstDataTime(null) + setHasSwitchedPeriod(false) + try { + const res = await apiService.cryptoTailStrategy.monitorInit(selectedStrategyId) + if (res.data.code === 0 && res.data.data) { + setInitData(res.data.data) + } else { + setInitData(null) + } + } catch (e) { + console.error('Failed to init monitor:', e) + setInitData(null) + } finally { + setInitLoading(false) + } + } + initMonitor() + }, [selectedStrategyId]) + + // WebSocket 订阅 + const handlePushData = useCallback((data: CryptoTailMonitorPushData) => { + if (data.strategyId !== selectedStrategyId) return + setPushData(data) + + const btcPrice = data.currentPriceBtc != null && data.currentPriceBtc !== '' + ? parseFloat(data.currentPriceBtc) + : null + const marketUp = data.currentPriceUp != null && data.currentPriceUp !== '' + ? parseFloat(data.currentPriceUp) + : null + const marketDown = data.currentPriceDown != null && data.currentPriceDown !== '' + ? parseFloat(data.currentPriceDown) + : null + const hasBtc = btcPrice != null && !Number.isNaN(btcPrice) + const hasMarket = (marketUp != null && !Number.isNaN(marketUp)) || (marketDown != null && !Number.isNaN(marketDown)) + if (!hasBtc && !hasMarket) return + + const newPoint: PriceDataPoint = { + time: data.timestamp, + btcPrice: hasBtc ? btcPrice : null, + marketPriceUp: hasMarket && marketUp != null && !Number.isNaN(marketUp) ? marketUp : null, + marketPriceDown: hasMarket && marketDown != null && !Number.isNaN(marketDown) ? marketDown : null + } + + // 用 ref 检测周期切换，避免因依赖 initData 导致回调频繁重建 + const pushPeriod = data.periodStartUnix + const lastPeriod = lastPeriodStartRef.current + + if (pushPeriod != null && pushPeriod !== lastPeriod) { + // 新周期或首次推送：更新 ref 和 initData，清空历史 + lastPeriodStartRef.current = pushPeriod + // 如果之前已经有周期数据，说明是周期切换，标记为已切换 + if (lastPeriod != null) { + setHasSwitchedPeriod(true) + } + // 周期切换时重置首次数据时间 + setFirstDataTime(newPoint.time) + setInitData(prev => prev ? { ...prev, periodStartUnix: pushPeriod } : null) + setPriceHistory([newPoint]) + } else { + // 同周期：追加数据 + // 记录首次数据时间（仅在中途进入且未切换过周期时记录） + setFirstDataTime(prev => { + if (prev == null) { + return newPoint.time + } + return prev + }) + setPriceHistory(prev => { + const maxPoints = 300 + const newHistory = [...prev, newPoint] + return newHistory.slice(-maxPoints) + }) + } + }, [selectedStrategyId]) + + const channel = selectedStrategyId ? `crypto_tail_monitor_${selectedStrategyId}` : '' + useWebSocketSubscription(channel, handlePushData) + + // 图表容器仅在 initData 存在时渲染，故在更新图表时懒初始化 + useEffect(() => { + const handleResize = () => { + chartInstance.current?.resize() + marketChartInstance.current?.resize() + } + window.addEventListener('resize', handleResize) + return () => { + window.removeEventListener('resize', handleResize) + chartInstance.current?.dispose() + chartInstance.current = null + marketChartInstance.current?.dispose() + marketChartInstance.current = null + } + }, []) + + // 更新图表：分时图为 BTC 价格 USDC + useEffect(() => { + if (!initData) return + if (chartRef.current && !chartInstance.current) { + chartInstance.current = echarts.init(chartRef.current) + } + if (!chartInstance.current) return + + const periodStartMs = (initData.periodStartUnix ?? 0) * 1000 + const periodEndMs = periodStartMs + (initData.intervalSeconds ?? 300) * 1000 + + // data.timestamp 为毫秒，firstDataTime 已是 ms，无需再乘 1000 + const firstDataMs = firstDataTime != null ? firstDataTime : null + const isMidEntry = firstDataMs != null && !hasSwitchedPeriod && firstDataMs > periodStartMs + // 中途进入时横轴起点为进入时刻，否则为周期起点 + const xAxisMin = isMidEntry ? firstDataMs : periodStartMs + + const btcData: [number, number | null][] = priceHistory.length > 0 + ? priceHistory.map(p => [p.time, p.btcPrice]) + : [] + const openBtc = pushData?.openPriceBtc ?? initData.openPriceBtc + const openBtcNum = openBtc != null ? parseFloat(openBtc) : null + + const hasAnyBtcData = btcData.some(([, v]) => v != null && !Number.isNaN(v)) + const btcPlaceholderTime = xAxisMin + const displayBtcData: [number, number | null][] = hasAnyBtcData + ? btcData + : (openBtcNum != null ? [[btcPlaceholderTime, openBtcNum]] : []) + const minSpreadUpRaw = pushData?.minSpreadLineUp ?? initData.autoMinSpreadUp + const minSpreadDownRaw = pushData?.minSpreadLineDown ?? initData.autoMinSpreadDown + const minSpreadUp = minSpreadUpRaw != null && minSpreadUpRaw !== '' ? parseFloat(minSpreadUpRaw) : null + const minSpreadDown = minSpreadDownRaw != null && minSpreadDownRaw !== '' ? parseFloat(minSpreadDownRaw) : null + + const validPrices = displayBtcData.flatMap(([, v]) => (v != null && !Number.isNaN(v) ? [v] : [])) + const defaultRange = 500 + let yMin: number | undefined + let yMax: number | undefined + if (validPrices.length > 0) { + const dataMin = Math.min(...validPrices) + const dataMax = Math.max(...validPrices) + const dataRange = dataMax - dataMin + const minRange = Math.max(Math.abs(dataMax) * 0.01, 10) + const range = Math.max(dataRange, minRange) + const padding = range * 0.25 + yMin = dataMin - padding + yMax = dataMax + padding + } else if (openBtcNum != null) { + const spread = minSpreadUp ?? minSpreadDown ?? defaultRange + const halfRange = spread * 1.5 + yMin = openBtcNum - halfRange + yMax = openBtcNum + halfRange + } + + const markLineData: Array<{ name: string; yAxis: number; lineStyle: { type: 'dashed'; color: string } }> = [] + if (openBtcNum != null && !Number.isNaN(openBtcNum)) { + markLineData.push({ + name: t('cryptoTailMonitor.chart.openPrice'), + yAxis: openBtcNum, + lineStyle: { type: 'dashed', color: '#999' } + }) + } + if (openBtcNum != null && minSpreadUp != null && !Number.isNaN(minSpreadUp)) { + markLineData.push({ + name: t('cryptoTailMonitor.chart.minSpreadLine') + ' Up', + yAxis: openBtcNum + minSpreadUp, + lineStyle: { type: 'dashed', color: '#ff4d4f' } + }) + } + if (openBtcNum != null && minSpreadDown != null && !Number.isNaN(minSpreadDown)) { + markLineData.push({ + name: t('cryptoTailMonitor.chart.minSpreadLine') + ' Down', + yAxis: openBtcNum - minSpreadDown, + lineStyle: { type: 'dashed', color: '#ff4d4f' } + }) + } + + const option: EChartsOption = { + tooltip: { + trigger: 'axis', + formatter: (params: unknown) => { + const arr = params as Array<{ seriesName: string; name: string; value: number | [number, number] }> + const priceParam = arr.find(p => p.seriesName === t('cryptoTailMonitor.chart.price')) + if (!priceParam) return '' + const val = Array.isArray(priceParam.value) ? priceParam.value[1] : priceParam.value + if (val == null || Number.isNaN(val)) return '' + const ts = priceParam.name as unknown as number + const offsetSec = Math.floor(ts / 1000) - (initData?.periodStartUnix ?? 0) + const mins = Math.floor(offsetSec / 60) + const secs = offsetSec % 60 + const timeStr = `${mins.toString().padStart(2, '0')}:${secs.toString().padStart(2, '0')}` + return ` +
+
${t('cryptoTailMonitor.chart.time')}: ${timeStr}
+
${t('cryptoTailMonitor.chart.price')}: ${Number(val).toFixed(2)} USDC
+
+ ` + } + }, + legend: { + show: true, + top: 0 + }, + grid: { + left: '3%', + right: '4%', + bottom: '3%', + top: '12%', + containLabel: true + }, + xAxis: { + type: 'time', + min: xAxisMin, + max: periodEndMs, + boundaryGap: false, + axisLabel: { + formatter: (val: number) => { + const offsetSec = Math.floor(val / 1000) - (initData.periodStartUnix ?? 0) + const mins = Math.floor(offsetSec / 60) + const secs = offsetSec % 60 + return `${mins.toString().padStart(2, '0')}:${secs.toString().padStart(2, '0')}` + } + } + }, + yAxis: { + type: 'value', + scale: true, + min: yMin, + max: yMax, + axisLabel: { + formatter: (value: number) => value.toFixed(0) + } + }, + series: [ + { + name: t('cryptoTailMonitor.chart.price'), + type: 'line', + data: displayBtcData, + smooth: true, + symbol: displayBtcData.length === 1 ? 'circle' : 'none', + symbolSize: 4, + lineStyle: { width: 2, color: '#1890ff' }, + areaStyle: { + color: { + type: 'linear', + x: 0, + y: 0, + x2: 0, + y2: 1, + colorStops: [ + { offset: 0, color: 'rgba(24, 144, 255, 0.3)' }, + { offset: 1, color: 'rgba(24, 144, 255, 0.05)' } + ] + } + }, + markLine: markLineData.length > 0 ? { data: markLineData } : undefined + } + ] + } + + chartInstance.current.setOption(option, true) + chartInstance.current.resize() + }, [priceHistory, initData, pushData, firstDataTime, hasSwitchedPeriod, t]) + + // 更新市场分时图：Polymarket 价格 0-1 + useEffect(() => { + if (!initData) return + if (marketChartRef.current && !marketChartInstance.current) { + marketChartInstance.current = echarts.init(marketChartRef.current) + } + if (!marketChartInstance.current) return + + const periodStartMs = (initData.periodStartUnix ?? 0) * 1000 + const periodEndMs = periodStartMs + (initData.intervalSeconds ?? 300) * 1000 + + // data.timestamp 为毫秒，firstDataTime 已是 ms，无需再乘 1000 + const firstDataMs = firstDataTime != null ? firstDataTime : null + const isMidEntry = firstDataMs != null && !hasSwitchedPeriod && firstDataMs > periodStartMs + const xAxisMin = isMidEntry ? firstDataMs : periodStartMs + + const toMs = (t: number) => (t > 0 && t < 1e12 ? t * 1000 : t) + const marketUpData: [number, number | null][] = priceHistory.length > 0 + ? priceHistory.map(p => [toMs(p.time), p.marketPriceUp]) + : [] + const marketDownData: [number, number | null][] = priceHistory.length > 0 + ? priceHistory.map(p => [toMs(p.time), p.marketPriceDown]) + : [] + + const minPrice = parseFloat(initData.minPrice) + const maxPrice = parseFloat(initData.maxPrice) + const midPrice = (minPrice + maxPrice) / 2 + const isValid = (v: number | null): v is number => v != null && !Number.isNaN(v) + const validUp: [number, number][] = marketUpData.filter((point): point is [number, number] => isValid(point[1])) + const validDown: [number, number][] = marketDownData.filter((point): point is [number, number] => isValid(point[1])) + const hasAnyMarketData = validUp.length > 0 || validDown.length > 0 + const placeholderTime = xAxisMin + const finalMarketUp: [number, number][] = hasAnyMarketData ? validUp : [[placeholderTime, midPrice]] + const finalMarketDown: [number, number][] = hasAnyMarketData ? validDown : [[placeholderTime, midPrice]] + + const option: EChartsOption = { + tooltip: { + trigger: 'axis', + formatter: (params: unknown) => { + const arr = params as Array<{ seriesName: string; name: string | number; value: number | [number, number] }> + const upParam = arr.find(p => p.seriesName === t('cryptoTailMonitor.chart.marketUp')) + const downParam = arr.find(p => p.seriesName === t('cryptoTailMonitor.chart.marketDown')) + const rawTime = arr[0]?.name + const timeStr = typeof rawTime === 'number' && initData + ? (() => { + const offsetSec = Math.floor(rawTime / 1000) - (initData.periodStartUnix ?? 0) + const mins = Math.floor(offsetSec / 60) + const secs = offsetSec % 60 + return `${mins.toString().padStart(2, '0')}:${secs.toString().padStart(2, '0')}` + })() + : String(rawTime ?? '') + let html = `
${t('cryptoTailMonitor.chart.time')}: ${timeStr}
` + const upVal = Array.isArray(upParam?.value) ? upParam?.value[1] : upParam?.value + const downVal = Array.isArray(downParam?.value) ? downParam?.value[1] : downParam?.value + if (upVal != null && !Number.isNaN(upVal)) html += `
Up: ${Number(upVal).toFixed(4)}
` + if (downVal != null && !Number.isNaN(downVal)) html += `
Down: ${Number(downVal).toFixed(4)}
` + html += '
' + return html + } + }, + legend: { + show: true, + top: 0, + data: [t('cryptoTailMonitor.chart.marketUp'), t('cryptoTailMonitor.chart.marketDown')] + }, + grid: { + left: '3%', + right: '4%', + bottom: '3%', + top: '15%', + containLabel: true + }, + xAxis: { + type: 'time', + min: xAxisMin, + max: periodEndMs, + boundaryGap: false, + axisLabel: { + formatter: (val: number) => { + const offsetSec = Math.floor(val / 1000) - (initData.periodStartUnix ?? 0) + const mins = Math.floor(offsetSec / 60) + const secs = offsetSec % 60 + return `${mins.toString().padStart(2, '0')}:${secs.toString().padStart(2, '0')}` + } + } + }, + yAxis: { + type: 'value', + min: 0, + max: 1, + interval: 0.2, + axisLabel: { formatter: (v: number) => v.toFixed(1) } + }, + series: [ + { + name: t('cryptoTailMonitor.chart.marketUp'), + type: 'line', + data: finalMarketUp, + smooth: true, + symbol: 'circle', + symbolSize: 4, + showSymbol: true, + connectNulls: true, + lineStyle: { width: 2, color: '#1890ff' }, + itemStyle: { color: '#1890ff' }, + markArea: { + silent: true, + itemStyle: { color: 'rgba(82, 196, 26, 0.12)' }, + data: [[{ yAxis: minPrice }, { yAxis: maxPrice }]] + } + }, + { + name: t('cryptoTailMonitor.chart.marketDown'), + type: 'line', + data: finalMarketDown, + smooth: true, + symbol: 'circle', + symbolSize: 4, + showSymbol: true, + connectNulls: true, + lineStyle: { width: 2, color: '#fa8c16' }, + itemStyle: { color: '#fa8c16' } + } + ] + } + + marketChartInstance.current.setOption(option, true) + marketChartInstance.current.resize() + }, [priceHistory, initData, firstDataTime, hasSwitchedPeriod, t]) + + // 格式化剩余时间 + const formatRemainingTime = (seconds: number): string => { + const mins = Math.floor(seconds / 60) + const secs = seconds % 60 + return `${mins}:${secs.toString().padStart(2, '0')}` + } + + // 显示 BTC 价格（最新价、价差、开盘价均为 USDC） + const openPrice = pushData?.openPriceBtc ?? initData?.openPriceBtc + const currentPrice = pushData?.currentPriceBtc + const currentSpread = pushData?.spreadBtc + const minSpreadUpStr = pushData?.minSpreadLineUp ?? initData?.autoMinSpreadUp + const minSpreadDownStr = pushData?.minSpreadLineDown ?? initData?.autoMinSpreadDown + const minSpreadUpVal = minSpreadUpStr != null && minSpreadUpStr !== '' ? parseFloat(minSpreadUpStr) : null + const minSpreadDownVal = minSpreadDownStr != null && minSpreadDownStr !== '' ? parseFloat(minSpreadDownStr) : null + const minSpreadLineNum = [minSpreadUpVal, minSpreadDownVal].filter((v): v is number => v != null && !Number.isNaN(v)) + const spreadBelowThreshold = currentSpread != null && currentSpread !== '' && minSpreadLineNum.length > 0 && + parseFloat(currentSpread) < Math.min(...minSpreadLineNum) + + return ( +
+ + {t('cryptoTailMonitor.title')} + + + {/* 顶部控制区 */} + + + + {t('cryptoTailMonitor.selectStrategy')} +