fix(gateApi): 修复止损订单数量参数类型及优化WebSocket心跳机制

Administrator

10 hours ago cd6a9821d0e9e4f7f13d7b5dab00f22b267cc222

fix(gateApi): 修复止损订单数量参数类型及优化WebSocket心跳机制

- 修复placeStopLossOrders方法中qty参数类型从int改为String
- 优化多仓和空仓止损逻辑，支持持仓不足一个网格量时按剩余张数挂止损
- 更新WebSocket心跳检测机制，调整超时时间为20秒并增加主动换线功能
- 添加连接最大存活时长限制（55分钟），避免中间设备强制断连问题
- 改进WebSocket连接管理，增加连接建立时间跟踪和主动关闭方法

 src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java

@@ -2325,14 +2325,17 @@
        int baseQty = Integer.parseInt(config.getBaseQuantity());
        int gridCount = baseQty / qty;
        GridElement entryElem = GridElement.findById(gridId);
        if (gridCount <= 0 || filledQty < qty) {
        if (gridCount <= 0 || filledQty <= 0) {
            // 早期返回需重置防重入标记
            if (entryElem != null) { entryElem.setExtendStopLossInProgress(false); }
            return;
        }

        // [Gate] 止损只针对初始开仓仓位(baseQuantity)，加仓部分只做止盈
        int totalOrders = filledQty / qty;
        // [Gate] 止损只针对初始开仓仓位(baseQuantity)，加仓部分只做止盈；
        // 持仓不足一个网格量时也按剩余张数挂一张止损（手动重建场景，避免裸奔）
        int slSize = Math.min(filledQty, baseQty);
        int totalOrders = slSize / qty;
        int remainder = slSize % qty;
        int[] orders = new int[gridCount];

        if (gridCount == 1) {
@@ -2356,6 +2359,7 @@
        for (int i = 0; i < gridCount; i++) {
            if (orders[i] > 0) { totalOrdersToPlace += orders[i]; }
        }
        if (remainder > 0) { totalOrdersToPlace += 1; }
        if (totalOrdersToPlace == 0) {
            if (entryElem != null) { entryElem.setExtendStopLossInProgress(false); }
            return;
@@ -2366,13 +2370,23 @@
        for (int i = 0; i < gridCount; i++) {
            if (orders[i] > 0) {
                int sgId = gridId - 2 - i;
                placeStopLossOrders(sgId, orders[i], qty, FuturesPriceTrigger.RuleEnum.NUMBER_2,
                placeStopLossOrders(sgId, orders[i], String.valueOf(qty), FuturesPriceTrigger.RuleEnum.NUMBER_2,
                        ORDER_TYPE_CLOSE_LONG, true, remainCount, entryElem);
                if (detail.length() > 0) {
                    detail.append("  ");
                }
                detail.append(sgId).append("x").append(orders[i]);
            }
        }
        // 不足一个网格量的余数：单独挂一张止损
        if (remainder > 0) {
            int sgId = gridId - 2 - totalOrders;
            placeStopLossOrders(sgId, 1, String.valueOf(remainder), FuturesPriceTrigger.RuleEnum.NUMBER_2,
                    ORDER_TYPE_CLOSE_LONG, true, remainCount, entryElem);
            if (detail.length() > 0) {
                detail.append("  ");
            }
            detail.append(sgId).append("x").append(remainder);
        }

        log.info("[Gate] 多仓追挂止损, 成交{}张, gridId:{}, 网格数:{}, 分配:{}",
@@ -2388,14 +2402,17 @@
        int baseQty = Integer.parseInt(config.getBaseQuantity());
        int gridCount = baseQty / qty;
        GridElement entryElem = GridElement.findById(gridId);
        if (gridCount <= 0 || filledQty < qty) {
        if (gridCount <= 0 || filledQty <= 0) {
            // 早期返回需重置防重入标记
            if (entryElem != null) { entryElem.setExtendStopLossInProgress(false); }
            return;
        }

        // [Gate] 止损只针对初始开仓仓位(baseQuantity)，加仓部分只做止盈
        int totalOrders = filledQty / qty;
        // [Gate] 止损只针对初始开仓仓位(baseQuantity)，加仓部分只做止盈；
        // 持仓不足一个网格量时也按剩余张数挂一张止损（手动重建场景，避免裸奔）
        int slSize = Math.min(filledQty, baseQty);
        int totalOrders = slSize / qty;
        int remainder = slSize % qty;
        int[] orders = new int[gridCount];

        if (gridCount == 1) {
@@ -2419,6 +2436,7 @@
        for (int i = 0; i < gridCount; i++) {
            if (orders[i] > 0) { totalOrdersToPlace += orders[i]; }
        }
        if (remainder > 0) { totalOrdersToPlace += 1; }
        if (totalOrdersToPlace == 0) {
            if (entryElem != null) { entryElem.setExtendStopLossInProgress(false); }
            return;
@@ -2429,13 +2447,23 @@
        for (int i = 0; i < gridCount; i++) {
            if (orders[i] > 0) {
                int sgId = gridId + 2 + i;
                placeStopLossOrders(sgId, orders[i], qty, FuturesPriceTrigger.RuleEnum.NUMBER_1,
                placeStopLossOrders(sgId, orders[i], String.valueOf(qty), FuturesPriceTrigger.RuleEnum.NUMBER_1,
                        ORDER_TYPE_CLOSE_SHORT, false, remainCount, entryElem);
                if (detail.length() > 0) {
                    detail.append("  ");
                }
                detail.append(sgId).append("x").append(orders[i]);
            }
        }
        // 不足一个网格量的余数：单独挂一张止损
        if (remainder > 0) {
            int sgId = gridId + 2 + totalOrders;
            placeStopLossOrders(sgId, 1, String.valueOf(remainder), FuturesPriceTrigger.RuleEnum.NUMBER_1,
                    ORDER_TYPE_CLOSE_SHORT, false, remainCount, entryElem);
            if (detail.length() > 0) {
                detail.append("  ");
            }
            detail.append(sgId).append("x").append(remainder);
        }

        log.info("[Gate] 空仓追挂止损, 成交{}张, gridId:{}, 网格数:{}, 分配:{}",
@@ -2473,7 +2501,7 @@
    /**
     * 在指定网格位置挂 count 个独立止损单，每个 size 张。
     */
    private void placeStopLossOrders(int gridId, int count, int qty,
    private void placeStopLossOrders(int gridId, int count, String qtyStr,
                                      FuturesPriceTrigger.RuleEnum rule,
                                      String orderType, boolean isLong,
                                      AtomicInteger remainCount,
@@ -2506,8 +2534,8 @@
            return;
        }
        BigDecimal triggerPrice = elem.getGridPrice();
        String size = isLong ? negate(qtyStr) : qtyStr;
        for (int i = 0; i < count; i++) {
            String size = isLong ? negate(String.valueOf(qty)) : String.valueOf(qty);
            int finalGridId = gridId;
            int finalI = i;
            executor.placeTakeProfit(

 src/main/java/com/xcong/excoin/modules/gateApi/GateKlineWebSocketClient.java

@@ -15,6 +15,7 @@
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

/**
@@ -39,12 +40,15 @@
 *     4. update/all           → 遍历 channelHandlers → handler.handleMessage(response)
 * </pre>
 *
 * <h3>心跳与断线检测（双层）</h3>
 * <h3>心跳与断线检测（主动健康检查，依据 Gate.io 官方 ping/pong 语义）</h3>
 * <ol>
 *   <li>TCP 层：Java-WebSocket 自带 ping/pong 保活（{@code connectionLostTimeout = 30s}），
 *       半开连接（对端静默丢包、无 FIN/RST）会在 30 秒内被库判定丢失并触发 onClose。</li>
 *   <li>应用层：{@code futures.ping}/{@code futures.pong}。10 秒未收到任何消息 → 发 ping；
 *       45 秒未收到任何消息 → 判定连接死亡，主动 close → onClose → 重连。</li>
 *       半开连接（对端静默丢包、无 FIN/RST）会被库判定丢失并触发 onClose。</li>
 *   <li>应用层：{@code futures.ping}/{@code futures.pong}。
 *       静默 10 秒 → 发 futures.ping 探测（可重置服务端关闭计数，官方文档语义）；
 *       静默 20 秒（ping 也无任何回应）→ 判定连接不健康，主动 close → onClose → 重连。</li>
 *   <li>主动换线：连接存活满 55 分钟主动优雅重连，绕开链路中间设备约 1 小时的强制断连，
 *       把被动盲区变为计划内切换。</li>
 * </ol>
 *
 * <h3>重连策略</h3>
@@ -63,12 +67,21 @@

    private static final String FUTURES_PING = "futures.ping";
    private static final String FUTURES_PONG = "futures.pong";
    /** 静默超过该秒数 → 发送 futures.ping */
    /** 静默超过该秒数 → 发送 futures.ping 探测（官方文档：客户端 ping 可重置服务端关闭计数） */
    private static final int HEARTBEAT_TIMEOUT = 10;
    /** 静默超过该秒数 → 判定连接死亡，主动断开重连 */
    private static final int DEAD_TIMEOUT_SECONDS = 45;
    /** Java-WebSocket 库层 ping/pong 保活超时（秒），用于检测半开 TCP 连接 */
    /**
     * 静默超过该秒数（且已发出过探测 ping）→ 判定连接不健康，主动断开重连。
     * 即：10s 静默发 ping，再等 10s 仍无任何消息（连 pong 都没有）→ 判定死亡。
     */
    private static final int DEAD_TIMEOUT_SECONDS = 20;
    /** Java-WebSocket 库层 ping/pong 保活超时（秒），用于检测半开 TCP 连接（兜底） */
    private static final int CONNECTION_LOST_TIMEOUT_SECONDS = 30;
    /**
     * 连接最大存活时长（秒）：到达后主动优雅断开重连（"主动换线"）。
     * 用于绕开链路中间设备（腾讯云出口 NAT / 国际出口）约 1 小时的连接强制断连，
     * 把被动盲区变成计划内换线。设为 0 可禁用。
     */
    private static final int MAX_CONNECTION_AGE_SECONDS = 55 * 60;
    /** 重连初始延迟（毫秒） */
    private static final long INITIAL_RECONNECT_DELAY_MS = 5000;
    /** 重连最大延迟（毫秒），指数退避封顶值 */
@@ -85,6 +98,8 @@
    private volatile ScheduledFuture<?> pongTimeoutFuture;
    /** 最后收到消息的时间戳（毫秒） */
    private final AtomicReference<Long> lastMessageTime = new AtomicReference<>(System.currentTimeMillis());
    /** 当前连接建立时间（毫秒），用于主动换线 */
    private final AtomicLong connectionOpenedTime = new AtomicLong(System.currentTimeMillis());

    /** 连接状态 */
    private final AtomicBoolean isConnected = new AtomicBoolean(false);
@@ -221,6 +236,8 @@
                    isConnected.set(true);
                    isConnecting.set(false);
                    reconnectAttempt.set(0);
                    connectionOpenedTime.set(System.currentTimeMillis());
                    lastMessageTime.set(System.currentTimeMillis());
                    if (sharedExecutor != null && !sharedExecutor.isShutdown()) {
                        for (GateChannelHandler handler : channelHandlers) {
                            handler.setSubscribed(false);
@@ -357,24 +374,49 @@
    }

    /**
     * 心跳检查（由 25s 周期任务 + 每次收到消息后 10s 的一次性任务触发）：
     * 心跳检查（由 25s 周期任务 + 每次收到消息后 10s 的一次性任务触发）。
     * 主动健康检查逻辑（依据 Gate.io 官方文档 ping/pong 语义）：
     * <ol>
     *   <li>静默 ≥ {@link #DEAD_TIMEOUT_SECONDS} 秒 → 判定连接死亡（半开 TCP / 对端失联），主动断开触发重连；</li>
     *   <li>静默 ≥ {@link #HEARTBEAT_TIMEOUT} 秒 → 发送 futures.ping 探测。</li>
     *   <li>连接存活 ≥ {@link #MAX_CONNECTION_AGE_SECONDS} 秒 → 主动优雅断开重连（主动换线，
     *       绕开链路中间设备约 1 小时的强制断连，把被动盲区变为计划内切换）；</li>
     *   <li>静默 ≥ {@link #HEARTBEAT_TIMEOUT} 秒 → 发送 futures.ping 探测，并安排 10 秒后复查；</li>
     *   <li>静默 ≥ {@link #DEAD_TIMEOUT_SECONDS} 秒（探测 ping 也得不到任何回应，包括 pong）→
     *       判定连接不健康，主动断开触发重连。</li>
     * </ol>
     */
    private void checkHeartbeatTimeout() {
        if (!isConnected.get() || destroyed.get()) return;
        long silentMs = System.currentTimeMillis() - lastMessageTime.get();
        if (silentMs >= DEAD_TIMEOUT_SECONDS * 1000L) {
            log.warn("[WS] 已 {} 秒未收到任何消息，判定连接死亡，主动断开重连", silentMs / 1000);
            if (webSocketClient != null) {
                try { webSocketClient.close(); } catch (Exception e) { log.warn("[WS] 主动断开失败", e); }
            }
        long now = System.currentTimeMillis();
        long silentMs = now - lastMessageTime.get();

        // 1) 主动换线：到达连接最大存活时长
        long ageMs = now - connectionOpenedTime.get();
        if (MAX_CONNECTION_AGE_SECONDS > 0 && ageMs >= MAX_CONNECTION_AGE_SECONDS * 1000L) {
            log.warn("[WS] 连接已存活 {} 秒，达到主动换线阈值，主动断开重连", ageMs / 1000);
            closeSocket();
            return;
        }
        if (silentMs >= HEARTBEAT_TIMEOUT * 1000L) {

        // 2) 探测：静默 10s → 发 futures.ping，并安排复查
        if (silentMs >= HEARTBEAT_TIMEOUT * 1000L && silentMs < DEAD_TIMEOUT_SECONDS * 1000L) {
            sendPing();
            resetHeartbeatTimer();
            return;
        }

        // 3) 判定不健康：20s 静默（ping 后仍无 pong/任何消息）→ 断开重连
        if (silentMs >= DEAD_TIMEOUT_SECONDS * 1000L) {
            log.warn("[WS] 已 {} 秒未收到任何消息（ping 无回应），判定连接不健康，断开重连", silentMs / 1000);
            closeSocket();
        }
    }

    /**
     * 主动关闭 WebSocket（非阻塞），触发 onClose → 重连流程。
     */
    private void closeSocket() {
        if (webSocketClient != null) {
            try { webSocketClient.close(); } catch (Exception e) { log.warn("[WS] 主动断开失败", e); }
        }
    }