pom.xml

@@ -35,6 +35,14 @@
    </properties>

    <dependencies>

        <!-- Source: https://mvnrepository.com/artifact/io.gate/gate-api -->
        <dependency>
            <groupId>io.gate</groupId>
            <artifactId>gate-api</artifactId>
            <version>7.2.71</version>
            <scope>compile</scope>
        </dependency>
    
        <dependency>
            <groupId>ripple</groupId>

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

New file
@@ -0,0 +1,75 @@
package com.xcong.excoin.modules.gateApi;

// Import classes:
import io.gate.gateapi.ApiClient;
import io.gate.gateapi.ApiException;
import io.gate.gateapi.Configuration;
import io.gate.gateapi.GateApiException;
import io.gate.gateapi.api.FuturesApi;
import io.gate.gateapi.auth.*;
import io.gate.gateapi.models.*;
import io.gate.gateapi.api.AccountApi;

public class Example {
    public static void main(String[] args) {
        ApiClient defaultClient = Configuration.getDefaultApiClient();
        defaultClient.setBasePath("https://api-testnet.gateapi.io/api/v4");
//        defaultClient.setBasePath("https://api.gateio.ws/api/v4");

        // Configure APIv4 authorization: apiv4
        defaultClient.setApiKeySecret("d90ca272391992b8e74f8f92cedb21ec", "1861e4f52de4bb53369ea3208d9ede38ece4777368030f96c77d27934c46c274");

        try {

            String contract = "ETH_USDT";
            String settle = "usdt";
            //保证金模式 isolated/cross
            String marginMode = "cross";
            AccountApi accountApi = new AccountApi(defaultClient);
            AccountDetail accountDetail = accountApi.getAccountDetail();
            System.out.println(accountDetail.toString());


            /**
             * 获取账户余额
             */
            FuturesApi futuresApi = new FuturesApi(defaultClient);
            FuturesAccount futuresAccount = futuresApi.listFuturesAccounts(settle);
            String available = futuresAccount.getAvailable();
            String result = "可用余额：" + available;
            System.out.println(result);

            /**
             * 设置仓位模式
             * 可选值：single, dual, dual_plus，分别表示单向、双向、分仓
             */

            String position_mode = "dual";
            String positionMode = futuresAccount.getPositionMode();
            if (!position_mode.equals(positionMode)){
                futuresApi.setPositionMode(settle, position_mode);
            }

            /**
             * 设置杠杆倍数
             * 设置合理的杠杆倍数，不能为0
             */
            String leverage = "25";
            futuresApi.updateContractPositionLeverageCall(
                    settle,
                    contract,
                    leverage,
                    marginMode,
                    position_mode,
                    null);
        } catch (GateApiException e) {
            System.err.println(String.format("Gate api exception, label: %s, message: %s", e.getErrorLabel(), e.getMessage()));
            e.printStackTrace();
        } catch (ApiException e) {
            System.err.println("Exception when calling AccountApi#getAccountDetail");
            System.err.println("Status code: " + e.getCode());
            System.err.println("Response headers: " + e.getResponseHeaders());
            e.printStackTrace();
        }
    }
}

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

New file
@@ -0,0 +1,417 @@
package com.xcong.excoin.modules.gateApi;

import io.gate.gateapi.ApiClient;
import io.gate.gateapi.ApiException;
import io.gate.gateapi.GateApiException;
import io.gate.gateapi.api.FuturesApi;
import io.gate.gateapi.models.FuturesAccount;
import io.gate.gateapi.models.FuturesInitialOrder;
import io.gate.gateapi.models.FuturesOrder;
import io.gate.gateapi.models.FuturesPriceTrigger;
import io.gate.gateapi.models.FuturesPriceTriggeredOrder;
import io.gate.gateapi.models.TriggerOrderResponse;
import lombok.extern.slf4j.Slf4j;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.List;

/**
 * Gate 网格交易服务类，使用 gate-api SDK 进行合约下单。
 * 策略：多空双开 → 设置止盈止损点位 → 网格循环交易
 *
 * 测试参数：
 *   品种: XAU_USDT（黄金）
 *   杠杆: 100x（全仓）
 *   数量: 0.01 XAU
 *   网格: 0.0035（千分之三点五）
 *   整体止盈: 0.5 USDT
 *   循环次数: 3
 *   报警: 本金亏损 15%（初始本金 50 USDT）
 *
 * @author Administrator
 */
@Slf4j
public class GateGridTradeService {

    private final ApiClient apiClient;
    private final FuturesApi futuresApi;
    private static final String SETTLE = "usdt";

    private final String contract;
    private final String leverage;
    private final String marginMode;
    private final BigDecimal gridRate;
    private final BigDecimal overallTp;
    private final int maxCycles;
    private final BigDecimal maxLoss;
    private final String quantity;
    private final String positionMode;

    private volatile boolean strategyActive = false;
    private int currentCycle = 0;
    private BigDecimal totalProfit = BigDecimal.ZERO;
    private BigDecimal longEntryPrice;
    private BigDecimal shortEntryPrice;
    private Long longOrderId;
    private Long shortOrderId;

    private volatile BigDecimal lastClosePrice;

    public GateGridTradeService(String apiKey, String apiSecret,
                                 String contract, String leverage,
                                String marginMode,String positionMode,
                                 BigDecimal gridRate, BigDecimal overallTp,
                                 int maxCycles, BigDecimal maxLoss,
                                String quantity) {
        this.contract = contract;
        this.leverage = leverage;
        this.marginMode = marginMode;
        this.gridRate = gridRate;
        this.overallTp = overallTp;
        this.maxCycles = maxCycles;
        this.maxLoss = maxLoss;
        this.quantity = quantity;
        this.positionMode = positionMode;

        this.apiClient = new ApiClient();
        this.apiClient.setBasePath("https://api-testnet.gateapi.io/api/v4");
        this.apiClient.setApiKeySecret(apiKey, apiSecret);
        this.futuresApi = new FuturesApi(apiClient);
    }

    /**
     * 初始化账户：设置持仓模式 + 杠杆
     */
    public void init() {
        try {
            futuresApi.updateContractPositionLeverageCall(
                    SETTLE, contract, leverage, marginMode, positionMode, null);
            log.info("[GateGrid] 已设置杠杆: {}x, 保证金模式: {}", leverage, marginMode);

            FuturesAccount account = futuresApi.listFuturesAccounts(SETTLE);
            log.info("[GateGrid] 账户可用余额: {}, 总资产: {}",
                    account.getAvailable(), account.getTotal());
            String positionModeSet = account.getPositionMode();
            if (!positionMode.equals(positionModeSet)){
                futuresApi.setPositionMode(SETTLE, positionMode);
            }
            log.info("[GateGrid] 已设置双向持仓模式");
        } catch (GateApiException e) {
            log.error("[GateGrid] 初始化失败, label: {}, msg: {}", e.getErrorLabel(), e.getMessage());
        } catch (ApiException e) {
            log.error("[GateGrid] 初始化API调用失败, code: {}", e.getCode());
        }
    }

    /**
     * 启动网格策略
     */
    public void startGrid() {
        if (strategyActive) {
            log.warn("[GateGrid] 策略已在运行中");
            return;
        }
        strategyActive = true;
        currentCycle = 0;
        totalProfit = BigDecimal.ZERO;
        log.info("[GateGrid] 网格策略启动, cycle: {}", currentCycle + 1);
        dualOpenPositions();
    }

    /**
     * 停止网格策略
     */
    public void stopGrid() {
        strategyActive = false;
        closeAllPositions();
        log.info("[GateGrid] 网格策略已停止, 总盈亏: {}, 循环: {}", totalProfit, currentCycle);
    }

    /**
     * K线回调：收到新的收盘价
     */
    public void onKline(BigDecimal closePrice) {
        lastClosePrice = closePrice;
        if (!strategyActive) {
            return;
        }
        checkPositions(closePrice);
    }

    /**
     * 多空双开
     */
    private void dualOpenPositions() {
        try {
            FuturesOrder longOrder = new FuturesOrder();
            longOrder.setContract(contract);
            longOrder.setSize(quantity);
            longOrder.setPrice("0");
            longOrder.setTif(FuturesOrder.TifEnum.IOC);
            longOrder.setText("t-grid-long-" + (currentCycle + 1));
            FuturesOrder longResult = futuresApi.createFuturesOrder(SETTLE, longOrder, null);
            longOrderId = longResult.getId();
            longEntryPrice = safeDecimal(longResult.getFillPrice());
            log.info("[GateGrid] 开多成功, price: {}, id: {}", longEntryPrice, longOrderId);
            placeLongTpSl(longEntryPrice);

            FuturesOrder shortOrder = new FuturesOrder();
            shortOrder.setContract(contract);
            shortOrder.setSize(negateQuantity(quantity));
            shortOrder.setPrice("0");
            shortOrder.setTif(FuturesOrder.TifEnum.IOC);
            shortOrder.setText("t-grid-short-" + (currentCycle + 1));
            FuturesOrder shortResult = futuresApi.createFuturesOrder(SETTLE, shortOrder, null);
            shortOrderId = shortResult.getId();
            shortEntryPrice = safeDecimal(shortResult.getFillPrice());
            log.info("[GateGrid] 开空成功, price: {}, id: {}", shortEntryPrice, shortOrderId);
            placeShortTpSl(shortEntryPrice);

            printGridInfo();
        } catch (GateApiException e) {
            log.error("[GateGrid] 双开失败, label: {}, msg: {}", e.getErrorLabel(), e.getMessage());
            strategyActive = false;
        } catch (Exception e) {
            log.error("[GateGrid] 双开异常", e);
            strategyActive = false;
        }
    }

    private void placeLongTpSl(BigDecimal entryPrice) {
        BigDecimal tpPrice = entryPrice.multiply(BigDecimal.ONE.add(gridRate)).setScale(1, RoundingMode.HALF_UP);
        placePriceTriggeredOrder(tpPrice, FuturesPriceTrigger.RuleEnum.NUMBER_1, "close-long-position", "close_long");
        log.info("[GateGrid] 多头止盈已设置, TP:{}", tpPrice);
    }

    private void placeShortTpSl(BigDecimal entryPrice) {
        BigDecimal tpPrice = entryPrice.multiply(BigDecimal.ONE.subtract(gridRate)).setScale(1, RoundingMode.HALF_UP);
        placePriceTriggeredOrder(tpPrice, FuturesPriceTrigger.RuleEnum.NUMBER_2, "close-short-position", "close_short");
        log.info("[GateGrid] 空头止盈已设置, TP:{}", tpPrice);
    }

    private void placePriceTriggeredOrder(BigDecimal triggerPrice,
                                           FuturesPriceTrigger.RuleEnum rule,
                                           String orderType,
                                           String autoSize) {
        try {
            FuturesPriceTrigger trigger = new FuturesPriceTrigger();
            trigger.setStrategyType(FuturesPriceTrigger.StrategyTypeEnum.NUMBER_0);
            trigger.setPriceType(FuturesPriceTrigger.PriceTypeEnum.NUMBER_0);
            trigger.setPrice(triggerPrice.toString());
            trigger.setRule(rule);
            trigger.setExpiration(0);

            FuturesInitialOrder initial = new FuturesInitialOrder();
            initial.setContract(contract);
            initial.setSize(0L);
            initial.setPrice("0");
            initial.setTif(FuturesInitialOrder.TifEnum.IOC);
            initial.setReduceOnly(true);
            initial.setAutoSize(autoSize);

            FuturesPriceTriggeredOrder order = new FuturesPriceTriggeredOrder();
            order.setTrigger(trigger);
            order.setInitial(initial);
            order.setOrderType(orderType);

            TriggerOrderResponse response = futuresApi.createPriceTriggeredOrder(SETTLE, order);
            log.info("[GateGrid] 止盈条件单已创建, triggerPrice:{}, rule:{}, orderType:{}, autoSize:{}, id:{}",
                    triggerPrice, rule, orderType, autoSize, response.getId());
        } catch (Exception e) {
            log.error("[GateGrid] 止盈条件单创建失败, triggerPrice:{}, rule:{}, orderType:{}, autoSize:{}",
                    triggerPrice, rule, orderType, autoSize, e);
        }
    }

    /**
     * 检查多空仓位是否触及止盈止损
     */
    private void checkPositions(BigDecimal currentPrice) {
        if (longEntryPrice == null || shortEntryPrice == null) {
            return;
        }

        BigDecimal longTp = longEntryPrice.multiply(BigDecimal.ONE.add(gridRate)).setScale(1, RoundingMode.HALF_UP);
        BigDecimal longSl = longEntryPrice.multiply(BigDecimal.ONE.subtract(gridRate)).setScale(1, RoundingMode.HALF_UP);
        BigDecimal shortTp = shortEntryPrice.multiply(BigDecimal.ONE.subtract(gridRate)).setScale(1, RoundingMode.HALF_UP);
        BigDecimal shortSl = shortEntryPrice.multiply(BigDecimal.ONE.add(gridRate)).setScale(1, RoundingMode.HALF_UP);

        System.out.println("========== Gate 网格状态 ==========");
        System.out.println("当前价格: " + currentPrice);
        System.out.println("多头入场: " + longEntryPrice + " TP: " + longTp + " SL: " + longSl);
        System.out.println("空头入场: " + shortEntryPrice + " TP: " + shortTp + " SL: " + shortSl);
        System.out.println("累计盈亏: " + totalProfit + " 循环: " + currentCycle + "/" + maxCycles);
        System.out.println("===================================");

        // 多头止盈
        if (currentPrice.compareTo(longTp) >= 0) {
            log.info("[GateGrid] 多头止盈触发! entry:{}, current:{}", longEntryPrice, currentPrice);
            BigDecimal cnt = new BigDecimal(quantity);
            BigDecimal profit = currentPrice.subtract(longEntryPrice).multiply(cnt);
            totalProfit = totalProfit.add(profit);
            log.info("[GateGrid] 多头止盈 profit:{}, totalProfit:{}", profit, totalProfit);
            closeLongPosition();
            closeShortPosition();
            currentCycle++;
            checkStopConditions();
            return;
        }
        // 多头止损
        if (currentPrice.compareTo(longSl) <= 0) {
            log.info("[GateGrid] 多头止损触发! entry:{}, current:{}", longEntryPrice, currentPrice);
            BigDecimal cnt = new BigDecimal(quantity);
            BigDecimal loss = longEntryPrice.subtract(currentPrice).multiply(cnt);
            totalProfit = totalProfit.subtract(loss);
            log.info("[GateGrid] 多头止损 loss:{}, totalProfit:{}", loss, totalProfit);
            closeLongPosition();
            currentCycle++;
            checkStopConditions();
            return;
        }
        // 空头止盈
        if (currentPrice.compareTo(shortTp) <= 0) {
            log.info("[GateGrid] 空头止盈触发! entry:{}, current:{}", shortEntryPrice, currentPrice);
            BigDecimal cnt = new BigDecimal(quantity);
            BigDecimal profit = shortEntryPrice.subtract(currentPrice).multiply(cnt);
            totalProfit = totalProfit.add(profit);
            log.info("[GateGrid] 空头止盈 profit:{}, totalProfit:{}", profit, totalProfit);
            closeShortPosition();
            closeLongPosition();
            currentCycle++;
            checkStopConditions();
            return;
        }
        // 空头止损
        if (currentPrice.compareTo(shortSl) >= 0) {
            log.info("[GateGrid] 空头止损触发! entry:{}, current:{}", shortEntryPrice, currentPrice);
            BigDecimal cnt = new BigDecimal(quantity);
            BigDecimal loss = currentPrice.subtract(shortEntryPrice).multiply(cnt);
            totalProfit = totalProfit.subtract(loss);
            log.info("[GateGrid] 空头止损 loss:{}, totalProfit:{}", loss, totalProfit);
            closeShortPosition();
            currentCycle++;
            checkStopConditions();
        }
    }

    private void checkStopConditions() {
        if (totalProfit.compareTo(overallTp) >= 0) {
            log.info("[GateGrid] 达到整体止盈 {} USDT，停止策略", overallTp);
            strategyActive = false;
            return;
        }
        if (BigDecimal.ZERO.subtract(totalProfit).compareTo(maxLoss) >= 0) {
            log.info("[GateGrid] 亏损 {} 达到上限 {} USDT，停止策略", totalProfit.negate(), maxLoss);
            strategyActive = false;
            return;
        }
        if (currentCycle >= maxCycles) {
            log.info("[GateGrid] 达到最大循环次数 {}，停止策略", maxCycles);
            strategyActive = false;
            return;
        }
        log.info("[GateGrid] 进入下一轮循环: {}", currentCycle + 1);
        dualOpenPositions();
    }

    private void closeLongPosition() {
        if (longEntryPrice == null) {
            return;
        }
        try {
            FuturesOrder closeOrder = new FuturesOrder();
            closeOrder.setContract(contract);
            closeOrder.setSize(negateQuantity(quantity));
            closeOrder.setPrice("0");
            closeOrder.setTif(FuturesOrder.TifEnum.IOC);
            closeOrder.setReduceOnly(true);
            closeOrder.setText("t-grid-close-long");
            FuturesOrder result = futuresApi.createFuturesOrder(SETTLE, closeOrder, null);
            log.info("[GateGrid] 平多成功, id: {}, fillPrice: {}", result.getId(), result.getFillPrice());
        } catch (Exception e) {
            log.error("[GateGrid] 平多失败", e);
        }
        longEntryPrice = null;
        longOrderId = null;
    }

    private void closeShortPosition() {
        if (shortEntryPrice == null) {
            return;
        }
        try {
            FuturesOrder closeOrder = new FuturesOrder();
            closeOrder.setContract(contract);
            closeOrder.setSize(quantity);
            closeOrder.setPrice("0");
            closeOrder.setTif(FuturesOrder.TifEnum.IOC);
            closeOrder.setReduceOnly(true);
            closeOrder.setText("t-grid-close-short");
            FuturesOrder result = futuresApi.createFuturesOrder(SETTLE, closeOrder, null);
            log.info("[GateGrid] 平空成功, id: {}, fillPrice: {}", result.getId(), result.getFillPrice());
        } catch (Exception e) {
            log.error("[GateGrid] 平空失败", e);
        }
        shortEntryPrice = null;
        shortOrderId = null;
    }

    private void closeAllPositions() {
        closeLongPosition();
        closeShortPosition();
    }

    private void printGridInfo() {
        BigDecimal longTp = BigDecimal.ZERO;
        BigDecimal longSl = BigDecimal.ZERO;
        BigDecimal shortTp = BigDecimal.ZERO;
        BigDecimal shortSl = BigDecimal.ZERO;
        if (longEntryPrice != null) {
            longTp = longEntryPrice.multiply(BigDecimal.ONE.add(gridRate)).setScale(1, RoundingMode.HALF_UP);
            longSl = longEntryPrice.multiply(BigDecimal.ONE.subtract(gridRate)).setScale(1, RoundingMode.HALF_UP);
        }
        if (shortEntryPrice != null) {
            shortTp = shortEntryPrice.multiply(BigDecimal.ONE.subtract(gridRate)).setScale(1, RoundingMode.HALF_UP);
            shortSl = shortEntryPrice.multiply(BigDecimal.ONE.add(gridRate)).setScale(1, RoundingMode.HALF_UP);
        }
        System.out.println("========== Gate 网格开仓 ==========");
        System.out.println("合约: " + contract + " 杠杆: " + leverage + "x " + marginMode);
        System.out.println("多头入场: " + longEntryPrice + " TP: " + longTp + " SL: " + longSl);
        System.out.println("空头入场: " + shortEntryPrice + " TP: " + shortTp + " SL: " + shortSl);
        System.out.println("数量: " + quantity + " 网格间距: " + gridRate.multiply(new BigDecimal("100")) + "%");
        System.out.println("整体止盈: " + overallTp + " USDT 最大循环: " + maxCycles);
        System.out.println("最大亏损: " + maxLoss + " USDT");
        System.out.println("=====================================");
    }

    private String negateQuantity(String qty) {
        if (qty.startsWith("-")) {
            return qty.substring(1);
        }
        return "-" + qty;
    }

    private BigDecimal safeDecimal(String val) {
        if (val == null || val.isEmpty()) {
            return BigDecimal.ZERO;
        }
        return new BigDecimal(val);
    }

    public BigDecimal getLastClosePrice() {
        return lastClosePrice;
    }

    public boolean isStrategyActive() {
        return strategyActive;
    }

    public BigDecimal getTotalProfit() {
        return totalProfit;
    }

    public int getCurrentCycle() {
        return currentCycle;
    }
}

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

New file
@@ -0,0 +1,500 @@
package com.xcong.excoin.modules.gateApi;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.json.JSONException;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xcong.excoin.modules.blackchain.service.DateUtil;
import com.xcong.excoin.modules.okxNewPrice.celue.CaoZuoService;
import com.xcong.excoin.modules.okxNewPrice.indicator.macdAndMatrategy.MacdEmaStrategy;
import com.xcong.excoin.modules.okxNewPrice.indicator.macdAndMatrategy.MacdMaStrategy;
import com.xcong.excoin.modules.okxNewPrice.okxWs.InstrumentsWs;
import com.xcong.excoin.modules.okxNewPrice.okxWs.TradeOrderWs;
import com.xcong.excoin.modules.okxNewPrice.okxWs.enums.CoinEnums;
import com.xcong.excoin.modules.okxNewPrice.okxWs.param.Kline;
import com.xcong.excoin.modules.okxNewPrice.okxWs.param.TradeRequestParam;
import com.xcong.excoin.modules.okxNewPrice.okxWs.wanggeList.WangGeListService;
import com.xcong.excoin.modules.okxNewPrice.okxpi.config.ExchangeInfoEnum;
import com.xcong.excoin.modules.okxNewPrice.okxpi.config.ExchangeLoginService;
import com.xcong.excoin.modules.okxNewPrice.utils.SSLConfig;
import com.xcong.excoin.modules.okxNewPrice.utils.WsParamBuild;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import java.math.BigDecimal;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * Gate K线 WebSocket 客户端类，用于连接 Gate 的 WebSocket 接口，
 * 实时获取并处理 K线（candlestick）数据。
 * 同时支持心跳检测、自动重连以及异常恢复机制。
 * @author Administrator
 */
@Slf4j
public class GateKlineWebSocketClient {
    private final CaoZuoService caoZuoService;
    private final GateWebSocketClientManager clientManager;
    private final WangGeListService wangGeListService;

    private WebSocketClient webSocketClient;
    private ScheduledExecutorService heartbeatExecutor;
    private volatile ScheduledFuture<?> pongTimeoutFuture;
    private final AtomicReference<Long> lastMessageTime = new AtomicReference<>(System.currentTimeMillis());

    // 连接状态标志
    private final AtomicBoolean isConnected = new AtomicBoolean(false);
    private final AtomicBoolean isConnecting = new AtomicBoolean(false);
    private final AtomicBoolean isInitialized = new AtomicBoolean(false);

    private static final String CHANNEL = "futures.candlesticks";
    private static final String FUTURES_PING = "futures.ping";
    private static final String FUTURES_PONG = "futures.pong";
    private static final String GATE_INTERVAL = "1m";
    private static final String GATE_CONTRACT = "XAUT_USDT";

    private GateGridTradeService gridTradeService;

    // 心跳超时时间(秒)，小于30秒
    private static final int HEARTBEAT_TIMEOUT = 10;

    // 共享线程池用于重连等异步任务
    private final ExecutorService sharedExecutor = Executors.newCachedThreadPool(r -> {
        Thread t = new Thread(r, "gate-kline-worker");
        t.setDaemon(true);
        return t;
    });

    public GateKlineWebSocketClient(CaoZuoService caoZuoService,
                                    GateWebSocketClientManager clientManager,
                                    WangGeListService wangGeListService,
                                    GateGridTradeService gridTradeService
    ) {
        this.caoZuoService = caoZuoService;
        this.clientManager = clientManager;
        this.wangGeListService = wangGeListService;
        this.gridTradeService = gridTradeService;
    }

    /**
     * 初始化方法，创建并初始化WebSocket客户端实例
     */
    public void init() {
        if (!isInitialized.compareAndSet(false, true)) {
            log.warn("GateKlineWebSocketClient 已经初始化过，跳过重复初始化");
            return;
        }
        connect();
        startHeartbeat();
    }

    /**
     * 销毁方法，关闭WebSocket连接和相关资源
     */
    public void destroy() {
        log.info("开始销毁GateKlineWebSocketClient");

        if (webSocketClient != null && webSocketClient.isOpen()) {
            unsubscribeKlineChannels();
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.warn("取消订阅等待被中断");
            }
        }

        if (sharedExecutor != null && !sharedExecutor.isShutdown()) {
            sharedExecutor.shutdown();
        }

        if (webSocketClient != null && webSocketClient.isOpen()) {
            try {
                webSocketClient.closeBlocking();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.warn("关闭WebSocket连接时被中断");
            }
        }

        shutdownExecutorGracefully(heartbeatExecutor);
        if (pongTimeoutFuture != null) {
            pongTimeoutFuture.cancel(true);
        }
        shutdownExecutorGracefully(sharedExecutor);

        log.info("GateKlineWebSocketClient销毁完成");
    }

    private static final String WS_URL_MONIPAN = "wss://ws-testnet.gate.com/v4/ws/futures/usdt";
    private static final String WS_URL_SHIPAN = "wss://fx-ws.gateio.ws/v4/ws/usdt";
    private static final boolean isAccountType = false;

    /**
     * 建立与 Gate WebSocket 服务器的连接。
     * 设置回调函数以监听连接打开、接收消息、关闭和错误事件。
     */
    private void connect() {
        // 避免重复连接
        if (isConnecting.get()) {
            log.info("连接已在进行中，跳过重复连接请求");
            return;
        }
        
        if (!isConnecting.compareAndSet(false, true)) {
            log.info("连接已在进行中，跳过重复连接请求");
            return;
        }
        
        try {
            SSLConfig.configureSSL();
            System.setProperty("https.protocols", "TLSv1.2,TLSv1.3");
            String WS_URL = WS_URL_MONIPAN;
            if (isAccountType){
                WS_URL = WS_URL_SHIPAN;
            }
            URI uri = new URI(WS_URL);
            
            // 关闭之前的连接（如果存在）
            if (webSocketClient != null) {
                try {
                    webSocketClient.closeBlocking();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.warn("关闭之前连接时被中断");
                }
            }
            
            webSocketClient = new WebSocketClient(uri) {
                @Override
                public void onOpen(ServerHandshake handshake) {
                    log.info("Gate Kline WebSocket连接成功");
                    isConnected.set(true);
                    isConnecting.set(false);
                    
                    // 检查应用是否正在关闭
                    if (sharedExecutor != null && !sharedExecutor.isShutdown()) {
                        resetHeartbeatTimer();
                        subscribeKlineChannels();
                        subscribePingChannels();
                    } else {
                        log.warn("应用正在关闭，忽略WebSocket连接成功回调");
                    }
                }

                @Override
                public void onMessage(String message) {
                    lastMessageTime.set(System.currentTimeMillis());
                    handleWebSocketMessage(message);
                    resetHeartbeatTimer();
                }

                @Override
                public void onClose(int code, String reason, boolean remote) {
                    log.warn("Gate Kline WebSocket连接关闭: code={}, reason={}", code, reason);
                    isConnected.set(false);
                    isConnecting.set(false);
                    cancelPongTimeout();

                    if (sharedExecutor != null && !sharedExecutor.isShutdown() && !sharedExecutor.isTerminated()) {
                        sharedExecutor.execute(() -> {
                            try {
                                reconnectWithBackoff();
                            } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                                log.error("重连线程被中断", e);
                            } catch (Exception e) {
                                log.error("重连失败", e);
                            }
                        });
                    } else {
                        log.warn("共享线程池已关闭，无法执行重连任务");
                    }
                }

                @Override
                public void onError(Exception ex) {
                    log.error("Gate Kline WebSocket发生错误", ex);
                    isConnected.set(false);
                }
            };

            webSocketClient.connect();
        } catch (URISyntaxException e) {
            log.error("WebSocket URI格式错误", e);
            isConnecting.set(false);
        }
    }

    /**
     * 订阅K线频道。
     * 构造 Gate 格式的订阅请求并发送给服务端。
     * payload: ["1m", "BTC_USDT"] (间隔, 合约名)
     */
    private void subscribeKlineChannels() {
        JSONObject subscribeMsg = new JSONObject();
        subscribeMsg.put("time", System.currentTimeMillis() / 1000);
        subscribeMsg.put("channel", CHANNEL);
        subscribeMsg.put("event", "subscribe");

        JSONArray payload = new JSONArray();
        payload.add(GATE_INTERVAL);
        payload.add(GATE_CONTRACT);
        subscribeMsg.put("payload", payload);

        webSocketClient.send(subscribeMsg.toJSONString());
        log.info("已发送 K线频道订阅请求，合约: {}, 周期: {}", GATE_CONTRACT, GATE_INTERVAL);
    }
    private void subscribePingChannels() {
        JSONObject subscribeMsg = new JSONObject();
        subscribeMsg.put("time", System.currentTimeMillis() / 1000);
        subscribeMsg.put("channel", FUTURES_PING);
        webSocketClient.send(subscribeMsg.toJSONString());
        log.info("已发送 futures.ping");
    }

    private void unsubscribeKlineChannels() {
        JSONObject unsubscribeMsg = new JSONObject();
        unsubscribeMsg.put("time", System.currentTimeMillis() / 1000);
        unsubscribeMsg.put("channel", CHANNEL);
        unsubscribeMsg.put("event", "unsubscribe");
        JSONArray payload = new JSONArray();
        payload.add(GATE_INTERVAL);
        payload.add(GATE_CONTRACT);
        unsubscribeMsg.put("payload", payload);
        webSocketClient.send(unsubscribeMsg.toJSONString());
        log.info("已发送 K线频道取消订阅请求，合约: {}, 周期: {}", GATE_CONTRACT, GATE_INTERVAL);
    }

    /**
     * 处理从 WebSocket 收到的消息。
     * 包括订阅确认、错误响应、心跳响应以及实际的数据推送。
     *
     * @param message 来自 WebSocket 的原始字符串消息
     */
    private void handleWebSocketMessage(String message) {
        try {
            JSONObject response = JSON.parseObject(message);
            String channel = response.getString("channel");
            String event = response.getString("event");

            if (FUTURES_PONG.equals(channel)) {
                log.debug("收到futures.pong响应");
                cancelPongTimeout();
            } else if ("subscribe".equals(event)) {
                log.info("{} 频道订阅成功: {}", channel, response.getJSONObject("result"));
            } else if ("unsubscribe".equals(event)) {
                log.info("{} 频道取消订阅成功", channel);
            } else if ("error".equals(event)) {
                JSONObject error = response.getJSONObject("error");
                log.error("{} 频道错误: code={}, msg={}",
                        channel,
                        error != null ? error.getInteger("code") : "N/A",
                        error != null ? error.getString("message") : response.getString("msg"));
            } else if ("update".equals(event) || "all".equals(event)) {
                processPushDataV2(response);
            }
        } catch (Exception e) {
            log.error("处理WebSocket消息失败: {}", message, e);
        }
    }

    /**
     * 解析并处理K线推送数据。
     * 控制台输出K线数据，并在K线完结时触发策略和量化操作。
     *
     * @param response 包含K线数据的 JSON 对象
     */
    private void processPushDataV2(JSONObject response) {
        try {
            /**
             * Gate K线推送格式：
             * {
             *   "time": 1542162490,
             *   "time_ms": 1542162490123,
             *   "channel": "futures.candlesticks",
             *   "event": "update",
             *   "result": [
             *     {
             *       "t": 1545129300,
             *       "v": "27525555",
             *       "c": "95.4",
             *       "h": "96.9",
             *       "l": "89.5",
             *       "o": "94.3",
             *       "n": "1m_BTC_USD",
             *       "a": "314732.87412",
             *       "w": false
             *     }
             *   ]
             * }
             */
            String channel = response.getString("channel");
            if (!CHANNEL.equals(channel)) {
                return;
            }

            JSONArray resultArray = response.getJSONArray("result");
            if (resultArray == null || resultArray.isEmpty()) {
                log.warn("K线频道数据为空");
                return;
            }

            JSONObject data = resultArray.getJSONObject(0);
            BigDecimal openPx = new BigDecimal(data.getString("o"));
            BigDecimal highPx = new BigDecimal(data.getString("h"));
            BigDecimal lowPx = new BigDecimal(data.getString("l"));
            BigDecimal closePx = new BigDecimal(data.getString("c"));
            BigDecimal vol = new BigDecimal(data.getString("v"));
            BigDecimal baseVol = new BigDecimal(data.getString("a"));
            String name = data.getString("n");
            long t = data.getLong("t");
            boolean windowClosed = data.getBooleanValue("w");
            String time = DateUtil.TimeStampToDateTime(t);

            log.info("========== Gate K线数据 ==========");
            log.info("名称(n): " + name);
            log.info("时间  : " + time);
            log.info("开盘(o): " + openPx);
            log.info("最高(h): " + highPx);
            log.info("最低(l): " + lowPx);
            log.info("收盘(c): " + closePx);
            log.info("成交量(v): " + vol);
            log.info("成交额(a): " + baseVol);
            log.info("K线完结(w): " + windowClosed);
            log.info("==================================");

            if (gridTradeService != null) {
                gridTradeService.onKline(closePx);
            }
        } catch (Exception e) {
            log.error("处理 K线频道推送数据失败", e);
        }
    }

    /**
     * 启动心跳检测任务。
     * 使用 ScheduledExecutorService 定期检查是否需要发送 ping 请求来维持连接。
     */
    private void startHeartbeat() {
        if (heartbeatExecutor != null && !heartbeatExecutor.isTerminated()) {
            heartbeatExecutor.shutdownNow();
        }

        heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "gate-kline-heartbeat");
            t.setDaemon(true);
            return t;
        });

        heartbeatExecutor.scheduleWithFixedDelay(this::checkHeartbeatTimeout, 25, 25, TimeUnit.SECONDS);
    }

    /**
     * 重置心跳计时器。
     * 当收到新消息或发送 ping 后取消当前超时任务并重新安排下一次超时检查。
     */
    private synchronized void resetHeartbeatTimer() {
        cancelPongTimeout();

        if (heartbeatExecutor != null && !heartbeatExecutor.isShutdown()) {
            pongTimeoutFuture = heartbeatExecutor.schedule(this::checkHeartbeatTimeout,
                    HEARTBEAT_TIMEOUT, TimeUnit.SECONDS);
        }
    }

    /**
     * 检查心跳超时情况。
     * 若长时间未收到任何消息则主动发送 ping 请求保持连接活跃。
     */
    private void checkHeartbeatTimeout() {
        // 只有在连接状态下才检查心跳
        if (!isConnected.get()) {
            return;
        }
        
        long currentTime = System.currentTimeMillis();
        long lastTime = lastMessageTime.get();

        if (currentTime - lastTime >= HEARTBEAT_TIMEOUT * 1000L) {
            sendPing();
        }
    }

    /**
     * 发送 ping 请求至 WebSocket 服务端。
     * 用于维持长连接有效性。
     */
    private void sendPing() {
        try {
            if (webSocketClient != null && webSocketClient.isOpen()) {
                JSONObject pingMsg = new JSONObject();
                pingMsg.put("time", System.currentTimeMillis() / 1000);
                pingMsg.put("channel", FUTURES_PING);
                webSocketClient.send(pingMsg.toJSONString());
                log.debug("发送futures.ping请求");
            }
        } catch (Exception e) {
            log.warn("发送ping失败", e);
        }
    }

    /**
     * 取消当前的心跳超时任务。
     * 在收到 pong 或其他有效消息时调用此方法避免不必要的断开重连。
     */
    private synchronized void cancelPongTimeout() {
        if (pongTimeoutFuture != null && !pongTimeoutFuture.isDone()) {
            pongTimeoutFuture.cancel(true);
        }
    }

    /**
     * 执行 WebSocket 重连操作。
     * 在连接意外中断后尝试重新建立连接。
     */
    private void reconnectWithBackoff() throws InterruptedException {
        int attempt = 0;
        int maxAttempts = 3;
        long delayMs = 5000;

        while (attempt < maxAttempts) {
            try {
                Thread.sleep(delayMs);
                connect();
                return;
            } catch (Exception e) {
                log.warn("第{}次重连失败", attempt + 1, e);
                delayMs *= 2;
                attempt++;
            }
        }

        log.error("超过最大重试次数({})仍未连接成功", maxAttempts);
    }

    /**
     * 优雅关闭线程池
     */
    private void shutdownExecutorGracefully(ExecutorService executor) {
        if (executor == null || executor.isTerminated()) {
            return;
        }
        try {
            executor.shutdown();
            if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            executor.shutdownNow();
        }
    }
}

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

New file
@@ -0,0 +1,19 @@
package com.xcong.excoin.modules.gateApi;

import com.xcong.excoin.modules.okxNewPrice.OkxWebSocketClientManager;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class GateWebSocketClientMain {
    public static void main(String[] args) throws InterruptedException {
        // 使用Spring上下文初始化管理器
        ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
        GateWebSocketClientManager manager = context.getBean(GateWebSocketClientManager.class);

        // 运行一段时间以观察结果
        Thread.sleep(1200000000L); // 运行一小时
        
        // 关闭连接
        manager.destroy();
    }
}

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

New file
@@ -0,0 +1,86 @@
package com.xcong.excoin.modules.gateApi;

import com.xcong.excoin.modules.okxNewPrice.celue.CaoZuoService;
import com.xcong.excoin.modules.okxNewPrice.okxWs.wanggeList.WangGeListService;
import com.xcong.excoin.utils.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.math.BigDecimal;

/**
 * 管理 Gate WebSocket 客户端和网格交易服务实例
 */
@Slf4j
@Component
public class GateWebSocketClientManager {
    @Autowired
    private CaoZuoService caoZuoService;
    @Autowired
    private WangGeListService wangGeListService;

    private GateKlineWebSocketClient klinePriceClient;
    private GateGridTradeService gridTradeService;

    private static final String API_KEY = "d90ca272391992b8e74f8f92cedb21ec";
    private static final String API_SECRET = "1861e4f52de4bb53369ea3208d9ede38ece4777368030f96c77d27934c46c274";

    @PostConstruct
    public void init() {
        log.info("开始初始化GateWebSocketClientManager");

        try {
            gridTradeService = new GateGridTradeService(
                    API_KEY, API_SECRET,
                    "XAUT_USDT",
                    "30",
                    "cross",
                    "dual",
                    new BigDecimal("0.0035"),
                    new BigDecimal("0.5"),
                    3,
                    new BigDecimal("7.5"),
                    "10"
            );
            gridTradeService.init();

            klinePriceClient = new GateKlineWebSocketClient(caoZuoService, this, wangGeListService,gridTradeService);
            klinePriceClient.init();
            log.info("已初始化GateKlineWebSocketClient");

            gridTradeService.startGrid();
        } catch (Exception e) {
            log.error("初始化GateWebSocketClientManager失败", e);
        }
    }

    @PreDestroy
    public void destroy() {
        log.info("开始销毁GateWebSocketClientManager");

        if (gridTradeService != null) {
            gridTradeService.stopGrid();
        }
        if (klinePriceClient != null) {
            try {
                klinePriceClient.destroy();
                log.info("已销毁GateKlineWebSocketClient");
            } catch (Exception e) {
                log.error("销毁GateKlineWebSocketClient失败", e);
            }
        }

        log.info("GateWebSocketClientManager销毁完成");
    }

    public GateKlineWebSocketClient getKlineWebSocketClient() {
        return klinePriceClient;
    }

    public GateGridTradeService getGridTradeService() {
        return gridTradeService;
    }
}

 src/main/java/com/xcong/excoin/modules/gateApi/gate-websocket.txt

New file
Diff too large

 src/main/resources/application-test.yml

@@ -7,9 +7,9 @@
  profiles:
    active: app
  datasource:
    url: jdbc:mysql://47.76.217.51:3306/db_base?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2b8
    username: db_base
    password: P@ssw0rd!123
    url: jdbc:mysql://120.27.238.55:3406/db_base?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2b8
    username: ct_test
    password: 123456
    driver-class-name: com.mysql.jdbc.Driver
    type: com.alibaba.druid.pool.DruidDataSource
    druid:
@@ -48,13 +48,13 @@
  ## redis配置
  redis:
    # Redis数据库索引（默认为 0）
    database: 13
    database: 8
    # Redis服务器地址
    host: 47.76.217.51
    host: 120.27.238.55
    # Redis服务器连接端口
    port: 6379
    port: 6479
    # Redis 密码
    password: lianghua1!qaz2@WSX
    password: d3y6dsdl;f.327
    lettuce:
      pool:
        # 连接池中的最小空闲连接
@@ -69,10 +69,10 @@
    timeout: 500000

  rabbitmq:
    host: 47.76.217.51
    host: 120.27.238.55
    port: 5672
    username: lianghua20210816
    password: lianghua20210816
    username: ct_rabbit
    password: 123456
    publisher-confirm-type: correlated


@@ -102,7 +102,7 @@
  rabbit-consumer: false
  block-job: false
  websocket: false
  quant: true
  quant: false

aliyun:
  oss: