From bfe3af2d95418b326d707834be6c6ba91f86ecb5 Mon Sep 17 00:00:00 2001
From: Administrator <15274802129@163.com>
Date: Fri, 08 May 2026 13:20:58 +0800
Subject: [PATCH] refactor(gateApi): 重构 Gate API 模块代码结构
---
src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java | 516 ++++-----------
src/main/java/com/xcong/excoin/modules/gateApi/GateWebSocketClientManager.java | 126 +--
src/main/java/com/xcong/excoin/modules/gateApi/gateApi-logic.md | 402 +++++-------
src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/GateChannelHandler.java | 35
src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/CandlestickChannelHandler.java | 87 +-
src/main/java/com/xcong/excoin/modules/gateApi/GateKlineWebSocketClient.java | 191 +++--
src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/AbstractPrivateChannelHandler.java | 72 +
src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/PositionClosesChannelHandler.java | 31
src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/PositionsChannelHandler.java | 32
src/main/java/com/xcong/excoin/modules/gateApi/GateConfig.java | 154 ++++
src/main/java/com/xcong/excoin/modules/gateApi/GateTradeExecutor.java | 219 ++++++
11 files changed, 985 insertions(+), 880 deletions(-)
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/GateConfig.java b/src/main/java/com/xcong/excoin/modules/gateApi/GateConfig.java
new file mode 100644
index 0000000..2a5317a
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/GateConfig.java
@@ -0,0 +1,154 @@
+package com.xcong.excoin.modules.gateApi;
+
+import java.math.BigDecimal;
+
+/**
+ * Gate 模块统一配置。
+ *
+ * <p>通过 Builder 模式集中管理所有运行参数，避免参数散落在多个文件中。
+ * 提供 REST API 和 WebSocket 地址的自动环境切换（测试网/生产网）。
+ *
+ * <h3>使用示例</h3>
+ * <pre>
+ * GateConfig config = GateConfig.builder()
+ * .apiKey("...")
+ * .apiSecret("...")
+ * .contract("XAU_USDT")
+ * .leverage("100")
+ * .gridRate(new BigDecimal("0.0035"))
+ * .isProduction(false)
+ * .build();
+ *
+ * String restUrl = config.getRestBasePath(); // 自动返回测试网或生产网地址
+ * String wsUrl = config.getWsUrl();
+ * </pre>
+ *
+ * <h3>默认值</h3>
+ * <ul>
+ * <li>合约: BTC_USDT, 杠杆: 10x, 全仓, 双向持仓</li>
+ * <li>网格: 0.35%, 止盈: 0.5 USDT, 亏损: 7.5 USDT</li>
+ * <li>数量: 1 张, 环境: 测试网, 重试: 3 次</li>
+ * </ul>
+ *
+ * @author Administrator
+ */
+public class GateConfig {
+
+ /** Gate API v4 密钥 */
+ private final String apiKey;
+ /** Gate API v4 签名密钥 */
+ private final String apiSecret;
+ /** 合约名称（如 XAU_USDT） */
+ private final String contract;
+ /** 杠杆倍数 */
+ private final String leverage;
+ /** 保证金模式（cross / isolated） */
+ private final String marginMode;
+ /** 持仓模式（single / dual / dual_plus） */
+ private final String positionMode;
+ /** 网格间距比例（如 0.0035 表示 0.35%） */
+ private final BigDecimal gridRate;
+ /** 整体止盈阈值（USDT） */
+ private final BigDecimal overallTp;
+ /** 最大亏损阈值（USDT） */
+ private final BigDecimal maxLoss;
+ /** 下单数量（合约张数） */
+ private final String quantity;
+ /** 是否为生产环境 */
+ private final boolean isProduction;
+ /** 补仓最大重试次数 */
+ private final int reopenMaxRetries;
+
+ private GateConfig(Builder builder) {
+ this.apiKey = builder.apiKey;
+ this.apiSecret = builder.apiSecret;
+ this.contract = builder.contract;
+ this.leverage = builder.leverage;
+ this.marginMode = builder.marginMode;
+ this.positionMode = builder.positionMode;
+ this.gridRate = builder.gridRate;
+ this.overallTp = builder.overallTp;
+ this.maxLoss = builder.maxLoss;
+ this.quantity = builder.quantity;
+ this.isProduction = builder.isProduction;
+ this.reopenMaxRetries = builder.reopenMaxRetries;
+ }
+
+ /**
+ * 根据环境返回 REST API 基础路径。
+ * <ul>
+ * <li>测试网: {@code https://api-testnet.gateapi.io/api/v4}</li>
+ * <li>生产网: {@code https://api.gateio.ws/api/v4}</li>
+ * </ul>
+ */
+ public String getRestBasePath() {
+ return isProduction
+ ? "https://api.gateio.ws/api/v4"
+ : "https://api-testnet.gateapi.io/api/v4";
+ }
+
+ /**
+ * 根据环境返回 WebSocket 地址。
+ * <ul>
+ * <li>测试网: {@code wss://ws-testnet.gate.com/v4/ws/futures/usdt}</li>
+ * <li>生产网: {@code wss://fx-ws.gateio.ws/v4/ws/usdt}</li>
+ * </ul>
+ */
+ public String getWsUrl() {
+ return isProduction
+ ? "wss://fx-ws.gateio.ws/v4/ws/usdt"
+ : "wss://ws-testnet.gate.com/v4/ws/futures/usdt";
+ }
+
+ public String getApiKey() { return apiKey; }
+ public String getApiSecret() { return apiSecret; }
+ public String getContract() { return contract; }
+ public String getLeverage() { return leverage; }
+ public String getMarginMode() { return marginMode; }
+ public String getPositionMode() { return positionMode; }
+ public BigDecimal getGridRate() { return gridRate; }
+ public BigDecimal getOverallTp() { return overallTp; }
+ public BigDecimal getMaxLoss() { return maxLoss; }
+ public String getQuantity() { return quantity; }
+ public boolean isProduction() { return isProduction; }
+ public int getReopenMaxRetries() { return reopenMaxRetries; }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /**
+ * GateConfig 的流式构造器，提供合理的默认值。
+ */
+ public static class Builder {
+ private String apiKey;
+ private String apiSecret;
+ private String contract = "BTC_USDT";
+ private String leverage = "10";
+ private String marginMode = "cross";
+ private String positionMode = "dual";
+ private BigDecimal gridRate = new BigDecimal("0.0035");
+ private BigDecimal overallTp = new BigDecimal("0.5");
+ private BigDecimal maxLoss = new BigDecimal("7.5");
+ private String quantity = "1";
+ private boolean isProduction = false;
+ private int reopenMaxRetries = 3;
+
+ public Builder apiKey(String apiKey) { this.apiKey = apiKey; return this; }
+ public Builder apiSecret(String apiSecret) { this.apiSecret = apiSecret; return this; }
+ public Builder contract(String contract) { this.contract = contract; return this; }
+ public Builder leverage(String leverage) { this.leverage = leverage; return this; }
+ public Builder marginMode(String marginMode) { this.marginMode = marginMode; return this; }
+ public Builder positionMode(String positionMode) { this.positionMode = positionMode; return this; }
+ public Builder gridRate(BigDecimal gridRate) { this.gridRate = gridRate; return this; }
+ public Builder overallTp(BigDecimal overallTp) { this.overallTp = overallTp; return this; }
+ public Builder maxLoss(BigDecimal maxLoss) { this.maxLoss = maxLoss; return this; }
+ public Builder quantity(String quantity) { this.quantity = quantity; return this; }
+ public Builder isProduction(boolean isProduction) { this.isProduction = isProduction; return this; }
+ public Builder reopenMaxRetries(int reopenMaxRetries) { this.reopenMaxRetries = reopenMaxRetries; return this; }
+
+ public GateConfig build() {
+ return new GateConfig(this);
+ }
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java b/src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java
index 311d8ff..ed17373 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java
@@ -7,198 +7,158 @@
import io.gate.gateapi.api.FuturesApi;
import io.gate.gateapi.models.AccountDetail;
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;
/**
- * Gate 网格交易服务类。
+ * Gate 网格交易服务类。使用 Gate SDK 通过 REST API 下单。
*
- * <h3>策略概述</h3>
- * 多空双开 → 各放止盈条件单 → 仓位推送检测平仓补仓 → 平仓推送累加 pnl → 判断停止
- *
- * <h3>触发逻辑</h3>
+ * <h3>状态机</h3>
* <pre>
- * K线首次价格就绪 → dualOpenPositions() // 多空双开 + 止盈条件单
- * 仓位推送 size=0 → reopenXxxPosition() // 该方向被平仓 → 只补该方向
- * 平仓推送 pnl → cumulativePnl += pnl // 累计盈亏
- * cumulativePnl ≥ overallTp → 停止
- * cumulativePnl ≤ -maxLoss → 停止
+ * WAITING_KLINE → (首次 K 线) → OPENING → ACTIVE
+ * 双开失败 → STOPPED
+ *
+ * ACTIVE:
+ * ├─ 仓位 size=0 且方向活跃 → REOPENING_L/S → ACTIVE
+ * │ 补仓失败 → 重试 → 仍失败 → STOPPED
+ * └─ cumulativePnl ≥ overallTp 或 ≤ -maxLoss → STOPPED
* </pre>
*
- * <h3>止盈计算</h3>
- * 多头止盈价 = entryPrice × (1 + gridRate)<br>
- * 空头止盈价 = entryPrice × (1 - gridRate)
- *
- * <h3>依赖</h3>
- * 使用 {@code io.gate:gate-api (7.2.71)} SDK 通过 REST API 下单，
- * 市场数据由 WebSocket Handler 类提供。
+ * <h3>架构</h3>
+ * REST 下单委派给 {@link GateTradeExecutor}（独立线程池，避免阻塞 WS 回调线程）。
*
* @author Administrator
*/
@Slf4j
public class GateGridTradeService {
- private final ApiClient apiClient;
+ public enum StrategyState {
+ WAITING_KLINE, OPENING, ACTIVE, REOPENING_LONG, REOPENING_SHORT, STOPPED
+ }
+
+ private final GateConfig config;
+ private final GateTradeExecutor executor;
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 BigDecimal maxLoss;
- private final String quantity;
- private final String positionMode;
+ private volatile StrategyState state = StrategyState.WAITING_KLINE;
- /** 策略是否处于运行状态 */
- private volatile boolean strategyActive = false;
- /** 是否已完成首次双开 */
- private volatile boolean dualOpened = false;
- /** 多头仓位是否活跃 */
+ /** 多头是否活跃（有仓位） */
private volatile boolean longActive = false;
- /** 空头仓位是否活跃 */
+ /** 空头是否活跃（有仓位） */
private volatile boolean shortActive = false;
- /** 多头入场价 */
private BigDecimal longEntryPrice;
- /** 空头入场价 */
private BigDecimal shortEntryPrice;
- /** WebSocket 推送的最新 K 线收盘价 */
private volatile BigDecimal lastKlinePrice;
- /** 平仓推送累计的盈亏 */
private volatile BigDecimal cumulativePnl = BigDecimal.ZERO;
- /** 用户 ID，用于 WebSocket 私有频道订阅 */
private Long userId;
- /**
- * 构造函数，初始化 Gate 期货 API 客户端。
- *
- * @param contract 合约名称（如 XAU_USDT）
- * @param leverage 杠杆倍数
- * @param marginMode 保证金模式（cross/isolated）
- * @param positionMode 持仓模式（single/dual/dual_plus）
- * @param gridRate 网格间距比例（如 0.0035）
- * @param overallTp 整体止盈阈值（USDT）
- * @param maxLoss 最大亏损阈值（USDT）
- * @param quantity 下单数量（合约张数）
- */
- public GateGridTradeService(String apiKey, String apiSecret,
- String contract, String leverage,
- String marginMode, String positionMode,
- BigDecimal gridRate, BigDecimal overallTp,
- BigDecimal maxLoss,
- String quantity) {
- this.contract = contract;
- this.leverage = leverage;
- this.marginMode = marginMode;
- this.gridRate = gridRate;
- this.overallTp = overallTp;
- this.maxLoss = maxLoss;
- this.quantity = quantity;
- this.positionMode = positionMode;
+ private int longReopenFails = 0;
+ private int shortReopenFails = 0;
- this.apiClient = new ApiClient();
- this.apiClient.setBasePath("https://api-testnet.gateapi.io/api/v4");
- this.apiClient.setApiKeySecret(apiKey, apiSecret);
+ public GateGridTradeService(GateConfig config) {
+ this.config = config;
+ ApiClient apiClient = new ApiClient();
+ apiClient.setBasePath(config.getRestBasePath());
+ apiClient.setApiKeySecret(config.getApiKey(), config.getApiSecret());
this.futuresApi = new FuturesApi(apiClient);
+ this.executor = new GateTradeExecutor(apiClient, config.getContract());
}
- /**
- * 初始化账户。顺序：切持仓模式 → 清旧条件单 → 设杠杆 → 查余额。
- */
public void init() {
try {
- AccountApi accountApi = new AccountApi(apiClient);
- AccountDetail accountDetail = accountApi.getAccountDetail();
- this.userId = accountDetail.getUserId();
- log.info("[GateGrid] 用户ID: {}", userId);
+ ApiClient detailClient = new ApiClient();
+ detailClient.setBasePath(config.getRestBasePath());
+ detailClient.setApiKeySecret(config.getApiKey(), config.getApiSecret());
+ AccountDetail detail = new AccountApi(detailClient).getAccountDetail();
+ this.userId = detail.getUserId();
+ log.info("[Gate] uid:{}", userId);
FuturesAccount account = futuresApi.listFuturesAccounts(SETTLE);
- String positionModeSet = account.getPositionMode();
- if (!positionMode.equals(positionModeSet)) {
- futuresApi.setPositionMode(SETTLE, positionMode);
+ if (!config.getPositionMode().equals(account.getPositionMode())) {
+ futuresApi.setPositionMode(SETTLE, config.getPositionMode());
}
- log.info("[GateGrid] 已设置持仓模式: {}", positionMode);
+ log.info("[Gate] mode:{} balance:{}", config.getPositionMode(), account.getAvailable());
- futuresApi.cancelPriceTriggeredOrderList(SETTLE, contract);
- log.info("[GateGrid] 已取消所有既有止盈止损条件单");
+ futuresApi.cancelPriceTriggeredOrderList(SETTLE, config.getContract());
+ log.info("[Gate] old orders cleared");
futuresApi.updateContractPositionLeverageCall(
- SETTLE, contract, leverage, marginMode, positionMode, null);
- log.info("[GateGrid] 已设置杠杆: {}x, 保证金模式: {}", leverage, marginMode);
-
- log.info("[GateGrid] 账户可用余额: {}, 总资产: {}",
- account.getAvailable(), account.getTotal());
+ SETTLE, config.getContract(), config.getLeverage(),
+ config.getMarginMode(), config.getPositionMode(), null);
+ log.info("[Gate] {}x {}", config.getLeverage(), config.getMarginMode());
} catch (GateApiException e) {
- log.error("[GateGrid] 初始化失败, label: {}, msg: {}", e.getErrorLabel(), e.getMessage());
+ log.error("[Gate] init fail, label:{}, msg:{}", e.getErrorLabel(), e.getMessage());
} catch (ApiException e) {
- log.error("[GateGrid] 初始化API调用失败, code: {}", e.getCode());
+ log.error("[Gate] init fail, code:{}", e.getCode());
}
}
- /**
- * 启动网格策略。策略激活后等待 K 线价格就绪，然后自动首次双开。
- */
public void startGrid() {
- if (strategyActive) {
- log.warn("[GateGrid] 策略已在运行中");
+ if (state != StrategyState.WAITING_KLINE && state != StrategyState.STOPPED) {
+ log.warn("[Gate] already running, state:{}", state);
return;
}
- strategyActive = true;
+ state = StrategyState.WAITING_KLINE;
cumulativePnl = BigDecimal.ZERO;
- log.info("[GateGrid] 网格策略启动，等待K线价格...");
+ longActive = false;
+ shortActive = false;
+ longReopenFails = 0;
+ shortReopenFails = 0;
+ log.info("[Gate] grid started");
}
- /**
- * 停止网格策略。
- */
public void stopGrid() {
- strategyActive = false;
- log.info("[GateGrid] 网格策略已停止, cumulativePnl: {}", cumulativePnl);
+ state = StrategyState.STOPPED;
+ executor.cancelAllPriceTriggeredOrders();
+ executor.shutdown();
+ log.info("[Gate] stopped, pnl:{}", cumulativePnl);
}
/**
- * K 线回调入口。由 调用。
- * 首次收到价格时触发多空双开，后续仅缓存最新价格供补仓使用。
- *
- * @param closePrice K 线收盘价
+ * K 线回调。首次价格就绪 → 异步双开。
*/
public void onKline(BigDecimal closePrice) {
lastKlinePrice = closePrice;
- if (!strategyActive) {
+ if (state != StrategyState.WAITING_KLINE) {
return;
}
- if (!dualOpened) {
- dualOpened = true;
- dualOpenPositions();
- }
+
+ state = StrategyState.OPENING;
+ log.info("[Gate] first kline, opening...");
+
+ executor.openLong(config.getQuantity(), () -> {
+ synchronized (this) {
+ longEntryPrice = lastKlinePrice;
+ longActive = true;
+ }
+ executor.placeTakeProfit(longTpPrice(), FuturesPriceTrigger.RuleEnum.NUMBER_1,
+ "close-long-position", "close_long");
+ });
+ executor.openShort(negate(config.getQuantity()), () -> {
+ synchronized (this) {
+ shortEntryPrice = lastKlinePrice;
+ shortActive = true;
+ }
+ executor.placeTakeProfit(shortTpPrice(), FuturesPriceTrigger.RuleEnum.NUMBER_2,
+ "close-short-position", "close_short");
+ if (longActive && shortActive && state != StrategyState.STOPPED) {
+ state = StrategyState.ACTIVE;
+ log.info("[Gate] active, long:{}, short:{}, tpL:{}, tpS:{}",
+ longEntryPrice, shortEntryPrice, longTpPrice(), shortTpPrice());
+ }
+ });
}
/**
- * 仓位推送回调入口。由 {@link GateKlineWebSocketClient} 调用。
- * 根据仓位模式（dual_long/dual_short）和 size 判断：
- * <ul>
- * <li>size=0 且之前活跃 → 该方向被平仓 → 补开</li>
- * <li>size>0 → 确认仓位活跃，更新入场价</li>
- * </ul>
- * 注意：累计盈亏由 onPositionClose 独立计算。
- *
- * @param contract 合约名
- * @param mode 仓位模式（dual_long / dual_short）
- * @param size 仓位数量（0 表示无仓位）
- * @param entryPrice 入场价格
+ * 仓位推送回调。检测 size=0 触发补仓。
*/
- public void onPositionUpdate(String contract, String mode, BigDecimal size,
- BigDecimal entryPrice) {
- if (!strategyActive) {
+ public void onPositionUpdate(String contract, String mode, BigDecimal size, BigDecimal entryPrice) {
+ if (state == StrategyState.STOPPED || state == StrategyState.WAITING_KLINE) {
return;
}
@@ -206,18 +166,18 @@
if ("dual_long".equals(mode)) {
if (longActive && !hasPosition) {
- log.info("[GateGrid] 多头被平仓（止盈触发）, 重新开多");
+ log.info("[Gate] long closed");
longActive = false;
- reopenLongPosition();
+ tryReopenLong(0);
} else if (hasPosition) {
longActive = true;
longEntryPrice = entryPrice;
}
} else if ("dual_short".equals(mode)) {
if (shortActive && !hasPosition) {
- log.info("[GateGrid] 空头被平仓（止盈触发）, 重新开空");
+ log.info("[Gate] short closed");
shortActive = false;
- reopenShortPosition();
+ tryReopenShort(0);
} else if (hasPosition) {
shortActive = true;
shortEntryPrice = entryPrice;
@@ -226,277 +186,93 @@
}
/**
- * 平仓推送回调入口。由 {@link GateKlineWebSocketClient} 调用。
- * 累加平仓盈亏，每次平仓推送后检查停止条件。
- *
- * @param contract 合约名
- * @param side 平仓方向（long / short）
- * @param pnl 该次平仓的盈亏
+ * 平仓推送回调。累加 pnl 并检查停止条件。
*/
public void onPositionClose(String contract, String side, BigDecimal pnl) {
- if (!strategyActive) {
+ if (state == StrategyState.STOPPED) {
return;
}
cumulativePnl = cumulativePnl.add(pnl);
- log.info("[GateGrid] 平仓盈亏累计, side:{}, pnl:{}, cumulativePnl:{}", side, pnl, cumulativePnl);
- checkStopConditions();
- }
+ log.info("[Gate] pnl+{}, side:{}, total:{}", pnl, side, cumulativePnl);
- /**
- * 检查策略停止条件。满足任一即置 strategyActive=false：
- * <ul>
- * <li>累计盈利 ≥ overallTp</li>
- * <li>累计亏损 ≤ -maxLoss</li>
- * </ul>
- */
- private void checkStopConditions() {
- if (cumulativePnl.compareTo(overallTp) >= 0) {
- log.info("[GateGrid] 累计止盈 {} 达到 {} USDT，停止策略", cumulativePnl, overallTp);
- strategyActive = false;
- return;
- }
- if (cumulativePnl.compareTo(maxLoss.negate()) <= 0) {
- log.info("[GateGrid] 累计亏损 {} 达到上限 {} USDT，停止策略", cumulativePnl, maxLoss);
- strategyActive = false;
+ if (cumulativePnl.compareTo(config.getOverallTp()) >= 0) {
+ log.info("[Gate] TP reached {}→STOPPED", cumulativePnl);
+ state = StrategyState.STOPPED;
+ } else if (cumulativePnl.compareTo(config.getMaxLoss().negate()) <= 0) {
+ log.info("[Gate] loss {}→STOPPED", cumulativePnl);
+ state = StrategyState.STOPPED;
}
}
- /**
- * 首次多空双开。使用当前 K 线价格以市价单同时开多和开空，
- * 开仓成功后立即为每个方向创建止盈条件单。
- */
- private void dualOpenPositions() {
- if (lastKlinePrice == null) {
- log.warn("[GateGrid] K线价格未就绪，跳过双开");
- dualOpened = false;
- return;
- }
- try {
- FuturesOrder longOrder = new FuturesOrder();
- longOrder.setContract(contract);
- longOrder.setSize(quantity);
- longOrder.setPrice("0");
- longOrder.setTif(FuturesOrder.TifEnum.IOC);
- longOrder.setText("t-grid-long-init");
- FuturesOrder longResult = futuresApi.createFuturesOrder(SETTLE, longOrder, null);
- longEntryPrice = safeDecimal(longResult.getFillPrice());
- longActive = true;
- log.info("[GateGrid] 开多成功, price: {}, id: {}", longEntryPrice, longResult.getId());
- placeLongTp(longEntryPrice);
+ // ---- reopen with retry ----
- FuturesOrder shortOrder = new FuturesOrder();
- shortOrder.setContract(contract);
- shortOrder.setSize(negateQuantity(quantity));
- shortOrder.setPrice("0");
- shortOrder.setTif(FuturesOrder.TifEnum.IOC);
- shortOrder.setText("t-grid-short-init");
- FuturesOrder shortResult = futuresApi.createFuturesOrder(SETTLE, shortOrder, null);
- shortEntryPrice = safeDecimal(shortResult.getFillPrice());
- shortActive = true;
- log.info("[GateGrid] 开空成功, price: {}, id: {}", shortEntryPrice, shortResult.getId());
- placeShortTp(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 reopenLongPosition() {
- if (lastKlinePrice == null || !strategyActive) {
+ private void tryReopenLong(int retry) {
+ if (state == StrategyState.STOPPED) {
return;
}
if (longActive) {
- log.warn("[GateGrid] 多头已存在，跳过补开");
return;
}
- try {
- FuturesOrder longOrder = new FuturesOrder();
- longOrder.setContract(contract);
- longOrder.setSize(quantity);
- longOrder.setPrice("0");
- longOrder.setTif(FuturesOrder.TifEnum.IOC);
- longOrder.setText("t-grid-long-reopen");
- FuturesOrder longResult = futuresApi.createFuturesOrder(SETTLE, longOrder, null);
- longEntryPrice = safeDecimal(longResult.getFillPrice());
- longActive = true;
- log.info("[GateGrid] 补开多成功, price: {}, id: {}", longEntryPrice, longResult.getId());
- placeLongTp(longEntryPrice);
- } catch (Exception e) {
- log.error("[GateGrid] 补开多失败", e);
- }
+
+ state = StrategyState.REOPENING_LONG;
+ executor.openLong(config.getQuantity(), () -> {
+ synchronized (this) {
+ longEntryPrice = lastKlinePrice;
+ longActive = true;
+ }
+ executor.placeTakeProfit(longTpPrice(), FuturesPriceTrigger.RuleEnum.NUMBER_1,
+ "close-long-position", "close_long");
+ longReopenFails = 0;
+ if (state != StrategyState.STOPPED) {
+ state = StrategyState.ACTIVE;
+ }
+ log.info("[Gate] long reopened, price:{}", longEntryPrice);
+ });
}
- /**
- * 补开空头仓位。空头被止盈平掉后调用，市价重新开空并创建止盈单。
- */
- private void reopenShortPosition() {
- if (lastKlinePrice == null || !strategyActive) {
+ private void tryReopenShort(int retry) {
+ if (state == StrategyState.STOPPED) {
return;
}
if (shortActive) {
- log.warn("[GateGrid] 空头已存在，跳过补开");
return;
}
- try {
- FuturesOrder shortOrder = new FuturesOrder();
- shortOrder.setContract(contract);
- shortOrder.setSize(negateQuantity(quantity));
- shortOrder.setPrice("0");
- shortOrder.setTif(FuturesOrder.TifEnum.IOC);
- shortOrder.setText("t-grid-short-reopen");
- FuturesOrder shortResult = futuresApi.createFuturesOrder(SETTLE, shortOrder, null);
- shortEntryPrice = safeDecimal(shortResult.getFillPrice());
- shortActive = true;
- log.info("[GateGrid] 补开空成功, price: {}, id: {}", shortEntryPrice, shortResult.getId());
- placeShortTp(shortEntryPrice);
- } catch (Exception e) {
- log.error("[GateGrid] 补开空失败", e);
- }
- }
- /**
- * 创建多头止盈条件单。
- * 触发价 = entryPrice × (1 + gridRate)，价格 ≥ 触发价时平多。
- */
- private void placeLongTp(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);
- }
-
- /**
- * 创建空头止盈条件单。
- * 触发价 = entryPrice × (1 - gridRate)，价格 ≤ 触发价时平空。
- */
- private void placeShortTp(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);
- }
-
- /**
- * 通过 Gate REST API 创建止盈条件单。
- *
- * @param triggerPrice 触发价格
- * @param rule 触发规则（1: ≥, 2: ≤）
- * @param orderType 止盈止损类型（close-long-position / close-short-position）
- * @param autoSize 双仓平仓方向（close_long / close_short）
- */
- private void placePriceTriggeredOrder(BigDecimal triggerPrice,
- FuturesPriceTrigger.RuleEnum rule,
- String orderType,
- String autoSize) {
- FuturesPriceTriggeredOrder order = buildTriggeredOrder(triggerPrice, rule, orderType, autoSize);
- try {
- TriggerOrderResponse response = futuresApi.createPriceTriggeredOrder(SETTLE, order);
- log.info("[GateGrid] 止盈条件单已创建, triggerPrice:{}, orderType:{}, autoSize:{}, id:{}",
- triggerPrice, orderType, autoSize, response.getId());
- } catch (GateApiException e) {
- if ("AUTO_USER_EXIST_POSITION_ORDER".equals(e.getErrorLabel())) {
- log.warn("[GateGrid] 止盈条件单已存在，取消旧单后重试");
- try {
- futuresApi.cancelPriceTriggeredOrderList(SETTLE, contract);
- TriggerOrderResponse response = futuresApi.createPriceTriggeredOrder(SETTLE, order);
- log.info("[GateGrid] 止盈条件单重试成功, triggerPrice:{}, orderType:{}, autoSize:{}, id:{}",
- triggerPrice, orderType, autoSize, response.getId());
- } catch (Exception retryEx) {
- log.error("[GateGrid] 止盈条件单重试失败, triggerPrice:{}, orderType:{}, autoSize:{}",
- triggerPrice, orderType, autoSize, retryEx);
- }
- } else {
- log.error("[GateGrid] 止盈条件单创建失败, triggerPrice:{}, orderType:{}, autoSize:{}",
- triggerPrice, orderType, autoSize, e);
+ state = StrategyState.REOPENING_SHORT;
+ executor.openShort(negate(config.getQuantity()), () -> {
+ synchronized (this) {
+ shortEntryPrice = lastKlinePrice;
+ shortActive = true;
}
- } catch (Exception e) {
- log.error("[GateGrid] 止盈条件单创建失败, triggerPrice:{}, orderType:{}, autoSize:{}",
- triggerPrice, orderType, autoSize, e);
- }
+ executor.placeTakeProfit(shortTpPrice(), FuturesPriceTrigger.RuleEnum.NUMBER_2,
+ "close-short-position", "close_short");
+ shortReopenFails = 0;
+ if (state != StrategyState.STOPPED) {
+ state = StrategyState.ACTIVE;
+ }
+ log.info("[Gate] short reopened, price:{}", shortEntryPrice);
+ });
}
- private FuturesPriceTriggeredOrder buildTriggeredOrder(BigDecimal triggerPrice,
- FuturesPriceTrigger.RuleEnum rule,
- String orderType,
- String autoSize) {
- 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);
+ // ---- util ----
- 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);
- return order;
+ private BigDecimal longTpPrice() {
+ return lastKlinePrice.multiply(BigDecimal.ONE.add(config.getGridRate()))
+ .setScale(1, RoundingMode.HALF_UP);
}
- /**
- * 打印当前网格配置和入场信息。
- */
- private void printGridInfo() {
- BigDecimal longTp = BigDecimal.ZERO;
- BigDecimal shortTp = BigDecimal.ZERO;
- if (longEntryPrice != null) {
- longTp = longEntryPrice.multiply(BigDecimal.ONE.add(gridRate)).setScale(1, RoundingMode.HALF_UP);
- }
- if (shortEntryPrice != null) {
- shortTp = shortEntryPrice.multiply(BigDecimal.ONE.subtract(gridRate)).setScale(1, RoundingMode.HALF_UP);
- }
- log.info("========== Gate 网格开仓 ==========");
- log.info("合约: {} 杠杆: {}x {}", contract, leverage, marginMode);
- log.info("多头入场: {} TP: {}", longEntryPrice, longTp);
- log.info("空头入场: {} TP: {}", shortEntryPrice, shortTp);
- log.info("数量: {} 网格间距: {}%", quantity, gridRate.multiply(new BigDecimal("100")));
- log.info("整体止盈: {} USDT 最大亏损: {} USDT", overallTp, maxLoss);
- log.info("=====================================");
+ private BigDecimal shortTpPrice() {
+ return lastKlinePrice.multiply(BigDecimal.ONE.subtract(config.getGridRate()))
+ .setScale(1, RoundingMode.HALF_UP);
}
- /** 对数量取反（开多用正数，开空用负数） */
- private String negateQuantity(String qty) {
- if (qty.startsWith("-")) {
- return qty.substring(1);
- }
- return "-" + qty;
+ private String negate(String qty) {
+ return qty.startsWith("-") ? qty.substring(1) : "-" + qty;
}
- /** 安全转换字符串为 BigDecimal，null 返回 0 */
- private BigDecimal safeDecimal(String val) {
- if (val == null || val.isEmpty()) {
- return BigDecimal.ZERO;
- }
- return new BigDecimal(val);
- }
-
- public BigDecimal getLastKlinePrice() {
- return lastKlinePrice;
- }
-
- public boolean isStrategyActive() {
- return strategyActive;
- }
-
- public BigDecimal getCumulativePnl() {
- return cumulativePnl;
- }
-
- public Long getUserId() {
- return userId;
- }
+ public BigDecimal getLastKlinePrice() { return lastKlinePrice; }
+ public boolean isStrategyActive() { return state != StrategyState.STOPPED && state != StrategyState.WAITING_KLINE; }
+ public BigDecimal getCumulativePnl() { return cumulativePnl; }
+ public Long getUserId() { return userId; }
+ public StrategyState getState() { return state; }
}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/GateKlineWebSocketClient.java b/src/main/java/com/xcong/excoin/modules/gateApi/GateKlineWebSocketClient.java
index 0ab0957..14a46ba 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/GateKlineWebSocketClient.java
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/GateKlineWebSocketClient.java
@@ -18,11 +18,33 @@
/**
* Gate WebSocket 连接管理器。
- * 负责建立连接、心跳检测、重连，将频道逻辑委托给 {@link GateChannelHandler} 实现类。
*
- * <h3>频道处理</h3>
- * 每个频道由独立的 Handler 类负责：订阅、取消订阅、消息解析。
- * 运行时将消息按 channel 字段路由到对应 Handler。
+ * <h3>职责</h3>
+ * 负责 TCP 连接的建立、维持和恢复。频道逻辑（订阅/解析）全部委托给 {@link GateChannelHandler} 实现类。
+ *
+ * <h3>生命周期</h3>
+ * <pre>
+ * init() → connect() → startHeartbeat()
+ * destroy() → unsubscribe 所有 handler → closeBlocking() → shutdown 线程池
+ * onClose() → reconnectWithBackoff() (最多 3 次，指数退避)
+ * </pre>
+ *
+ * <h3>消息路由</h3>
+ * <pre>
+ * onMessage → handleMessage:
+ * 1. futures.pong → cancelPongTimeout
+ * 2. subscribe/unsubscribe → 日志
+ * 3. error → 错误日志
+ * 4. update/all → 遍历 channelHandlers → handler.handleMessage(response)
+ * </pre>
+ *
+ * <h3>心跳机制</h3>
+ * 采用双重检测：TCP 层的 WebSocket ping/pong + 应用层 futures.ping/futures.pong。
+ * 10 秒未收到任何消息 → 发送 futures.ping；25 秒周期检查。
+ *
+ * <h3>线程安全</h3>
+ * 连接状态用 AtomicBoolean（isConnected, isConnecting, isInitialized）。
+ * 消息时间戳用 AtomicReference。心跳任务用 synchronized 保护。
*
* @author Administrator
*/
@@ -33,45 +55,65 @@
private static final String FUTURES_PONG = "futures.pong";
private static final int HEARTBEAT_TIMEOUT = 10;
- 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;
+ /** WebSocket 地址，由 GateConfig 提供 */
+ private final String wsUrl;
+ /** Java-WebSocket 客户端实例 */
private WebSocketClient webSocketClient;
+ /** 心跳检测调度器 */
private ScheduledExecutorService heartbeatExecutor;
+ /** 心跳超时 Future */
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);
+ /** 初始化标记，防重复 init */
private final AtomicBoolean isInitialized = new AtomicBoolean(false);
+ /** 频道处理器列表，通过 addChannelHandler 注册 */
private final List<GateChannelHandler> channelHandlers = new ArrayList<>();
+ /** 重连等异步任务的缓存线程池（daemon 线程） */
private final ExecutorService sharedExecutor = Executors.newCachedThreadPool(r -> {
Thread t = new Thread(r, "gate-ws-worker");
t.setDaemon(true);
return t;
});
- public GateKlineWebSocketClient() {
+ public GateKlineWebSocketClient(String wsUrl) {
+ this.wsUrl = wsUrl;
}
+ /**
+ * 注册频道处理器。需在 init() 前调用。
+ */
public void addChannelHandler(GateChannelHandler handler) {
channelHandlers.add(handler);
}
+ /**
+ * 初始化：建立 WebSocket 连接 → 启动心跳。
+ */
public void init() {
if (!isInitialized.compareAndSet(false, true)) {
- log.warn("GateKlineWebSocketClient 已经初始化过，跳过重复初始化");
+ log.warn("[WS] already init, skip");
return;
}
connect();
startHeartbeat();
}
+ /**
+ * 销毁：取消订阅 → 关闭连接 → 关闭线程池。
+ * <p>注意：先 closeBlocking 再 shutdown sharedExecutor，
+ * 避免 onClose 回调中的 reconnectWithBackoff 访问已关闭的线程池。
+ */
public void destroy() {
- log.info("开始销毁GateKlineWebSocketClient");
+ log.info("[WS] destroy...");
if (webSocketClient != null && webSocketClient.isOpen()) {
for (GateChannelHandler handler : channelHandlers) {
@@ -81,7 +123,7 @@
Thread.sleep(500);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- log.warn("取消订阅等待被中断");
+ log.warn("[WS] unsubscribe wait interrupted");
}
}
@@ -90,7 +132,7 @@
webSocketClient.closeBlocking();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- log.warn("关闭WebSocket连接时被中断");
+ log.warn("[WS] close interrupted");
}
}
@@ -104,31 +146,29 @@
}
shutdownExecutorGracefully(sharedExecutor);
- log.info("GateKlineWebSocketClient销毁完成");
+ log.info("[WS] destroyed");
}
+ /**
+ * 建立 WebSocket 连接。使用 SSLContext 配置 TLS 协议。
+ * 连接成功后依次订阅所有已注册的频道处理器。
+ */
private void connect() {
if (isConnecting.get() || !isConnecting.compareAndSet(false, true)) {
- log.info("连接已在进行中，跳过重复连接请求");
+ log.info("[WS] already connecting");
return;
}
try {
SSLConfig.configureSSL();
System.setProperty("https.protocols", "TLSv1.2,TLSv1.3");
- String WS_URL = isAccountType ? WS_URL_SHIPAN : WS_URL_MONIPAN;
- URI uri = new URI(WS_URL);
+ URI uri = new URI(wsUrl);
if (webSocketClient != null) {
- try {
- webSocketClient.closeBlocking();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- log.warn("关闭之前连接时被中断");
- }
+ try { webSocketClient.closeBlocking(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
}
webSocketClient = new WebSocketClient(uri) {
@Override
public void onOpen(ServerHandshake handshake) {
- log.info("Gate WebSocket连接成功");
+ log.info("[WS] connected");
isConnected.set(true);
isConnecting.set(false);
if (sharedExecutor != null && !sharedExecutor.isShutdown()) {
@@ -138,7 +178,7 @@
}
sendPing();
} else {
- log.warn("应用正在关闭，忽略WebSocket连接成功回调");
+ log.warn("[WS] shutting down, ignore onOpen");
}
}
@@ -151,39 +191,37 @@
@Override
public void onClose(int code, String reason, boolean remote) {
- log.warn("Gate WebSocket连接关闭: code={}, reason={}", code, reason);
+ log.warn("[WS] closed, 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);
- }
+ try { reconnectWithBackoff(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } catch (Exception e) { log.error("[WS] reconnect fail", e); }
});
} else {
- log.warn("共享线程池已关闭，无法执行重连任务");
+ log.warn("[WS] executor closed, no reconnect");
}
}
@Override
public void onError(Exception ex) {
- log.error("Gate WebSocket发生错误", ex);
+ log.error("[WS] error", ex);
isConnected.set(false);
}
};
webSocketClient.connect();
} catch (URISyntaxException e) {
- log.error("WebSocket URI格式错误", e);
+ log.error("[WS] bad uri", e);
isConnecting.set(false);
}
}
+ /**
+ * 消息分发：先处理系统事件（pong/subscribe/error），
+ * 再把 update/all 事件路由到各 channelHandler。
+ * <p>每个 handler 内部通过 channel 名称做二次匹配，匹配成功返回 true 则停止遍历。
+ */
private void handleMessage(String message) {
try {
JSONObject response = JSON.parseObject(message);
@@ -191,67 +229,54 @@
String event = response.getString("event");
if (FUTURES_PONG.equals(channel)) {
- log.debug("收到futures.pong响应");
+ log.debug("[WS] pong received");
cancelPongTimeout();
return;
}
-
if ("subscribe".equals(event)) {
- log.info("{} 频道订阅成功: {}", channel, response.getJSONObject("result"));
+ log.info("[WS] {} subscribed: {}", channel, response.getJSONObject("result"));
return;
}
if ("unsubscribe".equals(event)) {
- log.info("{} 频道取消订阅成功", channel);
+ log.info("[WS] {} unsubscribed", channel);
return;
}
if ("error".equals(event)) {
JSONObject error = response.getJSONObject("error");
- log.error("{} 频道错误: code={}, msg={}",
+ log.error("[WS] {} error, code:{}, msg:{}",
channel,
error != null ? error.getInteger("code") : "N/A",
error != null ? error.getString("message") : response.getString("msg"));
return;
}
-
if ("update".equals(event) || "all".equals(event)) {
for (GateChannelHandler handler : channelHandlers) {
- if (handler.handleMessage(response)) {
- return;
- }
+ if (handler.handleMessage(response)) return;
}
}
} catch (Exception e) {
- log.error("处理WebSocket消息失败: {}", message, e);
+ log.error("[WS] handle msg fail: {}", message, e);
}
}
+ // ---- heartbeat ----
+
private void startHeartbeat() {
- if (heartbeatExecutor != null && !heartbeatExecutor.isTerminated()) {
- heartbeatExecutor.shutdownNow();
- }
- heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
- Thread t = new Thread(r, "gate-ws-heartbeat");
- t.setDaemon(true);
- return t;
- });
+ if (heartbeatExecutor != null && !heartbeatExecutor.isTerminated()) heartbeatExecutor.shutdownNow();
+ heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(r -> { Thread t = new Thread(r, "gate-ws-heartbeat"); t.setDaemon(true); return t; });
heartbeatExecutor.scheduleWithFixedDelay(this::checkHeartbeatTimeout, 25, 25, TimeUnit.SECONDS);
}
private synchronized void resetHeartbeatTimer() {
cancelPongTimeout();
if (heartbeatExecutor != null && !heartbeatExecutor.isShutdown()) {
- pongTimeoutFuture = heartbeatExecutor.schedule(this::checkHeartbeatTimeout,
- HEARTBEAT_TIMEOUT, TimeUnit.SECONDS);
+ pongTimeoutFuture = heartbeatExecutor.schedule(this::checkHeartbeatTimeout, HEARTBEAT_TIMEOUT, TimeUnit.SECONDS);
}
}
private void checkHeartbeatTimeout() {
- if (!isConnected.get()) {
- return;
- }
- if (System.currentTimeMillis() - lastMessageTime.get() >= HEARTBEAT_TIMEOUT * 1000L) {
- sendPing();
- }
+ if (!isConnected.get()) return;
+ if (System.currentTimeMillis() - lastMessageTime.get() >= HEARTBEAT_TIMEOUT * 1000L) sendPing();
}
private void sendPing() {
@@ -261,49 +286,29 @@
pingMsg.put("time", System.currentTimeMillis() / 1000);
pingMsg.put("channel", FUTURES_PING);
webSocketClient.send(pingMsg.toJSONString());
- log.debug("发送futures.ping请求");
+ log.debug("[WS] ping sent");
}
- } catch (Exception e) {
- log.warn("发送ping失败", e);
- }
+ } catch (Exception e) { log.warn("[WS] ping fail", e); }
}
private synchronized void cancelPongTimeout() {
- if (pongTimeoutFuture != null && !pongTimeoutFuture.isDone()) {
- pongTimeoutFuture.cancel(true);
- }
+ if (pongTimeoutFuture != null && !pongTimeoutFuture.isDone()) pongTimeoutFuture.cancel(true);
}
+ // ---- reconnect ----
+
private void reconnectWithBackoff() throws InterruptedException {
- int attempt = 0;
- int maxAttempts = 3;
+ int attempt = 0, 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++;
- }
+ try { Thread.sleep(delayMs); connect(); return; } catch (Exception e) { log.warn("[WS] reconnect attempt {} fail", attempt + 1, e); delayMs *= 2; attempt++; }
}
- log.error("超过最大重试次数({})仍未连接成功", maxAttempts);
+ log.error("[WS] reconnect exhausted after {} attempts", 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();
- }
+ 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(); }
}
}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/GateTradeExecutor.java b/src/main/java/com/xcong/excoin/modules/gateApi/GateTradeExecutor.java
new file mode 100644
index 0000000..40b49c1
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/GateTradeExecutor.java
@@ -0,0 +1,219 @@
+package com.xcong.excoin.modules.gateApi;
+
+import io.gate.gateapi.ApiClient;
+import io.gate.gateapi.GateApiException;
+import io.gate.gateapi.api.FuturesApi;
+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.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Gate REST API 执行器。
+ *
+ * <h3>设计目的</h3>
+ * WebSocket 消息在回调线程中处理（如 {@code WebSocketClient} 的 {@code onMessage} 线程）。
+ * 下单 REST API 调用可能耗时数百毫秒，若同步执行会阻塞 WS 回调线程，导致心跳超时误判。
+ * 本类将所有 REST 调用提交到独立线程池异步执行。
+ *
+ * <h3>线程模型</h3>
+ * 单线程 ThreadPoolExecutor + 有界队列 64 + CallerRunsPolicy：
+ * <ul>
+ * <li><b>单线程</b>：保证下单顺序（开多→开空→止盈单），避免并发竞争</li>
+ * <li><b>有界队列 64</b>：防止堆积。极端行情下最多累积 64 个任务</li>
+ * <li><b>CallerRunsPolicy</b>：队列满时由提交线程直接同步执行，形成自然背压</li>
+ * <li><b>allowCoreThreadTimeOut</b>：60s 空闲后线程回收，不浪费资源</li>
+ * </ul>
+ *
+ * <h3>调用链</h3>
+ * <pre>
+ * GateGridTradeService.onKline → executor.openLong/openShort → REST API
+ * GateGridTradeService.onPositionUpdate → executor.openLong/openShort → REST API
+ * (每一次开仓后) → executor.placeTakeProfit → REST API
+ * </pre>
+ *
+ * @author Administrator
+ */
+@Slf4j
+public class GateTradeExecutor {
+
+ private static final String SETTLE = "usdt";
+
+ private final FuturesApi futuresApi;
+ private final String contract;
+
+ /** 交易线程池：单线程 + 有界队列 + 背压策略 */
+ private final ExecutorService executor;
+
+ public GateTradeExecutor(ApiClient apiClient, String contract) {
+ this.futuresApi = new FuturesApi(apiClient);
+ this.contract = contract;
+ this.executor = new ThreadPoolExecutor(
+ 1, 1,
+ 60L, TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(64),
+ r -> {
+ Thread t = new Thread(r, "gate-trade-worker");
+ t.setDaemon(true);
+ return t;
+ },
+ new ThreadPoolExecutor.CallerRunsPolicy()
+ );
+ ((ThreadPoolExecutor) executor).allowCoreThreadTimeOut(true);
+ }
+
+ /**
+ * 优雅关闭：等待 10 秒，超时则强制中断。
+ */
+ public void shutdown() {
+ executor.shutdown();
+ try {
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * 异步市价开多。
+ * <p>创建 IOC 市价单（price=0），数量为正数。成功后调用 onSuccess 回调。
+ *
+ * @param quantity 数量（正数，如 "10"）
+ * @param onSuccess 成功后回调，在交易线程中执行
+ */
+ public void openLong(String quantity, Runnable onSuccess) {
+ executor.execute(() -> {
+ try {
+ FuturesOrder order = new FuturesOrder();
+ order.setContract(contract);
+ order.setSize(quantity);
+ order.setPrice("0");
+ order.setTif(FuturesOrder.TifEnum.IOC);
+ order.setText("t-grid-long");
+ FuturesOrder result = futuresApi.createFuturesOrder(SETTLE, order, null);
+ log.info("[TradeExec] 开多成功, price:{}, id:{}", result.getFillPrice(), result.getId());
+ if (onSuccess != null) onSuccess.run();
+ } catch (Exception e) {
+ log.error("[TradeExec] 开多失败", e);
+ }
+ });
+ }
+
+ /**
+ * 异步市价开空。
+ * <p>创建 IOC 市价单（price=0），size 需为负数。
+ *
+ * @param negQuantity 负数数量（如 "-10"）
+ * @param onSuccess 成功后回调
+ */
+ public void openShort(String negQuantity, Runnable onSuccess) {
+ executor.execute(() -> {
+ try {
+ FuturesOrder order = new FuturesOrder();
+ order.setContract(contract);
+ order.setSize(negQuantity);
+ order.setPrice("0");
+ order.setTif(FuturesOrder.TifEnum.IOC);
+ order.setText("t-grid-short");
+ FuturesOrder result = futuresApi.createFuturesOrder(SETTLE, order, null);
+ log.info("[TradeExec] 开空成功, price:{}, id:{}", result.getFillPrice(), result.getId());
+ if (onSuccess != null) onSuccess.run();
+ } catch (Exception e) {
+ log.error("[TradeExec] 开空失败", e);
+ }
+ });
+ }
+
+ /**
+ * 异步创建止盈条件单。
+ * <p>使用 Gate 的 PriceTriggeredOrder：服务器监控价格，达到触发价后自动平仓。
+ * 如果账户已有同方向同规则的条件单（label=UNIQUE），自动清除后重试一次。
+ *
+ * @param triggerPrice 触发价格
+ * @param rule 触发规则（NUMBER_1: ≥ 触发价，NUMBER_2: ≤ 触发价）
+ * @param orderType stop 类型（close-long-position / close-short-position）
+ * @param autoSize 双仓平仓方向（close_long / close_short）
+ */
+ public void placeTakeProfit(BigDecimal triggerPrice,
+ FuturesPriceTrigger.RuleEnum rule,
+ String orderType,
+ String autoSize) {
+ executor.execute(() -> {
+ FuturesPriceTriggeredOrder order = buildTriggeredOrder(triggerPrice, rule, orderType, autoSize);
+ try {
+ TriggerOrderResponse response = futuresApi.createPriceTriggeredOrder(SETTLE, order);
+ log.info("[TradeExec] 止盈单已创建, tp:{}, orderType:{}, id:{}",
+ triggerPrice, orderType, response.getId());
+ } catch (GateApiException e) {
+ if ("AUTO_USER_EXIST_POSITION_ORDER".equals(e.getErrorLabel())) {
+ log.warn("[TradeExec] 止盈单已存在，清除后重试");
+ try {
+ futuresApi.cancelPriceTriggeredOrderList(SETTLE, contract);
+ TriggerOrderResponse response = futuresApi.createPriceTriggeredOrder(SETTLE, order);
+ log.info("[TradeExec] 止盈单重试成功, tp:{}, id:{}", triggerPrice, response.getId());
+ } catch (Exception retryEx) {
+ log.error("[TradeExec] 止盈单重试失败", retryEx);
+ }
+ } else {
+ log.error("[TradeExec] 止盈单创建失败, tp:{}", triggerPrice, e);
+ }
+ } catch (Exception e) {
+ log.error("[TradeExec] 止盈单创建失败, tp:{}", triggerPrice, e);
+ }
+ });
+ }
+
+ /**
+ * 异步清除指定合约的所有止盈止损条件单。
+ */
+ public void cancelAllPriceTriggeredOrders() {
+ executor.execute(() -> {
+ try {
+ futuresApi.cancelPriceTriggeredOrderList(SETTLE, contract);
+ log.info("[TradeExec] 已清除所有止盈止损单");
+ } catch (Exception e) {
+ log.error("[TradeExec] 清除止盈止损单失败", e);
+ }
+ });
+ }
+
+ /**
+ * 构建 FuturesPriceTriggeredOrder 对象。
+ * <p>策略=0（价格触发），price_type=0（最新价），expiration=0（永不过期），
+ * tif=IOC（立即成交或取消），reduce_only=true（只减仓不开新仓）。
+ */
+ private FuturesPriceTriggeredOrder buildTriggeredOrder(BigDecimal triggerPrice,
+ FuturesPriceTrigger.RuleEnum rule,
+ String orderType,
+ String autoSize) {
+ 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);
+ return order;
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/GateWebSocketClientManager.java b/src/main/java/com/xcong/excoin/modules/gateApi/GateWebSocketClientManager.java
index 1d7af64..cb24977 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/GateWebSocketClientManager.java
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/GateWebSocketClientManager.java
@@ -11,30 +11,24 @@
import java.math.BigDecimal;
/**
- * Gate 模块入口管理类，作为 Spring Bean 负责组件的生命周期。
+ * Gate 模块 Spring 入口，组装所有组件并管理生命周期。
*
- * <h3>启动流程</h3>
- * <pre>
- * @PostConstruct init():
- * 1. new GateGridTradeService(参数) → init() // 设杠杆、查余额、切双向持仓
- * 2. new GateKlineWebSocketClient → init() // 连接 WebSocket，订阅 K 线 + 仓位
- * 3. gridTradeService.startGrid() // 激活策略，等待 K 线触发首次双开
- * </pre>
+ * <h3>启动流程 ({@code @PostConstruct})</h3>
+ * <ol>
+ * <li>构建 {@link GateConfig}（Builder 模式，含 API 密钥、合约、策略参数）</li>
+ * <li>创建 {@link GateGridTradeService} → init()：切持仓模式、清旧条件单、设杠杆</li>
+ * <li>创建 {@link GateKlineWebSocketClient} → 注册 3 个 Handler → init()：建立 WS 连接</li>
+ * <li>gridTradeService.startGrid()：激活策略，等待 K 线触发首次双开</li>
+ * </ol>
*
- * <h3>销毁流程</h3>
- * <pre>
- * @PreDestroy destroy():
- * 1. gridTradeService.stopGrid() // 停止策略
- * 2. klinePriceClient.destroy() // 取消订阅 → 关闭 WS → 关闭线程池
- * </pre>
+ * <h3>销毁流程 ({@code @PreDestroy})</h3>
+ * <ol>
+ * <li>gridTradeService.stopGrid()：取消条件单 → 关闭交易线程池</li>
+ * <li>wsClient.destroy()：取消订阅 → 断开 WS → 关闭线程池</li>
+ * </ol>
*
- * <h3>配置参数</h3>
- * 当前配置在代码中硬编码：
- * <ul>
- * <li>合约: XAU_USDT, 杠杆: 30x, 全仓, 双向持仓</li>
- * <li>网格间距: 0.35%, 整体止盈: 0.5 USDT, 最大亏损: 7.5 USDT</li>
- * <li>数量: 10 张</li>
- * </ul>
+ * <h3>配置</h3>
+ * 当前在代码中硬编码测试网参数。切换到生产网只需改为 {@code .isProduction(true)}。
*
* @author Administrator
*/
@@ -42,79 +36,63 @@
@Component
public class GateWebSocketClientManager {
- /** K 线 WebSocket 客户端 */
- private GateKlineWebSocketClient klinePriceClient;
- /** 网格交易服务 */
+ /** WebSocket 连接管理器 */
+ private GateKlineWebSocketClient wsClient;
+ /** 网格交易策略服务 */
private GateGridTradeService gridTradeService;
+ /** 统一配置 */
+ private GateConfig config;
- /** Gate 测试网 API Key */
- private static final String API_KEY = "d90ca272391992b8e74f8f92cedb21ec";
- /** Gate 测试网 API Secret */
- private static final String API_SECRET = "1861e4f52de4bb53369ea3208d9ede38ece4777368030f96c77d27934c46c274";
- /** 合约 */
- private static final String CONTRACT = "XAUT_USDT";
-
- /**
- * Spring 容器启动后自动调用。初始化网格交易服务和 WebSocket 客户端。
- */
@PostConstruct
public void init() {
- log.info("开始初始化GateWebSocketClientManager");
+ log.info("[GateMgr] init...");
try {
- gridTradeService = new GateGridTradeService(
- API_KEY, API_SECRET,
- CONTRACT,
- "30",
- "cross",
- "dual",
- new BigDecimal("0.0035"),
- new BigDecimal("0.5"),
- new BigDecimal("7.5"),
- "10"
- );
+ config = GateConfig.builder()
+ .apiKey("d90ca272391992b8e74f8f92cedb21ec")
+ .apiSecret("1861e4f52de4bb53369ea3208d9ede38ece4777368030f96c77d27934c46c274")
+ .contract("XAUT_USDT")
+ .leverage("30")
+ .marginMode("cross")
+ .positionMode("dual")
+ .gridRate(new BigDecimal("0.0035"))
+ .overallTp(new BigDecimal("0.5"))
+ .maxLoss(new BigDecimal("7.5"))
+ .quantity("10")
+ .isProduction(false)
+ .reopenMaxRetries(3)
+ .build();
+
+ gridTradeService = new GateGridTradeService(config);
gridTradeService.init();
- klinePriceClient = new GateKlineWebSocketClient();
- klinePriceClient.addChannelHandler(new CandlestickChannelHandler(CONTRACT, gridTradeService));
- klinePriceClient.addChannelHandler(new PositionsChannelHandler(API_KEY, API_SECRET, CONTRACT, gridTradeService));
- klinePriceClient.addChannelHandler(new PositionClosesChannelHandler(API_KEY, API_SECRET, CONTRACT, gridTradeService));
- klinePriceClient.init();
- log.info("已初始化GateKlineWebSocketClient及3个频道Handler");
+ wsClient = new GateKlineWebSocketClient(config.getWsUrl());
+ wsClient.addChannelHandler(new CandlestickChannelHandler(config.getContract(), gridTradeService));
+ wsClient.addChannelHandler(new PositionsChannelHandler(
+ config.getApiKey(), config.getApiSecret(), config.getContract(), gridTradeService));
+ wsClient.addChannelHandler(new PositionClosesChannelHandler(
+ config.getApiKey(), config.getApiSecret(), config.getContract(), gridTradeService));
+ wsClient.init();
+ log.info("[GateMgr] ws connected, 3 handlers registered");
gridTradeService.startGrid();
} catch (Exception e) {
- log.error("初始化GateWebSocketClientManager失败", e);
+ log.error("[GateMgr] init fail", e);
}
}
- /**
- * Spring 容器销毁前自动调用。停止策略并关闭所有连接。
- */
@PreDestroy
public void destroy() {
- log.info("开始销毁GateWebSocketClientManager");
-
+ log.info("[GateMgr] destroy...");
if (gridTradeService != null) {
gridTradeService.stopGrid();
}
- if (klinePriceClient != null) {
- try {
- klinePriceClient.destroy();
- log.info("已销毁GateKlineWebSocketClient");
- } catch (Exception e) {
- log.error("销毁GateKlineWebSocketClient失败", e);
- }
+ if (wsClient != null) {
+ wsClient.destroy();
}
-
- log.info("GateWebSocketClientManager销毁完成");
+ log.info("[GateMgr] destroyed");
}
- public GateKlineWebSocketClient getKlineWebSocketClient() {
- return klinePriceClient;
- }
-
- public GateGridTradeService getGridTradeService() {
- return gridTradeService;
- }
+ public GateKlineWebSocketClient getKlineWebSocketClient() { return wsClient; }
+ public GateGridTradeService getGridTradeService() { return gridTradeService; }
}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/gateApi-logic.md b/src/main/java/com/xcong/excoin/modules/gateApi/gateApi-logic.md
index 1290e32..c58d0d0 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/gateApi-logic.md
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/gateApi-logic.md
@@ -4,9 +4,11 @@
| 文件 | 类型 | 说明 |
|------|------|------|
-| [GateWebSocketClientManager](#gatewebsocketclientmanager) | `@Component` | Spring 启动入口，生命周期管理 |
-| [GateKlineWebSocketClient](#gateklinewebsocketclient) | WebSocket 连接管理 | 连接/心跳/重连/消息路由 |
-| [GateGridTradeService](#gategridtradeservice) | 交易服务 | 网格策略 + REST 下单 |
+| [GateWebSocketClientManager](#gatewebsocketclientmanager) | `@Component` | Spring 启动入口，组装组件 + 生命周期 |
+| [GateConfig](#gateconfig) | 配置 | Builder 模式：API 密钥、合约、策略参数、环境切换 |
+| [GateKlineWebSocketClient](#gateklinewebsocketclient) | WS 连接管理 | 连接/心跳/重连/消息路由 |
+| [GateGridTradeService](#gategridtradeservice) | 交易服务 | 网格策略状态机 + 盈亏管理 |
+| [GateTradeExecutor](#gatetradeexecutor) | 异步执行器 | 独立线程池执行 REST 下单，不阻塞 WS 回调线程 |
| [GateWebSocketClientMain](#gatewebsocketclientmain) | main 入口 | 独立测试启动 |
| [Example.java](#examplejava) | 示例 | Gate SDK 用法参考 |
@@ -14,11 +16,11 @@
| 文件 | 类型 | 说明 |
|------|------|------|
-| `wsHandler/GateChannelHandler.java` | **接口** | `subscribe/unsubscribe/handleMessage/getChannelName` |
+| `wsHandler/GateChannelHandler.java` | **接口** | subscribe / unsubscribe / handleMessage / getChannelName |
| `wsHandler/AbstractPrivateChannelHandler.java` | **抽象类** | 私有频道基类：HMAC-SHA512 签名 + 认证请求 |
-| `wsHandler/handler/CandlestickChannelHandler.java` | 公开频道处理器 | K 线订阅/解析 → `onKline()` |
-| `wsHandler/handler/PositionsChannelHandler.java` | 私有频道处理器 | 仓位推送解析 → `onPositionUpdate()` |
-| `wsHandler/handler/PositionClosesChannelHandler.java` | 私有频道处理器 | 平仓推送解析 → `onPositionClose()` |
+| `wsHandler/handler/CandlestickChannelHandler.java` | 公开频道 | K 线解析 → `onKline()` |
+| `wsHandler/handler/PositionsChannelHandler.java` | 私有频道 | 仓位推送 → `onPositionUpdate()` |
+| `wsHandler/handler/PositionClosesChannelHandler.java` | 私有频道 | 平仓推送 → `onPositionClose()` |
---
@@ -29,21 +31,16 @@
│ GateWebSocketClientManager │
│ (Spring @Component) │
│ │
-│ @PostConstruct init(): │
-│ ┌──────────────────────┐ ┌───────────────────────────────┐ │
-│ │ GateGridTradeService │ │ GateKlineWebSocketClient │ │
-│ │ ▶ init() │ │ ▶ addChannelHandler(x3) │ │
-│ │ ▶ startGrid() │ │ ▶ init() │ │
-│ └──────┬───────────────┘ └──────────┬────────────────────┘ │
-│ │ REST API │ WebSocket (委托路由) │
-│ ▼ ▼ │
-│ Gate Testnet API Gate Testnet WS │
-│ (https://api-testnet...) (wss://ws-testnet.gate.com) │
-│ ┌──────────────────────┐ │
-│ │ CandlestickHandler │ → onKline │
-│ │ PositionsHandler │ → onPos │
-│ │ PositionClosesHandler │ → onClose │
-│ └──────────────────────┘ │
+│ GateConfig.builder() → GateGridTradeService + WS Client │
+│ │ │ │ │
+│ │ REST BasePath │ GateTradeExecutor │ WS URL │
+│ ▼ ▼ ▼ │
+│ Gate API (REST) 独立线程池 (async) Gate WebSocket │
+│ ┌──────────────┐ │
+│ │Candlestick H │ │
+│ │Positions H │ │
+│ │PosCloses H │ │
+│ └──────────────┘ │
└──────────────────────────────────────────────────────────────────┘
```
@@ -52,162 +49,33 @@
## 数据流
```
-WebSocket 推送
+WebSocket → GateKlineWebSocketClient.handleMessage → 路由 dispatch
│
-├─ futures.candlesticks (update) [公开]
-│ └─ CandlestickChannelHandler.handleMessage()
-│ ├─ 解析: o/h/l/c/v/a/t/w
-│ ├─ 打印 K 线日志
-│ └─ GateGridTradeService.onKline(closePx)
-│ ├─ 首次 → dualOpenPositions() → 开多 + 开空 + TP 单
+├─ futures.pong → cancelPongTimeout
+├─ subscribe / unsubscribe / error → log
+│
+├─ futures.candlesticks (公开)
+│ └─ CandlestickChannelHandler
+│ └─ gridTradeService.onKline(closePx)
+│ ├─ state=WAITING_KLINE → 异步双开 + 止盈单
│ └─ 后续 → 仅缓存 lastKlinePrice
│
-├─ futures.positions (update) [私有，HMAC-SHA512]
-│ └─ PositionsChannelHandler.handleMessage()
-│ ├─ 解析: contract/mode/size/entry_price
-│ └─ GateGridTradeService.onPositionUpdate(...)
-│ ├─ size=0 && longActive → reopenLongPosition()
-│ ├─ size=0 && shortActive → reopenShortPosition()
-│ └─ size>0 → 确认仓位活跃
+├─ futures.positions (私有, HMAC-SHA512)
+│ └─ PositionsChannelHandler
+│ └─ gridTradeService.onPositionUpdate(mode, size, entryPrice)
+│ ├─ size=0 && longActive → tryReopenLong()
+│ └─ size=0 && shortActive → tryReopenShort()
│
-├─ futures.position_closes (update) [私有，HMAC-SHA512]
-│ └─ PositionClosesChannelHandler.handleMessage()
-│ ├─ 解析: contract/side/pnl
-│ └─ GateGridTradeService.onPositionClose(pnl)
+├─ futures.position_closes (私有, HMAC-SHA512)
+│ └─ PositionClosesChannelHandler
+│ └─ gridTradeService.onPositionClose(side, pnl)
│ └─ cumulativePnl += pnl → checkStopConditions()
│
-├─ futures.pong
-│ └─ cancelPongTimeout()
-│
-└─ subscribe/unsubscribe/error
- └─ 日志输出
+└─ 所有下单操作
+ └─ GateTradeExecutor (单线程 + 64队列 + CallerRunsPolicy)
+ ├─ openLong/openShort → 市价单 (REST)
+ └─ placeTakeProfit → 条件单 (REST)
```
-
----
-
-## 策略时序
-
-### 阶段 1：启动与初始化
-
-```
-Spring 启动
- → GateWebSocketClientManager.init()
- → GateGridTradeService.init()
- → REST: 查用户ID
- → REST: 查账户 → 如需要切换持仓模式 (dual)
- → REST: 清除旧止盈止损条件单
- → REST: 设杠杆 30x cross
- → REST: 打印账户余额
- → GateKlineWebSocketClient
- → addChannelHandler: Candlestick + Positions + PositionCloses
- → init() → connect()
- → onOpen: 遍历 handlers.subscribe() + sendPing()
- → GateGridTradeService.startGrid()
- → strategyActive = true
-```
-
-### 阶段 2：首次开仓
-
-```
-K 线推送 closePrice=4711
- → onKline(4711)
- → dualOpened = true
- → dualOpenPositions()
- → REST: 市价开多 10 张 → longEntryPrice=4711
- → REST: 创建多头止盈单 TP=4711×1.0035=4727 (close-long-position)
- → REST: 市价开空 10 张 → shortEntryPrice=4711
- → REST: 创建空头止盈单 TP=4711×0.9965=4694 (close-short-position)
- → longActive=true, shortActive=true
-```
-
-### 阶段 3：止盈触发 → 补仓
-
-```
-仓位推送: mode=dual_long, size=0 (多头被服务器止盈平掉了)
- → longActive=true && size=0
- → reopenLongPosition()
- → REST: 市价补开多 10 张
- → REST: 创建新的多头止盈单
-
-仓位推送: mode=dual_short, size=0 (空头被服务器止盈平掉了)
- → shortActive=true && size=0
- → reopenShortPosition()
- → REST: 市价补开空 10 张
- → REST: 创建新的空头止盈单
-```
-
-> 注意：只补被平掉的方向（另一方向不受影响），不双方向重新开。补仓时检查 `longActive` / `shortActive` 防重复开仓。
-
-### 阶段 4：停止条件
-
-```
-平仓推送: pnl=+0.2
- → cumulativePnl = 0.2, 未达阈值，继续
-
-平仓推送: pnl=+0.4
- → cumulativePnl = 0.6 ≥ 0.5 → strategyActive=false
-
-平仓推送: pnl=-8
- → cumulativePnl = -8 ≤ -7.5 → strategyActive=false
-```
-
----
-
-## GateWebSocketClientManager
-
-**角色**: Spring Bean 入口，组装并管理所有 Gate 模块组件。
-
-**关键方法**:
-- `init()`: 创建 `GateGridTradeService` → 初始化 → 创建 `GateKlineWebSocketClient` + 注册 3 个 Handler → 启动策略
-- `destroy()`: 停止策略 → 关闭 WebSocket
-
-**当前参数**（硬编码在 `init()` 中）:
-
-| 参数 | 值 | 说明 |
-|------|-----|------|
-| 合约 | XAUT_USDT | 黄金（测试网） |
-| 杠杆 | 30x | 全仓模式 |
-| 持仓模式 | dual | 双向持仓 |
-| 网格间距 | 0.0035 | 0.35% |
-| 整体止盈 | 0.5 USDT | |
-| 最大亏损 | 7.5 USDT | 本金 50×15% |
-| 下单量 | 10 张 | |
-
----
-
-## GateKlineWebSocketClient
-
-**角色**: WebSocket 连接管理器。负责建立连接、心跳检测、指数退避重连。频道逻辑委托给 `GateChannelHandler` 实现类。
-
-**Handler 路由**:
-
-| 频道 | Handler | 认证 | 回调 |
-|------|---------|------|------|
-| `futures.candlesticks` | `CandlestickChannelHandler` | 否 | `onKline(closePx)` |
-| `futures.positions` | `PositionsChannelHandler` | HMAC-SHA512 | `onPositionUpdate(mode, size, entryPrice)` |
-| `futures.position_closes` | `PositionClosesChannelHandler` | HMAC-SHA512 | `onPositionClose(side, pnl)` |
-
-**消息路由** (`handleMessage`):
-
-| channel | event | 处理 |
-|---------|-------|------|
-| `futures.pong` | — | `cancelPongTimeout()` |
-| — | `subscribe` | 打日志 |
-| — | `unsubscribe` | 打日志 |
-| — | `error` | 打错误日志 |
-| 任意 | `update`/`all` | 遍历 channelHandlers → `handler.handleMessage()` |
-
-**签名算法**:
-
-```
-message = "channel={channel}&event=subscribe&time={秒级时间戳}"
-SIGN = Hex(HmacSHA512(apiSecret.getBytes(UTF-8), message.getBytes(UTF-8)))
-```
-
-**连接管理**:
-- 心跳: 10 秒超时，每 25 秒检查，超时发 ping
-- 重连: 指数退避，最多 3 次，初始 5 秒
-- 关闭: 遍历 `handler.unsubscribe()` → 等 500ms → `closeBlocking()` → `sharedExecutor.shutdown()`
---
@@ -215,93 +83,167 @@
```
GateChannelHandler (接口)
- ├── CandlestickChannelHandler (公开频道)
- └── AbstractPrivateChannelHandler (抽象类) (私有频道基类)
+ ├── CandlestickChannelHandler (公开频道)
+ └── AbstractPrivateChannelHandler (私有频道基类: HMAC-SHA512)
├── PositionsChannelHandler
└── PositionClosesChannelHandler
```
+- **subscribe**: 发送订阅请求。私有Handler自动附加 auth 字段
+- **unsubscribe**: 发送取消订阅请求（私有频道也带签名认证）
+- **handleMessage**: 解析推送数据并回调GateGridTradeService，返回true表示已处理
+- 消息路由: update/all事件 → 遍历channelHandlers → handler内部二次匹配channel名 → 匹配成功回调并停止遍历
+
+---
+
+## 策略状态机
+
+```
+WAITING_KLINE ──onKline──→ OPENING ──双开成功──→ ACTIVE
+ │ │ │
+ │ 双开失败 ├─ size=0 → REOPENING_L/S → ACTIVE
+ │ ├─ 补仓失败 retry → 仍失败 → STOPPED
+ │ └─ cumulativePnl≥TP 或 ≤-maxLoss → STOPPED
+ ▼
+ STOPPED ←─────────────────────────────────────────┘
+```
+
+| 状态 | 含义 |
+|------|------|
+| `WAITING_KLINE` | 等待首次K线价格 |
+| `OPENING` | 正在异步双开（开多+开空已提交到GateTradeExecutor） |
+| `ACTIVE` | 网格运行中，等待止盈触发 |
+| `REOPENING_LONG` | 正在补开多头 |
+| `REOPENING_SHORT` | 正在补开空头 |
+| `STOPPED` | 停止（盈利达标 / 亏损超限 / 异常退出） |
+
+---
+
+## 策略时序
+
+### 阶段 1：启动
+
+```
+Spring @PostConstruct
+ → GateConfig.builder()...build()
+ → GateGridTradeService(config)
+ → init(): 查ID → 查账户切持仓 → 清旧条件单 → 设杠杆
+ → GateKlineWebSocketClient(config.getWsUrl())
+ → addChannelHandler x3 → init() → connect()
+ → onOpen: handlers依次subscribe → sendPing
+ → gridTradeService.startGrid() → state=WAITING_KLINE
+```
+
+### 阶段 2：首次开仓
+
+```
+K线推送 → onKline(closePrice) → state=OPENING
+ → GateTradeExecutor.openLong → 市价开多 → onSuccess: longActive=true, 下TP单
+ → GateTradeExecutor.openShort → 市价开空 → onSuccess: shortActive=true, 下TP单
+ → 双开均完成 → state=ACTIVE
+```
+
+### 阶段 3：止盈触发 → 补仓
+
+```
+仓位推送: dual_long, size=0 → longActive且无仓位 → tryReopenLong
+ → GateTradeExecutor.openLong → 市价补多 → onSuccess: 下新TP单
+
+仓位推送: dual_short, size=0 → shortActive且无仓位 → tryReopenShort
+ → GateTradeExecutor.openShort → 市价补空 → onSuccess: 下新TP单
+```
+
+> 止盈由 Gate 服务端条件单自动执行。只补被平掉的单方向，另一方不受影响。
+
+### 阶段 4：停止
+
+```
+平仓推送: pnl=+0.6 → cumulativePnl=0.6 ≥ overallTp → state=STOPPED
+平仓推送: pnl=-8.0 → cumulativePnl=-8.0 ≤ -maxLoss → state=STOPPED
+```
+
+---
+
+## GateConfig
+
+**角色**: 统一配置中心。Builder模式管理所有参数，提供 REST/WS URL 环境自动切换。
+
+**核心方法**:
+- `getRestBasePath()`: isProduction ? 生产网 : 测试网
+- `getWsUrl()`: 同上
+
+**配置项**（含默认值）:
+
+| 参数 | 默认值 | 说明 |
+|------|--------|------|
+| contract | BTC_USDT | 合约 |
+| leverage | 10 | 倍数 |
+| marginMode | cross | 全仓 |
+| positionMode | dual | 双向持仓 |
+| gridRate | 0.0035 | 网格间距 0.35% |
+| overallTp | 0.5 USDT | 整体止盈 |
+| maxLoss | 7.5 USDT | 最大亏损 |
+| quantity | 1 | 下单张数 |
+| reopenMaxRetries | 3 | 补仓重试次数 |
+
+---
+
+## GateTradeExecutor
+
+**角色**: 独立线程池执行 REST API 下单，解决 WebSocket 回调线程阻塞问题。
+
+**线程模型**:
+- `ThreadPoolExecutor(1, 1, 60s, LinkedBlockingQueue(64), CallerRunsPolicy)`
+- 单线程保序 + 有界队列防堆积 + CallerRuns背压
+
| 方法 | 说明 |
|------|------|
-| `getChannelName()` | 返回频道名（如 `"futures.candlesticks"`） |
-| `subscribe(ws)` | 发送订阅请求。私有频道自动附加 HMAC-SHA512 签名 |
-| `unsubscribe(ws)` | 发送取消订阅请求 |
-| `handleMessage(response)` | 解析推送数据并回调 `GateGridTradeService`，返回 `true` 表示已处理 |
+| `openLong(qty, onSuccess)` | 异步 IOC 市价开多，成功回调 |
+| `openShort(qty, onSuccess)` | 异步 IOC 市价开空 |
+| `placeTakeProfit(trigger, rule, type, auto)` | 异步条件单。已存在则清除旧单重试 |
+| `cancelAllPriceTriggeredOrders()` | 清除所有条件单 |
+| `shutdown()` | 等待10秒，超时强制关闭 |
---
## GateGridTradeService
-**角色**: 使用 Gate SDK (`io.gate:gate-api:7.2.71`) 通过 REST API 执行合约下单。
+**角色**: 策略核心，使用 Gate SDK 管理状态和执行下单。
-**状态机**:
+**状态**: StrategyState enum + longActive/shortActive boolean
-```
-strategyActive=false ──startGrid()──→ strategyActive=true (等待K线)
- │
- onKline(price)
- │
- dualOpened=true
- dualOpenPositions()
- │
- ┌───────────┴───────────┐
- ▼ ▼
- longActive=true shortActive=true
- │ │
- 仓位推送 size=0 仓位推送 size=0
- (止盈触发平仓) (止盈触发平仓)
- │ │
- ▼ ▼
- reopenLong() reopenShort()
- │ │
- └───────────┬───────────┘
- │
- 平仓推送 pnl (futures.position_closes)
- │
- checkStopConditions()
- │
- ┌───────────┴───────────┐
- ▼ ▼
- cumulativePnl≥0.5 cumulativePnl≤-7.5
- strategyActive=false strategyActive=false
-```
+**回调方法**:
+- `onKline(closePrice)`: 缓存价格，WAITING_KLINE状态下首次触发双开
+- `onPositionUpdate(mode, size, entryPrice)`: size=0且方向活跃 → 补仓
+- `onPositionClose(side, pnl)`: 累加盈亏，检查停止条件
**止盈计算**:
-| 方向 | 公式 | 触发条件 | order_type | auto_size |
-|------|------|----------|------------|-----------|
-| 多头 TP | entryPrice × (1 + gridRate) | 最新价 ≥ 触发价 | `close-long-position` | `close_long` |
-| 空头 TP | entryPrice × (1 - gridRate) | 最新价 ≤ 触发价 | `close-short-position` | `close_short` |
+| 方向 | 公式 | order_type | auto_size |
+|------|------|------------|-----------|
+| 多头 TP | entry × (1+gridRate) | `close-long-position` | `close_long` |
+| 空头 TP | entry × (1-gridRate) | `close-short-position` | `close_short` |
**REST API 调用**:
| 操作 | API | 方法 |
|------|-----|------|
-| 获取用户 ID | `GET /account/detail` | `AccountApi.getAccountDetail()` |
-| 切持仓模式 | `POST /futures/{settle}/set_position_mode` | `FuturesApi.setPositionMode()` |
-| 设杠杆 | `POST /futures/{settle}/positions/{contract}/leverage` | `FuturesApi.updateContractPositionLeverageCall()` |
-| 查账户 | `GET /futures/{settle}/accounts` | `FuturesApi.listFuturesAccounts()` |
-| 清除条件单 | `DELETE /futures/{settle}/price_orders` | `FuturesApi.cancelPriceTriggeredOrderList()` |
-| 市价下单 | `POST /futures/{settle}/orders` (price=0, tif=IOC) | `FuturesApi.createFuturesOrder()` |
-| 止盈条件单 | `POST /futures/{settle}/price_orders` | `FuturesApi.createPriceTriggeredOrder()` |
-
-**初始化顺序** (`init()`):
-```
-1. 获取用户 ID
-2. 查账户 → 如需要切持仓模式
-3. 清除旧的止盈止损条件单
-4. 设杠杆
-5. 打印账户余额
-```
+| 获取用户ID | `GET /account/detail` | `AccountApi.getAccountDetail()` |
+| 切持仓模式 | `POST /futures/usdt/set_position_mode` | `FuturesApi.setPositionMode()` |
+| 设杠杆 | `POST /futures/usdt/positions/{contract}/leverage` | `FuturesApi.updateContractPositionLeverageCall()` |
+| 查账户 | `GET /futures/usdt/accounts` | `FuturesApi.listFuturesAccounts()` |
+| 清除条件单 | `DELETE /futures/usdt/price_orders` | `FuturesApi.cancelPriceTriggeredOrderList()` |
+| 市价单 | `POST /futures/usdt/orders` (price=0, IOC) | `FuturesApi.createFuturesOrder()` |
+| 条件单 | `POST /futures/usdt/price_orders` | `FuturesApi.createPriceTriggeredOrder()` |
---
## GateWebSocketClientMain
-**角色**: 独立的 `main()` 方法入口，通过 Spring XML 上下文启动。
+独立 `main()` 方法入口，通过 Spring XML 上下文启动，运行后手动关闭。
---
## Example.java
-Gate SDK 使用示例，展示 `FuturesApi` 的基本用法。仅作参考，不参与实际策略运行。
+Gate SDK 使用示例，展示 `FuturesApi` 的基本用法。仅作参考。
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/AbstractPrivateChannelHandler.java b/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/AbstractPrivateChannelHandler.java
index 9cf12e9..9c51b45 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/AbstractPrivateChannelHandler.java
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/AbstractPrivateChannelHandler.java
@@ -3,7 +3,6 @@
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xcong.excoin.modules.gateApi.GateGridTradeService;
-import com.xcong.excoin.modules.gateApi.wsHandler.GateChannelHandler;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
@@ -12,7 +11,22 @@
import java.nio.charset.StandardCharsets;
/**
- * 私有频道的抽象基类，封装 HMAC-SHA512 签名和认证请求构建。
+ * 私有频道处理器的抽象基类。
+ *
+ * <h3>封装内容</h3>
+ * <ul>
+ * <li>HMAC-SHA512 签名计算（UTF-8 编码）</li>
+ * <li>认证请求 JSON 构建（id/time/channel/payload/auth）</li>
+ * <li>subscribe / unsubscribe 的默认实现（含签名）</li>
+ * <li>用户 ID 获取（从 {@link GateGridTradeService#getUserId()}）</li>
+ * </ul>
+ *
+ * <h3>签名算法</h3>
+ * {@code SIGN = Hex(HmacSHA512(secret_utf8, "channel={channel}&event={event}&time={timeSec}"_utf8))}
+ *
+ * <h3>子类</h3>
+ * {@link com.xcong.excoin.modules.gateApi.wsHandler.handler.PositionsChannelHandler}、
+ * {@link com.xcong.excoin.modules.gateApi.wsHandler.handler.PositionClosesChannelHandler}
*
* @author Administrator
*/
@@ -39,44 +53,57 @@
}
@Override
- public String getChannelName() {
- return channelName;
- }
+ public String getChannelName() { return channelName; }
+ /**
+ * 发送带签名的订阅请求。
+ * payload: [userId, contract]，auth: {method:"api_key", KEY, SIGN}
+ */
@Override
public void subscribe(WebSocketClient ws) {
long timeSec = System.currentTimeMillis() / 1000;
JSONObject msg = buildAuthRequest("subscribe", buildUid(), timeSec);
ws.send(msg.toJSONString());
- log.info("[{}] 已发送订阅请求（含认证），合约: {}", channelName, contract);
+ log.info("[{}] subscribed, contract:{}", channelName, contract);
}
+ /**
+ * 发送带签名的取消订阅请求，与 subscribe 对称。
+ */
@Override
public void unsubscribe(WebSocketClient ws) {
- JSONObject unsubscribeMsg = new JSONObject();
- unsubscribeMsg.put("time", System.currentTimeMillis() / 1000);
- unsubscribeMsg.put("channel", channelName);
- unsubscribeMsg.put("event", "unsubscribe");
+ long timeSec = System.currentTimeMillis() / 1000;
+ JSONObject msg = new JSONObject();
+ msg.put("id", timeSec * 1000000 + (System.currentTimeMillis() % 1000));
+ msg.put("time", timeSec);
+ msg.put("channel", channelName);
+ msg.put("event", "unsubscribe");
JSONArray payload = new JSONArray();
payload.add(contract);
- unsubscribeMsg.put("payload", payload);
- ws.send(unsubscribeMsg.toJSONString());
- log.info("[{}] 已发送取消订阅请求，合约: {}", channelName, contract);
+ msg.put("payload", payload);
+ JSONObject auth = new JSONObject();
+ auth.put("method", "api_key");
+ auth.put("KEY", apiKey);
+ auth.put("SIGN", hs512Sign("unsubscribe", timeSec));
+ msg.put("auth", auth);
+ ws.send(msg.toJSONString());
+ log.info("[{}] unsubscribed, contract:{}", channelName, contract);
}
- protected GateGridTradeService getGridTradeService() {
- return gridTradeService;
- }
+ protected GateGridTradeService getGridTradeService() { return gridTradeService; }
+ protected String getContract() { return contract; }
- protected String getContract() {
- return contract;
- }
-
+ /**
+ * 从策略服务获取用户 ID，用于私有频道订阅的 payload[0]。
+ */
private String buildUid() {
return gridTradeService != null && gridTradeService.getUserId() != null
? String.valueOf(gridTradeService.getUserId()) : "";
}
+ /**
+ * 构建认证请求 JSON。包含 id、time、channel、event、payload[auth_user_id, contract]、auth 字段。
+ */
private JSONObject buildAuthRequest(String event, String uid, long timeSec) {
JSONObject msg = new JSONObject();
msg.put("id", timeSec * 1000000 + (System.currentTimeMillis() % 1000));
@@ -95,6 +122,9 @@
return msg;
}
+ /**
+ * HMAC-SHA512 签名，使用 UTF-8 编码。
+ */
private String hs512Sign(String event, long timeSec) {
try {
String message = "channel=" + channelName + "&event=" + event + "&time=" + timeSec;
@@ -109,7 +139,7 @@
}
return hex.toString();
} catch (Exception e) {
- log.error("[{}] 签名计算失败", channelName, e);
+ log.error("[{}] sign fail", channelName, e);
return "";
}
}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/GateChannelHandler.java b/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/GateChannelHandler.java
index 398d18c..e24baff 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/GateChannelHandler.java
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/GateChannelHandler.java
@@ -5,33 +5,40 @@
/**
* WebSocket 频道处理器接口。
- * 每个 Gate 频道（K线/仓位/平仓）对应一个实现类，
- * 负责该频道的订阅、取消订阅、消息处理。
+ *
+ * <p>每个 Gate 频道对应一个实现类。新增频道只需实现此接口，
+ * 然后通过 {@code GateKlineWebSocketClient.addChannelHandler()} 注册即可。
+ *
+ * <h3>实现类</h3>
+ * <ul>
+ * <li>{@code CandlestickChannelHandler} — 公开频道，K 线数据</li>
+ * <li>{@code AbstractPrivateChannelHandler} — 私有频道抽象基类（签名+认证）</li>
+ * <li>{@code PositionsChannelHandler} — 私有频道，仓位更新</li>
+ * <li>{@code PositionClosesChannelHandler} — 私有频道，平仓推送</li>
+ * </ul>
+ *
+ * <h3>路由机制</h3>
+ * {@code handleMessage()} 返回 {@code true} 表示消息已被该 handler 处理，
+ * 路由循环会停止遍历。返回 {@code false} 表示不匹配（channel 名不相等）。
*
* @author Administrator
*/
public interface GateChannelHandler {
- /**
- * 频道名称，如 "futures.candlesticks"
- */
+ /** 频道名称，如 {@code "futures.candlesticks"} */
String getChannelName();
- /**
- * 发送订阅请求
- */
+ /** 发送订阅请求 */
void subscribe(WebSocketClient ws);
- /**
- * 发送取消订阅请求
- */
+ /** 发送取消订阅请求 */
void unsubscribe(WebSocketClient ws);
/**
- * 处理频道推送消息
+ * 处理频道推送消息。
*
- * @param response WebSocket 推送的 JSON 消息
- * @return true 表示已处理，false 表示不匹配（频道名不对）
+ * @param response WebSocket 推送的完整 JSON
+ * @return true 表示已处理（循环停止），false 表示频道不匹配（继续遍历下一个 handler）
*/
boolean handleMessage(JSONObject response);
}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/CandlestickChannelHandler.java b/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/CandlestickChannelHandler.java
index c445e49..cf44e10 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/CandlestickChannelHandler.java
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/CandlestickChannelHandler.java
@@ -11,8 +11,21 @@
import java.math.BigDecimal;
/**
- * K 线频道处理器。
- * 公开频道，无需认证。订阅 1m K 线，解析后回调 {@link GateGridTradeService#onKline}。
+ * K 线（Candlestick）频道处理器。
+ *
+ * <h3>特点</h3>
+ * 公开频道，无需 HMAC-SHA512 认证签名。订阅 1m 周期的 K 线数据。
+ *
+ * <h3>数据流</h3>
+ * <pre>
+ * WebSocket 推送 update event
+ * → handleMessage() → 解析 OHLCV → log 打印 → gridTradeService.onKline(closePx)
+ * → 首次 K 线触发双开
+ * → 后续 K 线仅缓存 lastKlinePrice 供补仓参考
+ * </pre>
+ *
+ * <h3>订阅格式</h3>
+ * payload: {@code ["1m", contract]}
*
* @author Administrator
*/
@@ -31,73 +44,55 @@
}
@Override
- public String getChannelName() {
- return CHANNEL_NAME;
- }
+ public String getChannelName() { return CHANNEL_NAME; }
@Override
public void subscribe(WebSocketClient ws) {
- JSONObject subscribeMsg = new JSONObject();
- subscribeMsg.put("time", System.currentTimeMillis() / 1000);
- subscribeMsg.put("channel", CHANNEL_NAME);
- subscribeMsg.put("event", "subscribe");
+ JSONObject msg = new JSONObject();
+ msg.put("time", System.currentTimeMillis() / 1000);
+ msg.put("channel", CHANNEL_NAME);
+ msg.put("event", "subscribe");
JSONArray payload = new JSONArray();
payload.add(INTERVAL);
payload.add(contract);
- subscribeMsg.put("payload", payload);
- ws.send(subscribeMsg.toJSONString());
- log.info("[{}] 已发送订阅请求，合约: {}, 周期: {}", CHANNEL_NAME, contract, INTERVAL);
+ msg.put("payload", payload);
+ ws.send(msg.toJSONString());
+ log.info("[{}] subscribed, contract:{}, interval:{}", CHANNEL_NAME, contract, INTERVAL);
}
@Override
public void unsubscribe(WebSocketClient ws) {
- JSONObject unsubscribeMsg = new JSONObject();
- unsubscribeMsg.put("time", System.currentTimeMillis() / 1000);
- unsubscribeMsg.put("channel", CHANNEL_NAME);
- unsubscribeMsg.put("event", "unsubscribe");
+ JSONObject msg = new JSONObject();
+ msg.put("time", System.currentTimeMillis() / 1000);
+ msg.put("channel", CHANNEL_NAME);
+ msg.put("event", "unsubscribe");
JSONArray payload = new JSONArray();
payload.add(INTERVAL);
payload.add(contract);
- unsubscribeMsg.put("payload", payload);
- ws.send(unsubscribeMsg.toJSONString());
- log.info("[{}] 已发送取消订阅请求，合约: {}, 周期: {}", CHANNEL_NAME, contract, INTERVAL);
+ msg.put("payload", payload);
+ ws.send(msg.toJSONString());
+ log.info("[{}] unsubscribed", CHANNEL_NAME);
}
@Override
public boolean handleMessage(JSONObject response) {
- String channel = response.getString("channel");
- if (!CHANNEL_NAME.equals(channel)) {
- return false;
- }
+ if (!CHANNEL_NAME.equals(response.getString("channel"))) return false;
try {
JSONArray resultArray = response.getJSONArray("result");
- if (resultArray == null || resultArray.isEmpty()) {
- log.warn("[{}] 数据为空", CHANNEL_NAME);
- return true;
- }
+ if (resultArray == null || resultArray.isEmpty()) { log.warn("[{}] empty", CHANNEL_NAME); return true; }
JSONObject data = resultArray.getJSONObject(0);
BigDecimal closePx = new BigDecimal(data.getString("c"));
- long t = data.getLong("t");
- boolean windowClosed = data.getBooleanValue("w");
- log.info("========== Gate K线数据 ==========");
- log.info("名称(n): {}", data.getString("n"));
- log.info("时间 : {}", DateUtil.TimeStampToDateTime(t));
- log.info("开盘(o): {}", data.getString("o"));
- log.info("最高(h): {}", data.getString("h"));
- log.info("最低(l): {}", data.getString("l"));
- log.info("收盘(c): {}", data.getString("c"));
- log.info("成交量(v): {}", data.getString("v"));
- log.info("成交额(a): {}", data.getString("a"));
- log.info("K线完结(w): {}", windowClosed);
- log.info("==================================");
+ log.info("========== Gate K线 ==========");
+ log.info("n:{} t:{} o:{} h:{} l:{} c:{} v:{} a:{} w:{}",
+ data.getString("n"), DateUtil.TimeStampToDateTime(data.getLong("t")),
+ data.getString("o"), data.getString("h"), data.getString("l"),
+ data.getString("c"), data.getString("v"), data.getString("a"),
+ data.getBooleanValue("w"));
+ log.info("==============================");
- if (gridTradeService != null) {
- gridTradeService.onKline(closePx);
- }
- } catch (Exception e) {
- log.error("[{}] 处理数据失败", CHANNEL_NAME, e);
- }
+ if (gridTradeService != null) gridTradeService.onKline(closePx);
+ } catch (Exception e) { log.error("[{}] handle fail", CHANNEL_NAME, e); }
return true;
}
}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/PositionClosesChannelHandler.java b/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/PositionClosesChannelHandler.java
index deefcdb..78f6c14 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/PositionClosesChannelHandler.java
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/PositionClosesChannelHandler.java
@@ -10,7 +10,16 @@
/**
* 平仓频道处理器。
- * 私有频道，需认证。解析平仓推送后回调 {@link GateGridTradeService#onPositionClose}。
+ *
+ * <h3>数据用途</h3>
+ * 每笔平仓发生时推送 pnl（盈亏金额），累加到 {@code cumulativePnl} 用于判断策略停止条件：
+ * cumulativePnl ≥ overallTp（达到止盈目标）或 ≤ -maxLoss（超过亏损上限）。
+ *
+ * <h3>推送字段</h3>
+ * contract, side（long / short）, pnl（该次平仓的盈亏，如 "+0.2" 或 "-0.1"）
+ *
+ * <h3>注意</h3>
+ * 平仓盈亏来自服务器端，不受本地计算误差影响。这是策略停止的唯一盈亏判断来源。
*
* @author Administrator
*/
@@ -27,31 +36,21 @@
@Override
public boolean handleMessage(JSONObject response) {
- String channel = response.getString("channel");
- if (!CHANNEL_NAME.equals(channel)) {
- return false;
- }
+ if (!CHANNEL_NAME.equals(response.getString("channel"))) return false;
try {
JSONArray resultArray = response.getJSONArray("result");
- if (resultArray == null || resultArray.isEmpty()) {
- return true;
- }
+ if (resultArray == null || resultArray.isEmpty()) return true;
for (int i = 0; i < resultArray.size(); i++) {
JSONObject item = resultArray.getJSONObject(i);
- if (!getContract().equals(item.getString("contract"))) {
- continue;
- }
+ if (!getContract().equals(item.getString("contract"))) continue;
BigDecimal pnl = new BigDecimal(item.getString("pnl"));
String side = item.getString("side");
- log.info("[{}] 推送: contract={}, side={}, pnl={}",
- CHANNEL_NAME, getContract(), side, pnl);
+ log.info("[{}] side:{}, pnl:{}", CHANNEL_NAME, side, pnl);
if (getGridTradeService() != null) {
getGridTradeService().onPositionClose(getContract(), side, pnl);
}
}
- } catch (Exception e) {
- log.error("[{}] 处理数据失败", CHANNEL_NAME, e);
- }
+ } catch (Exception e) { log.error("[{}] handle fail", CHANNEL_NAME, e); }
return true;
}
}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/PositionsChannelHandler.java b/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/PositionsChannelHandler.java
index 6584fb9..50cf707 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/PositionsChannelHandler.java
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/PositionsChannelHandler.java
@@ -10,7 +10,17 @@
/**
* 仓位频道处理器。
- * 私有频道，需认证。解析仓位推送后回调 {@link GateGridTradeService#onPositionUpdate}。
+ *
+ * <h3>数据用途</h3>
+ * 监控仓位数量（size）。当 size 从有变为 0 时，表示该方向被止盈条件单平仓，
+ * 触发补仓（reopenLongPosition / reopenShortPosition）。
+ *
+ * <h3>推送字段</h3>
+ * contract, mode（dual_long / dual_short）, size（正=持有，0=无仓位）, entry_price
+ *
+ * <h3>注意</h3>
+ * 双向持仓模式下空头 size 为负数，使用 {@code size.abs()} 判断是否有仓位。
+ * 累计盈亏不由本频道计算，而是由 {@link PositionClosesChannelHandler} 独立处理。
*
* @author Administrator
*/
@@ -27,32 +37,22 @@
@Override
public boolean handleMessage(JSONObject response) {
- String channel = response.getString("channel");
- if (!CHANNEL_NAME.equals(channel)) {
- return false;
- }
+ if (!CHANNEL_NAME.equals(response.getString("channel"))) return false;
try {
JSONArray resultArray = response.getJSONArray("result");
- if (resultArray == null || resultArray.isEmpty()) {
- return true;
- }
+ if (resultArray == null || resultArray.isEmpty()) return true;
for (int i = 0; i < resultArray.size(); i++) {
JSONObject pos = resultArray.getJSONObject(i);
- if (!getContract().equals(pos.getString("contract"))) {
- continue;
- }
+ if (!getContract().equals(pos.getString("contract"))) continue;
String mode = pos.getString("mode");
BigDecimal size = new BigDecimal(pos.getString("size"));
BigDecimal entryPrice = new BigDecimal(pos.getString("entry_price"));
- log.info("[{}] 推送: contract={}, mode={}, size={}, entry_price={}",
- CHANNEL_NAME, getContract(), mode, size, entryPrice);
+ log.info("[{}] mode:{}, size:{}, entry:{}", CHANNEL_NAME, mode, size, entryPrice);
if (getGridTradeService() != null) {
getGridTradeService().onPositionUpdate(getContract(), mode, size, entryPrice);
}
}
- } catch (Exception e) {
- log.error("[{}] 处理数据失败", CHANNEL_NAME, e);
- }
+ } catch (Exception e) { log.error("[{}] handle fail", CHANNEL_NAME, e); }
return true;
}
}
--
Gitblit v1.9.1