feat: enhance copy trading logic and UI

- fix: correct copyRatio scaling (divide by 100) - fix: enforce minimum order size with round-up logic - feat: customize Telegram notification icons - feat: optimize frontend order list (active loading, refresh buttons) - refactor: optimize OnChain WebSocket connection management
2026-01-04 20:36:41 +08:00
parent 2bb8cbc564
commit 3e667b70fd
15 changed files with 630 additions and 594 deletions
@@ -383,9 +383,9 @@ class PositionCheckService(
                        if (ordersToMarkAsSold.isNotEmpty()) {
                            // 有订单创建时间超过2分钟，认为仓位已被出售
                            try {
-                                val currentPrice = getCurrentMarketPrice(marketId, outcomeIndex)
-                                updateOrdersAsSold(ordersToMarkAsSold, currentPrice, copyTrading.id, marketId, outcomeIndex)
-                                logger.debug("仓位不存在且订单创建时间超过2分钟，标记为已卖出: marketId=$marketId, outcomeIndex=$outcomeIndex, orderCount=${ordersToMarkAsSold.size}")
+                            val currentPrice = getCurrentMarketPrice(marketId, outcomeIndex)
+                            updateOrdersAsSold(ordersToMarkAsSold, currentPrice, copyTrading.id, marketId, outcomeIndex)
+                            logger.debug("仓位不存在且订单创建时间超过2分钟，标记为已卖出: marketId=$marketId, outcomeIndex=$outcomeIndex, orderCount=${ordersToMarkAsSold.size}")
                            } catch (e: Exception) {
                                logger.warn("无法获取市场价格，跳过标记为已卖出: marketId=$marketId, outcomeIndex=$outcomeIndex, error=${e.message}")
                                // 无法获取价格时，跳过该市场的处理，等待下次检查时再试
@@ -420,9 +420,9 @@ class PositionCheckService(
                        
                        // 如果已成交数量 > 0，按FIFO顺序匹配订单
                        try {
-                            val currentPrice = getCurrentMarketPrice(marketId, outcomeIndex)
-                            updateOrdersAsSoldByFIFO(orders, soldQuantity, currentPrice,
-                                copyTrading.id, marketId, outcomeIndex)
+                        val currentPrice = getCurrentMarketPrice(marketId, outcomeIndex)
+                        updateOrdersAsSoldByFIFO(orders, soldQuantity, currentPrice,
+                            copyTrading.id, marketId, outcomeIndex)
                        } catch (e: Exception) {
                            logger.warn("无法获取市场价格，跳过FIFO匹配: marketId=$marketId, outcomeIndex=$outcomeIndex, error=${e.message}")
                            // 无法获取价格时，跳过该市场的处理，等待下次检查时再试
@@ -715,7 +715,7 @@ class BlockchainService(
    
    /**
     * 从链上查询市场条件（Condition）的结算结果
-     * 通过调用 ConditionalTokens 合约的 getCondition 函数获取 payouts
+     * 通过调用 ConditionalTokens 合约的 conditions mapping 和 payoutNumerators mapping
     * 
     * @param conditionId 市场条件ID（bytes32，必须是 0x 开头的 66 位十六进制字符串）
     * @return Result<Pair<payoutDenominator, payouts>>
@@ -734,44 +734,101 @@ class BlockchainService(
            
            val rpcApi = polygonRpcApi
            
-            // 构建 getCondition(bytes32) 函数调用
-            // 函数签名: getCondition(bytes32)
-            val functionSelector = EthereumUtils.getFunctionSelector("getCondition(bytes32)")
+            // 1. 调用 conditions(bytes32) 获取 outcomeSlotCount 和 payoutDenominator
+            // 注意：这是一个公开的 mapping，Solidity 自动生成的 getter
+            // 函数签名: conditions(bytes32) returns (uint outcomeSlotCount, uint payoutDenominator)
+            val conditionsFunctionSelector = EthereumUtils.getFunctionSelector("conditions(bytes32)")
            val encodedConditionId = EthereumUtils.encodeBytes32(conditionId)
-            val data = functionSelector + encodedConditionId
+            val conditionsData = conditionsFunctionSelector + encodedConditionId
            
            // 构建 JSON-RPC 请求
-            val rpcRequest = JsonRpcRequest(
+            val conditionsRequest = JsonRpcRequest(
                method = "eth_call",
                params = listOf(
                    mapOf(
                        "to" to conditionalTokensAddress,
-                        "data" to data
+                        "data" to conditionsData
                    ),
                    "latest"
                )
            )
            
            // 发送 RPC 请求
-            val response = rpcApi.call(rpcRequest)
+            val conditionsResponse = rpcApi.call(conditionsRequest)
            
-            if (!response.isSuccessful || response.body() == null) {
-                return Result.failure(Exception("RPC 请求失败: ${response.code()} ${response.message()}"))
+            if (!conditionsResponse.isSuccessful || conditionsResponse.body() == null) {
+                return Result.failure(Exception("RPC 请求失败 (conditions): ${conditionsResponse.code()} ${conditionsResponse.message()}"))
            }
            
-            val rpcResponse = response.body()!!
+            val conditionsRpcResponse = conditionsResponse.body()!!
            
            // 检查错误
-            if (rpcResponse.error != null) {
-                return Result.failure(Exception("RPC 错误: ${rpcResponse.error.message}"))
+            if (conditionsRpcResponse.error != null) {
+                // 记录完整的错误信息，包括 code 和 data
+                val errorMsg = "RPC 错误 (code=${conditionsRpcResponse.error.code}): ${conditionsRpcResponse.error.message}, data=${conditionsRpcResponse.error.data}"
+                logger.error("查询市场条件(conditions)出现RPC错误: conditionId=$conditionId, $errorMsg")
+                logger.debug("RPC 请求详情: to=$conditionalTokensAddress, data=$conditionsData")
+                return Result.failure(Exception(errorMsg))
            }
            
            // 使用 Gson 解析 result（JsonElement）
-            val hexResult = rpcResponse.result?.asString 
+            val hexResult = conditionsRpcResponse.result?.asString 
                ?: return Result.failure(Exception("RPC 响应格式错误: result 为空"))
            
-            // 解析 ABI 编码的返回结果
-            val (payoutDenominator, payouts) = EthereumUtils.decodeConditionResult(hexResult)
+            // 解析返回的 (outcomeSlotCount, payoutDenominator)
+            val cleanHex = hexResult.removePrefix("0x")
+            val outcomeSlotCountHex = cleanHex.substring(0, 64)
+            val payoutDenominatorHex = cleanHex.substring(64, 128)
+            
+            val outcomeSlotCount = BigInteger(outcomeSlotCountHex, 16).toInt()
+            val payoutDenominator = BigInteger(payoutDenominatorHex, 16)
+            
+            // 如果 outcomeSlotCount 为 0，说明市场尚未创建或不存在
+            if (outcomeSlotCount <= 0) {
+                logger.debug("市场尚未创建或不存在: conditionId=$conditionId, outcomeSlotCount=$outcomeSlotCount")
+                return Result.success(Pair(BigInteger.ZERO, emptyList()))
+            }
+            
+            // 如果 payoutDenominator 为 0，说明市场尚未结算
+            if (payoutDenominator == BigInteger.ZERO) {
+                logger.debug("市场尚未结算: conditionId=$conditionId, payoutDenominator=$payoutDenominator")
+                return Result.success(Pair(BigInteger.ZERO, emptyList()))
+            }
+            
+            // 2. 查询每个 outcome 的 payoutNumerators
+            val payouts = mutableListOf<BigInteger>()
+            for (i in 0 until outcomeSlotCount) {
+                val payoutNumeratorsFunctionSelector = EthereumUtils.getFunctionSelector("payoutNumerators(bytes32,uint256)")
+                val encodedIndex = EthereumUtils.encodeUint256(BigInteger.valueOf(i.toLong()))
+                val payoutNumeratorsData = payoutNumeratorsFunctionSelector + encodedConditionId + encodedIndex
+                
+                val payoutRequest = JsonRpcRequest(
+                    method = "eth_call",
+                    params = listOf(
+                        mapOf(
+                            "to" to conditionalTokensAddress,
+                            "data" to payoutNumeratorsData
+                        ),
+                        "latest"
+                    )
+                )
+                
+                val payoutResponse = rpcApi.call(payoutRequest)
+                if (!payoutResponse.isSuccessful || payoutResponse.body() == null) {
+                    logger.warn("查询 payoutNumerators 失败: index=$i")
+                    continue
+                }
+                
+                val payoutRpcResponse = payoutResponse.body()!!
+                if (payoutRpcResponse.error != null) {
+                    logger.warn("查询 payoutNumerators 错误: index=$i, error=${payoutRpcResponse.error.message}")
+                    continue
+                }
+                
+                val payoutHex = payoutRpcResponse.result?.asString ?: "0x0"
+                val payout = EthereumUtils.decodeUint256(payoutHex)
+                payouts.add(payout)
+            }
            
            Result.success(Pair(payoutDenominator, payouts))
        } catch (e: Exception) {
@@ -31,7 +31,8 @@ class MarketPriceService(
     * 获取当前市场最新价
     * 优先级：
     * 1. 链上查询市场结算结果（如果已结算，返回 1.0 或 0.0）
-     * 2. CLOB API 查询订单簿价格（最准确，使用 bestBid）
+     * 2. CLOB API 查询订单簿价格（最准确，优先使用，使用 bestBid）
+     * 3. Gamma Market API 查询市场价格（快速，作为备选）
     * 
     * 价格会被截位到 4 位小数（向下截断，不四舍五入），用于显示和后续计算
     * 
@@ -48,15 +49,22 @@ class MarketPriceService(
            return chainPrice.setScale(4, java.math.RoundingMode.DOWN)
        }
        
-        // 2. 从 CLOB API 查询订单簿价格（最准确）
+        // 2. 从 CLOB API 查询订单簿价格（最准确，优先使用）
        val orderbookPrice = getPriceFromClobOrderbook(marketId, outcomeIndex)
        if (orderbookPrice != null) {
            // 截位到 4 位小数（向下截断，不四舍五入）
            return orderbookPrice.setScale(4, java.math.RoundingMode.DOWN)
        }
        
+        // 3. 从 Gamma Market API 查询市场价格（作为备选）
+        val marketPrice = getPriceFromGammaMarket(marketId, outcomeIndex)
+        if (marketPrice != null) {
+            // 截位到 4 位小数（向下截断，不四舍五入）
+            return marketPrice.setScale(4, java.math.RoundingMode.DOWN)
+        }
+        
        // 如果所有数据源都失败，抛出异常
-        val errorMsg = "无法获取市场价格: marketId=$marketId, outcomeIndex=$outcomeIndex (链上查询和订单簿查询均失败)"
+        val errorMsg = "无法获取市场价格: marketId=$marketId, outcomeIndex=$outcomeIndex (链上查询、订单簿查询和 Market API 均失败)"
        logger.error(errorMsg)
        throw IllegalStateException(errorMsg)
    }
@@ -106,6 +114,61 @@ class MarketPriceService(
        }
    }
    
+    /**
+     * 从 Gamma Market API 获取价格
+     * 使用 outcomePrices 字段，格式通常为 JSON 字符串 "[\"0.5\", \"0.5\"]"
+     * 如果查询失败或 outcomePrices 为空，返回 null
+     */
+    private suspend fun getPriceFromGammaMarket(marketId: String, outcomeIndex: Int): BigDecimal? {
+        return try {
+            val gammaApi = retrofitFactory.createGammaApi()
+            val marketResponse = gammaApi.listMarkets(conditionIds = listOf(marketId))
+            
+            if (!marketResponse.isSuccessful || marketResponse.body() == null) {
+                logger.debug("Gamma Market API 查询失败: marketId=$marketId, code=${marketResponse.code()}")
+                return null
+            }
+            
+            val markets = marketResponse.body()!!
+            if (markets.isEmpty()) {
+                logger.debug("Gamma Market API 未找到市场: marketId=$marketId")
+                return null
+            }
+            
+            val market = markets.first()
+            
+            // 尝试从 outcomePrices 字段获取价格
+            val outcomePricesStr = market.outcomePrices
+            if (outcomePricesStr.isNullOrBlank()) {
+                logger.debug("Market outcomePrices 为空: marketId=$marketId")
+                return null
+            }
+            
+            // 解析 outcomePrices（通常是 JSON 数组字符串）
+            val outcomePrices = try {
+                // 移除首尾的方括号和引号，按逗号分割
+                val cleanStr = outcomePricesStr.trim().removeSurrounding("[", "]")
+                cleanStr.split(",").map { 
+                    it.trim().removeSurrounding("\"").toSafeBigDecimal() 
+                }
+            } catch (e: Exception) {
+                logger.warn("解析 outcomePrices 失败: marketId=$marketId, outcomePrices=$outcomePricesStr, error=${e.message}")
+                null
+            }
+            
+            if (outcomePrices != null && outcomeIndex < outcomePrices.size) {
+                val price = outcomePrices[outcomeIndex]
+                logger.debug("从 Gamma Market API 获取价格: marketId=$marketId, outcomeIndex=$outcomeIndex, price=$price")
+                return price
+            }
+            
+            null
+        } catch (e: Exception) {
+            logger.debug("Gamma Market API 查询异常: marketId=$marketId, outcomeIndex=$outcomeIndex, error=${e.message}")
+            null
+        }
+    }
+    
    
    /**
     * 从 CLOB API 查询订单簿价格
@@ -89,6 +89,8 @@ class OnChainWsService(
    private suspend fun handleLeaderTransaction(leaderId: Long, txHash: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) {
        val leader = monitoredLeaders[leaderId] ?: return
        
+        logger.debug("开始处理 Leader 交易: leaderId=$leaderId, txHash=$txHash, leaderAddress=${leader.leaderAddress}")
+        
        try {
            // 获取交易 receipt
            val receiptRequest = JsonRpcRequest(
@@ -98,11 +100,13 @@ class OnChainWsService(
            
            val receiptResponse = rpcApi.call(receiptRequest)
            if (!receiptResponse.isSuccessful || receiptResponse.body() == null) {
+                logger.warn("获取交易 receipt 失败: leaderId=$leaderId, txHash=$txHash, code=${receiptResponse.code()}")
                return
            }
            
            val receiptRpcResponse = receiptResponse.body()!!
            if (receiptRpcResponse.error != null || receiptRpcResponse.result == null) {
+                logger.warn("交易 receipt 错误: leaderId=$leaderId, txHash=$txHash, error=${receiptRpcResponse.error}")
                return
            }
            
@@ -118,8 +122,12 @@ class OnChainWsService(
            }
            
            // 解析 receipt 中的 Transfer 日志
-            val logs = receiptJson.getAsJsonArray("logs") ?: return
+            val logs = receiptJson.getAsJsonArray("logs") ?: run {
+                logger.warn("交易 receipt 中没有日志: leaderId=$leaderId, txHash=$txHash")
+                return
+            }
            val (erc20Transfers, erc1155Transfers) = OnChainWsUtils.parseReceiptTransfers(logs)
+            logger.debug("解析交易日志: leaderId=$leaderId, txHash=$txHash, erc20Transfers=${erc20Transfers.size}, erc1155Transfers=${erc1155Transfers.size}")
            
            // 解析交易信息
            val trade = OnChainWsUtils.parseTradeFromTransfers(
@@ -132,12 +140,15 @@ class OnChainWsService(
            )
            
            if (trade != null) {
+                logger.info("成功解析交易: leaderId=$leaderId, txHash=$txHash, side=${trade.side}, market=${trade.market}, size=${trade.size}")
                // 调用 processTrade 处理交易
                copyOrderTrackingService.processTrade(
                    leaderId = leaderId,
                    trade = trade,
                    source = "onchain-ws"
                )
+            } else {
+                logger.warn("无法解析交易（返回 null）: leaderId=$leaderId, txHash=$txHash, erc20Transfers=${erc20Transfers.size}, erc1155Transfers=${erc1155Transfers.size}")
            }
        } catch (e: Exception) {
            logger.error("处理 Leader 交易失败: leaderId=$leaderId, txHash=$txHash, ${e.message}", e)
@@ -207,6 +207,7 @@ object OnChainWsUtils {
            usdcRaw = usdcIn
        } else {
            // 无法判断交易方向
+            logger.debug("无法判断交易方向: txHash=$txHash, bestInId=$bestInId, bestInVal=$bestInVal, bestOutId=$bestOutId, bestOutVal=$bestOutVal, usdcOut=$usdcOut, usdcIn=$usdcIn")
            return null
        }
        
@@ -1,6 +1,7 @@
 package com.wrbug.polymarketbot.service.copytrading.monitor

 import com.google.gson.Gson
+import com.google.gson.JsonArray
 import com.google.gson.JsonObject
 import com.wrbug.polymarketbot.api.*
 import com.wrbug.polymarketbot.service.system.RpcNodeService
@@ -19,6 +20,7 @@ import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.stereotype.Service
 import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicInteger

 /**
 * 统一的链上 WebSocket 服务
@@ -38,27 +40,8 @@ class UnifiedOnChainWsService(
    
    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
    
-    // WebSocket 连接（唯一）
-    private var webSocket: WebSocket? = null
-    @Volatile
-    private var isConnected = false
-    
-    // 订阅ID计数器（用于请求 ID）
-    private var requestIdCounter = 0
-    
-    // 连接任务（确保只有一个连接任务在运行）
-    private var connectionJob: Job? = null
-    
-    // 存储所有订阅：subscriptionId -> 订阅信息
-    private val subscriptions = ConcurrentHashMap<String, SubscriptionInfo>()
-    
-    // 存储请求 ID 到订阅 ID 的映射：requestId -> subscriptionId
-    // 用于在收到订阅响应时，将 subscription ID 关联到对应的订阅
-    private val requestIdToSubscriptionId = ConcurrentHashMap<Int, String>()
-    
-    // 存储 RPC subscriptionId 到订阅 ID 的映射：rpcSubscriptionId -> subscriptionId
-    // 用于在收到日志通知时，知道是哪个订阅
-    private val rpcSubscriptionIdToSubscriptionId = ConcurrentHashMap<String, String>()
+    // 存储所有地址的连接：address -> AddressWsConnection
+    private val addressConnections = ConcurrentHashMap<String, AddressWsConnection>()
    
    /**
     * 订阅信息
@@ -88,31 +71,24 @@ class UnifiedOnChainWsService(
        callback: suspend (String, OkHttpClient, EthereumRpcApi) -> Unit
    ): Boolean {
        try {
-            // 如果已经订阅，先取消
-            if (subscriptions.containsKey(subscriptionId)) {
-                unsubscribe(subscriptionId)
+            val lowerAddress = address.lowercase()
+            
+            // 找到或创建该地址的连接
+            val connection = addressConnections.computeIfAbsent(lowerAddress) {
+                AddressWsConnection(it).apply { start() }
            }
            
            // 创建订阅信息
            val subscription = SubscriptionInfo(
                subscriptionId = subscriptionId,
-                address = address.lowercase(),
+                address = lowerAddress,
                entityType = entityType,
                entityId = entityId,
                callback = callback
            )
            
-            subscriptions[subscriptionId] = subscription
-            
-            // 如果已连接，立即订阅
-            if (isConnected) {
-                scope.launch {
-                    subscribeAddress(subscription)
-                }
-            } else {
-                // 如果未连接，启动连接
-                startConnection()
-            }
+            // 添加订阅
+            connection.addSubscription(subscription)
            
            logger.info("订阅地址监听: subscriptionId=$subscriptionId, address=$address, entityType=$entityType, entityId=$entityId")
            return true
@@ -126,434 +102,37 @@ class UnifiedOnChainWsService(
     * 取消订阅
     */
    fun unsubscribe(subscriptionId: String) {
-        val subscription = subscriptions.remove(subscriptionId)
-        
-        if (subscription != null && isConnected) {
-            // 取消该订阅的所有 RPC 订阅
-            scope.launch {
-                // 查找该订阅的所有 RPC subscriptionId
-                val rpcSubscriptionIds = rpcSubscriptionIdToSubscriptionId.entries
-                    .filter { it.value == subscriptionId }
-                    .map { it.key }
+        // 遍历所有连接找到含有该订阅的连接
+        for (connection in addressConnections.values) {
+            if (connection.hasSubscription(subscriptionId)) {
+                connection.removeSubscription(subscriptionId)
                
-                for (rpcSubId in rpcSubscriptionIds) {
-                    unsubscribeRpc(rpcSubId)
-                    rpcSubscriptionIdToSubscriptionId.remove(rpcSubId)
-                }
-            }
-            
-            logger.info("取消订阅: subscriptionId=$subscriptionId")
-        }
-        
-        // 如果没有订阅了，停止连接
-        if (subscriptions.isEmpty()) {
-            stop()
-        }
-    }
-    
-    /**
-     * 启动连接（如果还没有连接）
-     */
-    private fun startConnection() {
-        // 如果没有订阅，不启动连接
-        if (subscriptions.isEmpty()) {
-            return
-        }
-        
-        // 如果连接任务已经在运行，不重复启动
-        if (connectionJob != null && connectionJob!!.isActive) {
-            return
-        }
-        
-        // 启动连接任务
-        connectionJob = scope.launch {
-            startConnectionLoop()
-        }
-    }
-    
-    /**
-     * 启动连接循环
-     */
-    private suspend fun startConnectionLoop() {
-        while (scope.isActive) {
-            try {
-                // 如果没有订阅，停止连接
-                if (subscriptions.isEmpty()) {
-                    logger.info("没有订阅，停止连接")
-                    stop()
-                    break
+                // 如果该连接没有订阅了，停止并移除
+                if (connection.isSubscriptionsEmpty()) {
+                    connection.stop()
+                    addressConnections.remove(connection.address)
+                    logger.info("连接已无订阅，关闭连接: address=${connection.address}")
                }
                
-                // 如果已经连接，等待断开
-                if (isConnected && webSocket != null) {
-                    waitForDisconnect()
-                    continue
-                }
-                
-                // 获取可用的 RPC 节点
-                val wsUrl = rpcNodeService.getWsUrl()
-                val httpUrl = rpcNodeService.getHttpUrl()
-                
-                if (wsUrl.isBlank() || httpUrl.isBlank()) {
-                    logger.warn("没有可用的 RPC 节点，等待重试...")
-                    delay(reconnectDelay)
-                    continue
-                }
-                
-                logger.info("连接链上 WebSocket: $wsUrl (${subscriptions.size} 个订阅)")
-                
-                // 创建 HTTP 客户端（用于 RPC 调用）
-                val httpClient = createHttpClient()
-                
-                // 创建 RPC API 客户端
-                val rpcApi = retrofitFactory.createEthereumRpcApi(httpUrl)
-                
-                // 连接 WebSocket
-                connectWebSocket(wsUrl, httpClient, rpcApi)
-                
-                // 等待连接建立
-                waitForConnect()
-                
-                // 如果连接成功，订阅所有地址
-                if (isConnected) {
-                    logger.info("WebSocket 连接已建立，开始订阅")
-                    for (subscription in subscriptions.values) {
-                        subscribeAddress(subscription)
-                    }
-                    
-                    // 等待连接断开
-                    waitForDisconnect()
-                }
-                
-                // 连接断开后，如果没有订阅了，不再重连
-                if (subscriptions.isEmpty()) {
-                    logger.info("没有订阅，停止重连")
-                    break
-                }
-                
-                // 等待后重连
-                logger.info("WebSocket 连接断开，等待 ${reconnectDelay}ms 后重连")
-                delay(reconnectDelay)
-                
-            } catch (e: Exception) {
-                logger.error("连接异常: ${e.message}", e)
-                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) {
-        // 先关闭旧连接
-        webSocket?.close(1000, "重新连接")
-        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 连接成功")
-            }
-            
-            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
-            }
-        })
-    }
-    
-    /**
-     * 等待连接建立
-     */
-    private suspend fun waitForConnect() {
-        var waited = 0L
-        val timeout = 15000L  // 15秒超时
-        
-        while (!isConnected && waited < timeout) {
-            delay(100)
-            waited += 100
-        }
-        
-        if (!isConnected) {
-            logger.warn("WebSocket 连接超时，等待重连")
-        }
-    }
-    
-    /**
-     * 等待连接断开
-     */
-    private suspend fun waitForDisconnect() {
-        while (isConnected && scope.isActive) {
-            delay(1000)
-        }
-    }
-    
-    /**
-     * 订阅地址（为每个地址订阅 6 个事件）
-     */
-    private suspend fun subscribeAddress(subscription: SubscriptionInfo) {
-        if (webSocket == null || !isConnected) {
-            return
-        }
-        
-        val address = subscription.address
-        val walletTopic = OnChainWsUtils.addressToTopic32(address)
-        val subscriptionId = subscription.subscriptionId
-        
-        try {
-            // 订阅 USDC Transfer (from wallet)
-            subscribeLogs(OnChainWsUtils.USDC_CONTRACT, listOf(OnChainWsUtils.ERC20_TRANSFER_TOPIC, walletTopic), subscriptionId)
-            
-            // 订阅 USDC Transfer (to wallet)
-            subscribeLogs(OnChainWsUtils.USDC_CONTRACT, listOf(OnChainWsUtils.ERC20_TRANSFER_TOPIC, null, walletTopic), subscriptionId)
-            
-            // 订阅 ERC1155 TransferSingle (from wallet)
-            subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_SINGLE_TOPIC, null, walletTopic), subscriptionId)
-            
-            // 订阅 ERC1155 TransferSingle (to wallet)
-            subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_SINGLE_TOPIC, null, null, walletTopic), subscriptionId)
-            
-            // 订阅 ERC1155 TransferBatch (from wallet)
-            subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_BATCH_TOPIC, null, walletTopic), subscriptionId)
-            
-            // 订阅 ERC1155 TransferBatch (to wallet)
-            subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_BATCH_TOPIC, null, null, walletTopic), subscriptionId)
-            
-            logger.debug("已订阅地址: subscriptionId=$subscriptionId, address=$address")
-        } catch (e: Exception) {
-            logger.error("订阅地址失败: subscriptionId=$subscriptionId, address=$address, error=${e.message}", e)
-        }
-    }
-    
-    /**
-     * 订阅日志
-     */
-    private fun subscribeLogs(address: String, topics: List<String?>, subscriptionId: String) {
-        val ws = webSocket ?: return
-        
-        val params = mapOf(
-            "address" to address.lowercase(),
-            "topics" to topics.filterNotNull()
-        )
-        
-        val requestId = ++requestIdCounter
-        requestIdToSubscriptionId[requestId] = subscriptionId
-        
-        val request = mapOf(
-            "jsonrpc" to "2.0",
-            "id" to requestId,
-            "method" to "eth_subscribe",
-            "params" to listOf("logs", params)
-        )
-        
-        val message = gson.toJson(request)
-        ws.send(message)
-    }
-    
-    /**
-     * 取消 RPC 订阅
-     */
-    private fun unsubscribeRpc(rpcSubscriptionId: String) {
-        val ws = webSocket ?: return
-        
-        val requestId = ++requestIdCounter
-        val request = mapOf(
-            "jsonrpc" to "2.0",
-            "id" to requestId,
-            "method" to "eth_unsubscribe",
-            "params" to listOf(rpcSubscriptionId)
-        )
-        
-        val message = gson.toJson(request)
-        ws.send(message)
-    }
-    
-    /**
-     * 处理 WebSocket 消息
-     */
-    private suspend fun handleMessage(text: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) {
-        try {
-            val message = gson.fromJson(text, JsonObject::class.java)
-            
-            // 处理订阅响应
-            if (message.has("result") && message.has("id")) {
-                val requestId = message.get("id")?.asInt
-                val rpcSubscriptionId = message.get("result")?.asString
-                
-                if (requestId != null && rpcSubscriptionId != null) {
-                    val subscriptionId = requestIdToSubscriptionId.remove(requestId)
-                    if (subscriptionId != null) {
-                        // 保存 RPC subscriptionId 到订阅的映射
-                        rpcSubscriptionIdToSubscriptionId[rpcSubscriptionId] = subscriptionId
-                        logger.debug("订阅成功: subscriptionId=$subscriptionId, rpcSubscriptionId=$rpcSubscriptionId")
-                    }
-                }
+                logger.info("取消订阅: subscriptionId=$subscriptionId")
                return
            }
-            
-            // 处理日志通知
-            if (message.has("params")) {
-                val params = message.getAsJsonObject("params")
-                val subscriptionIdParam = params.get("subscription")?.asString
-                val result = params.getAsJsonObject("result")
-                
-                if (result != null) {
-                    val txHash = result.get("transactionHash")?.asString
-                    if (txHash != null && subscriptionIdParam != null) {
-                        // 根据 RPC subscriptionId 找到对应的订阅
-                        val subscriptionId = rpcSubscriptionIdToSubscriptionId[subscriptionIdParam]
-                        if (subscriptionId != null) {
-                            // 处理交易，分发给对应的订阅者
-                            processTransactionForSubscription(txHash, subscriptionId, httpClient, rpcApi)
-                        } else {
-                            // 如果没有找到订阅，可能是新订阅还未建立映射，尝试处理所有订阅
-                            processTransaction(txHash, httpClient, rpcApi)
-                        }
-                    }
-                }
-            }
-        } catch (e: Exception) {
-            logger.error("处理 WebSocket 消息失败: ${e.message}", e)
        }
    }
    
    /**
-     * 处理交易（为特定订阅）
-     * 直接调用订阅的回调
-     */
-    private suspend fun processTransactionForSubscription(
-        txHash: String,
-        subscriptionId: String,
-        httpClient: OkHttpClient,
-        rpcApi: EthereumRpcApi
-    ) {
-        val subscription = subscriptions[subscriptionId] ?: return
-        
-        try {
-            subscription.callback(txHash, httpClient, rpcApi)
-        } catch (e: Exception) {
-            logger.error("调用订阅回调失败: subscriptionId=$subscriptionId, txHash=$txHash, error=${e.message}", e)
-        }
-    }
-    
-    /**
-     * 处理交易（为所有订阅，用于兼容）
-     * 解析交易中的 Transfer 事件，分发给所有订阅者
-     */
-    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
-            val receiptJson = receiptRpcResponse.result.asJsonObject
-            
-            // 解析 receipt 中的 Transfer 日志
-            val logs = receiptJson.getAsJsonArray("logs") ?: return
-            val (erc20Transfers, erc1155Transfers) = OnChainWsUtils.parseReceiptTransfers(logs)
-            
-            // 为每个订阅检查是否匹配，如果匹配则调用回调
-            for (subscription in subscriptions.values) {
-                val address = subscription.address
-                
-                // 检查该地址是否参与了交易（通过检查 Transfer 日志）
-                val isInvolved = erc20Transfers.any { 
-                    it.from.lowercase() == address || it.to.lowercase() == address 
-                } || erc1155Transfers.any { 
-                    it.from.lowercase() == address || it.to.lowercase() == address 
-                }
-                
-                if (isInvolved) {
-                    // 该地址参与了交易，调用回调
-                    try {
-                        subscription.callback(txHash, httpClient, rpcApi)
-                    } catch (e: Exception) {
-                        logger.error("调用订阅回调失败: subscriptionId=${subscription.subscriptionId}, txHash=$txHash, error=${e.message}", e)
-                    }
-                }
-            }
-        } catch (e: Exception) {
-            logger.error("处理交易失败: txHash=$txHash, ${e.message}", e)
-        }
-    }
-    
-    /**
-     * 停止连接
+     * 停止所有服务
     */
    fun stop() {
-        connectionJob?.cancel()
-        connectionJob = null
-        
-        // 关闭 WebSocket 连接
-        webSocket?.close(1000, "停止监听")
-        webSocket = null
-        isConnected = false
-        
-        // 清空订阅信息
-        subscriptions.clear()
-        requestIdToSubscriptionId.clear()
-        rpcSubscriptionIdToSubscriptionId.clear()
+        for (connection in addressConnections.values) {
+            connection.stop()
+        }
+        addressConnections.clear()
    }
    
    @PostConstruct
    fun init() {
-        // 服务启动时不自动连接，等待有订阅时再连接
-        logger.info("统一链上 WebSocket 服务已初始化")
+        logger.info("统一链上 WebSocket 服务已初始化 (独立连接模式)")
    }
    
    @PreDestroy
@@ -561,5 +140,295 @@ class UnifiedOnChainWsService(
        stop()
        scope.cancel()
    }
+
+    /**
+     * 单个地址的 WebSocket 连接管理
+     */
+    inner class AddressWsConnection(val address: String) {
+        private var webSocket: WebSocket? = null
+        @Volatile
+        private var isConnected = false
+        
+        // 订阅ID计数器（用于请求 ID）
+        private var requestIdCounter = AtomicInteger(0)
+        
+        // 连接任务
+        private var connectionJob: Job? = null
+        
+        // 该连接下的所有订阅：subscriptionId -> SubscriptionInfo
+        // 理论上一个地址可能被多个业务订阅（如：既是被跟单者又是普通监控），虽然业务上通常只有一个
+        private val subscriptions = ConcurrentHashMap<String, SubscriptionInfo>()
+        
+        // 存储请求 ID 到订阅 ID 的映射：requestId -> subscriptionId
+        private val requestIdToSubscriptionId = ConcurrentHashMap<Int, String>()
+        
+        // 存储 RPC subscriptionId 到订阅 ID 的映射：rpcSubscriptionId -> subscriptionId
+        private val rpcSubscriptionIdToSubscriptionId = ConcurrentHashMap<String, String>()
+
+        fun start() {
+            if (connectionJob != null && connectionJob!!.isActive) return
+            connectionJob = scope.launch {
+                startConnectionLoop()
+            }
+        }
+
+        fun stop() {
+            connectionJob?.cancel()
+            connectionJob = null
+            webSocket?.close(1000, "停止监听")
+            webSocket = null
+            isConnected = false
+            subscriptions.clear()
+            requestIdToSubscriptionId.clear()
+            rpcSubscriptionIdToSubscriptionId.clear()
+        }
+
+        fun addSubscription(subscription: SubscriptionInfo) {
+            // 如果已经存在，先移除旧的
+            removeSubscription(subscription.subscriptionId)
+            subscriptions[subscription.subscriptionId] = subscription
+            
+            // 如果已经连接，立即发送链上订阅请求
+            if (isConnected) {
+                scope.launch {
+                    subscribeAddressOnChain(subscription)
+                }
+            }
+        }
+
+        fun removeSubscription(subscriptionId: String) {
+            subscriptions.remove(subscriptionId)
+            // 不需要显式发送 eth_unsubscribe，因为连接是 per-address 的，
+            // 只要只要连接还在，就保持该地址相关的所有 logs 订阅。
+            // 只有当所有 subscription 都移除了，连接才会关闭。
+        }
+
+        fun hasSubscription(subscriptionId: String): Boolean {
+            return subscriptions.containsKey(subscriptionId)
+        }
+
+        fun isSubscriptionsEmpty(): Boolean {
+            return subscriptions.isEmpty()
+        }
+
+        private suspend fun startConnectionLoop() {
+            while (scope.isActive) {
+                try {
+                    if (subscriptions.isEmpty()) {
+                        // 如果启动循环时还没订阅（不太可能，通常是先 addSubscription 再 start，或者是 start 后 addSubscription）
+                        // 或者订阅被清空了，外部应当掉 stop，但这里作为防守
+                        delay(1000) 
+                        continue
+                    }
+
+                    if (isConnected && webSocket != null) {
+                        waitForDisconnect()
+                        continue
+                    }
+
+                    // 获取可用的 RPC 节点
+                    val wsUrl = rpcNodeService.getWsUrl()
+                    val httpUrl = rpcNodeService.getHttpUrl()
+
+                    logger.info("[$address] 连接链上 WebSocket: $wsUrl")
+
+                    val httpClient = createHttpClient()
+                    val rpcApi = retrofitFactory.createEthereumRpcApi(httpUrl)
+
+                    connectWebSocket(wsUrl, httpClient, rpcApi)
+                    waitForConnect()
+
+                    if (isConnected) {
+                        logger.info("[$address] WebSocket 连接已建立，开始注册订阅")
+                        // 重新为所有订阅注册链上监听
+                        for (subscription in subscriptions.values) {
+                            subscribeAddressOnChain(subscription)
+                        }
+                        waitForDisconnect()
+                    }
+
+                    logger.info("[$address] WebSocket 连接断开，等待 ${reconnectDelay}ms 后重连")
+                    delay(reconnectDelay)
+
+                } catch (e: Exception) {
+                    logger.error("[$address] 连接异常: ${e.message}", e)
+                    delay(reconnectDelay)
+                }
+            }
+        }
+
+        private fun connectWebSocket(wsUrl: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) {
+            webSocket?.close(1000, "重新连接")
+            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("[$address] 链上 WebSocket 连接成功")
+                }
+
+                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("[$address] 链上 WebSocket 连接关闭: code=$code, reason=$reason")
+                }
+
+                override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
+                    isConnected = false
+                    logger.warn("[$address] 链上 WebSocket 连接已关闭: code=$code, reason=$reason")
+                }
+
+                override fun onFailure(webSocket: WebSocket, t: Throwable, response: okhttp3.Response?) {
+                    logger.error("[$address] 链上 WebSocket 连接失败: ${t.message}", t)
+                    isConnected = false
+                }
+            })
+        }
+
+        private suspend fun subscribeAddressOnChain(subscription: SubscriptionInfo) {
+            if (webSocket == null || !isConnected) return
+            
+            val walletTopic = OnChainWsUtils.addressToTopic32(address)
+            val subId = subscription.subscriptionId
+            
+            try {
+                // 订阅该地址相关的所有事件
+                // USDC Transfer (from/to)
+                subscribeLogs(OnChainWsUtils.USDC_CONTRACT, listOf(OnChainWsUtils.ERC20_TRANSFER_TOPIC, walletTopic), subId)
+                subscribeLogs(OnChainWsUtils.USDC_CONTRACT, listOf(OnChainWsUtils.ERC20_TRANSFER_TOPIC, null, walletTopic), subId)
+                
+                // ERC1155 TransferSingle (from/to)
+                subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_SINGLE_TOPIC, null, walletTopic), subId)
+                subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_SINGLE_TOPIC, null, null, walletTopic), subId)
+                
+                // ERC1155 TransferBatch (from/to)
+                subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_BATCH_TOPIC, null, walletTopic), subId)
+                subscribeLogs(OnChainWsUtils.ERC1155_CONTRACT, listOf(OnChainWsUtils.ERC1155_TRANSFER_BATCH_TOPIC, null, null, walletTopic), subId)
+                
+                logger.debug("[$address] 已发送链上订阅请求: subscriptionId=$subId")
+            } catch (e: Exception) {
+                logger.error("[$address] 发送链上订阅请求失败: error=${e.message}", e)
+            }
+        }
+
+        private fun subscribeLogs(contractAddress: String, topics: List<String?>, subscriptionId: String) {
+            val ws = webSocket ?: return
+            
+            val topicsArray = gson.toJsonTree(topics).asJsonArray
+            val logParams = JsonObject()
+            logParams.addProperty("address", contractAddress.lowercase())
+            logParams.add("topics", topicsArray)
+            
+            val requestId = requestIdCounter.incrementAndGet()
+            requestIdToSubscriptionId[requestId] = subscriptionId
+            
+            val request = JsonObject()
+            request.addProperty("jsonrpc", "2.0")
+            request.addProperty("id", requestId)
+            request.addProperty("method", "eth_subscribe")
+            val paramsArray = JsonArray()
+            paramsArray.add("logs")
+            paramsArray.add(logParams)
+            request.add("params", paramsArray)
+            
+            ws.send(gson.toJson(request))
+        }
+
+        private suspend fun handleMessage(text: String, httpClient: OkHttpClient, rpcApi: EthereumRpcApi) {
+            try {
+                val message = gson.fromJson(text, JsonObject::class.java)
+
+                // 1. 处理订阅响应 (eth_subscribe response)
+                if (message.has("result") && message.has("id")) {
+                    val requestId = message.get("id")?.asInt
+                    val rpcSubscriptionId = message.get("result")?.asString
+                    
+                    if (requestId != null && rpcSubscriptionId != null) {
+                        val subId = requestIdToSubscriptionId.remove(requestId)
+                        if (subId != null) {
+                            rpcSubscriptionIdToSubscriptionId[rpcSubscriptionId] = subId
+                            logger.debug("[$address] 链上订阅成功: mapped connection rpcSubId=$rpcSubscriptionId to localSubId=$subId")
+                        }
+                    }
+                    return
+                }
+
+                // 2. 处理日志通知 (eth_subscription)
+                val method = message.get("method")?.asString
+                if (method == "eth_subscription") {
+                    val params = message.getAsJsonObject("params") ?: return
+                    val rpcSubParam = params.get("subscription")?.asString
+                    val result = params.getAsJsonObject("result") ?: return
+                    val txHash = result.get("transactionHash")?.asString
+
+                    if (txHash != null && rpcSubParam != null) {
+                        // 找到触发此通知的本地订阅 ID
+                        // 因为我们在这个连接里只订阅了 this.address，所以理论上所有通知都跟这个 address 有关
+                        // 但我们需要找到对应的 callback
+                        val localSubId = rpcSubscriptionIdToSubscriptionId[rpcSubParam]
+                        
+                        if (localSubId != null) {
+                            val subscription = subscriptions[localSubId]
+                            if (subscription != null) {
+                                logger.info("[$address] 收到交易通知: txHash=$txHash, subId=$localSubId")
+                                runCatching {
+                                    subscription.callback(txHash, httpClient, rpcApi)
+                                }.onFailure { e ->
+                                    logger.error("[$address] 回调执行失败: ${e.message}", e)
+                                }
+                            }
+                        } else {
+                            // 找不到具体是哪个订阅请求触发的（可能是重启后之前的订阅残留？或者映射丢失？）
+                            // 在单地址单连接模式下，只要是这个 connection 收到的，肯定是关于这个 address 的
+                            // 我们可以尝试通知所有订阅者（通常一个地址只有一个订阅者，除非此地址既是Leader又是User）
+                            logger.warn("[$address] 未找到映射的订阅ID: rpcSubId=$rpcSubParam. 广播给所有订阅者.")
+                            subscriptions.values.forEach { sub ->
+                                runCatching {
+                                    sub.callback(txHash, httpClient, rpcApi)
+                                }.onFailure { e ->
+                                    logger.error("[$address] 广播回调执行失败: ${e.message}", e)
+                                }
+                            }
+                        }
+                    }
+                }
+            } catch (e: Exception) {
+                logger.error("[$address] 处理消息失败: ${e.message}", e)
+            }
+        }
+        
+        private suspend fun waitForConnect() {
+            var waited = 0L
+            val timeout = 15000L
+            while (!isConnected && waited < timeout) {
+                delay(100)
+                waited += 100
+            }
+            if (!isConnected) logger.warn("[$address] WebSocket 连接超时")
+        }
+
+        private suspend fun waitForDisconnect() {
+            while (isConnected && scope.isActive) {
+                delay(1000)
+            }
+        }
+        
+        private fun createHttpClient(): OkHttpClient {
+            val proxy = getProxyConfig()
+            val builder = createClient()
+            if (proxy != null) builder.proxy(proxy)
+            return builder.build()
+        }
+    }
 }

@@ -253,7 +253,7 @@ open class CopyOrderTrackingService(
                    // 先计算跟单金额（用于仓位检查）
                    // 注意：这里先计算金额，即使后续被过滤也会记录
                    val tradePrice = trade.price.toSafeBigDecimal()
-                    val buyQuantity = try {
+                    var buyQuantity = try {
                        calculateBuyQuantity(trade, copyTrading)
                    } catch (e: Exception) {
                        logger.warn("计算买入数量失败: ${e.message}", e)
@@ -369,10 +369,14 @@ open class CopyOrderTrackingService(
                    // 买入数量已在过滤检查前计算，这里直接使用
                    // 如果数量为0或负数，跳过
                    if (buyQuantity.lte(BigDecimal.ZERO)) {
-                        logger.warn("计算出的买入数量为0或负数，跳过: copyTradingId=${copyTrading.id}, tradeId=${trade.id}")
+                        logger.warn("计算得到的买入数量为0，跳过跟单: copyTradingId=${copyTrading.id}, tradeId=${trade.id}")
                        continue
                    }

+                    if (buyQuantity.lt(BigDecimal.ONE)) {
+                        logger.warn("计算得到的买入数量小于1，自动调整为1 (Polymarket 最小下单数量): copyTradingId=${copyTrading.id}, tradeId=${trade.id}, originalQuantity=$buyQuantity")
+                        buyQuantity = BigDecimal.ONE
+                    }
                    // 验证订单数量限制（仅比例模式）
                    var finalBuyQuantity = buyQuantity
                    if (copyTrading.copyMode == "RATIO") {
@@ -656,8 +660,8 @@ open class CopyOrderTrackingService(
    private fun calculateBuyQuantity(trade: TradeResponse, copyTrading: CopyTrading): BigDecimal {
        return when (copyTrading.copyMode) {
            "RATIO" -> {
-                // 比例模式：Leader 数量 × 比例
-                trade.size.toSafeBigDecimal().multi(copyTrading.copyRatio)
+                // 比例模式：Leader 数量 × (比例 / 100)
+                trade.size.toSafeBigDecimal().multi(copyTrading.copyRatio.div(100))
            }

            "FIXED" -> {
@@ -689,7 +693,7 @@ open class CopyOrderTrackingService(
        val leader = leaderRepository.findById(copyTrading.leaderId).orElse(null)
            ?: run {
                logger.warn("Leader 不存在，使用默认比例: leaderId=${copyTrading.leaderId}")
-                return leaderSellQuantity.multi(copyTrading.copyRatio)
+                return leaderSellQuantity.multi(copyTrading.copyRatio.div(100))
            }

        // 创建不需要认证的 CLOB API 客户端（用于查询公开的交易数据）
@@ -760,7 +764,7 @@ open class CopyOrderTrackingService(
        // 如果无法计算总比例（查询失败），使用默认比例
        if (totalLeaderQuantity.lte(BigDecimal.ZERO)) {
            logger.warn("无法计算总比例（Leader 买入数量为 0），使用默认比例: copyTradingId=${copyTrading.id}")
-            return leaderSellQuantity.multi(copyTrading.copyRatio)
+            return leaderSellQuantity.multi(copyTrading.copyRatio.div(100))
        }

        // 计算实际比例：跟单买入数量 / Leader 买入数量
@@ -835,15 +839,23 @@ open class CopyOrderTrackingService(
                )
            }
            "RATIO" -> {
-                // 比例模式：直接使用配置的 copyRatio
-                leaderSellTrade.size.toSafeBigDecimal().multi(copyTrading.copyRatio)
+                // 比例模式：直接使用配置的 copyRatio (需要除以100)
+                leaderSellTrade.size.toSafeBigDecimal().multi(copyTrading.copyRatio.div(100))
            }
            else -> {
                logger.warn("不支持的 copyMode: ${copyTrading.copyMode}，使用默认比例模式")
-                leaderSellTrade.size.toSafeBigDecimal().multi(copyTrading.copyRatio)
+                leaderSellTrade.size.toSafeBigDecimal().multi(copyTrading.copyRatio.div(100))
            }
        }

+        // 如果需要卖出的数量小于1（但大于0），自动调整为1（Polymarket 最小下单数量）
+        // 注意：如果实际持有数量不足1，后续的 totalMatched 检查会拦截
+        var finalNeedMatch = needMatch
+        if (finalNeedMatch.gt(BigDecimal.ZERO) && finalNeedMatch.lt(BigDecimal.ONE)) {
+            logger.warn("计算得到的卖出数量小于1，自动调整为1: copyTradingId=${copyTrading.id}, original=$needMatch")
+            finalNeedMatch = BigDecimal.ONE
+        }
+
        // 4. 获取tokenId（直接使用outcomeIndex，支持多元市场）
        val tokenIdResult = blockchainService.getTokenId(leaderSellTrade.market, leaderSellTrade.outcomeIndex)
        if (tokenIdResult.isFailure) {
@@ -867,7 +879,7 @@ open class CopyOrderTrackingService(
        // 6. 按FIFO顺序匹配，计算实际可以卖出的数量
        // 使用计算出的实际卖出价格（而不是 Leader 价格）来创建匹配明细
        var totalMatched = BigDecimal.ZERO
-        var remaining = needMatch
+        var remaining = finalNeedMatch
        val matchDetails = mutableListOf<SellMatchDetail>()

        for (order in unmatchedOrders) {
@@ -904,6 +916,11 @@ open class CopyOrderTrackingService(
            return
        }

+        if (totalMatched.lt(BigDecimal.ONE)) {
+            logger.warn("卖出数量小于1，跳过卖出 (Polymarket 最小下单数量为 1): copyTradingId=${copyTrading.id}, tradeId=${leaderSellTrade.id}, quantity=$totalMatched")
+            return
+        }
+
        // 7. 解密 API 凭证
        val apiSecret = try {
            decryptApiSecret(account)
@@ -143,14 +143,16 @@ class OrderStatusUpdateService(
            // 计算30秒前的时间戳
            val thirtySecondsAgo = System.currentTimeMillis() - 30000
            
-            // 查询30秒前创建的订单
-            val ordersToCheck = copyOrderTrackingRepository.findByCreatedAtBefore(thirtySecondsAgo)
+            // 查询30秒前创建的订单,并过滤掉已经完全匹配的订单
+            // 已经完全匹配的订单(status = "fully_matched")不需要再检查
+            val allOrdersToCheck = copyOrderTrackingRepository.findByCreatedAtBefore(thirtySecondsAgo)
+            val ordersToCheck = allOrdersToCheck.filter { it.status != "fully_matched" }
            
            if (ordersToCheck.isEmpty()) {
                return
            }
            
-            logger.debug("检查 ${ordersToCheck.size} 个30秒前创建的订单是否成交")
+            logger.debug("检查 ${ordersToCheck.size} 个30秒前创建的订单是否成交 (已过滤 ${allOrdersToCheck.size - ordersToCheck.size} 个已完全匹配的订单)")
            
            // 按账户分组，避免重复创建 API 客户端
            val ordersByAccount = ordersToCheck.groupBy { it.accountId }
@@ -686,6 +686,13 @@ class TelegramNotificationService(
            else -> side
        }
        
+        // 获取图标
+        val icon = when (side.uppercase()) {
+            "BUY" -> "🚀"
+            "SELL" -> "💰"
+            else -> "📣"
+        }
+        
        // 构建账户信息（格式：账户名(钱包地址)）
        val accountInfo = buildAccountInfo(accountName, walletAddress, unknownAccount)

@@ -761,7 +768,7 @@ class TelegramNotificationService(
        val priceDisplay = formatPrice(price)
        val sizeDisplay = formatQuantity(size)

-        return """✅ <b>$orderCreatedSuccess</b>
+        return """$icon <b>$orderCreatedSuccess</b>

 📊 <b>$orderInfo：</b>
 • $orderIdLabel: <code>${orderId ?: unknown}</code>
@@ -989,7 +996,7 @@ class TelegramNotificationService(
            "  • ${position.marketId.substring(0, 8)}... (${position.side}): $quantityDisplay shares = $valueDisplay USDC"
        }
        
-        return """✅ <b>$redeemSuccess</b>
+        return """💸 <b>$redeemSuccess</b>

 📊 <b>$redeemInfo：</b>
 • $accountLabel: $escapedAccountInfo