diff --git a/.cursor/rules/backend.mdc b/.cursor/rules/backend.mdc index dac0ae6..b7a3f64 100644 --- a/.cursor/rules/backend.mdc +++ b/.cursor/rules/backend.mdc @@ -13,7 +13,6 @@ path: backend/** ## 项目范围 - **平台**: 仅支持 Polymarket -- **分类**: 仅支持 `sports` 和 `crypto` - **包名**: `com.wrbug.polymarketbot` ## 实体类规范 @@ -109,6 +108,57 @@ if (total.lt(BigDecimal.ONE)) { } ``` +## JSON 解析规范 + +### 使用扩展函数（推荐） +- **优先**使用扩展函数 `fromJson()` 和 `toJson()`，禁止直接使用 `Gson` +- 扩展函数提供了类型安全的解析，使用方便 + +```kotlin +// ✅ 正确：使用扩展函数（推荐） +val json = "{\"name\":\"test\"}" +val obj = json.fromJson() +val jsonStr = obj.toJson() + +// ✅ 也可以：使用 JsonUtils 类（兼容旧代码） +val obj = jsonUtils.fromJson(json) + +// ❌ 错误：直接使用 Gson +val obj = gson.fromJson(json, MyDataClass::class.java) +``` + +### 扩展函数列表 +- `String?.fromJson()` - 解析 JSON 字符串为对象（支持泛型） +- `JsonElement?.fromJson()` - 解析 JsonElement 为对象（支持泛型） +- `Any?.toJson()` - 将对象转换为 JSON 字符串 +- `String?.parseStringArray()` - 解析 JSON 字符串数组 + +### JsonUtils 类 +- `JsonUtils` 主要用于初始化全局 Gson 实例，供扩展函数使用 +- 保留 `parseStringArray()` 方法用于兼容旧代码 +- 不推荐直接使用 `JsonUtils` 的方法，优先使用扩展函数 + +### Data Class 规范 +- **所有 data class 字段必须提供默认值** +- 可空字段使用 `? = null` +- 非空字段提供合适的默认值（空字符串、空集合、默认数值等） + +```kotlin +// ✅ 正确：所有字段都有默认值 +data class MyDto( + val name: String = "", + val age: Int = 0, + val tags: List = emptyList(), + val optional: String? = null +) + +// ❌ 错误：缺少默认值 +data class MyDto( + val name: String, // 缺少默认值 + val age: Int = 0 +) +``` + ## Side 判断规范 **禁止**使用 "YES"/"NO" 字符串判断 side @@ -165,6 +215,8 @@ return ResponseEntity.ok(ApiResponse.error(ErrorCode.PARAM_EMPTY, messageSource) - ❌ 禁止使用 `Double` 进行数值计算 - ❌ 禁止使用 `LocalDateTime` 存储时间 - ❌ 禁止实体类 ID 使用非空默认值 +- ❌ 禁止直接使用 `Gson`（必须使用 `JsonUtils`） +- ❌ 禁止 data class 字段缺少默认值 ### API 接口 - ❌ 禁止使用 GET/PUT/DELETE（统一使用 POST） diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/dto/ActivityTradeMessageDto.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/dto/ActivityTradeMessageDto.kt new file mode 100644 index 0000000..cdf192d --- /dev/null +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/dto/ActivityTradeMessageDto.kt @@ -0,0 +1,64 @@ +package com.wrbug.polymarketbot.dto + +import com.google.gson.annotations.SerializedName + +/** + * Activity WebSocket 交易消息 DTO + * 根据 Polymarket RTDS Activity WebSocket API 格式定义 + */ +data class ActivityTradeMessage( + val topic: String = "", // "activity" + val type: String = "", // "trades" + val timestamp: Long? = null, // 消息时间戳（可选） + @SerializedName("connection_id") + val connectionId: String? = null, // 连接 ID（可选，由服务器返回） + val payload: ActivityTradePayload = ActivityTradePayload() // 交易数据 +) + +/** + * Activity Trade Payload + */ +data class ActivityTradePayload( + val asset: String = "", // Token ID (用于下单) + + @SerializedName("conditionId") + val conditionId: String = "", // Market condition ID + + @SerializedName("eventSlug") + val eventSlug: String? = null, // 事件 slug + + val slug: String? = null, // 市场 slug + + val outcome: String? = null, // 结果方向 (Yes/No/Up/Down) + + @SerializedName("outcomeIndex") + val outcomeIndex: Int? = null, // 结果索引 (0=Yes/Up, 1=No/Down) - 优先使用此字段 + + val side: String = "", // 交易方向 (BUY/SELL) + + // price 和 size 可能是数字或字符串，使用 Any 类型，后续转换为 String + val price: Any? = null, // 交易价格 + + val size: Any? = null, // 交易数量 (shares) + + val timestamp: Any? = null, // Unix 时间戳（可能是秒或毫秒，可能是数字或字符串） + + @SerializedName("transactionHash") + val transactionHash: String? = null, // 交易哈希 + + val trader: ActivityTrader? = null, // 交易者信息对象（优先） + + @SerializedName("proxyWallet") + val proxyWallet: String? = null, // 交易者地址（fallback，如果 trader 不存在） + + val name: String? = null // 交易者名称（fallback，如果 trader 不存在） +) + +/** + * Activity Trader 信息 + */ +data class ActivityTrader( + val name: String? = null, // 交易者用户名（可选） + val address: String? = null // 交易者钱包地址 ⭐ 关键字段 +) + diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingMonitorService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingMonitorService.kt index 29a60bb..cf0c4cb 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingMonitorService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/CopyTradingMonitorService.kt @@ -14,7 +14,9 @@ import org.springframework.stereotype.Service /** * 跟单监听服务（主服务） * 管理所有Leader的交易监听 - * 使用链上 WebSocket 监听 Leader 的交易（实时，秒级延迟） + * 使用双重监听机制： + * 1. Activity WebSocket - 低延迟（< 100ms），优先使用 + * 2. OnChain WebSocket - 高可靠性（~2-3s），作为兜底 * 同时监听跟单账户的卖出/赎回事件（通过链上 WebSocket） */ @Service @@ -22,6 +24,7 @@ class CopyTradingMonitorService( private val copyTradingRepository: CopyTradingRepository, private val leaderRepository: LeaderRepository, private val accountRepository: AccountRepository, + private val activityWsService: PolymarketActivityWsService, private val onChainWsService: OnChainWsService, private val accountOnChainMonitorService: AccountOnChainMonitorService ) { @@ -50,14 +53,17 @@ class CopyTradingMonitorService( @PreDestroy fun destroy() { scope.cancel() - // 停止链上 WS 监听 + // 停止所有监听 + activityWsService.stop() onChainWsService.stop() accountOnChainMonitorService.stop() } /** * 启动监听 - * 启动链上 WebSocket 监听 Leader 的交易（实时，秒级延迟） + * 启动双重监听机制： + * 1. Activity WebSocket - 低延迟（< 100ms），优先使用 + * 2. OnChain WebSocket - 高可靠性（~2-3s），作为兜底 * 同时启动跟单账户的链上 WebSocket 监听（用于检测卖出/赎回事件） */ suspend fun startMonitoring() { @@ -80,10 +86,13 @@ class CopyTradingMonitorService( accountRepository.findById(accountId).orElse(null) } - // 4. 启动链上 WebSocket 监听 Leader 的交易（实时，秒级延迟） + // 4. 启动 Activity WebSocket 监听（优先，低延迟） + activityWsService.start(leaders) + + // 5. 启动链上 WebSocket 监听（兜底，高可靠性） onChainWsService.start(leaders) - // 5. 启动跟单账户的链上 WebSocket 监听（用于检测卖出/赎回事件） + // 6. 启动跟单账户的链上 WebSocket 监听（用于检测卖出/赎回事件） accountOnChainMonitorService.start(accounts) } @@ -100,7 +109,8 @@ class CopyTradingMonitorService( return } - // 添加到链上 WS 监听（如果不在列表中才添加） + // 同时添加到两种监听 + activityWsService.addLeader(leader) onChainWsService.addLeader(leader) } @@ -115,7 +125,8 @@ class CopyTradingMonitorService( return } - // 没有启用的跟单配置了，移除监听 + // 没有启用的跟单配置了，同时从两种监听移除 + activityWsService.removeLeader(leaderId) onChainWsService.removeLeader(leaderId) } @@ -130,6 +141,7 @@ class CopyTradingMonitorService( if (copyTradings.isNotEmpty()) { // 有启用的跟单配置，确保在监听列表中 + activityWsService.addLeader(leader) onChainWsService.addLeader(leader) // 更新账户监听（添加该配置关联的账户） @@ -141,7 +153,8 @@ class CopyTradingMonitorService( } } } else { - // 没有启用的跟单配置，移除监听 + // 没有启用的跟单配置，同时从两种监听移除 + activityWsService.removeLeader(leaderId) onChainWsService.removeLeader(leaderId) } } @@ -171,6 +184,7 @@ class CopyTradingMonitorService( */ suspend fun restartMonitoring() { // 停止所有监听 + activityWsService.stop() onChainWsService.stop() delay(1000) // 等待1秒 startMonitoring() diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/PolymarketActivityWsService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/PolymarketActivityWsService.kt new file mode 100644 index 0000000..cb094d4 --- /dev/null +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/monitor/PolymarketActivityWsService.kt @@ -0,0 +1,412 @@ +package com.wrbug.polymarketbot.service.copytrading.monitor + +import com.wrbug.polymarketbot.api.TradeResponse +import com.wrbug.polymarketbot.dto.ActivityTradeMessage +import com.wrbug.polymarketbot.dto.ActivityTradePayload +import com.wrbug.polymarketbot.entity.Leader +import com.wrbug.polymarketbot.repository.LeaderRepository +import com.wrbug.polymarketbot.service.copytrading.statistics.CopyOrderTrackingService +import com.wrbug.polymarketbot.util.fromJson +import com.wrbug.polymarketbot.websocket.PolymarketWebSocketClient +import jakarta.annotation.PreDestroy +import kotlinx.coroutines.* +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Value +import org.springframework.stereotype.Service +import java.math.BigDecimal +import java.util.concurrent.ConcurrentHashMap + +/** + * Polymarket Activity WebSocket 监听服务 + * 通过订阅全局 activity 交易流，客户端过滤 Leader 地址，实现实时交易检测 + * 延迟 < 100ms，适合快速跟单场景 + */ +@Service +class PolymarketActivityWsService( + private val copyOrderTrackingService: CopyOrderTrackingService, + private val leaderRepository: LeaderRepository +) { + + private val logger = LoggerFactory.getLogger(PolymarketActivityWsService::class.java) + + @Value("\${polymarket.websocket.activity.url:wss://ws-live-data.polymarket.com}") + private var websocketUrl: String = "wss://ws-live-data.polymarket.com" + + private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + + // 单例 WebSocket 客户端 + private var wsClient: PolymarketWebSocketClient? = null + + // 要监听的 Leader 地址集合（小写地址 -> leaderId） + private val monitoredAddresses = ConcurrentHashMap() + + // 是否已订阅 + @Volatile + private var isSubscribed = false + + /** + * 启动监听 + */ + fun start(leaders: List) { + monitoredAddresses.clear() + leaders.forEach { leader -> + val leaderId = leader.id + if (leaderId != null) { + monitoredAddresses[leader.leaderAddress.lowercase()] = leaderId + } + } + + if (monitoredAddresses.isEmpty()) { + logger.info("没有需要监听的 Leader，停止 Activity WebSocket") + stop() + return + } + + logger.info("启动 Activity WebSocket 监听，监控 ${monitoredAddresses.size} 个 Leader 地址") + connectAndSubscribe() + } + + /** + * 添加 Leader + */ + fun addLeader(leader: Leader) { + if (leader.id == null) { + logger.warn("Leader ID 为空，跳过: ${leader.leaderAddress}") + return + } + + val address = leader.leaderAddress.lowercase() + val existingLeaderId = monitoredAddresses[address] + + if (existingLeaderId != null && existingLeaderId == leader.id) { + logger.debug("Leader 已在监听列表中: ${leader.leaderName} (${address})") + return + } + + val leaderId = leader.id + if (leaderId != null) { + monitoredAddresses[address] = leaderId + logger.info("添加 Leader 到 Activity WS 监听: ${leader.leaderName} (${address})") + + // 如果 WebSocket 未连接，连接 + val client = wsClient + if (client == null || !client.isConnected()) { + connectAndSubscribe() + } + } + } + + /** + * 移除 Leader + */ + fun removeLeader(leaderId: Long) { + val addressToRemove = monitoredAddresses.entries + .find { it.value == leaderId }?.key + + if (addressToRemove != null) { + monitoredAddresses.remove(addressToRemove) + logger.info("从 Activity WS 监听移除 Leader: leaderId=$leaderId, address=$addressToRemove") + } + + // 如果没有 Leader 了，停止监听 + if (monitoredAddresses.isEmpty()) { + logger.info("没有 Leader 需要监听了，停止 Activity WebSocket") + stop() + } + } + + /** + * 连接并订阅 + */ + private fun connectAndSubscribe() { + val existingClient = wsClient + if (existingClient != null && existingClient.isConnected()) { + // 如果已连接但未订阅，发送订阅 + if (!isSubscribed) { + subscribeAllActivity() + } + return + } + + logger.info("连接 Activity WebSocket: $websocketUrl") + + val newClient = PolymarketWebSocketClient( + url = websocketUrl, + sessionId = "copy-trading-activity", + onMessage = { message -> handleMessage(message) }, + onOpen = { + logger.info("Activity WebSocket 连接成功") + subscribeAllActivity() + }, + onReconnect = { + logger.info("Activity WebSocket 重连成功，重新订阅") + subscribeAllActivity() + } + ) + + wsClient = newClient + + scope.launch { + try { + newClient.connect() + } catch (e: Exception) { + logger.error("连接 Activity WebSocket 失败", e) + } + } + } + + /** + * 订阅全局 activity + * 根据 @polymarket/real-time-data-client 的协议格式 + * 使用 "action": "subscribe" 而不是 "type": "subscribe" + */ + private fun subscribeAllActivity() { + val client = wsClient + if (client == null || !client.isConnected()) { + logger.warn("WebSocket 未连接，无法订阅") + return + } + + try { + // 根据 real-time-data-client 的协议格式 + // 订阅消息应包含 "action": "subscribe" 和 "subscriptions" 数组 + val subscribeMessage = """ + { + "action": "subscribe", + "subscriptions": [ + { + "topic": "activity", + "type": "trades" + } + ] + } + """.trimIndent() + + client.sendMessage(subscribeMessage) + isSubscribed = true + logger.info("Activity WebSocket 订阅成功（全局交易流）") + } catch (e: Exception) { + logger.error("订阅 Activity WebSocket 失败", e) + isSubscribed = false + } + } + + /** + * 处理消息 + */ + private fun handleMessage(message: String) { + try { + // 处理 PONG 响应 + if (message.trim() == "PONG" || message.trim() == "pong") { + return + } + + // 使用扩展函数解析消息 + val tradeMessage = message.fromJson() ?: run { + // 不是有效的 JSON 或格式不匹配，跳过 + logger.warn("无法解析为 ActivityTradeMessage，可能不是 activity 消息: ${message.take(200)}") + return + } + + // 检查是否是 activity trade 消息 + if (tradeMessage.topic != "activity" || tradeMessage.type != "trades") { + // 不是我们关心的消息，直接返回 + return + } + + val payload = tradeMessage.payload + + // 提取交易者地址 + val traderAddress = extractTraderAddress(payload) ?: run { + // 没有交易者地址，跳过 + logger.warn("Activity Trade 消息中没有交易者地址: trader=${payload.trader}, proxyWallet=${payload.proxyWallet}, asset=${payload.asset}") + return + } + + // 检查是否是我们监听的 Leader + val normalizedAddress = traderAddress.lowercase() + val leaderId = monitoredAddresses[normalizedAddress] ?: run { + return + } + + // 解析交易数据 + val trade = parseActivityTrade(payload, leaderId) + if (trade != null) { + logger.info("✅ 检测到 Leader 交易: leaderId=$leaderId, address=$traderAddress, side=${trade.side}, market=${trade.market}, size=${trade.size}") + + // 异步处理交易（避免阻塞消息处理） + scope.launch { + try { + copyOrderTrackingService.processTrade( + leaderId = leaderId, + trade = trade, + source = "activity-ws" + ) + } catch (e: Exception) { + logger.error("处理 Activity WS 交易失败: leaderId=$leaderId, tradeId=${trade.id}", e) + } + } + } else { + logger.warn("解析交易数据失败: leaderId=$leaderId, address=$traderAddress, asset=${payload.asset}, side=${payload.side}") + } + } catch (e: Exception) { + logger.error("处理 Activity WebSocket 消息失败: ${e.message}", e) + } + } + + /** + * 提取交易者地址 + * 优先检查 trader.address，fallback 到 proxyWallet + */ + private fun extractTraderAddress(payload: ActivityTradePayload): String? { + // 优先检查 trader.address + val address = payload.trader?.address + ?: payload.proxyWallet + + return address + } + + /** + * 解析 Activity Trade 为 TradeResponse + */ + private fun parseActivityTrade(payload: ActivityTradePayload, leaderId: Long): TradeResponse? { + return try { + // 提取必需字段并验证 + val asset = payload.asset + val conditionId = payload.conditionId + val sideRaw = payload.side + + if (asset.isBlank() || conditionId.isBlank() || sideRaw.isBlank()) { + logger.warn("Activity Trade 消息缺少必需字段: asset=$asset, conditionId=$conditionId, side=$sideRaw") + return null + } + + val side = sideRaw.uppercase() + + // 验证 side 必须是 BUY 或 SELL + if (side != "BUY" && side != "SELL") { + logger.warn("Activity Trade 消息 side 字段无效: side=$side") + return null + } + + // price 和 size 可能是数字或字符串，统一转换为字符串 + val price = convertToString(payload.price) ?: run { + logger.warn("Activity Trade 消息 price 字段无效: ${payload.price}") + return null + } + + val size = convertToString(payload.size) ?: run { + logger.warn("Activity Trade 消息 size 字段无效: ${payload.size}") + return null + } + + // 时间戳处理：可能是秒或毫秒，可能是数字或字符串 + val timestamp = when { + payload.timestamp == null -> System.currentTimeMillis().toString() + payload.timestamp is Number -> { + val ts = (payload.timestamp as Number).toLong() + // 如果时间戳小于 1e12（秒级），转换为毫秒 + if (ts < 1e12) (ts * 1000).toString() else ts.toString() + } + + payload.timestamp is String -> { + val tsStr = payload.timestamp as String + val ts = tsStr.toLongOrNull() ?: System.currentTimeMillis() + if (ts < 1e12) (ts * 1000).toString() else tsStr + } + + else -> System.currentTimeMillis().toString() + } + + // 解析 outcome 和 outcomeIndex + // 优先使用消息中的 outcomeIndex，如果没有则从 outcome 字符串解析 + val outcome = payload.outcome + val outcomeIndex = payload.outcomeIndex + ?: parseOutcomeIndex(outcome) + + // 使用 transactionHash 作为 trade ID，如果没有则生成 fallback ID + val tradeId = payload.transactionHash ?: "${leaderId}_${System.currentTimeMillis()}_${asset.take(10)}" + + TradeResponse( + id = tradeId, + market = conditionId, + side = side, + price = price, + size = size, + timestamp = timestamp, + user = null, // Activity WS 中不需要 + outcomeIndex = outcomeIndex, + outcome = outcome + ) + } catch (e: Exception) { + logger.error("解析 Activity Trade 失败: ${e.message}", e) + null + } + } + + /** + * 将 Any 类型的值转换为 String + * 支持 Number、String、BigDecimal 等类型 + */ + private fun convertToString(value: Any?): String? { + if (value == null) return null + + return when (value) { + is String -> value + is Number -> { + // 如果是数字，转换为 BigDecimal 再转为字符串（保持精度） + try { + BigDecimal(value.toString()).toPlainString() + } catch (e: Exception) { + value.toString() + } + } + + is BigDecimal -> value.toPlainString() + else -> value.toString() + } + } + + /** + * 解析 outcome 为 outcomeIndex + */ + private fun parseOutcomeIndex(outcome: String?): Int? { + return when (outcome?.uppercase()) { + "YES", "UP", "TRUE" -> 0 + "NO", "DOWN", "FALSE" -> 1 + else -> null + } + } + + /** + * 停止监听 + */ + fun stop() { + logger.info("停止 Activity WebSocket 监听") + wsClient?.closeConnection() + wsClient = null + isSubscribed = false + monitoredAddresses.clear() + } + + /** + * 获取连接状态 + */ + fun isConnected(): Boolean { + return wsClient?.isConnected() ?: false + } + + /** + * 获取监听的 Leader 数量 + */ + fun getMonitoredCount(): Int { + return monitoredAddresses.size + } + + @PreDestroy + fun destroy() { + stop() + scope.cancel() + } +} + diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/CopyOrderTrackingService.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/CopyOrderTrackingService.kt index 9da2419..0e47913 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/CopyOrderTrackingService.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/service/copytrading/statistics/CopyOrderTrackingService.kt @@ -124,13 +124,14 @@ open class CopyOrderTrackingService( suspend fun processTrade(leaderId: Long, trade: TradeResponse, source: String): Result { // 获取该交易的 Mutex（按交易ID锁定，不同交易可以并行处理） val mutex = getMutex(leaderId, trade.id) - + logger.debug("processTrade: ${trade.id}, $source") return mutex.withLock { try { // 1. 检查是否已处理（去重，包括失败状态） val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id) if (existingProcessed != null) { + logger.debug("processTrade: 重复 ${trade.id}, $source") if (existingProcessed.status == "FAILED") { return@withLock Result.success(Unit) } @@ -269,8 +270,9 @@ open class CopyOrderTrackingService( // 如果启用了关键字过滤或市场截止时间过滤，需要先获取市场信息 var marketTitle: String? = null var marketEndDate: Long? = null - val needMarketInfo = copyTrading.keywordFilterMode != "DISABLED" || copyTrading.maxMarketEndDate != null - + val needMarketInfo = + copyTrading.keywordFilterMode != "DISABLED" || copyTrading.maxMarketEndDate != null + if (needMarketInfo) { try { val market = marketService.getMarket(trade.market) diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/util/JsonUtils.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/util/JsonUtils.kt index d7e1306..af3107c 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/util/JsonUtils.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/util/JsonUtils.kt @@ -1,34 +1,154 @@ package com.wrbug.polymarketbot.util import com.google.gson.Gson +import com.google.gson.GsonBuilder +import com.google.gson.JsonElement import com.google.gson.reflect.TypeToken +import jakarta.annotation.PostConstruct import org.springframework.stereotype.Component /** * JSON 工具类 - * 用于解析 JSON 字符串 + * 用于初始化全局 Gson 实例，供扩展函数使用 + * + * 使用 Spring 注入的 Gson Bean（已在 GsonConfig 中配置为 lenient 模式） + * 初始化后设置全局 gson 实例，供扩展函数使用 */ @Component class JsonUtils( - private val gson: Gson + private val injectedGson: Gson ) { - + + @PostConstruct + fun init() { + // 设置全局 Gson 实例，供扩展函数使用 + _gson = injectedGson + } + /** * 解析 JSON 字符串数组 * @param jsonString JSON 字符串，如 "[\"Yes\", \"No\"]" * @return 字符串列表，如果解析失败返回空列表 */ fun parseStringArray(jsonString: String?): List { - if (jsonString.isNullOrBlank()) { - return emptyList() - } - - return try { - val listType = object : TypeToken>() {}.type - gson.fromJson>(jsonString, listType) ?: emptyList() + return jsonString?.parseStringArray() ?: emptyList() + } +} + +/** + * 全局 Gson 实例（用于扩展函数） + * 在 JsonUtils 初始化时设置 + */ +@Volatile +private var _gson: Gson? = null + +/** + * 获取全局 Gson 实例 + * 如果尚未初始化，创建默认实例 + */ +val gson: Gson + get() = _gson ?: GsonBuilder().setLenient().create().also { _gson = it } + +// ============================================================================ +// 扩展函数：JSON 序列化和反序列化 +// ============================================================================ + +/** + * 将对象转换为 JSON 字符串 + * + * @return JSON 字符串，如果对象为 null 则返回空字符串 + * + * @example + * ```kotlin + * val obj = MyDataClass(name = "test", value = 123) + * val json = obj.toJson() // {"name":"test","value":123} + * ``` + */ +fun Any?.toJson(): String { + return if (this == null) { + "" + } else { + try { + gson.toJson(this) } catch (e: Exception) { - emptyList() + "" } } } +/** + * 将 JSON 字符串解析为对象（支持泛型） + * + * @return 解析后的对象，如果解析失败返回 null + * + * @example + * ```kotlin + * val json = "{\"name\":\"test\",\"value\":123}" + * val obj = json.fromJson() // MyDataClass(name="test", value=123) + * + * // 对于泛型类型，自动使用 TypeToken + * val json = "[\"a\", \"b\", \"c\"]" + * val list = json.fromJson>() // ["a", "b", "c"] + * ``` + */ +inline fun String?.fromJson(): T? { + if (this.isNullOrBlank()) { + return null + } + + return try { + val typeToken = object : TypeToken() {} + gson.fromJson(this, typeToken.type) + } catch (e: Exception) { + null + } +} + +/** + * 将 JsonElement 解析为对象（支持泛型） + * + * @return 解析后的对象，如果解析失败返回 null + * + * @example + * ```kotlin + * val jsonElement: JsonElement = ... + * val obj = jsonElement.fromJson() + * ``` + */ +inline fun JsonElement?.fromJson(): T? { + if (this == null) { + return null + } + + return try { + val typeToken = object : TypeToken() {} + gson.fromJson(this, typeToken.type) + } catch (e: Exception) { + null + } +} + +/** + * 解析 JSON 字符串数组 + * + * @return 字符串列表，如果解析失败返回空列表 + * + * @example + * ```kotlin + * val json = "[\"Yes\", \"No\"]" + * val list = json.parseStringArray() // ["Yes", "No"] + * ``` + */ +fun String?.parseStringArray(): List { + if (this.isNullOrBlank()) { + return emptyList() + } + + return try { + val listType = object : TypeToken>() {}.type + gson.fromJson>(this, listType) ?: emptyList() + } catch (e: Exception) { + emptyList() + } +} + diff --git a/backend/src/main/kotlin/com/wrbug/polymarketbot/websocket/PolymarketWebSocketClient.kt b/backend/src/main/kotlin/com/wrbug/polymarketbot/websocket/PolymarketWebSocketClient.kt index 37e9584..83a0b1a 100644 --- a/backend/src/main/kotlin/com/wrbug/polymarketbot/websocket/PolymarketWebSocketClient.kt +++ b/backend/src/main/kotlin/com/wrbug/polymarketbot/websocket/PolymarketWebSocketClient.kt @@ -139,7 +139,8 @@ class PolymarketWebSocketClient( /** * 启动 PING 保活机制 - * 根据官方文档，每 10 秒发送一次 "PING" + * 根据 @polymarket/real-time-data-client，发送小写 "ping" + * 每 10 秒发送一次 */ private fun startPing() { stopPing() // 先停止之前的 PING 任务 @@ -149,7 +150,7 @@ class PolymarketWebSocketClient( delay(10000) // 10 秒 if (isConnected) { try { - sendMessage("PING") + sendMessage("ping") // 使用小写，与官方客户端保持一致 } catch (e: Exception) { logger.warn("发送 PING 失败: $sessionId, ${e.message}") break diff --git a/backend/src/main/resources/application.properties b/backend/src/main/resources/application.properties index 39089e4..7d012a0 100644 --- a/backend/src/main/resources/application.properties +++ b/backend/src/main/resources/application.properties @@ -37,6 +37,8 @@ logging.pattern.console=%d{yyyy-MM-dd HH:mm:ss} - %msg%n # Polymarket API 配置 polymarket.clob.base-url=https://clob.polymarket.com polymarket.rtds.ws-url=wss://ws-subscriptions-clob.polymarket.com +polymarket.websocket.url=wss://ws-live-data.polymarket.com +polymarket.websocket.activity.url=${POLYMARKET_WEBSOCKET_ACTIVITY_URL:wss://ws-live-data.polymarket.com} polymarket.data-api.base-url=https://data-api.polymarket.com polymarket.gamma.base-url=https://gamma-api.polymarket.com @@ -52,6 +54,10 @@ copy.trading.polling.interval=${COPY_TRADING_POLLING_INTERVAL:2000} # 是否启用轮询，默认true copy.trading.polling.enabled=${COPY_TRADING_POLLING_ENABLED:true} +# 跟单监听配置 +# 链上 WebSocket 重连延迟（毫秒），默认3秒 +copy.trading.onchain.ws.reconnect.delay=${COPY_TRADING_ONCHAIN_WS_RECONNECT_DELAY:3000} + # WebSocket 配置 websocket.heartbeat-timeout=${WEBSOCKET_HEARTBEAT_TIMEOUT:60000} diff --git a/docs/zh/copy-trading-dual-monitoring-plan.md b/docs/zh/copy-trading-dual-monitoring-plan.md new file mode 100644 index 0000000..0e9b9f1 --- /dev/null +++ b/docs/zh/copy-trading-dual-monitoring-plan.md @@ -0,0 +1,477 @@ +# 跟单双重监听方案（OnChain + Poly Activity WS） + +## 1. 方案概述 + +### 1.1 目标 +实现 **OnChain WebSocket + Polymarket Activity WebSocket** 双重监听机制，提高跟单系统的实时性和可靠性。 + +### 1.2 架构图 + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ CopyTradingMonitorService │ +│ (主监听协调服务) │ +└────────────┬──────────────────────────────────┬──────────────────┘ + │ │ + ▼ ▼ +┌────────────────────────┐ ┌──────────────────────────────┐ +│ PolymarketActivity │ │ OnChainWsService │ +│ WsService (新增) │ │ (已存在) │ +│ │ │ │ +│ - 订阅全局 activity │ │ - 订阅 Leader 地址的链上事件 │ +│ - 客户端地址过滤 │ │ - 监听 Transfer 事件 │ +│ - 延迟 < 100ms │ │ - 延迟 ~2-3s (出块时间) │ +│ - 数据来源：Poly WS │ │ - 数据来源：链上 RPC │ +└────────────┬───────────┘ └──────────────┬───────────────┘ + │ │ + └──────────────┬──────────────────────┘ + ▼ + ┌───────────────────────┐ + │ CopyOrderTrackingService│ + │ (统一处理服务) │ + │ │ + │ - 去重（leaderId+ │ + │ tradeId） │ + │ - 处理买入/卖出 │ + │ - 记录处理状态 │ + └────────────────────────┘ +``` + +### 1.3 优势对比 + +| 特性 | Activity WS | OnChain WS | 双重监听 | +|------|-------------|------------|----------| +| **延迟** | < 100ms | ~2-3s | < 100ms (优先) | +| **可靠性** | 中等（依赖 Poly 服务） | 高（链上数据） | 高（双重保障） | +| **数据完整性** | 完整（包含 market、price、size） | 需要解析 Transfer | 完整 | +| **兜底机制** | ❌ | ✅ | ✅ | +| **去重** | 需要 | 需要 | 统一去重 | + +## 2. 技术方案 + +### 2.1 新增服务：PolymarketActivityWsService + +**职责**： +- 订阅 Polymarket Activity WebSocket（全局交易流） +- 根据 Leader 地址列表进行客户端过滤 +- 将交易事件转换为 `TradeResponse` 格式 +- 调用 `CopyOrderTrackingService.processTrade()` 处理 + +**实现要点**： +1. 使用单个 WebSocket 连接订阅全局 activity（不传 filters） +2. 客户端维护 Leader 地址 Set，实时过滤交易 +3. 支持动态添加/移除 Leader（无需重连） +4. 自动重连机制 + +### 2.2 修改 CopyTradingMonitorService + +**职责**： +- 协调两种监听服务 +- 统一管理 Leader 的添加/移除 +- 启动时同时启动两种监听 + +### 2.3 去重机制 + +**现有机制**（CopyOrderTrackingService）： +- 使用 `leaderId + trade.id` 作为唯一键 +- 基于数据库表 `processed_trade` 去重 +- 使用 Mutex 保证线程安全 + +**双重监听场景**： +- Activity WS 和 OnChain WS 可能同时检测到同一笔交易 +- 去重机制确保只处理一次（第一个到达的） +- 通过 `source` 字段区分数据源（"activity-ws" / "onchain-ws"） + +## 3. 实现细节 + +### 3.1 PolymarketActivityWsService 设计 + +```kotlin +@Service +class PolymarketActivityWsService( + private val copyOrderTrackingService: CopyOrderTrackingService, + private val leaderRepository: LeaderRepository +) { + private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + + // 单例 WebSocket 客户端 + private var wsClient: PolymarketWebSocketClient? = null + + // 要监听的 Leader 地址集合（小写） + private val monitoredAddresses = ConcurrentHashMap() // address -> leaderId + + /** + * 启动监听 + */ + fun start(leaders: List) { + monitoredAddresses.clear() + leaders.forEach { leader -> + if (leader.id != null) { + monitoredAddresses[leader.leaderAddress.lowercase()] = leader.id!! + } + } + + if (monitoredAddresses.isEmpty()) { + stop() + return + } + + connectAndSubscribe() + } + + /** + * 添加 Leader + */ + fun addLeader(leader: Leader) { + if (leader.id == null) return + monitoredAddresses[leader.leaderAddress.lowercase()] = leader.id!! + + // 如果 WebSocket 未连接，连接 + if (wsClient == null) { + connectAndSubscribe() + } + } + + /** + * 移除 Leader + */ + fun removeLeader(leaderId: Long) { + val addressToRemove = monitoredAddresses.entries + .find { it.value == leaderId }?.key + addressToRemove?.let { monitoredAddresses.remove(it) } + + // 如果没有 Leader 了，停止监听 + if (monitoredAddresses.isEmpty()) { + stop() + } + } + + /** + * 连接并订阅 + */ + private fun connectAndSubscribe() { + if (wsClient != null && wsClient!!.isConnected()) { + return + } + + wsClient = PolymarketWebSocketClient( + url = "wss://ws-live-data.polymarket.com", + sessionId = "copy-trading-activity", + onMessage = { message -> handleMessage(message) }, + onOpen = { + subscribeAllActivity() + }, + onReconnect = { + subscribeAllActivity() + } + ) + + scope.launch { + try { + wsClient!!.connect() + } catch (e: Exception) { + logger.error("连接 Activity WebSocket 失败", e) + } + } + } + + /** + * 订阅全局 activity + */ + private fun subscribeAllActivity() { + val message = """ + { + "type": "subscribe", + "channel": "activity", + "topic": "trades" + } + """.trimIndent() + + wsClient?.sendMessage(message) + } + + /** + * 处理消息 + */ + private fun handleMessage(message: String) { + try { + if (message.trim() == "PONG") return + + val json = JsonParser.parseString(message).asJsonObject + + // 检查是否是 trade 事件 + val topic = json.get("topic")?.asString + val type = json.get("type")?.asString + if (topic != "activity" || type != "trades") { + return + } + + val payload = json.getAsJsonObject("payload") ?: return + + // 提取交易者地址 + val traderObj = payload.getAsJsonObject("trader") + val traderAddress = traderObj?.get("address")?.asString?.lowercase() + ?: payload.get("proxyWallet")?.asString?.lowercase() + ?: return + + // 检查是否是我们要监听的 Leader + val leaderId = monitoredAddresses[traderAddress] ?: return + + // 解析交易数据 + val trade = parseActivityTrade(payload, leaderId) + if (trade != null) { + scope.launch { + copyOrderTrackingService.processTrade( + leaderId = leaderId, + trade = trade, + source = "activity-ws" + ) + } + } + } catch (e: Exception) { + logger.error("处理 Activity WebSocket 消息失败", e) + } + } + + /** + * 解析 Activity Trade 为 TradeResponse + */ + private fun parseActivityTrade(json: JsonObject, leaderId: Long): TradeResponse? { + return try { + TradeResponse( + id = json.get("transactionHash")?.asString + ?: "${leaderId}_${System.currentTimeMillis()}", // fallback + market = json.get("conditionId")?.asString ?: return null, + side = json.get("side")?.asString?.uppercase() ?: return null, + price = json.get("price")?.asString ?: return null, + size = json.get("size")?.asString ?: return null, + timestamp = (json.get("timestamp")?.asLong ?: System.currentTimeMillis()).toString(), + user = null, // Activity WS 中不需要 + outcomeIndex = parseOutcomeIndex(json.get("outcome")?.asString), + outcome = json.get("outcome")?.asString + ) + } catch (e: Exception) { + logger.error("解析 Activity Trade 失败", e) + null + } + } + + private fun parseOutcomeIndex(outcome: String?): Int? { + return when (outcome?.uppercase()) { + "YES", "UP", "TRUE" -> 0 + "NO", "DOWN", "FALSE" -> 1 + else -> null + } + } + + fun stop() { + wsClient?.closeConnection() + wsClient = null + monitoredAddresses.clear() + } +} +``` + +### 3.2 修改 CopyTradingMonitorService + +```kotlin +@Service +class CopyTradingMonitorService( + private val copyTradingRepository: CopyTradingRepository, + private val leaderRepository: LeaderRepository, + private val accountRepository: AccountRepository, + private val onChainWsService: OnChainWsService, + private val accountOnChainMonitorService: AccountOnChainMonitorService, + private val activityWsService: PolymarketActivityWsService // 新增 +) { + + suspend fun startMonitoring() { + val enabledCopyTradings = copyTradingRepository.findByEnabledTrue() + if (enabledCopyTradings.isEmpty()) { + return + } + + val leaderIds = enabledCopyTradings.map { it.leaderId }.distinct() + val leaders = leaderIds.mapNotNull { leaderId -> + leaderRepository.findById(leaderId).orElse(null) + } + + val accountIds = enabledCopyTradings.map { it.accountId }.distinct() + val accounts = accountIds.mapNotNull { accountId -> + accountRepository.findById(accountId).orElse(null) + } + + // 1. 启动 Activity WebSocket 监听（优先，低延迟） + activityWsService.start(leaders) + + // 2. 启动链上 WebSocket 监听（兜底，高可靠性） + onChainWsService.start(leaders) + + // 3. 启动账户监听 + accountOnChainMonitorService.start(accounts) + } + + suspend fun addLeaderMonitoring(leaderId: Long) { + val leader = leaderRepository.findById(leaderId).orElse(null) ?: return + + val copyTradings = copyTradingRepository.findByLeaderIdAndEnabledTrue(leaderId) + if (copyTradings.isEmpty()) { + return + } + + // 同时添加到两种监听 + activityWsService.addLeader(leader) + onChainWsService.addLeader(leader) + } + + suspend fun removeLeaderMonitoring(leaderId: Long) { + val copyTradings = copyTradingRepository.findByLeaderIdAndEnabledTrue(leaderId) + if (copyTradings.isNotEmpty()) { + return + } + + // 同时从两种监听移除 + activityWsService.removeLeader(leaderId) + onChainWsService.removeLeader(leaderId) + } + + @PreDestroy + fun destroy() { + scope.cancel() + activityWsService.stop() + onChainWsService.stop() + accountOnChainMonitorService.stop() + } +} +``` + +### 3.3 WebSocket 客户端复用 + +**检查现有 PolymarketWebSocketClient**： +- 如果已存在，检查是否支持 Activity 订阅 +- 如果不存在，需要创建或引入依赖 + +**依赖选项**： +1. **使用现有实现**：检查 `PolymarketWebSocketClient` 是否支持 Activity +2. **引入 poly-sdk 的方式**：参考 poly-sdk 的 `RealTimeDataClient` +3. **自行实现**：基于 OkHttp WebSocket 或 Java-WebSocket + +## 4. 数据流 + +### 4.1 Activity WS 数据流 + +``` +Polymarket 服务器 + ↓ (WebSocket推送 activity trade) +PolymarketActivityWsService.handleMessage() + ↓ (提取 trader.address) +客户端过滤（monitoredAddresses.contains(traderAddress)） + ↓ (匹配到 Leader) +parseActivityTrade() → TradeResponse + ↓ +CopyOrderTrackingService.processTrade( + leaderId, trade, source="activity-ws" +) + ↓ (去重检查) +如果未处理 → 执行跟单逻辑 +如果已处理 → 跳过（可能是 OnChain WS 先到达） +``` + +### 4.2 OnChain WS 数据流（保持不变） + +``` +链上 RPC WebSocket + ↓ (推送 Transfer 事件) +UnifiedOnChainWsService + ↓ +OnChainWsService.handleLeaderTransaction() + ↓ (解析 Transfer → Trade) +CopyOrderTrackingService.processTrade( + leaderId, trade, source="onchain-ws" +) + ↓ (去重检查) +如果未处理 → 执行跟单逻辑 +如果已处理 → 跳过（可能是 Activity WS 先到达） +``` + +## 5. 配置项 + +### 5.1 application.properties + +```properties +# Polymarket WebSocket +polymarket.websocket.url=wss://ws-live-data.polymarket.com + +# 监听策略 +copy.trading.monitor.strategy=dual +# dual: 双重监听（Activity WS + OnChain WS） +# activity-only: 仅 Activity WS +# onchain-only: 仅 OnChain WS + +# Activity WS 配置 +copy.trading.activity.ws.enabled=true +copy.trading.activity.ws.reconnect.delay=3000 + +# OnChain WS 配置（已存在） +copy.trading.onchain.ws.reconnect.delay=3000 +``` + +## 6. 监控和日志 + +### 6.1 关键指标 + +- **Activity WS 连接状态** +- **OnChain WS 连接状态** +- **每种源的交易检测数量** +- **去重次数（重复检测）** +- **处理延迟（Activity WS vs OnChain WS）** + +### 6.2 日志示例 + +``` +[INFO] Activity WS 连接成功 +[INFO] Activity WS 订阅全局 activity 成功 +[INFO] 检测到交易: leaderId=1, address=0x1234..., source=activity-ws, latency=85ms +[INFO] 检测到交易: leaderId=1, address=0x1234..., source=onchain-ws, latency=2150ms +[WARN] 交易已处理（去重）: leaderId=1, tradeId=xxx, firstSource=activity-ws, duplicateSource=onchain-ws +``` + +## 7. 实施步骤 + +### Phase 1: 基础实现 +1. ✅ 创建 `PolymarketActivityWsService` +2. ✅ 实现 Activity WebSocket 连接和订阅 +3. ✅ 实现客户端地址过滤 +4. ✅ 实现交易解析和转换 + +### Phase 2: 集成 +5. ✅ 修改 `CopyTradingMonitorService` 集成 Activity WS +6. ✅ 测试双重监听和去重机制 +7. ✅ 添加配置项支持 + +### Phase 3: 优化 +8. ✅ 添加监控指标 +9. ✅ 性能优化（连接池、消息批处理等） +10. ✅ 错误处理和降级策略 + +### Phase 4: 验证 +11. ✅ 端到端测试 +12. ✅ 压力测试 +13. ✅ 生产环境灰度验证 + +## 8. 风险与缓解 + +| 风险 | 影响 | 缓解措施 | +|------|------|----------| +| Activity WS 服务不稳定 | 高 | OnChain WS 作为兜底 | +| 消息重复处理 | 中 | 数据库去重 + Mutex 锁 | +| 连接断开频繁 | 中 | 自动重连 + 指数退避 | +| 地址过滤性能 | 低 | 使用 ConcurrentHashMap，O(1) 查询 | + +## 9. 未来优化 + +1. **消息批处理**：将多个交易合并处理，减少数据库写入 +2. **智能切换**：根据 Activity WS 稳定性自动降级到 OnChain +3. **延迟监控**：实时监控两种源的延迟，选择最优源 +4. **缓存优化**：缓存 Leader 地址列表，减少查询 + diff --git a/docs/zh/polymarket-activity-websocket-api.md b/docs/zh/polymarket-activity-websocket-api.md new file mode 100644 index 0000000..7336b25 --- /dev/null +++ b/docs/zh/polymarket-activity-websocket-api.md @@ -0,0 +1,389 @@ +# Polymarket Activity WebSocket API 格式 + +## 1. 连接信息 + +### 1.1 WebSocket URL +``` +wss://ws-live-data.polymarket.com +``` + +这是 Polymarket 官方 RTDS (Real-Time Data Stream) 的 WebSocket 端点。 + +### 1.2 协议说明 +- 使用 **官方 `@polymarket/real-time-data-client`** 的协议格式 +- 消息格式为 JSON +- 支持自动重连 +- 需要 PING/PONG 保活（每 10 秒发送 PING） + +## 2. 订阅消息格式 + +### 2.1 订阅全局 Activity（所有交易） + +**订阅消息**： +```json +{ + "action": "subscribe", + "subscriptions": [ + { + "topic": "activity", + "type": "trades" + } + ] +} +``` + +**关键点**： +- `action`: `"subscribe"` - 订阅动作（根据 `@polymarket/real-time-data-client` 协议） +- `topic`: `"activity"` - 活动数据频道 +- `type`: `"trades"` - 交易类型（在 subscriptions 数组内） +- **不传 `filters` 字段** - 订阅所有市场的交易（空对象 `{}` 会被拒绝） + +### 2.2 订阅特定市场的 Activity（可选） + +如果需要过滤特定市场： +```json +{ + "action": "subscribe", + "subscriptions": [ + { + "topic": "activity", + "type": "trades", + "filters": "{\"market_slug\":\"trump-win-2024\"}" + } + ] +} +``` + +或者过滤特定事件： +```json +{ + "action": "subscribe", + "subscriptions": [ + { + "topic": "activity", + "type": "trades", + "filters": "{\"event_slug\":\"presidential-election-2024\"}" + } + ] +} +``` + +**注意**： +- `filters` 是 JSON 字符串（不是对象） +- 使用 `snake_case`（`market_slug`, `event_slug`） +- 对于 Copy Trading，我们**不传 filters**，订阅全局然后客户端过滤 + +### 2.3 取消订阅 + +```json +{ + "action": "unsubscribe", + "subscriptions": [ + { + "topic": "activity", + "type": "trades" + } + ] +} +``` + +## 3. 接收消息格式 + +### 3.1 Trade 消息结构 + +当有交易发生时，服务器会推送如下格式的消息： + +```json +{ + "topic": "activity", + "type": "trades", + "timestamp": 1704067200000, + "payload": { + "asset": "47632033502843656213...", // Token ID (用于下单) + "conditionId": "0xb82c6573...", // Market condition ID + "eventSlug": "aus-mct-per-2025-12-28", // 事件 slug + "slug": "aus-mct-per-draw", // 市场 slug + "outcome": "No", // 结果方向 (Yes/No) + "side": "BUY", // 交易方向 (BUY/SELL) + "size": 15.72, // 交易数量 (shares) + "price": 0.87, // 交易价格 + "timestamp": 1766913243, // Unix 时间戳 (秒) + "transactionHash": "0x921936dfc9...", // 交易哈希 + "trader": { // 交易者信息对象 + "name": "gabagool22", // 交易者用户名（可选） + "address": "0x6031B6eed1C97e..." // 交易者钱包地址 ⭐ 关键字段！ + } + } +} +``` + +**重要字段说明**： +- `payload.trader.address` - **交易者钱包地址**（用于过滤 Leader） +- `payload.trader.name` - 交易者用户名（可选） +- `payload.asset` - Token ID（用于下单） +- `payload.conditionId` - Market condition ID +- `payload.side` - BUY 或 SELL +- `payload.size` - 交易数量 +- `payload.price` - 交易价格 +- `payload.timestamp` - Unix 时间戳（秒） + +### 3.2 注意：字段命名可能不同 + +根据实测（poly-sdk 文档），有些情况下字段可能在顶层： +```json +{ + "topic": "activity", + "type": "trades", + "payload": { + "asset": "...", + "conditionId": "...", + "side": "BUY", + "price": 0.87, + "size": 15.72, + "transactionHash": "...", + "name": "gabagool22", // 可能在顶层 + "proxyWallet": "0x6031..." // 可能在顶层（而不是 trader.address） + } +} +``` + +**建议处理方式**：同时检查 `payload.trader?.address` 和 `payload.proxyWallet` + +## 4. PING/PONG 保活 + +### 4.1 发送 PING + +每 10 秒发送一次： +```json +"PING" +``` + +或： +```json +{ + "type": "ping" +} +``` + +### 4.2 接收 PONG + +服务器会回复： +```json +"PONG" +``` + +或： +```json +{ + "type": "pong" +} +``` + +## 5. Kotlin 实现示例 + +### 5.1 订阅消息 + +```kotlin +fun subscribeAllActivity() { + val subscribeMessage = """ + { + "type": "subscribe", + "subscriptions": [ + { + "topic": "activity", + "type": "trades" + } + ] + } + """.trimIndent() + + wsClient.sendMessage(subscribeMessage) +} +``` + +### 5.2 解析交易消息 + +```kotlin +private fun parseActivityTrade(json: JsonObject): ActivityTrade? { + val payload = json.getAsJsonObject("payload") ?: return null + + // 提取交易者地址（优先 trader.address，fallback 到 proxyWallet） + val traderObj = payload.getAsJsonObject("trader") + val traderAddress = traderObj?.get("address")?.asString + ?: payload.get("proxyWallet")?.asString + ?: return null + + val traderName = traderObj?.get("name")?.asString + ?: payload.get("name")?.asString + + return ActivityTrade( + asset = payload.get("asset")?.asString ?: return null, + conditionId = payload.get("conditionId")?.asString ?: return null, + eventSlug = payload.get("eventSlug")?.asString, + marketSlug = payload.get("slug")?.asString, + outcome = payload.get("outcome")?.asString, + side = payload.get("side")?.asString?.uppercase() ?: return null, + size = payload.get("size")?.asDouble ?: return null, + price = payload.get("price")?.asDouble ?: return null, + timestamp = payload.get("timestamp")?.asLong + ?.let { if (it < 1e12) it * 1000 else it } // 秒转毫秒 + ?: System.currentTimeMillis(), + transactionHash = payload.get("transactionHash")?.asString, + traderAddress = traderAddress.lowercase(), + traderName = traderName + ) +} +``` + +## 6. 与现有实现的对比 + +### 6.1 User Channel（当前使用） + +**URL**: `wss://ws-subscriptions-clob.polymarket.com/ws/user` +**订阅格式**: +```json +{ + "type": "subscribe", + "channel": "user", + "user": "0x1234..." +} +``` + +**限制**： +- ❌ 只能订阅**自己的**交易（需要 API 认证） +- ❌ 无法监听别人的交易 + +### 6.2 Activity Channel（新方案） + +**URL**: `wss://ws-live-data.polymarket.com` +**订阅格式**: +```json +{ + "type": "subscribe", + "subscriptions": [ + { + "topic": "activity", + "type": "trades" + } + ] +} +``` + +**优势**： +- ✅ 可以监听**所有**交易的全局流 +- ✅ 包含交易者地址，可以客户端过滤 +- ✅ 不需要认证 +- ✅ 延迟低（< 100ms） + +## 7. 完整的消息处理流程 + +### 7.1 连接和订阅 + +```kotlin +// 1. 连接到 WebSocket +val client = PolymarketWebSocketClient( + url = "wss://ws-live-data.polymarket.com", + sessionId = "copy-trading-activity", + onMessage = { message -> handleMessage(message) }, + onOpen = { + subscribeAllActivity() + }, + onReconnect = { + subscribeAllActivity() + } +) + +client.connect() + +// 2. 发送订阅消息 +fun subscribeAllActivity() { + val message = """ + { + "type": "subscribe", + "subscriptions": [ + { + "topic": "activity", + "type": "trades" + } + ] + } + """.trimIndent() + client.sendMessage(message) +} +``` + +### 7.2 处理消息 + +```kotlin +private fun handleMessage(message: String) { + if (message.trim() == "PONG") return + + val json = JsonParser.parseString(message).asJsonObject + + // 检查是否是 activity trade 消息 + val topic = json.get("topic")?.asString + val type = json.get("type")?.asString + + if (topic == "activity" && type == "trades") { + val payload = json.getAsJsonObject("payload") ?: return + + // 提取交易者地址 + val traderAddress = extractTraderAddress(payload) ?: return + + // 检查是否是我们监听的 Leader + val leaderId = monitoredAddresses[traderAddress.lowercase()] ?: return + + // 解析交易 + val trade = parseActivityTrade(payload) ?: return + + // 处理交易 + processTrade(leaderId, trade) + } +} + +private fun extractTraderAddress(payload: JsonObject): String? { + // 优先检查 trader.address + val traderObj = payload.getAsJsonObject("trader") + val address = traderObj?.get("address")?.asString + ?: payload.get("proxyWallet")?.asString + + return address?.lowercase() +} +``` + +## 8. 错误处理 + +### 8.1 连接错误 + +- 连接失败时自动重连（指数退避：3s → 6s → 12s → 最大 60s） +- 重连后重新发送订阅消息 + +### 8.2 订阅错误 + +如果订阅失败，服务器可能返回： +```json +{ + "type": "error", + "message": "Invalid subscription" +} +``` + +### 8.3 消息解析错误 + +- 如果字段缺失，记录警告日志并跳过 +- 如果地址格式错误，跳过该消息 +- 如果解析失败，记录错误但不中断连接 + +## 9. 性能考虑 + +### 9.1 消息频率 + +- Activity WebSocket 会推送**所有市场**的所有交易 +- 高频市场可能每秒收到数百条消息 +- 需要高效的客户端过滤（使用 `Set` 或 `Map`，O(1) 查找） + +### 9.2 内存管理 + +- 维护 Leader 地址 Set（通常 < 100 个地址） +- 消息处理使用协程异步处理，避免阻塞 +- 定期清理不再监听的地址 diff --git a/frontend/src/pages/CopyTradingOrders/AddModal.tsx b/frontend/src/pages/CopyTradingOrders/AddModal.tsx index c4f8c07..5cd3dbe 100644 --- a/frontend/src/pages/CopyTradingOrders/AddModal.tsx +++ b/frontend/src/pages/CopyTradingOrders/AddModal.tsx @@ -813,9 +813,9 @@ const AddModal: React.FC = ({ style={{ width: '60%' }} placeholder={t('copyTradingAdd.maxMarketEndDatePlaceholder') || '输入时间值（可选）'} parser={(value) => { - if (!value) return '' + if (!value) return 0 const num = parseInt(value.replace(/\D/g, ''), 10) - return isNaN(num) ? '' : num.toString() + return isNaN(num) ? 0 : num }} formatter={(value) => { if (!value && value !== 0) return '' diff --git a/frontend/src/pages/CopyTradingOrders/EditModal.tsx b/frontend/src/pages/CopyTradingOrders/EditModal.tsx index 1836f28..a4a67bb 100644 --- a/frontend/src/pages/CopyTradingOrders/EditModal.tsx +++ b/frontend/src/pages/CopyTradingOrders/EditModal.tsx @@ -731,9 +731,9 @@ const EditModal: React.FC = ({ style={{ width: '60%' }} placeholder={t('copyTradingEdit.maxMarketEndDatePlaceholder') || '输入时间值（可选）'} parser={(value) => { - if (!value) return '' + if (!value) return 0 const num = parseInt(value.replace(/\D/g, ''), 10) - return isNaN(num) ? '' : num.toString() + return isNaN(num) ? 0 : num }} formatter={(value) => { if (!value && value !== 0) return ''