feat: 实现链上 WebSocket 监听和并行监控策略
- 创建 OnChainWsService，支持通过 Polygon RPC eth_subscribe 实时监听链上交易 - 实现并行监控策略：链上 WS 和轮询同时运行，哪个数据先返回用哪个 - 支持通过 eth_unsubscribe 取消单个 Leader 的订阅 - 在 CopyOrderTrackingService 中添加 Mutex 保证线程安全 - 实现链上交易解析逻辑（USDC Transfer + ERC1155 Transfer） - 实现 Gamma API 元数据补齐逻辑 - 优化 WebSocket 连接管理：只创建一个连接，没有跟单配置时自动取消 - 跟单配置生效/失效时及时更新 WebSocket 订阅 - 使用 Gson 替换所有 JSON 解析，JsonRpcResponse.result 使用 JsonElement 类型 - 添加相关文档：copy-trading-logic-summary.md 和 copy-trading-monitor-strategy.md
This commit is contained in:
WrBug
@@ -1,5 +1,6 @@
|
||||
package com.wrbug.polymarketbot.api
|
||||
|
||||
import com.google.gson.JsonElement
|
||||
import retrofit2.Response
|
||||
import retrofit2.http.Body
|
||||
import retrofit2.http.POST
|
||||
@@ -29,10 +30,11 @@ data class JsonRpcRequest(
|
||||
|
||||
/**
|
||||
* JSON-RPC 响应
|
||||
* 使用 JsonElement 类型处理 result 字段，可以灵活处理字符串、对象、数组等类型
|
||||
*/
|
||||
data class JsonRpcResponse(
|
||||
val jsonrpc: String? = null,
|
||||
val result: String? = null,
|
||||
val result: JsonElement? = null, // 使用 JsonElement 类型，可以处理任意 JSON 类型
|
||||
val error: JsonRpcError? = null,
|
||||
val id: Int? = null
|
||||
)
|
||||
|
||||
+28
-10
@@ -117,7 +117,9 @@ class BlockchainService(
|
||||
throw Exception("RPC 错误: ${rpcResponse.error.message}")
|
||||
}
|
||||
|
||||
val hexResult = rpcResponse.result ?: throw Exception("RPC 响应格式错误: result 为空")
|
||||
// 使用 Gson 解析 result（JsonElement）
|
||||
val hexResult = rpcResponse.result?.asString
|
||||
?: throw Exception("RPC 响应格式错误: result 为空")
|
||||
|
||||
// 解析代理地址
|
||||
val proxyAddress = EthereumUtils.decodeAddress(hexResult)
|
||||
@@ -193,7 +195,9 @@ class BlockchainService(
|
||||
throw Exception("RPC 错误: ${rpcResponse.error.message}")
|
||||
}
|
||||
|
||||
val hexBalance = rpcResponse.result ?: throw Exception("RPC 响应格式错误: result 为空")
|
||||
// 使用 Gson 解析 result（JsonElement）
|
||||
val hexBalance = rpcResponse.result?.asString
|
||||
?: throw Exception("RPC 响应格式错误: result 为空")
|
||||
|
||||
// 将十六进制转换为 BigDecimal（USDC 有 6 位小数）
|
||||
val balanceWei = BigInteger(hexBalance.removePrefix("0x"), 16)
|
||||
@@ -288,7 +292,9 @@ class BlockchainService(
|
||||
return Result.failure(Exception("调用 getCollectionId 失败: ${collectionIdResult.error}"))
|
||||
}
|
||||
|
||||
val collectionId = collectionIdResult.result ?: return Result.failure(Exception("getCollectionId 返回结果为空"))
|
||||
// 使用 Gson 解析 result（JsonElement）
|
||||
val collectionId = collectionIdResult.result?.asString
|
||||
?: return Result.failure(Exception("getCollectionId 返回结果为空"))
|
||||
|
||||
// 2. 调用 getPositionId(collateralToken, collectionId)
|
||||
val getPositionIdSelector = EthereumUtils.getFunctionSelector("getPositionId(address,bytes32)")
|
||||
@@ -318,7 +324,9 @@ class BlockchainService(
|
||||
return Result.failure(Exception("调用 getPositionId 失败: ${positionIdResult.error}"))
|
||||
}
|
||||
|
||||
val tokenId = positionIdResult.result ?: return Result.failure(Exception("getPositionId 返回结果为空"))
|
||||
// 使用 Gson 解析 result（JsonElement）
|
||||
val tokenId = positionIdResult.result?.asString
|
||||
?: return Result.failure(Exception("getPositionId 返回结果为空"))
|
||||
val tokenIdBigInt = EthereumUtils.decodeUint256(tokenId)
|
||||
|
||||
Result.success(tokenIdBigInt.toString())
|
||||
@@ -454,7 +462,9 @@ class BlockchainService(
|
||||
return Result.failure(Exception("获取 Proxy nonce 失败: ${rpcResponse.error.message}"))
|
||||
}
|
||||
|
||||
val hexNonce = rpcResponse.result ?: return Result.failure(Exception("Proxy nonce 结果为空"))
|
||||
// 使用 Gson 解析 result（JsonElement）
|
||||
val hexNonce = rpcResponse.result?.asString
|
||||
?: return Result.failure(Exception("Proxy nonce 结果为空"))
|
||||
val nonce = EthereumUtils.decodeUint256(hexNonce)
|
||||
return Result.success(nonce)
|
||||
}
|
||||
@@ -482,7 +492,9 @@ class BlockchainService(
|
||||
return Result.failure(Exception("获取 nonce 失败: ${rpcResponse.error.message}"))
|
||||
}
|
||||
|
||||
val hexNonce = rpcResponse.result ?: return Result.failure(Exception("nonce 结果为空"))
|
||||
// 使用 Gson 解析 result（JsonElement）
|
||||
val hexNonce = rpcResponse.result?.asString
|
||||
?: return Result.failure(Exception("nonce 结果为空"))
|
||||
val nonce = EthereumUtils.decodeUint256(hexNonce)
|
||||
return Result.success(nonce)
|
||||
}
|
||||
@@ -508,7 +520,9 @@ class BlockchainService(
|
||||
return Result.failure(Exception("获取 gas price 失败: ${rpcResponse.error.message}"))
|
||||
}
|
||||
|
||||
val hexGasPrice = rpcResponse.result ?: return Result.failure(Exception("gas price 结果为空"))
|
||||
// 使用 Gson 解析 result（JsonElement）
|
||||
val hexGasPrice = rpcResponse.result?.asString
|
||||
?: return Result.failure(Exception("gas price 结果为空"))
|
||||
val gasPrice = EthereumUtils.decodeUint256(hexGasPrice)
|
||||
return Result.success(gasPrice)
|
||||
}
|
||||
@@ -582,7 +596,9 @@ class BlockchainService(
|
||||
return Result.failure(Exception("发送交易失败: ${rpcResponse.error.message}"))
|
||||
}
|
||||
|
||||
val txHash = rpcResponse.result ?: return Result.failure(Exception("交易哈希为空"))
|
||||
// 使用 Gson 解析 result（JsonElement）
|
||||
val txHash = rpcResponse.result?.asString
|
||||
?: return Result.failure(Exception("交易哈希为空"))
|
||||
return Result.success(txHash)
|
||||
}
|
||||
|
||||
@@ -611,7 +627,9 @@ class BlockchainService(
|
||||
return Result.failure(Exception("查询交易失败: ${txRpcResponse.error.message}"))
|
||||
}
|
||||
|
||||
val txResult = txRpcResponse.result ?: return Result.failure(Exception("交易结果为空"))
|
||||
// 使用 Gson 解析 result（JsonElement）
|
||||
val txResult = txRpcResponse.result?.toString()
|
||||
?: return Result.failure(Exception("交易结果为空"))
|
||||
|
||||
// 查询交易回执（包含内部调用和事件日志）
|
||||
val receiptRequest = JsonRpcRequest(
|
||||
@@ -628,7 +646,7 @@ class BlockchainService(
|
||||
val receiptResult = if (receiptRpcResponse.error != null) {
|
||||
"交易回执查询失败: ${receiptRpcResponse.error.message}"
|
||||
} else {
|
||||
receiptRpcResponse.result ?: "交易回执为空（可能还在打包中）"
|
||||
receiptRpcResponse.result?.toString() ?: "交易回执为空（可能还在打包中）"
|
||||
}
|
||||
|
||||
Result.success("交易信息:\n$txResult\n\n交易回执:\n$receiptResult")
|
||||
|
||||
+10
-9
@@ -151,13 +151,13 @@ class CopyTradingService(
|
||||
|
||||
val saved = copyTradingRepository.save(copyTrading)
|
||||
|
||||
// 如果跟单已启用，重新启动监听（确保状态完全同步）
|
||||
// 如果跟单已启用，更新 Leader 监听（增量更新，不重启所有监听）
|
||||
if (saved.enabled) {
|
||||
kotlinx.coroutines.runBlocking {
|
||||
try {
|
||||
monitorService.restartMonitoring()
|
||||
monitorService.updateLeaderMonitoring(saved.leaderId)
|
||||
} catch (e: Exception) {
|
||||
logger.error("重新启动跟单监听失败", e)
|
||||
logger.error("更新 Leader 监听失败", e)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -219,12 +219,12 @@ class CopyTradingService(
|
||||
|
||||
val saved = copyTradingRepository.save(updated)
|
||||
|
||||
// 重新启动监听（确保状态完全同步）
|
||||
// 更新 Leader 监听（增量更新，根据 enabled 状态决定添加或移除）
|
||||
kotlinx.coroutines.runBlocking {
|
||||
try {
|
||||
monitorService.restartMonitoring()
|
||||
monitorService.updateLeaderMonitoring(saved.leaderId)
|
||||
} catch (e: Exception) {
|
||||
logger.error("重新启动跟单监听失败", e)
|
||||
logger.error("更新 Leader 监听失败", e)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -322,14 +322,15 @@ class CopyTradingService(
|
||||
val copyTrading = copyTradingRepository.findById(copyTradingId).orElse(null)
|
||||
?: return Result.failure(IllegalArgumentException("跟单配置不存在"))
|
||||
|
||||
val leaderId = copyTrading.leaderId
|
||||
copyTradingRepository.delete(copyTrading)
|
||||
|
||||
// 重新启动监听（确保状态完全同步）
|
||||
// 更新 Leader 监听（检查该 Leader 是否还有其他启用的跟单配置）
|
||||
kotlinx.coroutines.runBlocking {
|
||||
try {
|
||||
monitorService.restartMonitoring()
|
||||
monitorService.removeLeaderMonitoring(leaderId)
|
||||
} catch (e: Exception) {
|
||||
logger.error("重新启动跟单监听失败", e)
|
||||
logger.error("更新 Leader 监听失败", e)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+42
-10
@@ -12,14 +12,15 @@ import org.springframework.stereotype.Service
|
||||
|
||||
/**
|
||||
* 跟单监听服务（主服务）
|
||||
* 管理所有Leader的交易监听（使用轮询方式）
|
||||
* 注意：WebSocket 需要认证才能订阅其他用户的交易，因此只使用轮询方式
|
||||
* 管理所有Leader的交易监听
|
||||
* 同时运行链上 WebSocket 监听和轮询监听（并行处理）
|
||||
*/
|
||||
@Service
|
||||
class CopyTradingMonitorService(
|
||||
private val copyTradingRepository: CopyTradingRepository,
|
||||
private val leaderRepository: LeaderRepository,
|
||||
private val pollingService: CopyTradingPollingService
|
||||
private val pollingService: CopyTradingPollingService,
|
||||
private val onChainWsService: OnChainWsService
|
||||
) {
|
||||
|
||||
private val logger = LoggerFactory.getLogger(CopyTradingMonitorService::class.java)
|
||||
@@ -46,12 +47,14 @@ class CopyTradingMonitorService(
|
||||
@PreDestroy
|
||||
fun destroy() {
|
||||
scope.cancel()
|
||||
// 只使用轮询，不使用WebSocket
|
||||
// 停止轮询和链上 WS 监听
|
||||
pollingService.stop()
|
||||
onChainWsService.stop()
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动监听
|
||||
* 同时启动链上 WebSocket 监听和轮询监听（并行运行）
|
||||
*/
|
||||
suspend fun startMonitoring() {
|
||||
// 1. 获取所有启用的跟单关系
|
||||
@@ -67,14 +70,17 @@ class CopyTradingMonitorService(
|
||||
leaderRepository.findById(leaderId).orElse(null)
|
||||
}
|
||||
|
||||
// 3. 同时启动链上 WebSocket 监听和轮询监听（并行运行）
|
||||
// 链上 WS 监听（实时，秒级延迟）
|
||||
onChainWsService.start(leaders)
|
||||
|
||||
// 3. 启动轮询监听（使用 /activity 接口，不需要认证）
|
||||
// 注意：WebSocket 需要认证才能订阅其他用户的交易，因此禁用WebSocket，只使用轮询
|
||||
// 轮询监听（延迟，2秒间隔，作为备份）
|
||||
pollingService.start(leaders)
|
||||
}
|
||||
|
||||
/**
|
||||
* 添加Leader监听（当创建新的跟单关系时调用）
|
||||
* 如果 Leader 已经在监听列表中，不重复添加
|
||||
*/
|
||||
suspend fun addLeaderMonitoring(leaderId: Long) {
|
||||
val leader = leaderRepository.findById(leaderId).orElse(null)
|
||||
@@ -85,28 +91,54 @@ class CopyTradingMonitorService(
|
||||
return
|
||||
}
|
||||
|
||||
// 只使用轮询，不使用WebSocket（需要认证）
|
||||
// 同时添加到链上 WS 监听和轮询监听（如果不在列表中才添加）
|
||||
onChainWsService.addLeader(leader)
|
||||
pollingService.addLeader(leader)
|
||||
}
|
||||
|
||||
/**
|
||||
* 移除Leader监听（当删除跟单关系时调用）
|
||||
* 移除Leader监听（当删除跟单关系或禁用时调用）
|
||||
* 检查该 Leader 是否还有其他启用的跟单配置
|
||||
*/
|
||||
suspend fun removeLeaderMonitoring(leaderId: Long) {
|
||||
val copyTradings = copyTradingRepository.findByLeaderIdAndEnabledTrue(leaderId)
|
||||
// 如果还有启用的跟单配置，不移除监听
|
||||
if (copyTradings.isNotEmpty()) {
|
||||
return
|
||||
}
|
||||
|
||||
// 只使用轮询，不使用WebSocket
|
||||
// 没有启用的跟单配置了，移除监听
|
||||
onChainWsService.removeLeader(leaderId)
|
||||
pollingService.removeLeader(leaderId)
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新Leader监听（当跟单配置状态改变时调用）
|
||||
* 根据当前状态决定添加或移除监听
|
||||
*/
|
||||
suspend fun updateLeaderMonitoring(leaderId: Long) {
|
||||
val copyTradings = copyTradingRepository.findByLeaderIdAndEnabledTrue(leaderId)
|
||||
val leader = leaderRepository.findById(leaderId).orElse(null)
|
||||
?: return
|
||||
|
||||
if (copyTradings.isNotEmpty()) {
|
||||
// 有启用的跟单配置，确保在监听列表中
|
||||
onChainWsService.addLeader(leader)
|
||||
pollingService.addLeader(leader)
|
||||
} else {
|
||||
// 没有启用的跟单配置，移除监听
|
||||
onChainWsService.removeLeader(leaderId)
|
||||
pollingService.removeLeader(leaderId)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 重新启动监听（当跟单关系状态改变时调用）
|
||||
* 注意：这个方法会停止所有监听并重新启动，建议使用 updateLeaderMonitoring 进行增量更新
|
||||
*/
|
||||
suspend fun restartMonitoring() {
|
||||
// 只使用轮询，不使用WebSocket
|
||||
// 停止所有监听
|
||||
onChainWsService.stop()
|
||||
pollingService.stop()
|
||||
delay(1000) // 等待1秒
|
||||
startMonitoring()
|
||||
|
||||
+925
@@ -0,0 +1,925 @@
|
||||
package com.wrbug.polymarketbot.service.copytrading.monitor
|
||||
|
||||
import com.google.gson.Gson
|
||||
import com.google.gson.JsonArray
|
||||
import com.google.gson.JsonObject
|
||||
import com.google.gson.reflect.TypeToken
|
||||
import com.wrbug.polymarketbot.api.*
|
||||
import com.wrbug.polymarketbot.entity.Leader
|
||||
import com.wrbug.polymarketbot.repository.LeaderRepository
|
||||
import com.wrbug.polymarketbot.service.copytrading.statistics.CopyOrderTrackingService
|
||||
import com.wrbug.polymarketbot.service.system.RpcNodeService
|
||||
import com.wrbug.polymarketbot.util.RetrofitFactory
|
||||
import com.wrbug.polymarketbot.util.createClient
|
||||
import com.wrbug.polymarketbot.util.getProxyConfig
|
||||
import jakarta.annotation.PreDestroy
|
||||
import kotlinx.coroutines.*
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Request
|
||||
import okhttp3.WebSocket
|
||||
import okhttp3.WebSocketListener
|
||||
import okio.ByteString
|
||||
import org.slf4j.LoggerFactory
|
||||
import org.springframework.beans.factory.annotation.Value
|
||||
import org.springframework.stereotype.Service
|
||||
import java.math.BigInteger
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
* 链上 WebSocket 监听服务
|
||||
* 通过 Polygon RPC 的 eth_subscribe 实时监听链上交易
|
||||
*/
|
||||
@Service
|
||||
class OnChainWsService(
|
||||
private val rpcNodeService: RpcNodeService,
|
||||
private val retrofitFactory: RetrofitFactory,
|
||||
private val copyOrderTrackingService: CopyOrderTrackingService,
|
||||
private val leaderRepository: LeaderRepository
|
||||
) {
|
||||
|
||||
private val logger = LoggerFactory.getLogger(OnChainWsService::class.java)
|
||||
|
||||
// Gson 实例，用于解析 JSON
|
||||
private val gson = Gson()
|
||||
|
||||
@Value("\${copy.trading.onchain.ws.reconnect.delay:3000}")
|
||||
private var reconnectDelay: Long = 3000 // 重连延迟（毫秒），默认3秒
|
||||
|
||||
private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
|
||||
|
||||
// 存储需要监听的Leader：leaderId -> Leader
|
||||
private val monitoredLeaders = ConcurrentHashMap<Long, Leader>()
|
||||
|
||||
// 存储每个 Leader 的订阅 ID：leaderId -> List<subscriptionId>
|
||||
// 每个 Leader 有 6 个订阅：USDC from/to, ERC1155 TransferSingle from/to, ERC1155 TransferBatch from/to
|
||||
private val leaderSubscriptions = ConcurrentHashMap<Long, MutableList<String>>()
|
||||
|
||||
// 存储请求 ID 到 Leader ID 的映射：requestId -> leaderId
|
||||
// 用于在收到订阅响应时，将 subscription ID 关联到对应的 Leader
|
||||
private val requestIdToLeaderId = ConcurrentHashMap<Int, Long>()
|
||||
|
||||
// WebSocket 连接
|
||||
private var webSocket: WebSocket? = null
|
||||
@Volatile
|
||||
private var isConnected = false
|
||||
|
||||
// 订阅ID计数器（用于请求 ID）
|
||||
private var requestIdCounter = 0
|
||||
|
||||
// 合约地址
|
||||
companion object {
|
||||
private const val USDC_CONTRACT = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174"
|
||||
private const val ERC1155_CONTRACT = "0x4d97dcd97ec945f40cf65f87097ace5ea0476045"
|
||||
private const val ERC20_TRANSFER_TOPIC = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
|
||||
private const val ERC1155_TRANSFER_SINGLE_TOPIC = "0xc3d58168c5ae7397731d063d5bbf3d657854427343f4c083240f7aacaa2d0f62"
|
||||
private const val ERC1155_TRANSFER_BATCH_TOPIC = "0x4a39dc06d4c0dbc64b70af90fd698a233a518aa5d07e595d983b8c0526c8f7fb"
|
||||
}
|
||||
|
||||
// 连接任务（确保只有一个连接任务在运行）
|
||||
private var connectionJob: Job? = null
|
||||
|
||||
/**
|
||||
* 启动链上 WebSocket 监听
|
||||
* 只创建一个 WebSocket 连接，为所有 Leader 订阅
|
||||
*/
|
||||
fun start(leaders: List<Leader>) {
|
||||
// 如果没有 Leader，不启动连接
|
||||
if (leaders.isEmpty()) {
|
||||
logger.info("没有需要监听的 Leader，不启动链上 WebSocket 连接")
|
||||
stop()
|
||||
return
|
||||
}
|
||||
|
||||
// 如果连接任务已经在运行，先停止旧任务
|
||||
if (connectionJob != null && connectionJob!!.isActive) {
|
||||
logger.info("停止旧的连接任务，准备重新启动")
|
||||
connectionJob?.cancel()
|
||||
connectionJob = null
|
||||
// 关闭旧连接
|
||||
webSocket?.close(1000, "重新启动")
|
||||
webSocket = null
|
||||
isConnected = false
|
||||
}
|
||||
|
||||
// 更新 Leader 列表
|
||||
monitoredLeaders.clear()
|
||||
leaders.forEach { leader ->
|
||||
addLeader(leader)
|
||||
}
|
||||
|
||||
// 启动连接任务（只创建一个）
|
||||
connectionJob = scope.launch {
|
||||
startConnection()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 添加Leader监听
|
||||
* 如果 Leader 已经在监听列表中，不重复添加
|
||||
* 如果已连接，立即订阅
|
||||
*/
|
||||
fun addLeader(leader: Leader) {
|
||||
if (leader.id == null) {
|
||||
logger.warn("Leader ID为空，跳过: ${leader.leaderAddress}")
|
||||
return
|
||||
}
|
||||
|
||||
val leaderId = leader.id!!
|
||||
|
||||
// 如果已经在监听列表中，不重复添加
|
||||
if (monitoredLeaders.containsKey(leaderId)) {
|
||||
logger.debug("Leader 已在监听列表中: ${leader.leaderName} (${leader.leaderAddress})")
|
||||
return
|
||||
}
|
||||
|
||||
monitoredLeaders[leaderId] = leader
|
||||
logger.info("添加 Leader 监听: ${leader.leaderName} (${leader.leaderAddress})")
|
||||
|
||||
// 如果已连接，立即订阅
|
||||
if (isConnected && webSocket != null) {
|
||||
scope.launch {
|
||||
subscribeLeader(leader)
|
||||
}
|
||||
} else {
|
||||
// 如果未连接，启动连接（如果连接任务未运行）
|
||||
if (connectionJob == null || !connectionJob!!.isActive) {
|
||||
connectionJob = scope.launch {
|
||||
startConnection()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 移除Leader监听
|
||||
* 通过 eth_unsubscribe 取消该 Leader 的所有订阅
|
||||
* 如果没有 Leader 了，关闭 WebSocket 连接
|
||||
*/
|
||||
fun removeLeader(leaderId: Long) {
|
||||
val leader = monitoredLeaders.remove(leaderId)
|
||||
if (leader != null) {
|
||||
logger.info("移除 Leader 监听: ${leader.leaderName} (${leader.leaderAddress})")
|
||||
}
|
||||
|
||||
// 取消该 Leader 的所有订阅
|
||||
val subscriptions = leaderSubscriptions.remove(leaderId)
|
||||
if (subscriptions != null && subscriptions.isNotEmpty() && isConnected && webSocket != null) {
|
||||
logger.info("取消 Leader ${leader?.leaderName} 的 ${subscriptions.size} 个订阅")
|
||||
subscriptions.forEach { subscriptionId ->
|
||||
unsubscribe(subscriptionId)
|
||||
}
|
||||
}
|
||||
|
||||
// 如果没有 Leader 了，关闭连接
|
||||
if (monitoredLeaders.isEmpty()) {
|
||||
logger.info("没有需要监听的 Leader，关闭链上 WebSocket 连接")
|
||||
stop()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 停止所有监听
|
||||
*/
|
||||
fun stop() {
|
||||
connectionJob?.cancel()
|
||||
connectionJob = null
|
||||
|
||||
// 取消所有订阅
|
||||
if (isConnected && webSocket != null) {
|
||||
leaderSubscriptions.values.flatten().forEach { subscriptionId ->
|
||||
unsubscribe(subscriptionId)
|
||||
}
|
||||
}
|
||||
|
||||
webSocket?.close(1000, "正常关闭")
|
||||
webSocket = null
|
||||
isConnected = false
|
||||
monitoredLeaders.clear()
|
||||
leaderSubscriptions.clear()
|
||||
requestIdToLeaderId.clear()
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动连接（带重连机制）
|
||||
* 只创建一个 WebSocket 连接
|
||||
*/
|
||||
private suspend fun startConnection() {
|
||||
while (scope.isActive) {
|
||||
try {
|
||||
// 检查是否有需要监听的 Leader
|
||||
if (monitoredLeaders.isEmpty()) {
|
||||
logger.info("没有需要监听的 Leader，停止连接")
|
||||
// 确保关闭连接
|
||||
webSocket?.close(1000, "没有 Leader")
|
||||
webSocket = null
|
||||
isConnected = false
|
||||
break
|
||||
}
|
||||
|
||||
// 如果已经连接，不需要重新连接
|
||||
if (isConnected && webSocket != null) {
|
||||
// 等待连接断开
|
||||
waitForDisconnect()
|
||||
// 连接断开后继续重连循环
|
||||
continue
|
||||
}
|
||||
|
||||
// 从后台配置获取 WS RPC URL
|
||||
val wsUrl = rpcNodeService.getWsUrl()
|
||||
val httpUrl = rpcNodeService.getHttpUrl()
|
||||
|
||||
logger.info("连接链上 WebSocket: $wsUrl (监听 ${monitoredLeaders.size} 个 Leader)")
|
||||
|
||||
// 创建 HTTP 客户端（用于 RPC 调用）
|
||||
val httpClient = createHttpClient()
|
||||
|
||||
// 创建 RPC API 客户端
|
||||
val rpcApi = retrofitFactory.createEthereumRpcApi(httpUrl)
|
||||
|
||||
// 连接 WebSocket（只创建一个连接，会先关闭旧连接）
|
||||
connectWebSocket(wsUrl, httpClient, rpcApi)
|
||||
|
||||
// 等待连接建立（最多等待 15 秒）
|
||||
// 注意：onOpen 回调是异步的，需要等待一段时间
|
||||
var waitCount = 0
|
||||
val maxWait = 15 // 最多等待 15 秒
|
||||
while (!isConnected && waitCount < maxWait && scope.isActive) {
|
||||
delay(1000)
|
||||
waitCount++
|
||||
// 每 3 秒打印一次日志，方便调试
|
||||
if (waitCount % 3 == 0) {
|
||||
logger.debug("等待 WebSocket 连接建立... (${waitCount}/${maxWait}秒)")
|
||||
}
|
||||
}
|
||||
|
||||
// 检查连接状态（同时检查 isConnected 和 webSocket 状态）
|
||||
val actuallyConnected = isConnected && webSocket != null
|
||||
|
||||
// 如果连接失败，等待重连延迟后继续
|
||||
if (!actuallyConnected) {
|
||||
logger.warn("WebSocket 连接超时或失败: isConnected=$isConnected, webSocket=${webSocket != null}, 等待重连")
|
||||
delay(reconnectDelay)
|
||||
continue
|
||||
}
|
||||
|
||||
logger.info("WebSocket 连接已建立，开始监听")
|
||||
|
||||
// 连接成功后持续监听
|
||||
waitForDisconnect()
|
||||
|
||||
// 连接断开后，如果没有 Leader 了，不再重连
|
||||
if (monitoredLeaders.isEmpty()) {
|
||||
logger.info("没有需要监听的 Leader，停止重连")
|
||||
break
|
||||
}
|
||||
|
||||
// 连接断开后，等待一下再重连（避免立即重连）
|
||||
logger.info("WebSocket 连接断开，等待 ${reconnectDelay}ms 后重连")
|
||||
delay(reconnectDelay)
|
||||
} catch (e: Exception) {
|
||||
// 如果没有 Leader 了，不再重连
|
||||
if (monitoredLeaders.isEmpty()) {
|
||||
logger.info("没有需要监听的 Leader，停止重连")
|
||||
// 确保关闭连接
|
||||
webSocket?.close(1000, "没有 Leader")
|
||||
webSocket = null
|
||||
isConnected = false
|
||||
break
|
||||
}
|
||||
logger.warn("链上 WebSocket 连接失败，等待重连: ${e.message}")
|
||||
// 确保关闭旧连接
|
||||
webSocket?.close(1000, "重连前关闭")
|
||||
webSocket = null
|
||||
isConnected = false
|
||||
delay(reconnectDelay)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建 HTTP 客户端
|
||||
*/
|
||||
private fun createHttpClient(): OkHttpClient {
|
||||
val proxy = getProxyConfig()
|
||||
val builder = createClient()
|
||||
|
||||
if (proxy != null) {
|
||||
builder.proxy(proxy)
|
||||
}
|
||||
|
||||
return builder.build()
|
||||
}
|
||||
|
||||
/**
|
||||
* 连接 WebSocket
|
||||
* 确保只创建一个连接，创建新连接前先关闭旧连接
|
||||
*/
|
||||
private fun connectWebSocket(wsUrl: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) {
|
||||
// 先关闭旧连接（如果存在）
|
||||
val oldWebSocket = webSocket
|
||||
if (oldWebSocket != null) {
|
||||
try {
|
||||
oldWebSocket.close(1000, "重新连接")
|
||||
} catch (e: Exception) {
|
||||
logger.debug("关闭旧 WebSocket 连接时出错: ${e.message}")
|
||||
}
|
||||
}
|
||||
webSocket = null
|
||||
isConnected = false
|
||||
|
||||
val request = Request.Builder()
|
||||
.url(wsUrl)
|
||||
.build()
|
||||
|
||||
// 创建新连接（只创建一个）
|
||||
webSocket = httpClient.newWebSocket(request, object : WebSocketListener() {
|
||||
override fun onOpen(webSocket: WebSocket, response: okhttp3.Response) {
|
||||
isConnected = true
|
||||
logger.info("链上 WebSocket 连接成功")
|
||||
|
||||
// 订阅所有 Leader
|
||||
scope.launch {
|
||||
monitoredLeaders.values.forEach { leader ->
|
||||
subscribeLeader(leader)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override fun onMessage(webSocket: WebSocket, text: String) {
|
||||
scope.launch {
|
||||
handleMessage(text, httpClient, rpcApi)
|
||||
}
|
||||
}
|
||||
|
||||
override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
|
||||
scope.launch {
|
||||
handleMessage(bytes.utf8(), httpClient, rpcApi)
|
||||
}
|
||||
}
|
||||
|
||||
override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
|
||||
isConnected = false
|
||||
logger.warn("链上 WebSocket 连接关闭: code=$code, reason=$reason")
|
||||
}
|
||||
|
||||
override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
|
||||
isConnected = false
|
||||
logger.warn("链上 WebSocket 连接已关闭: code=$code, reason=$reason")
|
||||
}
|
||||
|
||||
override fun onFailure(webSocket: WebSocket, t: Throwable, response: okhttp3.Response?) {
|
||||
logger.error("链上 WebSocket 连接失败: ${t.message}", t)
|
||||
isConnected = false
|
||||
// 注意：这里不直接重连，由 startConnection 循环处理重连逻辑
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* 等待连接断开
|
||||
*/
|
||||
private suspend fun waitForDisconnect() {
|
||||
while (isConnected && scope.isActive) {
|
||||
delay(1000)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 订阅 Leader 钱包地址
|
||||
* 每个 Leader 有 6 个订阅，保存订阅 ID 以便后续取消
|
||||
*/
|
||||
private suspend fun subscribeLeader(leader: Leader) {
|
||||
if (webSocket == null || !isConnected || leader.id == null) {
|
||||
return
|
||||
}
|
||||
|
||||
val walletAddress = leader.leaderAddress.lowercase()
|
||||
val walletTopic = addressToTopic32(walletAddress)
|
||||
val leaderId = leader.id!!
|
||||
|
||||
// 初始化该 Leader 的订阅列表
|
||||
if (!leaderSubscriptions.containsKey(leaderId)) {
|
||||
leaderSubscriptions[leaderId] = mutableListOf()
|
||||
}
|
||||
|
||||
try {
|
||||
// 订阅 USDC Transfer (from wallet)
|
||||
subscribeLogs(USDC_CONTRACT, listOf(ERC20_TRANSFER_TOPIC, walletTopic), leaderId)
|
||||
|
||||
// 订阅 USDC Transfer (to wallet)
|
||||
subscribeLogs(USDC_CONTRACT, listOf(ERC20_TRANSFER_TOPIC, null, walletTopic), leaderId)
|
||||
|
||||
// 订阅 ERC1155 TransferSingle (from wallet)
|
||||
subscribeLogs(ERC1155_CONTRACT, listOf(ERC1155_TRANSFER_SINGLE_TOPIC, null, walletTopic), leaderId)
|
||||
|
||||
// 订阅 ERC1155 TransferSingle (to wallet)
|
||||
subscribeLogs(ERC1155_CONTRACT, listOf(ERC1155_TRANSFER_SINGLE_TOPIC, null, null, walletTopic), leaderId)
|
||||
|
||||
// 订阅 ERC1155 TransferBatch (from wallet)
|
||||
subscribeLogs(ERC1155_CONTRACT, listOf(ERC1155_TRANSFER_BATCH_TOPIC, null, walletTopic), leaderId)
|
||||
|
||||
// 订阅 ERC1155 TransferBatch (to wallet)
|
||||
subscribeLogs(ERC1155_CONTRACT, listOf(ERC1155_TRANSFER_BATCH_TOPIC, null, null, walletTopic), leaderId)
|
||||
|
||||
logger.debug("已订阅 Leader 钱包地址: ${leader.leaderName} (${walletAddress})")
|
||||
} catch (e: Exception) {
|
||||
logger.error("订阅 Leader 失败: leaderId=$leaderId, address=$walletAddress", e)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 订阅日志
|
||||
* @param address 合约地址
|
||||
* @param topics 主题列表
|
||||
* @param leaderId Leader ID，用于关联订阅响应
|
||||
*/
|
||||
private fun subscribeLogs(address: String, topics: List<String?>, leaderId: Long) {
|
||||
val ws = webSocket ?: return
|
||||
|
||||
val params = mapOf(
|
||||
"address" to address.lowercase(),
|
||||
"topics" to topics
|
||||
)
|
||||
|
||||
val subscribeParams = listOf("logs", params)
|
||||
|
||||
val requestId = ++requestIdCounter
|
||||
|
||||
// 保存请求 ID 到 Leader ID 的映射
|
||||
requestIdToLeaderId[requestId] = leaderId
|
||||
|
||||
val request = mapOf(
|
||||
"jsonrpc" to "2.0",
|
||||
"id" to requestId,
|
||||
"method" to "eth_subscribe",
|
||||
"params" to subscribeParams
|
||||
)
|
||||
|
||||
val json = gson.toJson(request)
|
||||
ws.send(json)
|
||||
}
|
||||
|
||||
/**
|
||||
* 取消订阅
|
||||
* @param subscriptionId 订阅 ID
|
||||
*/
|
||||
private fun unsubscribe(subscriptionId: String) {
|
||||
val ws = webSocket ?: return
|
||||
|
||||
val request = mapOf(
|
||||
"jsonrpc" to "2.0",
|
||||
"id" to (++requestIdCounter),
|
||||
"method" to "eth_unsubscribe",
|
||||
"params" to listOf(subscriptionId)
|
||||
)
|
||||
|
||||
val json = gson.toJson(request)
|
||||
ws.send(json)
|
||||
logger.debug("已发送取消订阅请求: subscriptionId=$subscriptionId")
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理 WebSocket 消息
|
||||
*/
|
||||
private suspend fun handleMessage(message: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) {
|
||||
try {
|
||||
// 使用 Gson 解析消息
|
||||
val messageJson = gson.fromJson(message, JsonObject::class.java)
|
||||
|
||||
// 处理订阅响应（包含 subscription ID）
|
||||
val id = messageJson.get("id")
|
||||
if (id != null && !id.isJsonNull) {
|
||||
val result = messageJson.get("result")
|
||||
if (result != null && result.isJsonPrimitive && result.asJsonPrimitive.isString) {
|
||||
// 这是订阅响应，result 是 subscription ID（字符串）
|
||||
val requestId = id.asInt
|
||||
val subscriptionId = result.asString
|
||||
val leaderId = requestIdToLeaderId.remove(requestId)
|
||||
if (leaderId != null) {
|
||||
// 保存订阅 ID 到 Leader
|
||||
leaderSubscriptions.getOrPut(leaderId) { mutableListOf() }.add(subscriptionId)
|
||||
logger.debug("收到订阅响应: leaderId=$leaderId, subscriptionId=$subscriptionId")
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// 处理订阅通知（交易日志）
|
||||
val method = messageJson.get("method")?.asString
|
||||
if (method != "eth_subscription") {
|
||||
return
|
||||
}
|
||||
|
||||
val params = messageJson.getAsJsonObject("params") ?: return
|
||||
// result 是一个对象，包含日志信息
|
||||
val result = params.getAsJsonObject("result") ?: return
|
||||
|
||||
// 从 result 对象中获取 transactionHash（关键数据）
|
||||
val txHash = result.get("transactionHash")?.asString
|
||||
if (txHash.isNullOrEmpty()) {
|
||||
logger.debug("订阅通知中缺少 transactionHash，跳过处理")
|
||||
return
|
||||
}
|
||||
|
||||
// 处理交易
|
||||
processTransaction(txHash, httpClient, rpcApi)
|
||||
} catch (e: Exception) {
|
||||
logger.error("处理 WebSocket 消息失败: ${e.message}", e)
|
||||
logger.debug("消息内容: $message", e)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理交易
|
||||
*/
|
||||
private suspend fun processTransaction(txHash: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) {
|
||||
try {
|
||||
// 获取交易 receipt
|
||||
val receiptRequest = JsonRpcRequest(
|
||||
method = "eth_getTransactionReceipt",
|
||||
params = listOf(txHash)
|
||||
)
|
||||
|
||||
val receiptResponse = rpcApi.call(receiptRequest)
|
||||
if (!receiptResponse.isSuccessful || receiptResponse.body() == null) {
|
||||
return
|
||||
}
|
||||
|
||||
val receiptRpcResponse = receiptResponse.body()!!
|
||||
if (receiptRpcResponse.error != null || receiptRpcResponse.result == null) {
|
||||
return
|
||||
}
|
||||
|
||||
// 使用 Gson 解析 receipt JSON（result 是 JsonElement）
|
||||
val receiptJson = receiptRpcResponse.result.asJsonObject
|
||||
|
||||
// 获取区块号和时间戳
|
||||
val blockNumber = receiptJson.get("blockNumber")?.asString
|
||||
val blockTimestamp = if (blockNumber != null) {
|
||||
getBlockTimestamp(blockNumber, rpcApi)
|
||||
} else {
|
||||
null
|
||||
}
|
||||
|
||||
// 解析 receipt 中的 Transfer 日志
|
||||
val logs = receiptJson.getAsJsonArray("logs") ?: return
|
||||
val (erc20Transfers, erc1155Transfers) = parseReceiptTransfers(logs)
|
||||
|
||||
// 为每个 Leader 处理交易
|
||||
for (leader in monitoredLeaders.values) {
|
||||
val trade = parseTradeFromTransfers(
|
||||
txHash = txHash,
|
||||
timestamp = blockTimestamp,
|
||||
walletAddress = leader.leaderAddress,
|
||||
erc20Transfers = erc20Transfers,
|
||||
erc1155Transfers = erc1155Transfers
|
||||
)
|
||||
|
||||
if (trade != null) {
|
||||
// 调用 processTrade 处理交易（元数据已在 parseTradeFromTransfers 中补齐）
|
||||
copyOrderTrackingService.processTrade(
|
||||
leaderId = leader.id!!,
|
||||
trade = trade,
|
||||
source = "onchain-ws"
|
||||
)
|
||||
}
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
logger.error("处理交易失败: txHash=$txHash, ${e.message}", e)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 解析 receipt 中的 Transfer 日志
|
||||
*/
|
||||
private fun parseReceiptTransfers(logs: com.google.gson.JsonArray): Pair<List<Erc20Transfer>, List<Erc1155Transfer>> {
|
||||
val erc20 = mutableListOf<Erc20Transfer>()
|
||||
val erc1155 = mutableListOf<Erc1155Transfer>()
|
||||
|
||||
for (logElement in logs) {
|
||||
val log = logElement.asJsonObject
|
||||
val address = log.get("address")?.asString?.lowercase() ?: continue
|
||||
val topicsArray = log.getAsJsonArray("topics") ?: continue
|
||||
val topics = topicsArray.mapNotNull { it.asString }
|
||||
if (topics.isEmpty()) continue
|
||||
|
||||
val t0 = topics[0].lowercase()
|
||||
val data = log.get("data")?.asString ?: "0x"
|
||||
|
||||
// USDC ERC20 Transfer
|
||||
if (address == USDC_CONTRACT.lowercase() && t0 == ERC20_TRANSFER_TOPIC && topics.size >= 3) {
|
||||
val from = topicToAddress(topics[1])
|
||||
val to = topicToAddress(topics[2])
|
||||
val value = hexToBigInt(data)
|
||||
erc20.add(Erc20Transfer(from, to, value))
|
||||
continue
|
||||
}
|
||||
|
||||
// ERC1155 TransferSingle
|
||||
if (t0 == ERC1155_TRANSFER_SINGLE_TOPIC && topics.size >= 4) {
|
||||
val from = topicToAddress(topics[2])
|
||||
val to = topicToAddress(topics[3])
|
||||
val bytes = bytesFromHex(data)
|
||||
if (bytes.size >= 64) {
|
||||
val tokenId = sliceBigInt32(bytes, 0)
|
||||
val value = sliceBigInt32(bytes, 32)
|
||||
erc1155.add(Erc1155Transfer(from, to, tokenId, value))
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// ERC1155 TransferBatch
|
||||
if (t0 == ERC1155_TRANSFER_BATCH_TOPIC && topics.size >= 4) {
|
||||
val from = topicToAddress(topics[2])
|
||||
val to = topicToAddress(topics[3])
|
||||
val bytes = bytesFromHex(data)
|
||||
if (bytes.size < 64) continue
|
||||
|
||||
val offIds = sliceBigInt32(bytes, 0).toInt()
|
||||
val offVals = sliceBigInt32(bytes, 32).toInt()
|
||||
if (offIds + 32 > bytes.size || offVals + 32 > bytes.size) continue
|
||||
|
||||
val nIds = sliceBigInt32(bytes, offIds).toInt()
|
||||
val nVals = sliceBigInt32(bytes, offVals).toInt()
|
||||
if (nIds != nVals) continue
|
||||
|
||||
val idsStart = offIds + 32
|
||||
val valsStart = offVals + 32
|
||||
for (i in 0 until nIds) {
|
||||
val ib = idsStart + i * 32
|
||||
val vb = valsStart + i * 32
|
||||
if (ib + 32 > bytes.size || vb + 32 > bytes.size) break
|
||||
val tokenId = sliceBigInt32(bytes, ib)
|
||||
val value = sliceBigInt32(bytes, vb)
|
||||
erc1155.add(Erc1155Transfer(from, to, tokenId, value))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return Pair(erc20, erc1155)
|
||||
}
|
||||
|
||||
/**
|
||||
* 从 Transfer 日志解析交易信息
|
||||
*/
|
||||
private suspend fun parseTradeFromTransfers(
|
||||
txHash: String,
|
||||
timestamp: Long?,
|
||||
walletAddress: String,
|
||||
erc20Transfers: List<Erc20Transfer>,
|
||||
erc1155Transfers: List<Erc1155Transfer>
|
||||
): TradeResponse? {
|
||||
val wallet = walletAddress.lowercase()
|
||||
|
||||
// 计算 USDC 流入和流出
|
||||
val usdcOut = erc20Transfers.filter { it.from.lowercase() == wallet }
|
||||
.fold(BigInteger.ZERO) { acc, t -> acc + t.value }
|
||||
val usdcIn = erc20Transfers.filter { it.to.lowercase() == wallet }
|
||||
.fold(BigInteger.ZERO) { acc, t -> acc + t.value }
|
||||
|
||||
// 计算 ERC1155 流入和流出（按 tokenId 聚合）
|
||||
val inById = mutableMapOf<BigInteger, BigInteger>()
|
||||
val outById = mutableMapOf<BigInteger, BigInteger>()
|
||||
for (t in erc1155Transfers) {
|
||||
if (t.to.lowercase() == wallet) {
|
||||
inById[t.tokenId] = (inById[t.tokenId] ?: BigInteger.ZERO) + t.value
|
||||
}
|
||||
if (t.from.lowercase() == wallet) {
|
||||
outById[t.tokenId] = (outById[t.tokenId] ?: BigInteger.ZERO) + t.value
|
||||
}
|
||||
}
|
||||
|
||||
// 找到最大的流入和流出 tokenId
|
||||
fun best(map: Map<BigInteger, BigInteger>): Pair<BigInteger?, BigInteger> =
|
||||
map.entries.maxByOrNull { it.value }?.let { it.key to it.value } ?: (null to BigInteger.ZERO)
|
||||
|
||||
val (bestInId, bestInVal) = best(inById)
|
||||
val (bestOutId, bestOutVal) = best(outById)
|
||||
|
||||
// 判断交易方向
|
||||
var side: String? = null
|
||||
var asset: BigInteger? = null
|
||||
var sizeRaw = BigInteger.ZERO
|
||||
var usdcRaw = BigInteger.ZERO
|
||||
|
||||
if (bestInId != null && bestInVal > BigInteger.ZERO && usdcOut > BigInteger.ZERO) {
|
||||
// BUY: 收到 token，支付 USDC
|
||||
side = "BUY"
|
||||
asset = bestInId
|
||||
sizeRaw = bestInVal
|
||||
usdcRaw = usdcOut
|
||||
} else if (bestOutId != null && bestOutVal > BigInteger.ZERO && usdcIn > BigInteger.ZERO) {
|
||||
// SELL: 卖出 token，收到 USDC
|
||||
side = "SELL"
|
||||
asset = bestOutId
|
||||
sizeRaw = bestOutVal
|
||||
usdcRaw = usdcIn
|
||||
} else {
|
||||
// 无法判断交易方向
|
||||
return null
|
||||
}
|
||||
|
||||
// 计算价格和数量（USDC 有 6 位小数，shares 也有 6 位小数）
|
||||
val usdcSize = usdcRaw.toBigDecimal().divide(BigInteger("1000000").toBigDecimal(), 8, java.math.RoundingMode.DOWN)
|
||||
val size = sizeRaw.toBigDecimal().divide(BigInteger("1000000").toBigDecimal(), 8, java.math.RoundingMode.DOWN)
|
||||
val price = if (size.signum() > 0) {
|
||||
usdcSize.divide(size, 8, java.math.RoundingMode.DOWN)
|
||||
} else {
|
||||
return null
|
||||
}
|
||||
|
||||
// 尝试通过 Gamma API 查询市场信息（通过 tokenId）
|
||||
val marketInfo = fetchMarketByTokenId(asset.toString())
|
||||
|
||||
// 创建 TradeResponse
|
||||
return TradeResponse(
|
||||
id = txHash,
|
||||
market = marketInfo?.conditionId ?: "",
|
||||
side = side,
|
||||
price = price.toPlainString(),
|
||||
size = size.toPlainString(),
|
||||
timestamp = (timestamp ?: System.currentTimeMillis() / 1000).toString(),
|
||||
user = walletAddress,
|
||||
outcomeIndex = marketInfo?.outcomeIndex,
|
||||
outcome = marketInfo?.outcome
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* 通过 Gamma API 查询市场信息（通过 tokenId）
|
||||
*/
|
||||
private suspend fun fetchMarketByTokenId(tokenId: String): MarketInfo? {
|
||||
return try {
|
||||
// 使用 HTTP 请求直接调用 Gamma API（因为 Retrofit 接口可能不支持 clob_token_ids 参数）
|
||||
val httpClient = createHttpClient()
|
||||
val url = "https://gamma-api.polymarket.com/markets?clob_token_ids=$tokenId"
|
||||
|
||||
val request = okhttp3.Request.Builder()
|
||||
.url(url)
|
||||
.get()
|
||||
.build()
|
||||
|
||||
val response = httpClient.newCall(request).execute()
|
||||
if (!response.isSuccessful || response.body == null) {
|
||||
return null
|
||||
}
|
||||
|
||||
val responseBody = response.body!!.string()
|
||||
// 使用 Gson 解析市场列表
|
||||
val marketsType = object : TypeToken<List<JsonObject>>() {}.type
|
||||
val markets = gson.fromJson<List<JsonObject>>(responseBody, marketsType)
|
||||
|
||||
if (markets.isEmpty()) {
|
||||
return null
|
||||
}
|
||||
|
||||
val market = markets.first()
|
||||
|
||||
// 解析 clob_token_ids（可能是 JSON 字符串或数组）
|
||||
val clobTokenIdsRaw = market.get("clobTokenIds") ?: market.get("clob_token_ids")
|
||||
val clobTokenIds = when {
|
||||
clobTokenIdsRaw == null || clobTokenIdsRaw.isJsonNull -> null
|
||||
clobTokenIdsRaw.isJsonPrimitive && clobTokenIdsRaw.asJsonPrimitive.isString -> {
|
||||
// 尝试解析 JSON 字符串
|
||||
try {
|
||||
val listType = object : TypeToken<List<String>>() {}.type
|
||||
gson.fromJson<List<String>>(clobTokenIdsRaw.asString, listType)
|
||||
} catch (e: Exception) {
|
||||
null
|
||||
}
|
||||
}
|
||||
clobTokenIdsRaw.isJsonArray -> {
|
||||
clobTokenIdsRaw.asJsonArray.mapNotNull { it.asString }
|
||||
}
|
||||
else -> null
|
||||
}
|
||||
|
||||
// 解析 outcomes（可能是 JSON 字符串或数组）
|
||||
val outcomesRaw = market.get("outcomes")
|
||||
val outcomes = when {
|
||||
outcomesRaw == null || outcomesRaw.isJsonNull -> null
|
||||
outcomesRaw.isJsonPrimitive && outcomesRaw.asJsonPrimitive.isString -> {
|
||||
try {
|
||||
val listType = object : TypeToken<List<String>>() {}.type
|
||||
gson.fromJson<List<String>>(outcomesRaw.asString, listType)
|
||||
} catch (e: Exception) {
|
||||
null
|
||||
}
|
||||
}
|
||||
outcomesRaw.isJsonArray -> {
|
||||
outcomesRaw.asJsonArray.mapNotNull { it.asString }
|
||||
}
|
||||
else -> null
|
||||
}
|
||||
|
||||
// 查找 tokenId 在 clobTokenIds 中的索引
|
||||
val outcomeIndex = clobTokenIds?.indexOfFirst {
|
||||
it.equals(tokenId, ignoreCase = true)
|
||||
}?.takeIf { it >= 0 }
|
||||
|
||||
// 获取 outcome 名称
|
||||
val outcome = if (outcomeIndex != null && outcomes != null && outcomeIndex < outcomes.size) {
|
||||
outcomes[outcomeIndex]
|
||||
} else {
|
||||
null
|
||||
}
|
||||
|
||||
val conditionId = market.get("conditionId")?.asString ?: return null
|
||||
|
||||
MarketInfo(
|
||||
conditionId = conditionId,
|
||||
outcomeIndex = outcomeIndex,
|
||||
outcome = outcome
|
||||
)
|
||||
} catch (e: Exception) {
|
||||
logger.warn("通过 Gamma API 查询市场信息失败: tokenId=$tokenId, ${e.message}")
|
||||
null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 市场信息（从 Gamma API 获取）
|
||||
*/
|
||||
private data class MarketInfo(
|
||||
val conditionId: String,
|
||||
val outcomeIndex: Int?,
|
||||
val outcome: String?
|
||||
)
|
||||
|
||||
/**
|
||||
* 获取区块时间戳
|
||||
*/
|
||||
private suspend fun getBlockTimestamp(blockNumber: String, rpcApi: EthereumRpcApi): Long? {
|
||||
return try {
|
||||
val blockRequest = JsonRpcRequest(
|
||||
method = "eth_getBlockByNumber",
|
||||
params = listOf(blockNumber, false)
|
||||
)
|
||||
|
||||
val blockResponse = rpcApi.call(blockRequest)
|
||||
if (!blockResponse.isSuccessful || blockResponse.body() == null) {
|
||||
return null
|
||||
}
|
||||
|
||||
val blockRpcResponse = blockResponse.body()!!
|
||||
if (blockRpcResponse.error != null || blockRpcResponse.result == null) {
|
||||
return null
|
||||
}
|
||||
|
||||
// 使用 Gson 解析 block JSON（result 是 JsonElement）
|
||||
val blockJson = blockRpcResponse.result.asJsonObject
|
||||
val timestampHex = blockJson.get("timestamp")?.asString ?: return null
|
||||
hexToBigInt(timestampHex).toLong()
|
||||
} catch (e: Exception) {
|
||||
logger.warn("获取区块时间戳失败: ${e.message}")
|
||||
null
|
||||
}
|
||||
}
|
||||
|
||||
// 辅助函数
|
||||
private data class Erc20Transfer(val from: String, val to: String, val value: BigInteger)
|
||||
private data class Erc1155Transfer(val from: String, val to: String, val tokenId: BigInteger, val value: BigInteger)
|
||||
|
||||
private fun topicToAddress(topic: String): String {
|
||||
val t = topic.removePrefix("0x").lowercase()
|
||||
return "0x" + t.takeLast(40)
|
||||
}
|
||||
|
||||
private fun hexToBigInt(hex: String): BigInteger {
|
||||
val h = hex.removePrefix("0x")
|
||||
if (h.isEmpty()) return BigInteger.ZERO
|
||||
return BigInteger(h, 16)
|
||||
}
|
||||
|
||||
private fun bytesFromHex(hex: String): ByteArray {
|
||||
val s = hex.removePrefix("0x")
|
||||
if (s.isEmpty()) return ByteArray(0)
|
||||
val out = ByteArray(s.length / 2)
|
||||
var i = 0
|
||||
while (i < s.length) {
|
||||
out[i / 2] = s.substring(i, i + 2).toInt(16).toByte()
|
||||
i += 2
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
private fun sliceBigInt32(b: ByteArray, offset: Int): BigInteger {
|
||||
val sub = b.copyOfRange(offset, offset + 32)
|
||||
return BigInteger(1, sub)
|
||||
}
|
||||
|
||||
/**
|
||||
* 地址转换为 32 字节 topic（前 24 字节为 0，后 8 字节为地址）
|
||||
*/
|
||||
private fun addressToTopic32(address: String): String {
|
||||
val addr = address.removePrefix("0x").lowercase()
|
||||
return "0x" + "0".repeat(24) + addr
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
fun destroy() {
|
||||
stop()
|
||||
scope.cancel()
|
||||
}
|
||||
}
|
||||
|
||||
+101
-85
@@ -8,10 +8,13 @@ import com.wrbug.polymarketbot.repository.*
|
||||
import com.wrbug.polymarketbot.util.RetrofitFactory
|
||||
import com.wrbug.polymarketbot.util.*
|
||||
import kotlinx.coroutines.*
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import org.slf4j.LoggerFactory
|
||||
import org.springframework.dao.DataIntegrityViolationException
|
||||
import org.springframework.dao.DuplicateKeyException
|
||||
import java.sql.SQLException
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
import com.wrbug.polymarketbot.service.copytrading.configs.CopyTradingFilterService
|
||||
import com.wrbug.polymarketbot.service.copytrading.configs.FilterStatus
|
||||
import com.wrbug.polymarketbot.service.copytrading.orders.OrderSigningService
|
||||
@@ -53,11 +56,22 @@ open class CopyOrderTrackingService(
|
||||
// 协程作用域（用于异步发送通知）
|
||||
private val notificationScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
|
||||
|
||||
// 使用 Mutex 保证线程安全（按交易ID锁定）
|
||||
private val tradeMutexMap = ConcurrentHashMap<String, Mutex>()
|
||||
|
||||
// 订单创建重试配置
|
||||
companion object {
|
||||
private const val MAX_RETRY_ATTEMPTS = 2 // 最多重试次数（首次 + 1次重试）
|
||||
private const val RETRY_DELAY_MS = 3000L // 重试前等待时间（毫秒，3秒）
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取或创建 Mutex（按交易ID）
|
||||
*/
|
||||
private fun getMutex(leaderId: Long, tradeId: String): Mutex {
|
||||
val key = "${leaderId}_${tradeId}"
|
||||
return tradeMutexMap.getOrPut(key) { Mutex() }
|
||||
}
|
||||
|
||||
/**
|
||||
* 解密账户私钥
|
||||
@@ -102,96 +116,98 @@ open class CopyOrderTrackingService(
|
||||
/**
|
||||
* 处理交易事件（WebSocket 或轮询）
|
||||
* 根据交易方向调用相应的处理方法
|
||||
* 使用 Mutex 保证线程安全（单实例部署）
|
||||
*/
|
||||
@Transactional
|
||||
suspend fun processTrade(leaderId: Long, trade: TradeResponse, source: String): Result<Unit> {
|
||||
return try {
|
||||
// 1. 检查是否已处理（去重，包括失败状态）
|
||||
val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id)
|
||||
|
||||
if (existingProcessed != null) {
|
||||
if (existingProcessed.status == "FAILED") {
|
||||
return Result.success(Unit)
|
||||
}
|
||||
return Result.success(Unit)
|
||||
}
|
||||
|
||||
// 检查是否已记录为失败交易
|
||||
val failedTrade = failedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id)
|
||||
if (failedTrade != null) {
|
||||
return Result.success(Unit)
|
||||
}
|
||||
|
||||
// 2. 处理交易逻辑
|
||||
val result = when (trade.side.uppercase()) {
|
||||
"BUY" -> processBuyTrade(leaderId, trade)
|
||||
"SELL" -> processSellTrade(leaderId, trade)
|
||||
else -> {
|
||||
logger.warn("未知的交易方向: ${trade.side}")
|
||||
Result.failure(IllegalArgumentException("未知的交易方向: ${trade.side}"))
|
||||
}
|
||||
}
|
||||
|
||||
if (result.isFailure) {
|
||||
logger.error(
|
||||
"处理交易失败: leaderId=$leaderId, tradeId=${trade.id}, side=${trade.side}",
|
||||
result.exceptionOrNull()
|
||||
)
|
||||
return result
|
||||
}
|
||||
|
||||
// 3. 标记为已处理（成功状态）
|
||||
// 注意：并发情况下可能多个请求同时处理同一笔交易，需要处理唯一约束冲突
|
||||
// 获取该交易的 Mutex（按交易ID锁定，不同交易可以并行处理）
|
||||
val mutex = getMutex(leaderId, trade.id)
|
||||
|
||||
return mutex.withLock {
|
||||
try {
|
||||
val processed = ProcessedTrade(
|
||||
leaderId = leaderId,
|
||||
leaderTradeId = trade.id,
|
||||
tradeType = trade.side.uppercase(),
|
||||
source = source,
|
||||
status = "SUCCESS",
|
||||
processedAt = System.currentTimeMillis()
|
||||
)
|
||||
processedTradeRepository.save(processed)
|
||||
} catch (e: Exception) {
|
||||
// 检查是否是唯一键冲突异常（可能是 DataIntegrityViolationException、DuplicateKeyException 或 SQLException）
|
||||
if (isUniqueConstraintViolation(e)) {
|
||||
// 唯一约束冲突，说明已经处理过了（可能是并发请求）
|
||||
// 再次检查确认状态
|
||||
val existing = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id)
|
||||
if (existing != null) {
|
||||
if (existing.status == "FAILED") {
|
||||
logger.debug("交易已标记为失败，跳过处理: leaderId=$leaderId, tradeId=${trade.id}")
|
||||
return Result.success(Unit)
|
||||
}
|
||||
logger.debug("交易已处理（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existing.status}")
|
||||
return Result.success(Unit)
|
||||
} else {
|
||||
// 如果检查不到，可能是事务隔离级别问题，等待一下再查询
|
||||
delay(100)
|
||||
val existingAfterDelay =
|
||||
processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id)
|
||||
if (existingAfterDelay != null) {
|
||||
logger.debug("延迟查询到记录（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existingAfterDelay.status}")
|
||||
return Result.success(Unit)
|
||||
}
|
||||
// 如果还是查询不到，记录警告但不抛出异常（可能是其他约束冲突）
|
||||
logger.warn(
|
||||
"保存ProcessedTrade时发生唯一约束冲突，但查询不到记录: leaderId=$leaderId, tradeId=${trade.id}",
|
||||
e
|
||||
)
|
||||
// 不抛出异常，避免影响其他交易的处理
|
||||
return Result.success(Unit)
|
||||
}
|
||||
} else {
|
||||
// 其他类型的异常，重新抛出
|
||||
throw e
|
||||
}
|
||||
}
|
||||
// 1. 检查是否已处理（去重，包括失败状态）
|
||||
val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id)
|
||||
|
||||
Result.success(Unit)
|
||||
} catch (e: Exception) {
|
||||
logger.error("处理交易异常: leaderId=$leaderId, tradeId=${trade.id}", e)
|
||||
Result.failure(e)
|
||||
if (existingProcessed != null) {
|
||||
if (existingProcessed.status == "FAILED") {
|
||||
return@withLock Result.success(Unit)
|
||||
}
|
||||
return@withLock Result.success(Unit)
|
||||
}
|
||||
|
||||
// 检查是否已记录为失败交易
|
||||
val failedTrade = failedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id)
|
||||
if (failedTrade != null) {
|
||||
return@withLock Result.success(Unit)
|
||||
}
|
||||
|
||||
// 2. 处理交易逻辑
|
||||
val result = when (trade.side.uppercase()) {
|
||||
"BUY" -> processBuyTrade(leaderId, trade)
|
||||
"SELL" -> processSellTrade(leaderId, trade)
|
||||
else -> {
|
||||
logger.warn("未知的交易方向: ${trade.side}")
|
||||
Result.failure(IllegalArgumentException("未知的交易方向: ${trade.side}"))
|
||||
}
|
||||
}
|
||||
|
||||
if (result.isFailure) {
|
||||
logger.error(
|
||||
"处理交易失败: leaderId=$leaderId, tradeId=${trade.id}, side=${trade.side}",
|
||||
result.exceptionOrNull()
|
||||
)
|
||||
return@withLock result
|
||||
}
|
||||
|
||||
// 3. 标记为已处理（成功状态）
|
||||
// 由于使用了 Mutex，这里理论上不会出现并发冲突，但保留异常处理作为兜底
|
||||
try {
|
||||
val processed = ProcessedTrade(
|
||||
leaderId = leaderId,
|
||||
leaderTradeId = trade.id,
|
||||
tradeType = trade.side.uppercase(),
|
||||
source = source,
|
||||
status = "SUCCESS",
|
||||
processedAt = System.currentTimeMillis()
|
||||
)
|
||||
processedTradeRepository.save(processed)
|
||||
} catch (e: Exception) {
|
||||
// 检查是否是唯一键冲突异常（理论上不会发生，但保留作为兜底）
|
||||
if (isUniqueConstraintViolation(e)) {
|
||||
val existing = processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id)
|
||||
if (existing != null) {
|
||||
if (existing.status == "FAILED") {
|
||||
logger.debug("交易已标记为失败，跳过处理: leaderId=$leaderId, tradeId=${trade.id}")
|
||||
return@withLock Result.success(Unit)
|
||||
}
|
||||
logger.debug("交易已处理（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existing.status}")
|
||||
return@withLock Result.success(Unit)
|
||||
} else {
|
||||
// 如果检查不到，可能是事务隔离级别问题，等待一下再查询
|
||||
delay(100)
|
||||
val existingAfterDelay =
|
||||
processedTradeRepository.findByLeaderIdAndLeaderTradeId(leaderId, trade.id)
|
||||
if (existingAfterDelay != null) {
|
||||
logger.debug("延迟查询到记录（并发检测）: leaderId=$leaderId, tradeId=${trade.id}, status=${existingAfterDelay.status}")
|
||||
return@withLock Result.success(Unit)
|
||||
}
|
||||
logger.warn(
|
||||
"保存ProcessedTrade时发生唯一约束冲突，但查询不到记录: leaderId=$leaderId, tradeId=${trade.id}",
|
||||
e
|
||||
)
|
||||
return@withLock Result.success(Unit)
|
||||
}
|
||||
} else {
|
||||
// 其他类型的异常，重新抛出
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
Result.success(Unit)
|
||||
} catch (e: Exception) {
|
||||
logger.error("处理交易异常: leaderId=$leaderId, tradeId=${trade.id}", e)
|
||||
Result.failure(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+4
-4
@@ -655,7 +655,7 @@ class RelayClientService(
|
||||
}
|
||||
|
||||
val hexNonce = rpcResponse.result ?: return Result.failure(Exception("Proxy nonce 结果为空"))
|
||||
val nonce = EthereumUtils.decodeUint256(hexNonce)
|
||||
val nonce = EthereumUtils.decodeUint256(hexNonce.asString)
|
||||
return Result.success(nonce)
|
||||
}
|
||||
|
||||
@@ -679,7 +679,7 @@ class RelayClientService(
|
||||
}
|
||||
|
||||
val hexNonce = rpcResponse.result ?: return Result.failure(Exception("nonce 结果为空"))
|
||||
val nonce = EthereumUtils.decodeUint256(hexNonce)
|
||||
val nonce = EthereumUtils.decodeUint256(hexNonce.asString)
|
||||
return Result.success(nonce)
|
||||
}
|
||||
|
||||
@@ -703,7 +703,7 @@ class RelayClientService(
|
||||
}
|
||||
|
||||
val hexGasPrice = rpcResponse.result ?: return Result.failure(Exception("gas price 结果为空"))
|
||||
val gasPrice = EthereumUtils.decodeUint256(hexGasPrice)
|
||||
val gasPrice = EthereumUtils.decodeUint256(hexGasPrice.asString)
|
||||
return Result.success(gasPrice)
|
||||
}
|
||||
|
||||
@@ -774,7 +774,7 @@ class RelayClientService(
|
||||
}
|
||||
|
||||
val txHash = rpcResponse.result ?: return Result.failure(Exception("交易哈希为空"))
|
||||
return Result.success(txHash)
|
||||
return Result.success(txHash.asString)
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -446,7 +446,7 @@ class RpcNodeService(
|
||||
))
|
||||
}
|
||||
|
||||
val blockNumber = rpcResponse.result
|
||||
val blockNumber = rpcResponse.result?.asString
|
||||
if (blockNumber.isNullOrBlank()) {
|
||||
return Result.success(NodeCheckResult(
|
||||
status = NodeHealthStatus.UNHEALTHY,
|
||||
|
||||
@@ -0,0 +1,289 @@
|
||||
# 跟单买卖逻辑简易文档
|
||||
|
||||
## 一、跟单信号检测
|
||||
|
||||
### 1.1 监控方式
|
||||
|
||||
系统使用**轮询方式**监控 Leader 的交易活动：
|
||||
|
||||
- **数据源**：Polymarket Data API 的 `/activity` 接口
|
||||
- **轮询间隔**：默认 2 秒（可配置）
|
||||
- **查询参数**：
|
||||
- `user`: Leader 钱包地址
|
||||
- `type`: `["TRADE"]`（只查询交易类型）
|
||||
- `limit`: 100（每次最多查询 100 条）
|
||||
- `sortBy`: `TIMESTAMP`
|
||||
- `sortDirection`: `DESC`（按时间戳降序，最新的在前）
|
||||
|
||||
### 1.2 增量检测机制
|
||||
|
||||
通过 **diff 算法**检测新增交易：
|
||||
|
||||
1. **首次轮询**：
|
||||
- 查询最近 100 条交易
|
||||
- 缓存所有交易 ID（不处理）
|
||||
- 标记首次轮询完成
|
||||
|
||||
2. **后续轮询**：
|
||||
- 查询最近 100 条交易
|
||||
- 与缓存的交易 ID 集合进行 diff
|
||||
- 找出新增的交易 ID
|
||||
- 处理新增交易
|
||||
- 更新缓存（添加新增的交易 ID）
|
||||
|
||||
3. **去重机制**：
|
||||
- 使用 `leaderId + tradeId` 作为唯一标识
|
||||
- 在 `processed_trade` 表中记录已处理的交易
|
||||
- 避免重复处理同一笔交易
|
||||
|
||||
### 1.3 交易数据转换
|
||||
|
||||
从 `UserActivityResponse` 转换为 `TradeResponse`：
|
||||
|
||||
```kotlin
|
||||
TradeResponse(
|
||||
id = activity.transactionHash, // 交易ID（用于去重）
|
||||
market = activity.conditionId, // 市场ID
|
||||
side = activity.side, // "BUY" 或 "SELL"
|
||||
price = activity.price, // 交易价格
|
||||
size = activity.size, // 交易数量
|
||||
timestamp = activity.timestamp, // 时间戳（秒）
|
||||
user = activity.proxyWallet, // 用户钱包地址
|
||||
outcomeIndex = activity.outcomeIndex, // 结果索引（0=第一个outcome，1=第二个outcome）
|
||||
outcome = activity.outcome // 结果名称
|
||||
)
|
||||
```
|
||||
|
||||
### 1.4 信号触发流程
|
||||
|
||||
```
|
||||
轮询任务启动
|
||||
↓
|
||||
定期轮询所有 Leader（每 2 秒）
|
||||
↓
|
||||
查询 Leader 活动（/activity 接口）
|
||||
↓
|
||||
转换为 TradeResponse
|
||||
↓
|
||||
diff 检测新增交易
|
||||
↓
|
||||
调用 processTrade() 处理交易
|
||||
↓
|
||||
根据 side 字段判断：
|
||||
- "BUY" → processBuyTrade()
|
||||
- "SELL" → processSellTrade()
|
||||
```
|
||||
|
||||
## 二、订单构建
|
||||
|
||||
### 2.1 买入订单构建流程
|
||||
|
||||
#### 2.1.1 前置检查
|
||||
|
||||
1. **查找跟单关系**：
|
||||
- 查询所有启用且支持该 Leader 的跟单配置
|
||||
- 验证账户 API 凭证是否配置
|
||||
- 验证账户是否启用
|
||||
|
||||
2. **获取 Token ID**：
|
||||
- 使用 `outcomeIndex` 和 `market` 获取 tokenId
|
||||
- 支持多元市场（不限于 YES/NO）
|
||||
|
||||
3. **计算买入数量**：
|
||||
- **RATIO 模式**：`买入数量 = Leader 数量 × 跟单比例`
|
||||
- **FIXED 模式**：`买入数量 = 固定金额 / 买入价格`
|
||||
|
||||
4. **过滤条件检查**：
|
||||
- 价格区间检查
|
||||
- 仓位限制检查
|
||||
- 市场分类检查
|
||||
- 订单簿检查（获取最佳卖单价格）
|
||||
|
||||
5. **价格调整**：
|
||||
- 应用价格容忍度（`priceTolerance`）
|
||||
- 买入价格 = Leader 价格 × (1 + 容忍度)
|
||||
- 确保调整后的价格不低于最佳卖单价格
|
||||
|
||||
#### 2.1.2 订单签名
|
||||
|
||||
使用 `OrderSigningService.createAndSignOrder()` 创建并签名订单：
|
||||
|
||||
1. **计算订单金额**：
|
||||
- **BUY 订单**：
|
||||
- `makerAmount = price × size`（USDC 金额，最多 2 位小数）
|
||||
- `takerAmount = size`（shares 数量，最多 4 位小数）
|
||||
- 转换为 wei（6 位小数）
|
||||
|
||||
2. **生成订单参数**：
|
||||
- `salt`: 时间戳（毫秒）
|
||||
- `maker`: 代理钱包地址（proxyAddress）
|
||||
- `signer`: 从私钥推导的签名地址
|
||||
- `taker`: 零地址（0x0000...）
|
||||
- `tokenId`: 从 outcomeIndex 获取
|
||||
- `makerAmount`: 计算出的 maker 金额（wei）
|
||||
- `takerAmount`: 计算出的 taker 金额（wei）
|
||||
- `expiration`: "0"（永不过期）
|
||||
- `nonce`: "0"
|
||||
- `feeRateBps`: "0"
|
||||
- `side`: "BUY"
|
||||
- `signatureType`: 2（Browser Wallet）
|
||||
|
||||
3. **EIP-712 签名**：
|
||||
- 编码域分隔符（Exchange Contract + Chain ID）
|
||||
- 编码订单消息哈希
|
||||
- 计算结构化数据哈希
|
||||
- 使用私钥签名（r + s + v）
|
||||
|
||||
#### 2.1.3 创建订单请求
|
||||
|
||||
构建 `NewOrderRequest`：
|
||||
|
||||
```kotlin
|
||||
NewOrderRequest(
|
||||
order = signedOrder, // 签名的订单对象
|
||||
owner = account.apiKey, // API Key
|
||||
orderType = "FAK", // Fill-And-Kill（允许部分成交，未成交部分立即取消）
|
||||
deferExec = false // 立即执行
|
||||
)
|
||||
```
|
||||
|
||||
#### 2.1.4 提交订单
|
||||
|
||||
1. **创建 CLOB API 客户端**（带认证）：
|
||||
- 使用账户的 API Key、Secret、Passphrase
|
||||
- 解密 API 凭证
|
||||
|
||||
2. **调用 API 创建订单**：
|
||||
- `POST /orders`
|
||||
- 带重试机制（最多重试 2 次）
|
||||
- 每次重试都重新生成 salt 并重新签名
|
||||
|
||||
3. **记录订单跟踪**：
|
||||
- 保存到 `copy_order_tracking` 表
|
||||
- 记录买入订单 ID、数量、价格等信息
|
||||
- 状态：`filled`
|
||||
|
||||
### 2.2 卖出订单构建流程
|
||||
|
||||
#### 2.2.1 前置检查
|
||||
|
||||
1. **查找跟单关系**：
|
||||
- 查询所有启用且支持该 Leader 的跟单配置
|
||||
- 验证是否支持卖出（`supportSell = true`）
|
||||
|
||||
2. **计算需要匹配的数量**：
|
||||
- `需要匹配数量 = Leader 卖出数量 × 跟单比例`
|
||||
|
||||
3. **查找未匹配的买入订单**：
|
||||
- 使用 `outcomeIndex` 匹配（支持多元市场）
|
||||
- 按 FIFO 顺序（先进先出）
|
||||
- 查询 `copy_order_tracking` 表中未匹配的订单
|
||||
|
||||
4. **计算实际可卖出数量**：
|
||||
- 按 FIFO 顺序匹配
|
||||
- 支持部分匹配（一个买入订单可以被多次卖出匹配）
|
||||
|
||||
#### 2.2.2 价格计算
|
||||
|
||||
1. **优先使用订单簿 bestBid**：
|
||||
- 查询订单簿（`getOrderbookByTokenId`）
|
||||
- 获取最佳买单价格（bestBid）
|
||||
- 使用 bestBid 作为卖出价格
|
||||
|
||||
2. **备选方案**：
|
||||
- 如果获取订单簿失败，使用 Leader 价格
|
||||
- 卖出价格 = Leader 价格 × 0.9（固定按 90% 计算）
|
||||
|
||||
#### 2.2.3 订单签名
|
||||
|
||||
与买入订单类似，但：
|
||||
- `side`: "SELL"
|
||||
- **SELL 订单金额计算**：
|
||||
- `makerAmount = size`（shares 数量，最多 4 位小数）
|
||||
- `takerAmount = price × size`（USDC 金额，使用原始价格计算）
|
||||
|
||||
#### 2.2.4 创建订单请求
|
||||
|
||||
与买入订单相同：
|
||||
- `orderType`: "FAK"
|
||||
- `deferExec`: false
|
||||
|
||||
#### 2.2.5 提交订单
|
||||
|
||||
1. **调用 API 创建卖出订单**（带重试机制）
|
||||
|
||||
2. **更新买入订单状态**：
|
||||
- 更新 `copy_order_tracking` 表中的 `remainingQuantity`
|
||||
- 如果完全匹配，状态更新为 `fully_matched`
|
||||
- 如果部分匹配，状态更新为 `partially_matched`
|
||||
|
||||
3. **记录匹配关系**：
|
||||
- 保存到 `sell_match_record` 表（卖出匹配记录）
|
||||
- 保存到 `sell_match_detail` 表（匹配明细，包含盈亏计算）
|
||||
|
||||
## 三、关键配置
|
||||
|
||||
### 3.1 跟单配置参数
|
||||
|
||||
- `copyMode`: 跟单模式（"RATIO" 或 "FIXED"）
|
||||
- `copyRatio`: 跟单比例（RATIO 模式）
|
||||
- `fixedAmount`: 固定金额（FIXED 模式）
|
||||
- `priceTolerance`: 价格容忍度（买入时使用）
|
||||
- `supportSell`: 是否支持卖出
|
||||
- `enabled`: 是否启用
|
||||
|
||||
### 3.2 订单类型
|
||||
|
||||
- **FAK (Fill-And-Kill)**：
|
||||
- 允许部分成交
|
||||
- 未成交部分立即取消
|
||||
- 快速响应 Leader 交易，避免订单长期挂单导致价格不匹配
|
||||
|
||||
### 3.3 重试机制
|
||||
|
||||
- **最多重试次数**：2 次（首次 + 1 次重试）
|
||||
- **重试延迟**：3 秒
|
||||
- **重试策略**：每次重试都重新生成 salt 并重新签名，确保签名唯一性
|
||||
|
||||
## 四、数据流向
|
||||
|
||||
```
|
||||
Leader 交易（链上）
|
||||
↓
|
||||
Polymarket Data API (/activity)
|
||||
↓
|
||||
轮询服务（CopyTradingPollingService）
|
||||
↓
|
||||
交易处理服务（CopyOrderTrackingService）
|
||||
↓
|
||||
订单签名服务（OrderSigningService）
|
||||
↓
|
||||
CLOB API (POST /orders)
|
||||
↓
|
||||
订单跟踪表（copy_order_tracking）
|
||||
```
|
||||
|
||||
## 五、注意事项
|
||||
|
||||
1. **去重机制**：
|
||||
- 使用 `leaderId + tradeId` 作为唯一标识
|
||||
- 在 `processed_trade` 表中记录已处理的交易
|
||||
- 避免重复处理同一笔交易
|
||||
|
||||
2. **价格调整**：
|
||||
- 买入时应用价格容忍度（提高买入价格）
|
||||
- 卖出时优先使用订单簿 bestBid，失败则使用 Leader 价格的 90%
|
||||
|
||||
3. **订单簿检查**：
|
||||
- 买入前检查订单簿中是否有可匹配的卖单
|
||||
- 确保调整后的买入价格不低于最佳卖单价格
|
||||
|
||||
4. **FIFO 匹配**：
|
||||
- 卖出时按买入时间顺序匹配（先进先出）
|
||||
- 支持部分匹配
|
||||
|
||||
5. **错误处理**：
|
||||
- 订单创建失败时记录到 `failed_trade` 表
|
||||
- 支持重试机制（最多 2 次）
|
||||
- 发送失败通知（如果配置了 `pushFailedOrders`）
|
||||
|
||||
@@ -0,0 +1,653 @@
|
||||
# 跟单监听策略：链上 WebSocket + 轮询并行方案
|
||||
|
||||
## 一、方案概述
|
||||
|
||||
系统**同时运行**两种监听方式，并行处理，哪个数据先返回就用哪个：
|
||||
|
||||
1. **链上 WebSocket 监听**：通过 Polygon RPC 的 `eth_subscribe` 实时监听链上交易
|
||||
2. **轮询监听**：通过 Polymarket Data API 定期轮询交易记录
|
||||
|
||||
**核心特点**：
|
||||
- 两种方式**并行运行**，不互相排斥
|
||||
- WS 断开时**不断重试连接**，不停止轮询
|
||||
- 哪个数据先返回就用哪个，确保最快响应
|
||||
- 通过去重机制确保同一笔交易只处理一次
|
||||
|
||||
## 二、关键流程图
|
||||
|
||||
```
|
||||
系统启动
|
||||
↓
|
||||
同时启动两种监听方式
|
||||
↓
|
||||
├─→ 链上 WS 监听（并行）
|
||||
│ ↓
|
||||
│ 尝试连接 WS RPC
|
||||
│ ↓
|
||||
│ 连接成功？
|
||||
│ ├─→ 是 → 订阅 Leader 钱包地址
|
||||
│ │ (eth_subscribe: USDC Transfer + ERC1155 Transfer)
|
||||
│ │ ↓
|
||||
│ │ 实时接收交易日志
|
||||
│ │ ↓
|
||||
│ │ 解析交易 receipt
|
||||
│ │ ↓
|
||||
│ │ 转换为 TradeResponse
|
||||
│ │ ↓
|
||||
│ │ 调用 processTrade()（去重检查）
|
||||
│ │ ↓
|
||||
│ │ 连接断开？
|
||||
│ │ ├─→ 是 → 等待重连延迟 → 重试连接（循环）
|
||||
│ │ └─→ 否 → 继续监听
|
||||
│ │
|
||||
│ └─→ 否 → 等待重连延迟 → 重试连接（循环）
|
||||
│
|
||||
└─→ 轮询监听（并行）
|
||||
↓
|
||||
定期轮询 (每 2 秒)
|
||||
↓
|
||||
查询 /activity 接口
|
||||
↓
|
||||
diff 检测新增交易
|
||||
↓
|
||||
调用 processTrade()（去重检查）
|
||||
↓
|
||||
继续轮询（循环）
|
||||
```
|
||||
|
||||
## 三、并行处理流程
|
||||
|
||||
```
|
||||
交易发生
|
||||
↓
|
||||
├─→ WS 监听（实时，秒级）
|
||||
│ ↓
|
||||
│ 收到链上日志
|
||||
│ ↓
|
||||
│ 解析并转换为 TradeResponse
|
||||
│ ↓
|
||||
│ 调用 processTrade()
|
||||
│ ↓
|
||||
│ 去重检查（processed_trade 表）
|
||||
│ ├─→ 已处理 → 跳过
|
||||
│ └─→ 未处理 → 处理交易
|
||||
│
|
||||
└─→ 轮询监听（延迟，2秒间隔）
|
||||
↓
|
||||
轮询到新交易
|
||||
↓
|
||||
转换为 TradeResponse
|
||||
↓
|
||||
调用 processTrade()
|
||||
↓
|
||||
去重检查（processed_trade 表）
|
||||
├─→ 已处理 → 跳过（WS 已处理）
|
||||
└─→ 未处理 → 处理交易
|
||||
```
|
||||
|
||||
## 三、实现方案
|
||||
|
||||
### 3.1 服务架构
|
||||
|
||||
```
|
||||
CopyTradingMonitorService (主服务)
|
||||
├─→ OnChainWsService (链上 WS 监听，独立运行)
|
||||
└─→ CopyTradingPollingService (轮询监听，独立运行)
|
||||
```
|
||||
|
||||
**关键点**：
|
||||
- 两个服务**独立运行**，互不影响
|
||||
- 主服务负责启动和协调两个服务
|
||||
- 两个服务都调用同一个 `processTrade()` 方法
|
||||
- 去重由 `processTrade()` 内部处理
|
||||
|
||||
### 3.2 WS 重连机制
|
||||
|
||||
**重连策略**：
|
||||
1. WS 连接断开时，**不停止轮询服务**
|
||||
2. 等待重连延迟（如 3 秒）
|
||||
3. 自动重试连接
|
||||
4. 连接成功后重新订阅所有 Leader
|
||||
5. 如果连接失败，继续重试（无限重试）
|
||||
|
||||
**重连流程**：
|
||||
```
|
||||
WS 连接断开
|
||||
↓
|
||||
记录断开日志
|
||||
↓
|
||||
等待重连延迟（3秒）
|
||||
↓
|
||||
尝试重新连接
|
||||
↓
|
||||
连接成功？
|
||||
├─→ 是 → 重新订阅所有 Leader → 继续监听
|
||||
└─→ 否 → 等待重连延迟 → 继续重试（循环）
|
||||
```
|
||||
|
||||
**关键实现**：
|
||||
- 使用协程或后台线程持续重试
|
||||
- 不阻塞主流程
|
||||
- 轮询服务继续运行，不受 WS 状态影响
|
||||
|
||||
### 3.3 去重机制
|
||||
|
||||
**去重标识**：
|
||||
- 使用 `leaderId + transactionHash` 作为唯一标识
|
||||
- 在 `processed_trade` 表中记录已处理的交易
|
||||
|
||||
**去重流程**：
|
||||
```
|
||||
收到交易数据（WS 或轮询）
|
||||
↓
|
||||
调用 processTrade(leaderId, trade, source)
|
||||
↓
|
||||
检查 processed_trade 表
|
||||
├─→ 已存在 → 跳过处理（返回成功）
|
||||
└─→ 不存在 → 继续处理
|
||||
↓
|
||||
处理交易（创建订单等）
|
||||
↓
|
||||
保存到 processed_trade 表
|
||||
├─→ leaderId
|
||||
├─→ leaderTradeId (transactionHash)
|
||||
├─→ tradeType (BUY/SELL)
|
||||
├─→ source (onchain-ws / polling)
|
||||
└─→ status (SUCCESS/FAILED)
|
||||
```
|
||||
|
||||
**并发安全**：
|
||||
- 使用数据库唯一约束（`leaderId + leaderTradeId`）
|
||||
- 处理唯一约束冲突（并发情况下可能多个请求同时处理同一笔交易）
|
||||
- 如果冲突，再次查询确认状态
|
||||
|
||||
### 3.4 链上 WS 监听实现要点
|
||||
|
||||
**订阅参数**：
|
||||
- 钱包地址：Leader 的 `leaderAddress`
|
||||
- 订阅类型：
|
||||
- USDC Transfer（`from` 或 `to` 为钱包地址）
|
||||
- ERC1155 TransferSingle/Batch（`from` 或 `to` 为钱包地址）
|
||||
|
||||
**消息处理流程**：
|
||||
1. 接收 `eth_subscription` 消息
|
||||
2. 提取 `transactionHash`
|
||||
3. 调用 RPC 获取交易 receipt
|
||||
4. 解析 USDC Transfer 和 ERC1155 Transfer 日志
|
||||
5. 计算交易方向（BUY/SELL）、数量、价格
|
||||
6. 调用 Gamma API 补齐市场元数据（conditionId、outcomeIndex 等）
|
||||
7. 转换为 `TradeResponse`
|
||||
8. 调用 `processTrade(leaderId, trade, "onchain-ws")` 处理交易（去重由内部处理）
|
||||
|
||||
**重连实现**：
|
||||
```kotlin
|
||||
// 伪代码示例
|
||||
while (isActive) {
|
||||
try {
|
||||
// 尝试连接
|
||||
val ws = connectWebSocket()
|
||||
|
||||
// 订阅所有 Leader
|
||||
subscribeAllLeaders(ws)
|
||||
|
||||
// 监听消息
|
||||
ws.listen { message ->
|
||||
handleMessage(message)
|
||||
}
|
||||
|
||||
// 连接断开，等待重连
|
||||
waitReconnectDelay()
|
||||
} catch (e: Exception) {
|
||||
logger.warn("WS 连接失败，等待重连: ${e.message}")
|
||||
waitReconnectDelay()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 3.5 轮询监听实现要点
|
||||
|
||||
**轮询流程**：
|
||||
1. 定期查询 `/activity` 接口（每 2 秒）
|
||||
2. 通过 diff 检测新增交易
|
||||
3. 转换为 `TradeResponse`
|
||||
4. 调用 `processTrade(leaderId, trade, "polling")` 处理交易（去重由内部处理）
|
||||
|
||||
**关键点**：
|
||||
- 轮询服务**独立运行**，不受 WS 状态影响
|
||||
- 即使 WS 正常工作，轮询也继续运行（作为备份）
|
||||
- 去重机制确保不会重复处理同一笔交易
|
||||
|
||||
## 四、配置参数
|
||||
|
||||
### 4.1 RPC 配置获取
|
||||
|
||||
**RPC 配置从后台配置中读取**，通过 `RpcNodeService` 获取：
|
||||
|
||||
```kotlin
|
||||
// 获取 HTTP RPC URL
|
||||
val httpUrl = rpcNodeService.getHttpUrl()
|
||||
|
||||
// 获取 WebSocket RPC URL
|
||||
val wsUrl = rpcNodeService.getWsUrl()
|
||||
|
||||
// 获取可用节点配置（包含完整信息）
|
||||
val nodeResult = rpcNodeService.getAvailableNode()
|
||||
if (nodeResult.isSuccess) {
|
||||
val node = nodeResult.getOrNull()
|
||||
val httpUrl = node?.httpUrl
|
||||
val wsUrl = node?.wsUrl
|
||||
}
|
||||
```
|
||||
|
||||
**配置管理**：
|
||||
- **RPC 节点配置存储在数据库**（`rpc_node_config` 表），不从配置文件读取
|
||||
- 支持多个节点配置，按优先级选择
|
||||
- 支持健康检查，自动选择可用节点
|
||||
- 前端可以通过系统设置页面配置 RPC 节点
|
||||
- 配置变更后，WS 重连时会自动使用新配置
|
||||
|
||||
**配置字段**：
|
||||
- `httpUrl`: HTTP RPC 地址
|
||||
- `wsUrl`: WebSocket RPC 地址（可选）
|
||||
- `enabled`: 是否启用
|
||||
- `priority`: 优先级（数字越小优先级越高）
|
||||
- `lastCheckStatus`: 最后检查状态（HEALTHY/UNHEALTHY/UNKNOWN）
|
||||
|
||||
### 4.2 WS 连接配置
|
||||
|
||||
```properties
|
||||
# WS 连接超时（毫秒）
|
||||
polygen.ws.connect.timeout=5000
|
||||
|
||||
# WS 重连延迟（毫秒）
|
||||
polygen.ws.reconnect.delay=3000
|
||||
```
|
||||
|
||||
### 4.3 WS 重连配置
|
||||
|
||||
```properties
|
||||
# WS 重连延迟（毫秒）
|
||||
polygen.ws.reconnect.delay=3000
|
||||
|
||||
# WS 重连最大延迟（毫秒，指数退避上限）
|
||||
polygen.ws.reconnect.max.delay=60000
|
||||
|
||||
# WS 重连是否启用指数退避
|
||||
polygen.ws.reconnect.exponential.backoff=true
|
||||
```
|
||||
|
||||
### 4.4 合约地址配置
|
||||
|
||||
```properties
|
||||
# USDC 合约地址
|
||||
usdc.contract.address=0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174
|
||||
|
||||
# ERC1155 合约地址（Polymarket）
|
||||
erc1155.contract.address=0x4d97dcd97ec945f40cf65f87097ace5ea0476045
|
||||
```
|
||||
|
||||
## 五、并行方案优势
|
||||
|
||||
### 5.1 双重保障
|
||||
|
||||
- **实时性**：WS 提供秒级实时通知
|
||||
- **可靠性**：轮询作为备份，确保不遗漏交易
|
||||
- **容错性**：WS 断开时轮询继续工作，WS 恢复后自动继续
|
||||
|
||||
### 5.2 性能优化
|
||||
|
||||
- **最快响应**：哪个数据先返回就用哪个
|
||||
- **负载均衡**：两种方式并行，减少单点压力
|
||||
- **去重保护**：确保同一笔交易只处理一次
|
||||
|
||||
### 5.3 实现简单
|
||||
|
||||
- **无需切换逻辑**：两种方式始终运行
|
||||
- **独立管理**：WS 和轮询各自管理自己的状态
|
||||
- **易于维护**：逻辑清晰，易于调试
|
||||
|
||||
## 六、注意事项
|
||||
|
||||
1. **WS 重连策略**：
|
||||
- WS 断开时**不断重试**，不停止轮询
|
||||
- 使用指数退避策略，避免频繁重连
|
||||
- 记录重连日志，便于监控
|
||||
|
||||
2. **去重机制**：
|
||||
- 使用 `leaderId + transactionHash` 作为唯一标识
|
||||
- 在 `processTrade()` 方法内部统一处理去重
|
||||
- 使用数据库唯一约束确保并发安全
|
||||
|
||||
3. **性能考虑**：
|
||||
- WS 模式下需要为每个 Leader 订阅日志
|
||||
- 大量 Leader 时可能需要优化订阅策略（批量订阅）
|
||||
- 轮询间隔可以适当调整（默认 2 秒）
|
||||
|
||||
4. **错误处理**：
|
||||
- WS 连接失败不影响轮询服务
|
||||
- 记录详细的错误日志，便于排查问题
|
||||
- 两种方式的错误独立处理，互不影响
|
||||
|
||||
5. **元数据补齐**：
|
||||
- 链上数据不包含市场元数据（conditionId、outcomeIndex 等）
|
||||
- 需要通过 Gamma API 或内部元数据服务补齐
|
||||
- 轮询数据已包含元数据，无需额外补齐
|
||||
|
||||
6. **数据源标识**：
|
||||
- WS 数据：`source = "onchain-ws"`
|
||||
- 轮询数据：`source = "polling"`
|
||||
- 便于统计和分析不同数据源的处理情况
|
||||
|
||||
## 七、实现要点
|
||||
|
||||
### 7.1 主服务启动
|
||||
|
||||
```kotlin
|
||||
@Service
|
||||
class CopyTradingMonitorService(
|
||||
private val rpcNodeService: RpcNodeService,
|
||||
private val onChainWsService: OnChainWsService,
|
||||
private val pollingService: CopyTradingPollingService
|
||||
) {
|
||||
@PostConstruct
|
||||
fun init() {
|
||||
scope.launch {
|
||||
// 同时启动两种监听方式
|
||||
launch { onChainWsService.start() } // WS 监听（独立协程）
|
||||
launch { pollingService.start() } // 轮询监听（独立协程）
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 7.2 WS 服务实现（从后台配置获取 RPC）
|
||||
|
||||
```kotlin
|
||||
@Service
|
||||
class OnChainWsService(
|
||||
private val rpcNodeService: RpcNodeService,
|
||||
private val copyOrderTrackingService: CopyOrderTrackingService
|
||||
) {
|
||||
suspend fun start() {
|
||||
while (isActive) {
|
||||
try {
|
||||
// 从后台配置获取 WS RPC URL
|
||||
val wsUrl = rpcNodeService.getWsUrl()
|
||||
val httpUrl = rpcNodeService.getHttpUrl()
|
||||
|
||||
// 连接并订阅
|
||||
connectAndSubscribe(wsUrl, httpUrl)
|
||||
|
||||
// 连接成功后持续监听
|
||||
waitForDisconnect()
|
||||
} catch (e: Exception) {
|
||||
logger.warn("WS 连接失败，等待重连: ${e.message}")
|
||||
delay(reconnectDelay)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun connectAndSubscribe(wsUrl: String, httpUrl: String) {
|
||||
// 使用 wsUrl 和 httpUrl 连接和订阅
|
||||
// ...
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**关键点**：
|
||||
- 每次重连时都从 `RpcNodeService` 获取最新的 RPC 配置
|
||||
- 如果配置变更，下次重连时会自动使用新配置
|
||||
- 支持多个节点配置，自动选择可用节点
|
||||
|
||||
### 7.3 RPC 配置变更处理
|
||||
|
||||
**配置变更时的处理**：
|
||||
- WS 连接断开时，下次重连会从 `RpcNodeService` 获取最新配置
|
||||
- 如果当前使用的节点不可用，`getAvailableNode()` 会自动选择下一个可用节点
|
||||
- 支持动态切换节点，无需重启服务
|
||||
|
||||
**节点选择策略**：
|
||||
1. 优先使用 `lastCheckStatus = HEALTHY` 的节点
|
||||
2. 按 `priority` 排序，数字越小优先级越高
|
||||
3. 如果所有节点都不可用，返回失败（不降级到默认节点，因为默认节点可能也不可用）
|
||||
|
||||
### 7.4 线程安全保证（单实例推荐方案）
|
||||
|
||||
`processTrade` 方法需要保证线程安全，因为：
|
||||
- WS 和轮询可能同时处理同一笔交易
|
||||
- 多个协程可能并发调用 `processTrade`
|
||||
- 需要确保同一笔交易只处理一次
|
||||
|
||||
**推荐方案：使用 Mutex（应用级锁）**
|
||||
|
||||
对于单实例部署，**最轻量的方案是使用 Kotlin 协程的 Mutex**：
|
||||
- ✅ 无需额外依赖（不需要 Redis）
|
||||
- ✅ 性能开销小（内存锁，无网络开销）
|
||||
- ✅ 实现简单（Kotlin 标准库）
|
||||
- ✅ 协程友好（支持 suspend 函数）
|
||||
|
||||
**实现代码**：
|
||||
|
||||
```kotlin
|
||||
@Service
|
||||
open class CopyOrderTrackingService(
|
||||
// ... 其他依赖
|
||||
) {
|
||||
// 使用 Mutex 保证线程安全（按交易ID锁定）
|
||||
private val tradeMutexMap = ConcurrentHashMap<String, Mutex>()
|
||||
|
||||
/**
|
||||
* 获取或创建 Mutex（按交易ID）
|
||||
*/
|
||||
private fun getMutex(leaderId: Long, tradeId: String): Mutex {
|
||||
val key = "${leaderId}_${tradeId}"
|
||||
return tradeMutexMap.getOrPut(key) { Mutex() }
|
||||
}
|
||||
|
||||
/**
|
||||
* 清理不再使用的 Mutex（可选，避免内存泄漏）
|
||||
*/
|
||||
private fun cleanupMutex(leaderId: Long, tradeId: String) {
|
||||
val key = "${leaderId}_${tradeId}"
|
||||
// 延迟清理，避免频繁创建/删除
|
||||
// 可以定期清理或使用 WeakReference
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理交易事件（WebSocket 或轮询）
|
||||
* 使用 Mutex 保证线程安全
|
||||
*/
|
||||
@Transactional
|
||||
suspend fun processTrade(leaderId: Long, trade: TradeResponse, source: String): Result<Unit> {
|
||||
// 获取该交易的 Mutex
|
||||
val mutex = getMutex(leaderId, trade.id)
|
||||
|
||||
return mutex.withLock {
|
||||
try {
|
||||
// 1. 检查是否已处理（去重）
|
||||
val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId(
|
||||
leaderId,
|
||||
trade.id
|
||||
)
|
||||
|
||||
if (existingProcessed != null) {
|
||||
if (existingProcessed.status == "FAILED") {
|
||||
return@withLock Result.success(Unit)
|
||||
}
|
||||
return@withLock Result.success(Unit)
|
||||
}
|
||||
|
||||
// 检查是否已记录为失败交易
|
||||
val failedTrade = failedTradeRepository.findByLeaderIdAndLeaderTradeId(
|
||||
leaderId,
|
||||
trade.id
|
||||
)
|
||||
if (failedTrade != null) {
|
||||
return@withLock Result.success(Unit)
|
||||
}
|
||||
|
||||
// 2. 处理交易逻辑
|
||||
val result = when (trade.side.uppercase()) {
|
||||
"BUY" -> processBuyTrade(leaderId, trade)
|
||||
"SELL" -> processSellTrade(leaderId, trade)
|
||||
else -> {
|
||||
logger.warn("未知的交易方向: ${trade.side}")
|
||||
Result.failure(IllegalArgumentException("未知的交易方向: ${trade.side}"))
|
||||
}
|
||||
}
|
||||
|
||||
if (result.isFailure) {
|
||||
logger.error(
|
||||
"处理交易失败: leaderId=$leaderId, tradeId=${trade.id}, side=${trade.side}",
|
||||
result.exceptionOrNull()
|
||||
)
|
||||
return@withLock result
|
||||
}
|
||||
|
||||
// 3. 标记为已处理（成功状态）
|
||||
// 由于使用了 Mutex，这里不会出现并发冲突
|
||||
try {
|
||||
val processed = ProcessedTrade(
|
||||
leaderId = leaderId,
|
||||
leaderTradeId = trade.id,
|
||||
tradeType = trade.side.uppercase(),
|
||||
source = source,
|
||||
status = "SUCCESS",
|
||||
processedAt = System.currentTimeMillis()
|
||||
)
|
||||
processedTradeRepository.save(processed)
|
||||
} catch (e: Exception) {
|
||||
// 理论上不会发生，但保留异常处理作为兜底
|
||||
if (isUniqueConstraintViolation(e)) {
|
||||
val existing = processedTradeRepository.findByLeaderIdAndLeaderTradeId(
|
||||
leaderId,
|
||||
trade.id
|
||||
)
|
||||
if (existing != null) {
|
||||
logger.debug("交易已处理（并发检测）: leaderId=$leaderId, tradeId=${trade.id}")
|
||||
return@withLock Result.success(Unit)
|
||||
}
|
||||
} else {
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
Result.success(Unit)
|
||||
} catch (e: Exception) {
|
||||
logger.error("处理交易异常: leaderId=$leaderId, tradeId=${trade.id}", e)
|
||||
Result.failure(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**关键点**：
|
||||
1. **按交易ID锁定**：每个交易使用独立的 Mutex，不同交易可以并行处理
|
||||
2. **Mutex 复用**：使用 `ConcurrentHashMap` 缓存 Mutex，避免频繁创建
|
||||
3. **协程友好**：`Mutex.withLock` 是 suspend 函数，不会阻塞线程
|
||||
4. **性能优化**：只锁定同一笔交易，不影响其他交易的并发处理
|
||||
|
||||
**当前实现（基于数据库唯一约束，作为兜底）**：
|
||||
|
||||
```kotlin
|
||||
@Transactional
|
||||
suspend fun processTrade(leaderId: Long, trade: TradeResponse, source: String): Result<Unit> {
|
||||
// 1. 检查是否已处理（去重）
|
||||
val existingProcessed = processedTradeRepository.findByLeaderIdAndLeaderTradeId(
|
||||
leaderId,
|
||||
trade.id
|
||||
)
|
||||
|
||||
if (existingProcessed != null) {
|
||||
return Result.success(Unit) // 已处理，跳过
|
||||
}
|
||||
|
||||
// 2. 处理交易逻辑
|
||||
val result = when (trade.side.uppercase()) {
|
||||
"BUY" -> processBuyTrade(leaderId, trade)
|
||||
"SELL" -> processSellTrade(leaderId, trade)
|
||||
else -> Result.failure(IllegalArgumentException("未知的交易方向"))
|
||||
}
|
||||
|
||||
if (result.isFailure) {
|
||||
return result
|
||||
}
|
||||
|
||||
// 3. 标记为已处理（使用数据库唯一约束保证并发安全）
|
||||
try {
|
||||
val processed = ProcessedTrade(
|
||||
leaderId = leaderId,
|
||||
leaderTradeId = trade.id,
|
||||
tradeType = trade.side.uppercase(),
|
||||
source = source,
|
||||
status = "SUCCESS",
|
||||
processedAt = System.currentTimeMillis()
|
||||
)
|
||||
processedTradeRepository.save(processed)
|
||||
} catch (e: Exception) {
|
||||
// 处理唯一约束冲突（并发情况下可能发生）
|
||||
if (isUniqueConstraintViolation(e)) {
|
||||
// 再次检查确认状态
|
||||
val existing = processedTradeRepository.findByLeaderIdAndLeaderTradeId(
|
||||
leaderId,
|
||||
trade.id
|
||||
)
|
||||
if (existing != null) {
|
||||
logger.debug("交易已处理（并发检测）: leaderId=$leaderId, tradeId=${trade.id}")
|
||||
return Result.success(Unit)
|
||||
}
|
||||
} else {
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
return Result.success(Unit)
|
||||
}
|
||||
```
|
||||
|
||||
**线程安全机制**：
|
||||
|
||||
1. **数据库唯一约束**：
|
||||
- `UNIQUE KEY uk_leader_trade (leader_id, leader_trade_id)`
|
||||
- 确保同一笔交易只能插入一次
|
||||
- 即使两个线程同时通过检查，也只有一个能成功保存
|
||||
|
||||
2. **事务隔离**：
|
||||
- 使用 `@Transactional` 注解
|
||||
- 默认隔离级别（通常是 READ_COMMITTED）
|
||||
- 保证事务内的数据一致性
|
||||
|
||||
3. **异常处理**：
|
||||
- 捕获唯一约束冲突异常
|
||||
- 重新查询确认状态
|
||||
- 避免重复处理
|
||||
|
||||
**潜在问题**：
|
||||
|
||||
虽然数据库唯一约束可以防止重复插入，但在检查 `existingProcessed` 和保存 `processed` 之间存在时间窗口（TOCTOU），可能导致：
|
||||
- 两个线程同时通过检查
|
||||
- 两个线程都执行 `processBuyTrade` 或 `processSellTrade`
|
||||
- 虽然只有一个能成功保存 `processed`，但可能创建了重复的订单
|
||||
|
||||
**其他方案（不推荐，仅作参考）**：
|
||||
|
||||
**方案 1：使用数据库锁（SELECT FOR UPDATE）**
|
||||
- ❌ 需要修改 Repository 方法
|
||||
- ❌ 增加数据库负载
|
||||
- ❌ 不适合高并发场景
|
||||
|
||||
**方案 2：使用分布式锁（Redis，仅适用于多实例）**
|
||||
- ❌ 需要 Redis 依赖
|
||||
- ❌ 有网络开销
|
||||
- ❌ 单实例场景不需要
|
||||
|
||||
**总结**：
|
||||
|
||||
- ✅ **单实例部署推荐**：使用 **Mutex（应用级锁）**
|
||||
- 最轻量：无需额外依赖，性能开销小
|
||||
- 最简单：Kotlin 标准库，实现简单
|
||||
- 最高效：内存锁，无网络开销，支持协程并发
|
||||
- ❌ **不推荐**：仅依赖数据库唯一约束（存在 TOCTOU 问题，可能创建重复订单）
|
||||
- ❌ **不推荐**：数据库锁或分布式锁（单实例场景过于复杂）
|
||||
|
||||
Reference in New Issue
Block a user