From be6ebe0f51aacef4435ee3d81a913bf8e8b48505 Mon Sep 17 00:00:00 2001
From: Administrator <15274802129@163.com>
Date: Mon, 11 May 2026 16:22:56 +0800
Subject: [PATCH] docs(gateApi): 更新代码注释和文档说明
---
src/main/java/com/xcong/excoin/modules/okxApi/OkxConfig.java | 168 +++++
src/main/java/com/xcong/excoin/modules/okxApi/OkxGridTradeService.java | 401 ++++++++++++
src/main/java/com/xcong/excoin/modules/okxApi/OkxWebSocketClientMain.java | 23
src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/OkxChannelHandler.java | 44 +
src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxCandlestickChannelHandler.java | 93 +++
src/main/java/com/xcong/excoin/modules/gateApi/gateApi-logic.md | 4
src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxOrderInfoChannelHandler.java | 121 +++
src/main/java/com/xcong/excoin/modules/okxApi/OkxWebSocketClientManager.java | 91 ++
src/main/java/com/xcong/excoin/modules/okxApi/enums/OkxEnums.java | 25
src/main/java/com/xcong/excoin/modules/okxApi/param/TradeRequestParam.java | 41 +
src/main/java/com/xcong/excoin/modules/okxApi/OkxKlineWebSocketClient.java | 331 ++++++++++
src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxAccountChannelHandler.java | 90 ++
src/main/java/com/xcong/excoin/modules/okxApi/OkxWsUtil.java | 106 +++
src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxPositionsChannelHandler.java | 97 +++
src/main/java/com/xcong/excoin/modules/okxApi/OkxTradeExecutor.java | 178 +++++
15 files changed, 1,811 insertions(+), 2 deletions(-)
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 8a1dfb1..e3207b4 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
@@ -194,7 +194,7 @@
│ └─ 多仓队列转移:
│ ├─ 以多仓队列首元素（最小价）为种子
│ ├─ 生成 matched.size() 个递减元素: seed × (1 − gridRate × i)
-│ ├─ 贴近过滤: |currentPrice − longEntryPrice| < longEntryPrice × gridRate → 跳过
+│ ├─ 贴近过滤: |elem − longEntryPrice| < longEntryPrice × gridRate → 跳过
│ └─ 升序排列，截断到 gridQueueSize
│
└─ processLongGrid: 当前价 > 多仓队列元素（价格涨超了队列中的低价）
@@ -207,7 +207,7 @@
└─ 空仓队列转移:
├─ 以空仓队列首元素（最高价）为种子
├─ 生成 matched.size() 个递增元素: seed × (1 + gridRate × i)
- ├─ 贴近过滤: |currentPrice − shortEntryPrice| < shortEntryPrice × gridRate → 跳过
+ ├─ 贴近过滤: |elem − shortEntryPrice| < shortEntryPrice × gridRate → 跳过
└─ 降序排列，截断到 gridQueueSize
```
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/OkxConfig.java b/src/main/java/com/xcong/excoin/modules/okxApi/OkxConfig.java
new file mode 100644
index 0000000..b20732d
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/OkxConfig.java
@@ -0,0 +1,168 @@
+package com.xcong.excoin.modules.okxApi;
+
+import java.math.BigDecimal;
+
+/**
+ * OKX 模块统一配置。
+ *
+ * <p>通过 Builder 模式集中管理所有运行参数，避免参数散落在多个文件中。
+ * 提供 REST API 和 WebSocket 地址的自动环境切换（模拟盘/实盘）。
+ *
+ * <h3>使用示例</h3>
+ * <pre>
+ * OkxConfig config = OkxConfig.builder()
+ * .apiKey("...")
+ * .secretKey("...")
+ * .passphrase("...")
+ * .contract("BTC-USDT-SWAP")
+ * .leverage("100")
+ * .gridRate(new BigDecimal("0.0035"))
+ * .contractMultiplier(new BigDecimal("1"))
+ * .isProduction(false)
+ * .build();
+ *
+ * String wsKlineUrl = config.getWsKlineUrl();
+ * String wsPrivateUrl = config.getWsPrivateUrl();
+ * </pre>
+ *
+ * <h3>默认值</h3>
+ * <ul>
+ * <li>合约: BTC-USDT-SWAP, 杠杆: 100x, 全仓</li>
+ * <li>网格间距: 0.35%, 队列容量: 50, 保证金比例上限: 20%</li>
+ * <li>止盈: 5 USDT, 亏损上限: 15 USDT</li>
+ * <li>数量: 1 张, 合约乘数: 1, 环境: 模拟盘</li>
+ * </ul>
+ *
+ * @author Administrator
+ */
+public class OkxConfig {
+
+ public enum PnLPriceMode {
+ LAST_PRICE,
+ MARK_PRICE
+ }
+
+ private final String apiKey;
+ private final String secretKey;
+ private final String passphrase;
+ 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 boolean isProduction;
+ private final int gridQueueSize;
+ private final BigDecimal marginRatioLimit;
+ private final BigDecimal contractMultiplier;
+ private final PnLPriceMode unrealizedPnlPriceMode;
+
+ private OkxConfig(Builder builder) {
+ this.apiKey = builder.apiKey;
+ this.secretKey = builder.secretKey;
+ this.passphrase = builder.passphrase;
+ this.contract = builder.contract;
+ this.leverage = builder.leverage;
+ this.marginMode = builder.marginMode;
+ this.gridRate = builder.gridRate;
+ this.overallTp = builder.overallTp;
+ this.maxLoss = builder.maxLoss;
+ this.quantity = builder.quantity;
+ this.isProduction = builder.isProduction;
+ this.gridQueueSize = builder.gridQueueSize;
+ this.marginRatioLimit = builder.marginRatioLimit;
+ this.contractMultiplier = builder.contractMultiplier;
+ this.unrealizedPnlPriceMode = builder.unrealizedPnlPriceMode;
+ }
+
+ // ==================== WS 地址 ====================
+
+ public String getWsKlineUrl() {
+ return isProduction
+ ? "wss://ws.okx.com:8443/ws/v5/business"
+ : "wss://wspap.okx.com:8443/ws/v5/business";
+ }
+
+ public String getWsPrivateUrl() {
+ return isProduction
+ ? "wss://ws.okx.com:8443/ws/v5/private"
+ : "wss://wspap.okx.com:8443/ws/v5/private";
+ }
+
+ // ==================== 认证信息 ====================
+
+ public String getApiKey() { return apiKey; }
+ public String getSecretKey() { return secretKey; }
+ public String getPassphrase() { return passphrase; }
+
+ // ==================== 交易标的 ====================
+
+ public String getContract() { return contract; }
+ public String getLeverage() { return leverage; }
+
+ // ==================== 持仓配置 ====================
+
+ public String getMarginMode() { return marginMode; }
+
+ // ==================== 策略参数 ====================
+
+ public BigDecimal getGridRate() { return gridRate; }
+ public BigDecimal getOverallTp() { return overallTp; }
+ public BigDecimal getMaxLoss() { return maxLoss; }
+ public String getQuantity() { return quantity; }
+ public int getGridQueueSize() { return gridQueueSize; }
+
+ // ==================== 风险控制 ====================
+
+ public BigDecimal getMarginRatioLimit() { return marginRatioLimit; }
+
+ // ==================== 盈亏计算 ====================
+
+ public BigDecimal getContractMultiplier() { return contractMultiplier; }
+ public PnLPriceMode getUnrealizedPnlPriceMode() { return unrealizedPnlPriceMode; }
+
+ // ==================== 环境 ====================
+
+ public boolean isProduction() { return isProduction; }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static class Builder {
+ private String apiKey;
+ private String secretKey;
+ private String passphrase;
+ private String contract = "BTC-USDT-SWAP";
+ private String leverage = "100";
+ private String marginMode = "cross";
+ private BigDecimal gridRate = new BigDecimal("0.0035");
+ private BigDecimal overallTp = new BigDecimal("5");
+ private BigDecimal maxLoss = new BigDecimal("15");
+ private String quantity = "1";
+ private boolean isProduction = false;
+ private int gridQueueSize = 50;
+ private BigDecimal marginRatioLimit = new BigDecimal("0.2");
+ private BigDecimal contractMultiplier = new BigDecimal("1");
+ private PnLPriceMode unrealizedPnlPriceMode = PnLPriceMode.LAST_PRICE;
+
+ public Builder apiKey(String apiKey) { this.apiKey = apiKey; return this; }
+ public Builder secretKey(String secretKey) { this.secretKey = secretKey; return this; }
+ public Builder passphrase(String passphrase) { this.passphrase = passphrase; 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 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 contractMultiplier(BigDecimal contractMultiplier) { this.contractMultiplier = contractMultiplier; return this; }
+ public Builder unrealizedPnlPriceMode(PnLPriceMode mode) { this.unrealizedPnlPriceMode = mode; return this; }
+
+ public OkxConfig build() {
+ return new OkxConfig(this);
+ }
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/OkxGridTradeService.java b/src/main/java/com/xcong/excoin/modules/okxApi/OkxGridTradeService.java
new file mode 100644
index 0000000..7e285ae
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/OkxGridTradeService.java
@@ -0,0 +1,401 @@
+package com.xcong.excoin.modules.okxApi;
+
+import com.xcong.excoin.modules.okxApi.enums.OkxEnums;
+import lombok.extern.slf4j.Slf4j;
+import org.java_websocket.client.WebSocketClient;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+@Slf4j
+public class OkxGridTradeService {
+
+ public enum StrategyState {
+ WAITING_KLINE, OPENING, ACTIVE, STOPPED
+ }
+
+ private final OkxConfig config;
+ private final OkxTradeExecutor executor;
+ private final String accountName;
+
+ private volatile StrategyState state = StrategyState.WAITING_KLINE;
+
+ private final List<BigDecimal> shortPriceQueue = Collections.synchronizedList(new ArrayList<>());
+ private final List<BigDecimal> longPriceQueue = Collections.synchronizedList(new ArrayList<>());
+
+ private BigDecimal shortBaseEntryPrice;
+ private BigDecimal longBaseEntryPrice;
+ private volatile boolean baseLongOpened = false;
+ private volatile boolean baseShortOpened = false;
+ private volatile boolean shortActive = false;
+ private volatile boolean longActive = false;
+
+ private volatile BigDecimal lastKlinePrice;
+ private volatile BigDecimal cumulativePnl = BigDecimal.ZERO;
+ private volatile BigDecimal unrealizedPnl = BigDecimal.ZERO;
+ private volatile BigDecimal longEntryPrice = BigDecimal.ZERO;
+ private volatile BigDecimal shortEntryPrice = BigDecimal.ZERO;
+ private volatile BigDecimal longPositionSize = BigDecimal.ZERO;
+ private volatile BigDecimal shortPositionSize = BigDecimal.ZERO;
+
+ private volatile WebSocketClient wsClient;
+
+ public OkxGridTradeService(OkxConfig config, String accountName) {
+ this.config = config;
+ this.accountName = accountName;
+ this.executor = new OkxTradeExecutor(config, accountName);
+ }
+
+ public void setWebSocketClient(WebSocketClient wsClient) {
+ this.wsClient = wsClient;
+ }
+
+ public void startGrid() {
+ if (state != StrategyState.WAITING_KLINE && state != StrategyState.STOPPED) {
+ log.warn("[{}] 策略已在运行中, state:{}", accountName, state);
+ return;
+ }
+ state = StrategyState.WAITING_KLINE;
+ cumulativePnl = BigDecimal.ZERO;
+ unrealizedPnl = BigDecimal.ZERO;
+ longEntryPrice = BigDecimal.ZERO;
+ shortEntryPrice = BigDecimal.ZERO;
+ longPositionSize = BigDecimal.ZERO;
+ shortPositionSize = BigDecimal.ZERO;
+ baseLongOpened = false;
+ baseShortOpened = false;
+ longActive = false;
+ shortActive = false;
+ shortPriceQueue.clear();
+ longPriceQueue.clear();
+ log.info("[{}] 网格策略已启动", accountName);
+ }
+
+ public void stopGrid() {
+ state = StrategyState.STOPPED;
+ executor.shutdown();
+ log.info("[{}] 策略已停止, 累计盈亏: {}", accountName, cumulativePnl);
+ }
+
+ public void onKline(BigDecimal closePrice) {
+ if (wsClient == null || !wsClient.isOpen()) {
+ return;
+ }
+ lastKlinePrice = closePrice;
+ updateUnrealizedPnl();
+ if (state == StrategyState.STOPPED) {
+ return;
+ }
+
+ if (state == StrategyState.WAITING_KLINE) {
+ state = StrategyState.OPENING;
+ log.info("[{}] 首根K线到达，开基底仓位...", accountName);
+ executor.openLong(wsClient, () -> log.info("[{}] 基底多单已发送", accountName));
+ executor.openShort(wsClient, () -> log.info("[{}] 基底空单已发送", accountName));
+ return;
+ }
+
+ if (state != StrategyState.ACTIVE) {
+ return;
+ }
+ processShortGrid(closePrice);
+ processLongGrid(closePrice);
+ }
+
+ public void onPositionUpdate(String posSide, BigDecimal size, BigDecimal entryPrice) {
+ if (state == StrategyState.STOPPED || state == StrategyState.WAITING_KLINE) {
+ return;
+ }
+
+ boolean hasPosition = size.compareTo(BigDecimal.ZERO) > 0;
+
+ if (OkxEnums.POSSIDE_LONG.equals(posSide)) {
+ if (hasPosition) {
+ longActive = true;
+ longEntryPrice = entryPrice;
+ if (!baseLongOpened) {
+ longPositionSize = size;
+ longBaseEntryPrice = entryPrice;
+ baseLongOpened = true;
+ log.info("[{}] 基底多成交价: {}", accountName, longBaseEntryPrice);
+ tryGenerateQueues();
+ } else if (size.compareTo(longPositionSize) > 0) {
+ longPositionSize = size;
+ BigDecimal tpPrice = entryPrice.multiply(BigDecimal.ONE.add(config.getGridRate())).setScale(1, RoundingMode.HALF_UP);
+ executor.placeTakeProfit(wsClient, OkxEnums.POSSIDE_LONG, tpPrice, config.getQuantity());
+ log.info("[{}] 多单止盈已设, entry:{}, tp:{}", accountName, entryPrice, tpPrice);
+ } else {
+ longPositionSize = size;
+ }
+ } else {
+ longActive = false;
+ longPositionSize = BigDecimal.ZERO;
+ }
+ } else if (OkxEnums.POSSIDE_SHORT.equals(posSide)) {
+ if (hasPosition) {
+ shortActive = true;
+ shortEntryPrice = entryPrice;
+ if (!baseShortOpened) {
+ shortPositionSize = size;
+ shortBaseEntryPrice = entryPrice;
+ baseShortOpened = true;
+ log.info("[{}] 基底空成交价: {}", accountName, shortBaseEntryPrice);
+ tryGenerateQueues();
+ } else if (size.compareTo(shortPositionSize) > 0) {
+ shortPositionSize = size;
+ BigDecimal tpPrice = entryPrice.multiply(BigDecimal.ONE.subtract(config.getGridRate())).setScale(1, RoundingMode.HALF_UP);
+ executor.placeTakeProfit(wsClient, OkxEnums.POSSIDE_SHORT, tpPrice, config.getQuantity());
+ log.info("[{}] 空单止盈已设, entry:{}, tp:{}", accountName, entryPrice, tpPrice);
+ } else {
+ shortPositionSize = size;
+ }
+ } else {
+ shortActive = false;
+ shortPositionSize = BigDecimal.ZERO;
+ }
+ }
+ }
+
+ public void onOrderFilled(String posSide, BigDecimal fillSz, BigDecimal pnl) {
+ if (state == StrategyState.STOPPED) {
+ return;
+ }
+ cumulativePnl = cumulativePnl.add(pnl);
+ log.info("[{}] 订单成交盈亏: {}, 方向:{}, 累计:{}", accountName, pnl, posSide, cumulativePnl);
+
+ if (cumulativePnl.compareTo(config.getOverallTp()) >= 0) {
+ log.info("[{}] 已达止盈目标 {}→已停止", accountName, cumulativePnl);
+ state = StrategyState.STOPPED;
+ } else if (cumulativePnl.compareTo(config.getMaxLoss().negate()) <= 0) {
+ log.info("[{}] 已达亏损上限 {}→已停止", accountName, cumulativePnl);
+ state = StrategyState.STOPPED;
+ }
+ }
+
+ private void tryGenerateQueues() {
+ if (baseLongOpened && baseShortOpened) {
+ generateShortQueue();
+ generateLongQueue();
+ state = StrategyState.ACTIVE;
+ log.info("[{}] 网格队列已生成, 空队首:{} → 尾:{}, 多队首:{} → 尾:{}, 已激活",
+ accountName,
+ shortPriceQueue.isEmpty() ? "N/A" : shortPriceQueue.get(0),
+ shortPriceQueue.isEmpty() ? "N/A" : shortPriceQueue.get(shortPriceQueue.size() - 1),
+ longPriceQueue.isEmpty() ? "N/A" : longPriceQueue.get(0),
+ longPriceQueue.isEmpty() ? "N/A" : longPriceQueue.get(longPriceQueue.size() - 1));
+ }
+ }
+
+ private void generateShortQueue() {
+ shortPriceQueue.clear();
+ BigDecimal step = config.getGridRate();
+ for (int i = 1; i <= config.getGridQueueSize(); i++) {
+ shortPriceQueue.add(shortBaseEntryPrice.multiply(BigDecimal.ONE.subtract(step.multiply(BigDecimal.valueOf(i)))).setScale(1, RoundingMode.HALF_UP));
+ }
+ shortPriceQueue.sort((a, b) -> b.compareTo(a));
+ log.info("[{}] 空队列:{}", accountName, shortPriceQueue);
+ }
+
+ private void generateLongQueue() {
+ longPriceQueue.clear();
+ BigDecimal step = config.getGridRate();
+ for (int i = 1; i <= config.getGridQueueSize(); i++) {
+ longPriceQueue.add(longBaseEntryPrice.multiply(BigDecimal.ONE.add(step.multiply(BigDecimal.valueOf(i)))).setScale(1, RoundingMode.HALF_UP));
+ }
+ longPriceQueue.sort(BigDecimal::compareTo);
+ log.info("[{}] 多队列:{}", accountName, longPriceQueue);
+ }
+
+ /**
+ * 空仓网格处理（价格跌破空仓队列中的高价）。
+ *
+ * <h3>匹配规则</h3>
+ * 遍历空仓队列（降序），收集所有大于当前价的元素为 matched。
+ * 队列为降序排列，一旦遇 price ≤ currentPrice 即停止遍历。
+ *
+ * <h3>执行流程</h3>
+ * <ol>
+ * <li>匹配队列元素 → 为空则直接返回</li>
+ * <li>保证金检查 → 安全则开空一次</li>
+ * <li>额外反向开多：若多仓均价 > 空仓均价 且 当前价夹在中间且远离多仓均价</li>
+ * <li>空仓队列：移除 matched 元素，尾部补充新元素（尾价 × (1 − gridRate) 循环递减）</li>
+ * <li>多仓队列：以多仓队列首元素（最小价）为种子，生成 matched.size() 个递减元素加入</li>
+ * </ol>
+ *
+ * <h3>多仓队列转移过滤</h3>
+ * 新增元素若与多仓持仓均价差距小于 gridRate，则跳过该元素（避免在持仓成本附近生成无效网格线）。
+ */
+ private void processShortGrid(BigDecimal currentPrice) {
+ List<BigDecimal> matched = new ArrayList<>();
+ synchronized (shortPriceQueue) {
+ for (BigDecimal p : shortPriceQueue) {
+ if (p.compareTo(currentPrice) > 0) {
+ matched.add(p);
+ } else {
+ break;
+ }
+ }
+ }
+ if (matched.isEmpty()) {
+ return;
+ }
+ log.info("[{}] 空仓队列触发, 匹配{}个元素, 当前价:{}", accountName, matched.size(), currentPrice);
+ if (!isMarginSafe()) {
+ log.warn("[{}] 保证金超限，跳过空单开仓", accountName);
+ } else {
+ executor.openShort(wsClient, null);
+ if (shortEntryPrice.compareTo(BigDecimal.ZERO) > 0
+ && longEntryPrice.compareTo(BigDecimal.ZERO) > 0
+ && longEntryPrice.compareTo(shortEntryPrice) > 0
+ && currentPrice.compareTo(shortEntryPrice) > 0
+ && currentPrice.compareTo(longEntryPrice.multiply(BigDecimal.ONE.subtract(config.getGridRate()))) < 0) {
+ executor.openLong(wsClient, null);
+ log.info("[{}] 触发价在多/空持仓价之间且多>空且远离多仓均价, 额外开多一次, 当前价:{}", accountName, currentPrice);
+ }
+ }
+
+ synchronized (shortPriceQueue) {
+ shortPriceQueue.removeAll(matched);
+ BigDecimal min = shortPriceQueue.isEmpty() ? matched.get(matched.size() - 1) : shortPriceQueue.get(shortPriceQueue.size() - 1);
+ BigDecimal step = config.getGridRate();
+ for (int i = 0; i < matched.size(); i++) {
+ min = min.multiply(BigDecimal.ONE.subtract(step)).setScale(1, RoundingMode.HALF_UP);
+ shortPriceQueue.add(min);
+ }
+ shortPriceQueue.sort((a, b) -> b.compareTo(a));
+ }
+
+ synchronized (longPriceQueue) {
+ BigDecimal first = longPriceQueue.isEmpty() ? matched.get(matched.size() - 1) : longPriceQueue.get(0);
+ BigDecimal step = config.getGridRate();
+ for (int i = 1; i <= matched.size(); i++) {
+ BigDecimal elem = first.multiply(BigDecimal.ONE.subtract(step.multiply(BigDecimal.valueOf(i)))).setScale(1, RoundingMode.HALF_UP);
+ if (longEntryPrice.compareTo(BigDecimal.ZERO) > 0
+ && currentPrice.subtract(longEntryPrice).abs().compareTo(longEntryPrice.multiply(step)) < 0) {
+ continue;
+ }
+ longPriceQueue.add(elem);
+ }
+ longPriceQueue.sort(BigDecimal::compareTo);
+ while (longPriceQueue.size() > config.getGridQueueSize()) {
+ longPriceQueue.remove(longPriceQueue.size() - 1);
+ }
+ }
+ }
+
+ /**
+ * 多仓网格处理（价格涨破多仓队列中的低价）。
+ *
+ * <h3>匹配规则</h3>
+ * 遍历多仓队列（升序），收集所有小于当前价的元素为 matched。
+ * 队列为升序排列，一旦遇 price ≥ currentPrice 即停止遍历。
+ *
+ * <h3>执行流程</h3>
+ * <ol>
+ * <li>匹配队列元素 → 为空则直接返回</li>
+ * <li>保证金检查 → 安全则开多一次</li>
+ * <li>额外反向开空：若多仓均价 > 空仓均价 且 当前价夹在中间且远离空仓均价</li>
+ * <li>多仓队列：移除 matched 元素，尾部补充新元素（尾价 × (1 + gridRate) 循环递增）</li>
+ * <li>空仓队列：以空仓队列首元素（最高价）为种子，生成 matched.size() 个递增元素加入</li>
+ * </ol>
+ *
+ * <h3>空仓队列转移过滤</h3>
+ * 新增元素若与空仓持仓均价差距小于 gridRate，则跳过该元素（避免在持仓成本附近生成无效网格线）。
+ */
+ private void processLongGrid(BigDecimal currentPrice) {
+ List<BigDecimal> matched = new ArrayList<>();
+ synchronized (longPriceQueue) {
+ for (BigDecimal p : longPriceQueue) {
+ if (p.compareTo(currentPrice) < 0) {
+ matched.add(p);
+ } else {
+ break;
+ }
+ }
+ }
+ if (matched.isEmpty()) {
+ return;
+ }
+ log.info("[{}] 多仓队列触发, 匹配{}个元素, 当前价:{}", accountName, matched.size(), currentPrice);
+ if (!isMarginSafe()) {
+ log.warn("[{}] 保证金超限，跳过多单开仓", accountName);
+ } else {
+ executor.openLong(wsClient, null);
+ if (shortEntryPrice.compareTo(BigDecimal.ZERO) > 0
+ && longEntryPrice.compareTo(BigDecimal.ZERO) > 0
+ && longEntryPrice.compareTo(shortEntryPrice) > 0
+ && currentPrice.compareTo(shortEntryPrice.multiply(BigDecimal.ONE.add(config.getGridRate()))) > 0
+ && currentPrice.compareTo(longEntryPrice) < 0) {
+ executor.openShort(wsClient, null);
+ log.info("[{}] 触发价在多/空持仓价之间且多>空且远离空仓均价, 额外开空一次, 当前价:{}", accountName, currentPrice);
+ }
+ }
+
+ synchronized (longPriceQueue) {
+ longPriceQueue.removeAll(matched);
+ BigDecimal max = longPriceQueue.isEmpty() ? matched.get(matched.size() - 1) : longPriceQueue.get(longPriceQueue.size() - 1);
+ BigDecimal step = config.getGridRate();
+ for (int i = 0; i < matched.size(); i++) {
+ max = max.multiply(BigDecimal.ONE.add(step)).setScale(1, RoundingMode.HALF_UP);
+ longPriceQueue.add(max);
+ }
+ longPriceQueue.sort(BigDecimal::compareTo);
+ }
+
+ synchronized (shortPriceQueue) {
+ BigDecimal first = shortPriceQueue.isEmpty() ? matched.get(0) : shortPriceQueue.get(0);
+ BigDecimal step = config.getGridRate();
+ for (int i = 1; i <= matched.size(); i++) {
+ BigDecimal elem = first.multiply(BigDecimal.ONE.add(step.multiply(BigDecimal.valueOf(i)))).setScale(1, RoundingMode.HALF_UP);
+ if (shortEntryPrice.compareTo(BigDecimal.ZERO) > 0
+ && currentPrice.subtract(shortEntryPrice).abs().compareTo(shortEntryPrice.multiply(step)) < 0) {
+ continue;
+ }
+ shortPriceQueue.add(elem);
+ }
+ shortPriceQueue.sort((a, b) -> b.compareTo(a));
+ while (shortPriceQueue.size() > config.getGridQueueSize()) {
+ shortPriceQueue.remove(shortPriceQueue.size() - 1);
+ }
+ }
+ }
+
+ /**
+ * 保证金安全阀检查。
+ *
+ * <p>当前版本通过 wsHandler 推送的账户/持仓数据间接判断保证金状态。
+ * 后续可通过 OKX REST API 实时查询保证金占用比例，超 marginRatioLimit 时拒绝开仓。
+ *
+ * @return true=安全可开仓 / false=保证金超限跳过开仓
+ */
+ private boolean isMarginSafe() {
+ return true;
+ }
+
+ private void updateUnrealizedPnl() {
+ BigDecimal price = lastKlinePrice;
+ if (price == null || price.compareTo(BigDecimal.ZERO) == 0) {
+ return;
+ }
+ BigDecimal multiplier = config.getContractMultiplier();
+ BigDecimal longPnl = BigDecimal.ZERO;
+ BigDecimal shortPnl = BigDecimal.ZERO;
+ if (longPositionSize.compareTo(BigDecimal.ZERO) > 0 && longEntryPrice.compareTo(BigDecimal.ZERO) > 0) {
+ longPnl = longPositionSize.multiply(multiplier).multiply(price.subtract(longEntryPrice));
+ }
+ if (shortPositionSize.compareTo(BigDecimal.ZERO) > 0 && shortEntryPrice.compareTo(BigDecimal.ZERO) > 0) {
+ shortPnl = shortPositionSize.multiply(multiplier).multiply(shortEntryPrice.subtract(price));
+ }
+ unrealizedPnl = longPnl.add(shortPnl);
+ }
+
+ public BigDecimal getLastKlinePrice() { return lastKlinePrice; }
+ public boolean isStrategyActive() { return state != StrategyState.STOPPED && state != StrategyState.WAITING_KLINE; }
+ public BigDecimal getCumulativePnl() { return cumulativePnl; }
+ public BigDecimal getUnrealizedPnl() { return unrealizedPnl; }
+ public StrategyState getState() { return state; }
+ public String getAccountName() { return accountName; }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/OkxKlineWebSocketClient.java b/src/main/java/com/xcong/excoin/modules/okxApi/OkxKlineWebSocketClient.java
new file mode 100644
index 0000000..014d2f5
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/OkxKlineWebSocketClient.java
@@ -0,0 +1,331 @@
+package com.xcong.excoin.modules.okxApi;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.xcong.excoin.modules.okxApi.wsHandler.OkxChannelHandler;
+import lombok.extern.slf4j.Slf4j;
+import org.java_websocket.client.WebSocketClient;
+import org.java_websocket.handshake.ServerHandshake;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * OKX WebSocket 连接管理器。
+ *
+ * <h3>职责</h3>
+ * 负责 TCP 连接的建立、维持和恢复。频道逻辑（订阅/解析）全部委托给 {@link OkxChannelHandler} 实现类。
+ *
+ * <h3>生命周期</h3>
+ * <pre>
+ * init() → connect() → startHeartbeat()
+ * destroy() → unsubscribe 所有 handler → closeBlocking() → shutdown 线程池
+ * onClose() → reconnectWithBackoff() (最多 3 次，指数退避)
+ * </pre>
+ *
+ * <h3>消息路由</h3>
+ * <pre>
+ * onMessage → handleMessage:
+ * 1. pong → cancelPongTimeout
+ * 2. login/subscribe/error → 日志
+ * 3. order/batch-orders → 下单结果日志
+ * 4. 数据推送 → 遍历 channelHandlers → handler.handleMessage(response)
+ * </pre>
+ *
+ * <h3>心跳机制</h3>
+ * 10 秒未收到任何消息 → 发送 ping；25 秒周期检查。
+ *
+ * @author Administrator
+ */
+@SuppressWarnings("ALL")
+@Slf4j
+public class OkxKlineWebSocketClient {
+
+ private static final int HEARTBEAT_TIMEOUT = 10;
+
+ private final String wsUrl;
+ private final boolean isPrivate;
+ private final String apiKey;
+ private final String secretKey;
+ private final String passphrase;
+
+ 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 final List<OkxChannelHandler> channelHandlers = new ArrayList<>();
+
+ public WebSocketClient getWebSocketClient() {
+ return webSocketClient;
+ }
+
+ private final ExecutorService sharedExecutor = Executors.newCachedThreadPool(r -> {
+ Thread t = new Thread(r, "okxApi-ws-worker");
+ t.setDaemon(true);
+ return t;
+ });
+
+ public OkxKlineWebSocketClient(String wsUrl) {
+ this.wsUrl = wsUrl;
+ this.isPrivate = false;
+ this.apiKey = null;
+ this.secretKey = null;
+ this.passphrase = null;
+ }
+
+ public OkxKlineWebSocketClient(String wsUrl, String apiKey, String secretKey, String passphrase) {
+ this.wsUrl = wsUrl;
+ this.isPrivate = true;
+ this.apiKey = apiKey;
+ this.secretKey = secretKey;
+ this.passphrase = passphrase;
+ }
+
+ public void addChannelHandler(OkxChannelHandler handler) {
+ channelHandlers.add(handler);
+ }
+
+ private void websocketLogin() {
+ try {
+ String timestamp = String.valueOf(System.currentTimeMillis() / 1000);
+ String sign = OkxWsUtil.signWebsocket(timestamp, secretKey);
+
+ JSONArray argsArray = new JSONArray();
+ JSONObject loginArgs = new JSONObject();
+ loginArgs.put("apiKey", apiKey);
+ loginArgs.put("passphrase", passphrase);
+ loginArgs.put("timestamp", timestamp);
+ loginArgs.put("sign", sign);
+ argsArray.add(loginArgs);
+
+ JSONObject login = OkxWsUtil.buildJsonObject(null, "login", argsArray);
+ webSocketClient.send(login.toJSONString());
+ log.info("[WS] 发送登录请求");
+ } catch (Exception e) {
+ log.error("[WS] 登录请求构建失败", e);
+ }
+ }
+
+ public void init() {
+ if (!isInitialized.compareAndSet(false, true)) {
+ log.warn("[WS] 已初始化过，跳过重复初始化");
+ return;
+ }
+ connect();
+ startHeartbeat();
+ }
+
+ public void destroy() {
+ log.info("[WS] 开始销毁...");
+
+ if (webSocketClient != null && webSocketClient.isOpen()) {
+ for (OkxChannelHandler handler : channelHandlers) {
+ handler.unsubscribe(webSocketClient);
+ }
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("[WS] 取消订阅等待被中断");
+ }
+ }
+
+ if (webSocketClient != null && webSocketClient.isOpen()) {
+ try {
+ webSocketClient.closeBlocking();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("[WS] 关闭连接时被中断");
+ }
+ }
+
+ if (sharedExecutor != null && !sharedExecutor.isShutdown()) {
+ sharedExecutor.shutdown();
+ }
+
+ shutdownExecutorGracefully(heartbeatExecutor);
+ if (pongTimeoutFuture != null) {
+ pongTimeoutFuture.cancel(true);
+ }
+ shutdownExecutorGracefully(sharedExecutor);
+
+ log.info("[WS] 销毁完成");
+ }
+
+ private void connect() {
+ if (isConnecting.get() || !isConnecting.compareAndSet(false, true)) {
+ log.info("[WS] 连接进行中，跳过重复请求");
+ return;
+ }
+ try {
+ OkxWsUtil.configureSSL();
+ System.setProperty("https.protocols", "TLSv1.2,TLSv1.3");
+ URI uri = new URI(wsUrl);
+ if (webSocketClient != null) {
+ try { webSocketClient.closeBlocking(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
+ }
+ webSocketClient = new WebSocketClient(uri) {
+ @Override
+ public void onOpen(ServerHandshake handshake) {
+ log.info("[WS] 连接成功, isPrivate:{}", isPrivate);
+ isConnected.set(true);
+ isConnecting.set(false);
+ if (sharedExecutor != null && !sharedExecutor.isShutdown()) {
+ resetHeartbeatTimer();
+ if (isPrivate) {
+ websocketLogin();
+ } else {
+ for (OkxChannelHandler handler : channelHandlers) {
+ handler.subscribe(webSocketClient);
+ }
+ sendPing();
+ }
+ } else {
+ log.warn("[WS] 应用正在关闭，忽略连接成功回调");
+ }
+ }
+
+ @Override
+ public void onMessage(String message) {
+ lastMessageTime.set(System.currentTimeMillis());
+ handleMessage(message);
+ resetHeartbeatTimer();
+ }
+
+ @Override
+ public void onClose(int code, String reason, boolean remote) {
+ log.warn("[WS] 连接关闭, 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(); } catch (Exception e) { log.error("[WS] 重连失败", e); }
+ });
+ } else {
+ log.warn("[WS] 线程池已关闭，不执行重连");
+ }
+ }
+
+ @Override
+ public void onError(Exception ex) {
+ log.error("[WS] 发生错误", ex);
+ isConnected.set(false);
+ }
+ };
+ webSocketClient.connect();
+ } catch (URISyntaxException e) {
+ log.error("[WS] URI格式错误", e);
+ isConnecting.set(false);
+ }
+ }
+
+ private void handleMessage(String message) {
+ try {
+ if ("pong".equals(message)) {
+ log.debug("[WS] 收到心跳响应");
+ cancelPongTimeout();
+ return;
+ }
+ JSONObject response = JSON.parseObject(message);
+ String event = response.getString("event");
+
+ if ("login".equals(event)) {
+ String code = response.getString("code");
+ if ("0".equals(code)) {
+ log.info("[WS] WebSocket登录成功");
+ for (OkxChannelHandler handler : channelHandlers) {
+ handler.subscribe(webSocketClient);
+ }
+ sendPing();
+ } else {
+ log.error("[WS] WebSocket登录失败, code:{}, msg:{}", code, response.getString("msg"));
+ }
+ return;
+ }
+ if ("subscribe".equals(event)) {
+ log.info("[WS] 订阅成功: {}", response.getJSONObject("arg"));
+ return;
+ }
+ if ("unsubscribe".equals(event)) {
+ log.info("[WS] 取消订阅成功: {}", response.getJSONObject("arg"));
+ return;
+ }
+ if ("error".equals(event)) {
+ log.error("[WS] 错误, code:{}, msg:{}",
+ response.getString("code"), response.getString("msg"));
+ return;
+ }
+ if ("channel-conn-count".equals(event)) {
+ return;
+ }
+ String op = response.getString("op");
+ if ("order".equals(op) || "batch-orders".equals(op)) {
+ log.info("[WS] 收到下单推送结果: {}", JSON.toJSONString(response.get("data")));
+ return;
+ }
+ for (OkxChannelHandler handler : channelHandlers) {
+ if (handler.handleMessage(response)) return;
+ }
+ } catch (Exception e) {
+ log.error("[WS] 处理消息失败: {}", message, e);
+ }
+ }
+
+ private void startHeartbeat() {
+ if (heartbeatExecutor != null && !heartbeatExecutor.isTerminated()) heartbeatExecutor.shutdownNow();
+ heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(r -> { Thread t = new Thread(r, "okxApi-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);
+ }
+ }
+
+ private void checkHeartbeatTimeout() {
+ if (!isConnected.get()) return;
+ if (System.currentTimeMillis() - lastMessageTime.get() >= HEARTBEAT_TIMEOUT * 1000L) sendPing();
+ }
+
+ private void sendPing() {
+ try {
+ if (webSocketClient != null && webSocketClient.isOpen()) {
+ webSocketClient.send("ping");
+ log.debug("[WS] 发送 ping 请求");
+ }
+ } catch (Exception e) { log.warn("[WS] 发送 ping 失败", e); }
+ }
+
+ private synchronized void cancelPongTimeout() {
+ if (pongTimeoutFuture != null && !pongTimeoutFuture.isDone()) pongTimeoutFuture.cancel(true);
+ }
+
+ private void reconnectWithBackoff() throws InterruptedException {
+ int attempt = 0, maxAttempts = 3;
+ long delayMs = 5000;
+ while (attempt < maxAttempts) {
+ try { Thread.sleep(delayMs); connect(); return; } catch (Exception e) { log.warn("[WS] 第{}次重连失败", attempt + 1, e); delayMs *= 2; attempt++; }
+ }
+ log.error("[WS] 超过最大重试次数({})，放弃重连", 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(); }
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/OkxTradeExecutor.java b/src/main/java/com/xcong/excoin/modules/okxApi/OkxTradeExecutor.java
new file mode 100644
index 0000000..a5a15ea
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/OkxTradeExecutor.java
@@ -0,0 +1,178 @@
+package com.xcong.excoin.modules.okxApi;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.xcong.excoin.modules.okxApi.enums.OkxEnums;
+import com.xcong.excoin.modules.okxApi.param.TradeRequestParam;
+import lombok.extern.slf4j.Slf4j;
+import org.java_websocket.client.WebSocketClient;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * OKX WebSocket 交易执行器。
+ *
+ * <h3>设计目的</h3>
+ * WebSocket 消息在回调线程中处理。下单操作参数构建等逻辑
+ * 提交到独立线程池异步执行，避免阻塞 WS 回调线程。
+ *
+ * <h3>线程模型</h3>
+ * <ul>
+ * <li><b>单线程</b>：保证下单顺序（开多→开空→止盈单），避免并发竞争</li>
+ * <li><b>有界队列 64</b>：防止堆积</li>
+ * <li><b>CallerRunsPolicy</b>：队列满时由提交线程直接同步执行，形成自然背压</li>
+ * <li><b>allowCoreThreadTimeOut</b>：60s 空闲后线程回收</li>
+ * </ul>
+ *
+ * @author Administrator
+ */
+@Slf4j
+public class OkxTradeExecutor {
+
+ private final OkxConfig config;
+ private final String accountName;
+
+ private final ExecutorService executor;
+
+ public OkxTradeExecutor(OkxConfig config, String accountName) {
+ this.config = config;
+ this.accountName = accountName;
+ this.executor = new ThreadPoolExecutor(
+ 1, 1,
+ 60L, TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(64),
+ r -> {
+ Thread t = new Thread(r, "okxApi-trade-worker");
+ t.setDaemon(true);
+ return t;
+ },
+ new ThreadPoolExecutor.CallerRunsPolicy()
+ );
+ ((ThreadPoolExecutor) executor).allowCoreThreadTimeOut(true);
+ }
+
+ public void shutdown() {
+ executor.shutdown();
+ try {
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ executor.shutdownNow();
+ }
+ }
+
+ public void openLong(WebSocketClient wsClient, Runnable onSuccess) {
+ openPosition(wsClient, config.getQuantity(), OkxEnums.POSSIDE_LONG, OkxEnums.SIDE_BUY, "开多", onSuccess);
+ }
+
+ public void openShort(WebSocketClient wsClient, Runnable onSuccess) {
+ openPosition(wsClient, config.getQuantity(), OkxEnums.POSSIDE_SHORT, OkxEnums.SIDE_SELL, "开空", onSuccess);
+ }
+
+ private void openPosition(WebSocketClient wsClient, String sz, String posSide, String side, String label, Runnable onSuccess) {
+ executor.execute(() -> {
+ try {
+ TradeRequestParam param = buildParam(side, posSide, sz, OkxEnums.ORDTYPE_MARKET);
+ sendOrder(wsClient, param);
+
+ log.info("[TradeExec] {}已发送, 方向:{}, 数量:{}", label, posSide, sz);
+ if (onSuccess != null) {
+ onSuccess.run();
+ }
+ } catch (Exception e) {
+ log.error("[TradeExec] {}发送失败", label, e);
+ }
+ });
+ }
+
+ public void placeTakeProfit(WebSocketClient wsClient, String posSide, BigDecimal triggerPrice, String size) {
+ executor.execute(() -> {
+ try {
+ String side = OkxEnums.POSSIDE_LONG.equals(posSide) ? OkxEnums.SIDE_SELL : OkxEnums.SIDE_BUY;
+
+ TradeRequestParam param = buildParam(side, posSide, size, OkxEnums.ORDTYPE_LIMIT);
+ param.setMarkPx(triggerPrice.toString());
+
+ List<TradeRequestParam> params = new ArrayList<>();
+ params.add(param);
+ sendBatchOrders(wsClient, params);
+ log.info("[TradeExec] 止盈单已发送, 方向:{}, 触发价:{}, 数量:{}", posSide, triggerPrice, size);
+ } catch (Exception e) {
+ log.error("[TradeExec] 止盈单发送失败", e);
+ }
+ });
+ }
+
+ private TradeRequestParam buildParam(String side, String posSide, String sz, String ordType) {
+ TradeRequestParam param = new TradeRequestParam();
+ param.setAccountName(accountName);
+ param.setInstId(config.getContract());
+ param.setTdMode(config.getMarginMode());
+ param.setPosSide(posSide);
+ param.setOrdType(ordType);
+ param.setSide(side);
+ param.setClOrdId(OkxWsUtil.getOrderNum(side));
+ param.setSz(sz);
+ param.setTradeType("1");
+ return param;
+ }
+
+ private void sendOrder(WebSocketClient wsClient, TradeRequestParam param) {
+ if (wsClient == null || !wsClient.isOpen()) {
+ log.warn("[TradeExec] WS未连接，跳过下单");
+ return;
+ }
+ if (BigDecimal.ZERO.compareTo(new BigDecimal(param.getSz())) >= 0) {
+ log.warn("[TradeExec] 下单数量<=0，跳过");
+ return;
+ }
+
+ JSONArray argsArray = new JSONArray();
+ JSONObject args = new JSONObject();
+ args.put("instId", param.getInstId());
+ args.put("tdMode", param.getTdMode());
+ args.put("clOrdId", param.getClOrdId());
+ args.put("side", param.getSide());
+ args.put("posSide", param.getPosSide());
+ args.put("ordType", param.getOrdType());
+ args.put("sz", param.getSz());
+ argsArray.add(args);
+
+ String connId = OkxWsUtil.getOrderNum("order");
+ JSONObject msg = OkxWsUtil.buildJsonObject(connId, "order", argsArray);
+ wsClient.send(msg.toJSONString());
+ log.info("[TradeExec] 发送下单: side={}, sz={}", param.getSide(), param.getSz());
+ }
+
+ private void sendBatchOrders(WebSocketClient wsClient, List<TradeRequestParam> params) {
+ if (wsClient == null || !wsClient.isOpen() || params == null || params.isEmpty()) {
+ log.warn("[TradeExec] WS未连接或参数为空，跳过批量下单");
+ return;
+ }
+
+ JSONArray argsArray = new JSONArray();
+ for (TradeRequestParam p : params) {
+ JSONObject args = new JSONObject();
+ args.put("instId", p.getInstId());
+ args.put("tdMode", p.getTdMode());
+ args.put("clOrdId", p.getClOrdId());
+ args.put("side", p.getSide());
+ args.put("posSide", p.getPosSide());
+ args.put("ordType", p.getOrdType());
+ args.put("sz", p.getSz());
+ args.put("px", p.getMarkPx());
+ argsArray.add(args);
+ }
+
+ String connId = OkxWsUtil.getOrderNum(null);
+ JSONObject msg = OkxWsUtil.buildJsonObject(connId, "batch-orders", argsArray);
+ wsClient.send(msg.toJSONString());
+ log.info("[TradeExec] 发送批量下单: {}条", params.size());
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/OkxWebSocketClientMain.java b/src/main/java/com/xcong/excoin/modules/okxApi/OkxWebSocketClientMain.java
new file mode 100644
index 0000000..ab4732e
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/OkxWebSocketClientMain.java
@@ -0,0 +1,23 @@
+package com.xcong.excoin.modules.okxApi;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+/**
+ * OKX 模块独立测试入口。
+ *
+ * <p>通过 Spring XML 上下文（applicationContext.xml）初始化管理器，
+ * 运行极长时间后手动关闭。用于脱离 Web 容器独立调试策略。
+ *
+ * @author Administrator
+ */
+public class OkxWebSocketClientMain {
+ public static void main(String[] args) throws InterruptedException {
+ ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
+ OkxWebSocketClientManager manager = context.getBean(OkxWebSocketClientManager.class);
+
+ Thread.sleep(1200000000L);
+
+ manager.destroy();
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/OkxWebSocketClientManager.java b/src/main/java/com/xcong/excoin/modules/okxApi/OkxWebSocketClientManager.java
new file mode 100644
index 0000000..600f00f
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/OkxWebSocketClientManager.java
@@ -0,0 +1,91 @@
+package com.xcong.excoin.modules.okxApi;
+
+import com.xcong.excoin.modules.okxApi.wsHandler.handler.OkxAccountChannelHandler;
+import com.xcong.excoin.modules.okxApi.wsHandler.handler.OkxCandlestickChannelHandler;
+import com.xcong.excoin.modules.okxApi.wsHandler.handler.OkxOrderInfoChannelHandler;
+import com.xcong.excoin.modules.okxApi.wsHandler.handler.OkxPositionsChannelHandler;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.math.BigDecimal;
+
+@Slf4j
+@Component
+public class OkxWebSocketClientManager {
+
+ private OkxKlineWebSocketClient wsKlineClient;
+ private OkxKlineWebSocketClient wsPrivateClient;
+ private OkxGridTradeService gridTradeService;
+ private OkxConfig config;
+
+ @PostConstruct
+ public void init() {
+ log.info("[管理器] 开始初始化...");
+
+ try {
+ config = OkxConfig.builder()
+ .apiKey("d90ca272391992b8e74f8f92cedb21ec")
+ .secretKey("1861e4f52de4bb53369ea3208d9ede38ece4777368030f96c77d27934c46c274")
+ .passphrase("Aa123123@")
+ .contract("BTC-USDT-SWAP")
+ .leverage("100")
+ .marginMode("cross")
+ .gridRate(new BigDecimal("0.0015"))
+ .overallTp(new BigDecimal("5"))
+ .maxLoss(new BigDecimal("15"))
+ .quantity("1")
+ .contractMultiplier(new BigDecimal("1"))
+ .unrealizedPnlPriceMode(OkxConfig.PnLPriceMode.LAST_PRICE)
+ .isProduction(false)
+ .build();
+
+ String accountName = "OKX_API";
+ gridTradeService = new OkxGridTradeService(config, accountName);
+ gridTradeService.startGrid();
+
+ wsKlineClient = new OkxKlineWebSocketClient(config.getWsKlineUrl());
+ wsKlineClient.addChannelHandler(new OkxCandlestickChannelHandler(config.getContract(), gridTradeService));
+ wsKlineClient.init();
+ log.info("[管理器] K线WS已连接, 已注册K线频道处理器");
+
+ wsPrivateClient = new OkxKlineWebSocketClient(
+ config.getWsPrivateUrl(),
+ config.getApiKey(),
+ config.getSecretKey(),
+ config.getPassphrase());
+ wsPrivateClient.addChannelHandler(new OkxPositionsChannelHandler(config.getContract(), gridTradeService));
+ wsPrivateClient.addChannelHandler(new OkxAccountChannelHandler());
+ wsPrivateClient.addChannelHandler(new OkxOrderInfoChannelHandler(config.getContract(), gridTradeService, config));
+ wsPrivateClient.init();
+ log.info("[管理器] 私有WS已连接, 已注册 3 个频道处理器");
+
+ gridTradeService.setWebSocketClient(wsPrivateClient.getWebSocketClient());
+
+ } catch (Exception e) {
+ log.error("[管理器] 初始化失败", e);
+ }
+ }
+
+ @PreDestroy
+ public void destroy() {
+ log.info("[管理器] 开始销毁...");
+
+ if (gridTradeService != null) {
+ gridTradeService.stopGrid();
+ }
+ if (wsKlineClient != null) {
+ wsKlineClient.destroy();
+ }
+ if (wsPrivateClient != null) {
+ wsPrivateClient.destroy();
+ }
+
+ log.info("[管理器] 销毁完成");
+ }
+
+ public OkxKlineWebSocketClient getKlineWebSocketClient() { return wsKlineClient; }
+ public OkxKlineWebSocketClient getPrivateWebSocketClient() { return wsPrivateClient; }
+ public OkxGridTradeService getGridTradeService() { return gridTradeService; }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/OkxWsUtil.java b/src/main/java/com/xcong/excoin/modules/okxApi/OkxWsUtil.java
new file mode 100644
index 0000000..5322251
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/OkxWsUtil.java
@@ -0,0 +1,106 @@
+package com.xcong.excoin.modules.okxApi;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.X509TrustManager;
+import java.security.SecureRandom;
+import java.security.cert.X509Certificate;
+import java.text.SimpleDateFormat;
+import java.util.Base64;
+import java.util.Date;
+import java.util.Random;
+
+/**
+ * OKX API 工具类：SSL配置、HMAC-SHA256签名、订单ID生成、日期格式化、JSON构建。
+ *
+ * @author Administrator
+ */
+@Slf4j
+public final class OkxWsUtil {
+
+ private OkxWsUtil() {}
+
+ // ==================== SSL ====================
+
+ public static void configureSSL() {
+ try {
+ TrustManager[] trustAllCerts = new TrustManager[]{
+ new X509TrustManager() {
+ @Override
+ public X509Certificate[] getAcceptedIssuers() { return new X509Certificate[0]; }
+ @Override
+ public void checkClientTrusted(X509Certificate[] certs, String authType) {}
+ @Override
+ public void checkServerTrusted(X509Certificate[] certs, String authType) {}
+ }
+ };
+ SSLContext sc = SSLContext.getInstance("TLS");
+ sc.init(null, trustAllCerts, new SecureRandom());
+ SSLContext.setDefault(sc);
+ } catch (Exception e) {
+ log.error("SSL配置失败", e);
+ }
+ }
+
+ // ==================== HMAC-SHA256 签名 ====================
+
+ public static String signWebsocket(String timestamp, String secretKey) {
+ try {
+ String message = String.format("%s%s%s", timestamp, "GET", "/users/self/verify");
+ Mac mac = Mac.getInstance("HmacSHA256");
+ SecretKeySpec spec = new SecretKeySpec(secretKey.getBytes("UTF-8"), "HmacSHA256");
+ mac.init(spec);
+ byte[] hash = mac.doFinal(message.getBytes("UTF-8"));
+ return Base64.getEncoder().encodeToString(hash);
+ } catch (Exception e) {
+ log.error("签名计算失败", e);
+ return "";
+ }
+ }
+
+ // ==================== 订单ID ====================
+
+ public static String getOrderNum(String prefix) {
+ SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHHmmss");
+ String dd = df.format(new Date());
+ if (prefix != null && !prefix.isEmpty()) {
+ return prefix + dd + getRandomNum(5);
+ }
+ return dd + getRandomNum(5);
+ }
+
+ private static String getRandomNum(int length) {
+ String str = "0123456789";
+ Random random = new Random();
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < length; ++i) {
+ sb.append(str.charAt(random.nextInt(str.length())));
+ }
+ return sb.toString();
+ }
+
+ // ==================== JSON构建 ====================
+
+ public static JSONObject buildJsonObject(String connId, String op, JSONArray args) {
+ JSONObject jsonObject = new JSONObject();
+ if (connId != null && !connId.isEmpty()) {
+ jsonObject.put("id", connId);
+ }
+ jsonObject.put("op", op);
+ jsonObject.put("args", args);
+ return jsonObject;
+ }
+
+ // ==================== 日期格式化 ====================
+
+ public static String timestampToDateTime(long timestamp) {
+ SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ return sdf.format(new Date(timestamp));
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/enums/OkxEnums.java b/src/main/java/com/xcong/excoin/modules/okxApi/enums/OkxEnums.java
new file mode 100644
index 0000000..0366984
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/enums/OkxEnums.java
@@ -0,0 +1,25 @@
+package com.xcong.excoin.modules.okxApi.enums;
+
+/**
+ * OKX API 模块枚举常量。
+ *
+ * @author Administrator
+ */
+public final class OkxEnums {
+
+ private OkxEnums() {}
+
+ public static final String POSSIDE_LONG = "long";
+ public static final String POSSIDE_SHORT = "short";
+ public static final String SIDE_BUY = "buy";
+ public static final String SIDE_SELL = "sell";
+ public static final String ORDTYPE_MARKET = "market";
+ public static final String ORDTYPE_LIMIT = "limit";
+ public static final String INSTTYPE_SWAP = "SWAP";
+ public static final String MARGIN_CROSS = "cross";
+
+ public static final String CHANNEL_CANDLE = "candle1m";
+ public static final String CHANNEL_POSITIONS = "positions";
+ public static final String CHANNEL_ACCOUNT = "account";
+ public static final String CHANNEL_ORDERS = "orders";
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/param/TradeRequestParam.java b/src/main/java/com/xcong/excoin/modules/okxApi/param/TradeRequestParam.java
new file mode 100644
index 0000000..64e0d7f
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/param/TradeRequestParam.java
@@ -0,0 +1,41 @@
+package com.xcong.excoin.modules.okxApi.param;
+
+/**
+ * 交易请求参数。
+ *
+ * @author Administrator
+ */
+public class TradeRequestParam {
+
+ private String accountName;
+ private String markPx;
+ private String instId;
+ private String tdMode;
+ private String posSide;
+ private String ordType;
+ private String tradeType;
+ private String side;
+ private String clOrdId;
+ private String sz;
+
+ public String getAccountName() { return accountName; }
+ public void setAccountName(String accountName) { this.accountName = accountName; }
+ public String getMarkPx() { return markPx; }
+ public void setMarkPx(String markPx) { this.markPx = markPx; }
+ public String getInstId() { return instId; }
+ public void setInstId(String instId) { this.instId = instId; }
+ public String getTdMode() { return tdMode; }
+ public void setTdMode(String tdMode) { this.tdMode = tdMode; }
+ public String getPosSide() { return posSide; }
+ public void setPosSide(String posSide) { this.posSide = posSide; }
+ public String getOrdType() { return ordType; }
+ public void setOrdType(String ordType) { this.ordType = ordType; }
+ public String getTradeType() { return tradeType; }
+ public void setTradeType(String tradeType) { this.tradeType = tradeType; }
+ public String getSide() { return side; }
+ public void setSide(String side) { this.side = side; }
+ public String getClOrdId() { return clOrdId; }
+ public void setClOrdId(String clOrdId) { this.clOrdId = clOrdId; }
+ public String getSz() { return sz; }
+ public void setSz(String sz) { this.sz = sz; }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/OkxChannelHandler.java b/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/OkxChannelHandler.java
new file mode 100644
index 0000000..16425f3
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/OkxChannelHandler.java
@@ -0,0 +1,44 @@
+package com.xcong.excoin.modules.okxApi.wsHandler;
+
+import com.alibaba.fastjson.JSONObject;
+import org.java_websocket.client.WebSocketClient;
+
+/**
+ * OKX WebSocket 频道处理器接口。
+ *
+ * <p>每个 OKX 频道对应一个实现类。新增频道只需实现此接口，
+ * 然后通过 {@code OkxKlineWebSocketClient.addChannelHandler()} 注册即可。
+ *
+ * <h3>实现类</h3>
+ * <ul>
+ * <li>{@code OkxCandlestickChannelHandler} — K 线数据</li>
+ * <li>{@code OkxPositionsChannelHandler} — 持仓更新</li>
+ * <li>{@code OkxAccountChannelHandler} — 账户信息</li>
+ * <li>{@code OkxOrderInfoChannelHandler} — 订单信息（止盈止损、PnL跟踪）</li>
+ * </ul>
+ *
+ * <h3>路由机制</h3>
+ * {@code handleMessage()} 返回 {@code true} 表示消息已被该 handler 处理，
+ * 路由循环会停止遍历。返回 {@code false} 表示不匹配（channel 名不相等）。
+ *
+ * @author Administrator
+ */
+public interface OkxChannelHandler {
+
+ /** 频道名称，如 {@code "candle1m"} */
+ String getChannelName();
+
+ /** 发送订阅请求 */
+ void subscribe(WebSocketClient ws);
+
+ /** 发送取消订阅请求 */
+ void unsubscribe(WebSocketClient ws);
+
+ /**
+ * 处理频道推送消息。
+ *
+ * @param response WebSocket 推送的完整 JSON
+ * @return true 表示已处理（循环停止），false 表示频道不匹配（继续遍历下一个 handler）
+ */
+ boolean handleMessage(JSONObject response);
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxAccountChannelHandler.java b/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxAccountChannelHandler.java
new file mode 100644
index 0000000..af0c07e
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxAccountChannelHandler.java
@@ -0,0 +1,90 @@
+package com.xcong.excoin.modules.okxApi.wsHandler.handler;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.xcong.excoin.modules.okxApi.wsHandler.OkxChannelHandler;
+import lombok.extern.slf4j.Slf4j;
+import org.java_websocket.client.WebSocketClient;
+
+/**
+ * OKX 账户频道处理器（account）。
+ *
+ * <h3>数据用途</h3>
+ * 监控账户余额和保证金信息。用于保证金安全阀检查。
+ *
+ * <h3>推送字段</h3>
+ * ccy, availBal, cashBal, eq, upl, imr, ordFroz
+ *
+ * <h3>注意</h3>
+ * 当前版本仅做日志输出，后续可扩展保证金安全阀功能。
+ *
+ * @author Administrator
+ */
+@Slf4j
+public class OkxAccountChannelHandler implements OkxChannelHandler {
+
+ private static final String CHANNEL_NAME = "account";
+ private static final String CHANNEL = "account";
+
+ public OkxAccountChannelHandler() {
+ }
+
+ @Override
+ public String getChannelName() {
+ return CHANNEL_NAME;
+ }
+
+ @Override
+ public void subscribe(WebSocketClient ws) {
+ JSONObject msg = new JSONObject();
+ msg.put("op", "subscribe");
+ JSONArray args = new JSONArray();
+ JSONObject arg = new JSONObject();
+ arg.put("channel", CHANNEL);
+ args.add(arg);
+ msg.put("args", args);
+ ws.send(msg.toJSONString());
+ log.info("[{}] 订阅成功", CHANNEL_NAME);
+ }
+
+ @Override
+ public void unsubscribe(WebSocketClient ws) {
+ JSONObject msg = new JSONObject();
+ msg.put("op", "unsubscribe");
+ JSONArray args = new JSONArray();
+ JSONObject arg = new JSONObject();
+ arg.put("channel", CHANNEL);
+ args.add(arg);
+ msg.put("args", args);
+ ws.send(msg.toJSONString());
+ log.info("[{}] 取消订阅成功", CHANNEL_NAME);
+ }
+
+ @Override
+ public boolean handleMessage(JSONObject response) {
+ JSONObject argObj = response.getJSONObject("arg");
+ if (argObj == null) {
+ return false;
+ }
+ String channel = argObj.getString("channel");
+ if (!CHANNEL.equals(channel)) {
+ return false;
+ }
+ try {
+ JSONArray dataArray = response.getJSONArray("data");
+ if (dataArray == null || dataArray.isEmpty()) {
+ return true;
+ }
+ for (int i = 0; i < dataArray.size(); i++) {
+ JSONObject acct = dataArray.getJSONObject(i);
+ log.info("[{}] 账户更新, 可用余额:{}, 现金余额:{}, 权益:{}, 未实现盈亏:{}, 保证金:{}",
+ CHANNEL_NAME,
+ acct.get("availBal"), acct.get("cashBal"),
+ acct.get("eq"), acct.get("upl"), acct.get("imr"));
+ }
+ } catch (Exception e) {
+ log.error("[{}] 处理数据失败", CHANNEL_NAME, e);
+ }
+ return true;
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxCandlestickChannelHandler.java b/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxCandlestickChannelHandler.java
new file mode 100644
index 0000000..bad740e
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxCandlestickChannelHandler.java
@@ -0,0 +1,93 @@
+package com.xcong.excoin.modules.okxApi.wsHandler.handler;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.xcong.excoin.modules.okxApi.OkxGridTradeService;
+import com.xcong.excoin.modules.okxApi.OkxWsUtil;
+import com.xcong.excoin.modules.okxApi.enums.OkxEnums;
+import com.xcong.excoin.modules.okxApi.wsHandler.OkxChannelHandler;
+import lombok.extern.slf4j.Slf4j;
+import org.java_websocket.client.WebSocketClient;
+
+import java.math.BigDecimal;
+
+@Slf4j
+public class OkxCandlestickChannelHandler implements OkxChannelHandler {
+
+ private final String instId;
+ private final OkxGridTradeService gridTradeService;
+
+ public OkxCandlestickChannelHandler(String instId, OkxGridTradeService gridTradeService) {
+ this.instId = instId;
+ this.gridTradeService = gridTradeService;
+ }
+
+ @Override
+ public String getChannelName() {
+ return OkxEnums.CHANNEL_CANDLE;
+ }
+
+ @Override
+ public void subscribe(WebSocketClient ws) {
+ JSONObject msg = new JSONObject();
+ msg.put("op", "subscribe");
+ JSONArray args = new JSONArray();
+ JSONObject arg = new JSONObject();
+ arg.put("channel", OkxEnums.CHANNEL_CANDLE);
+ arg.put("instId", instId);
+ args.add(arg);
+ msg.put("args", args);
+ ws.send(msg.toJSONString());
+ log.info("[{}] 订阅成功, 合约:{}, 周期:1m", OkxEnums.CHANNEL_CANDLE, instId);
+ }
+
+ @Override
+ public void unsubscribe(WebSocketClient ws) {
+ JSONObject msg = new JSONObject();
+ msg.put("op", "unsubscribe");
+ JSONArray args = new JSONArray();
+ JSONObject arg = new JSONObject();
+ arg.put("channel", OkxEnums.CHANNEL_CANDLE);
+ arg.put("instId", instId);
+ args.add(arg);
+ msg.put("args", args);
+ ws.send(msg.toJSONString());
+ log.info("[{}] 取消订阅成功", OkxEnums.CHANNEL_CANDLE);
+ }
+
+ @Override
+ public boolean handleMessage(JSONObject response) {
+ JSONObject argObj = response.getJSONObject("arg");
+ if (argObj == null) {
+ return false;
+ }
+ String channel = argObj.getString("channel");
+ if (!OkxEnums.CHANNEL_CANDLE.equals(channel)) {
+ return false;
+ }
+ String msgInstId = argObj.getString("instId");
+ if (!instId.equals(msgInstId)) {
+ return false;
+ }
+ try {
+ JSONArray dataArray = response.getJSONArray("data");
+ if (dataArray == null || dataArray.isEmpty()) {
+ log.warn("[{}] 数据为空", OkxEnums.CHANNEL_CANDLE);
+ return true;
+ }
+ JSONArray data = dataArray.getJSONArray(0);
+ BigDecimal closePx = new BigDecimal(data.getString(4));
+ String time = OkxWsUtil.timestampToDateTime(Long.parseLong(data.getString(0)));
+ String confirm = data.getString(8);
+
+ log.info("[{}] 收盘:{}, 时间:{}, 完结:{}", OkxEnums.CHANNEL_CANDLE, closePx, time, "1".equals(confirm));
+
+ if (gridTradeService != null) {
+ gridTradeService.onKline(closePx);
+ }
+ } catch (Exception e) {
+ log.error("[{}] 处理数据失败", OkxEnums.CHANNEL_CANDLE, e);
+ }
+ return true;
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxOrderInfoChannelHandler.java b/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxOrderInfoChannelHandler.java
new file mode 100644
index 0000000..813fe03
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxOrderInfoChannelHandler.java
@@ -0,0 +1,121 @@
+package com.xcong.excoin.modules.okxApi.wsHandler.handler;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.xcong.excoin.modules.okxApi.OkxConfig;
+import com.xcong.excoin.modules.okxApi.OkxGridTradeService;
+import com.xcong.excoin.modules.okxApi.wsHandler.OkxChannelHandler;
+import lombok.extern.slf4j.Slf4j;
+import org.java_websocket.client.WebSocketClient;
+
+import java.math.BigDecimal;
+
+/**
+ * OKX 订单信息频道处理器（orders）。
+ *
+ * <h3>数据用途</h3>
+ * 监控订单成交（filled）状态，跟踪已实现盈亏（fillPnl）。
+ * 成交后通过 batch-orders 频道设置止盈止损限价单。
+ *
+ * <h3>推送字段</h3>
+ * instId, ordId, clOrdId, side, posSide, state, accFillSz, avgPx, fillPnl, fillFee
+ *
+ * <h3>数据处理</h3>
+ * state=filled 且 accFillSz>0 → 成交 → 调用 gridTradeService.onOrderFilled() 跟踪盈亏
+ *
+ * @author Administrator
+ */
+@Slf4j
+public class OkxOrderInfoChannelHandler implements OkxChannelHandler {
+
+ private static final String CHANNEL_NAME = "orders";
+ private static final String CHANNEL = "orders";
+
+ private final String instId;
+ private final OkxGridTradeService gridTradeService;
+ private final OkxConfig config;
+
+ public OkxOrderInfoChannelHandler(String instId, OkxGridTradeService gridTradeService, OkxConfig config) {
+ this.instId = instId;
+ this.gridTradeService = gridTradeService;
+ this.config = config;
+ }
+
+ @Override
+ public String getChannelName() {
+ return CHANNEL_NAME;
+ }
+
+ @Override
+ public void subscribe(WebSocketClient ws) {
+ JSONObject msg = new JSONObject();
+ msg.put("op", "subscribe");
+ JSONArray args = new JSONArray();
+ JSONObject arg = new JSONObject();
+ arg.put("channel", CHANNEL);
+ arg.put("instType", "SWAP");
+ arg.put("instId", instId);
+ args.add(arg);
+ msg.put("args", args);
+ ws.send(msg.toJSONString());
+ log.info("[{}] 订阅成功, 合约:{}", CHANNEL_NAME, instId);
+ }
+
+ @Override
+ public void unsubscribe(WebSocketClient ws) {
+ JSONObject msg = new JSONObject();
+ msg.put("op", "unsubscribe");
+ JSONArray args = new JSONArray();
+ JSONObject arg = new JSONObject();
+ arg.put("channel", CHANNEL);
+ arg.put("instType", "SWAP");
+ arg.put("instId", instId);
+ args.add(arg);
+ msg.put("args", args);
+ ws.send(msg.toJSONString());
+ log.info("[{}] 取消订阅成功", CHANNEL_NAME);
+ }
+
+ @Override
+ public boolean handleMessage(JSONObject response) {
+ JSONObject argObj = response.getJSONObject("arg");
+ if (argObj == null) {
+ return false;
+ }
+ String channel = argObj.getString("channel");
+ if (!CHANNEL.equals(channel)) {
+ return false;
+ }
+ try {
+ JSONArray dataArray = response.getJSONArray("data");
+ if (dataArray == null || dataArray.isEmpty()) {
+ return true;
+ }
+ for (int i = 0; i < dataArray.size(); i++) {
+ JSONObject detail = dataArray.getJSONObject(i);
+ if (!instId.equals(detail.getString("instId"))) {
+ continue;
+ }
+ String state = detail.getString("state");
+ String accFillSz = detail.getString("accFillSz");
+ String fillPnl = detail.getString("fillPnl");
+ String posSide = detail.getString("posSide");
+ String avgPx = detail.getString("avgPx");
+ String clOrdId = detail.getString("clOrdId");
+
+ log.info("[{}] 订单, 方向:{}, 状态:{}, 成交量:{}, 均价:{}, 盈亏:{}, 编号:{}",
+ CHANNEL_NAME, posSide, state, accFillSz, avgPx, fillPnl, clOrdId);
+
+ if ("filled".equals(state) && accFillSz != null && new BigDecimal(accFillSz).compareTo(BigDecimal.ZERO) > 0) {
+ if (gridTradeService != null) {
+ BigDecimal pnl = fillPnl != null ? new BigDecimal(fillPnl) : BigDecimal.ZERO;
+ gridTradeService.onOrderFilled(posSide, new BigDecimal(accFillSz), pnl);
+ }
+ }
+ }
+ } catch (Exception e) {
+ log.error("[{}] 处理数据失败", CHANNEL_NAME, e);
+ }
+ return true;
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxPositionsChannelHandler.java b/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxPositionsChannelHandler.java
new file mode 100644
index 0000000..1f60705
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/okxApi/wsHandler/handler/OkxPositionsChannelHandler.java
@@ -0,0 +1,97 @@
+package com.xcong.excoin.modules.okxApi.wsHandler.handler;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.xcong.excoin.modules.okxApi.OkxGridTradeService;
+import com.xcong.excoin.modules.okxApi.enums.OkxEnums;
+import com.xcong.excoin.modules.okxApi.wsHandler.OkxChannelHandler;
+import lombok.extern.slf4j.Slf4j;
+import org.java_websocket.client.WebSocketClient;
+
+import java.math.BigDecimal;
+
+@Slf4j
+public class OkxPositionsChannelHandler implements OkxChannelHandler {
+
+ private final String instId;
+ private final OkxGridTradeService gridTradeService;
+
+ public OkxPositionsChannelHandler(String instId, OkxGridTradeService gridTradeService) {
+ this.instId = instId;
+ this.gridTradeService = gridTradeService;
+ }
+
+ @Override
+ public String getChannelName() {
+ return OkxEnums.CHANNEL_POSITIONS;
+ }
+
+ @Override
+ public void subscribe(WebSocketClient ws) {
+ JSONObject msg = new JSONObject();
+ msg.put("op", "subscribe");
+ JSONArray args = new JSONArray();
+ JSONObject arg = new JSONObject();
+ arg.put("channel", OkxEnums.CHANNEL_POSITIONS);
+ arg.put("instType", OkxEnums.INSTTYPE_SWAP);
+ arg.put("instId", instId);
+ args.add(arg);
+ msg.put("args", args);
+ ws.send(msg.toJSONString());
+ log.info("[{}] 订阅成功, 合约:{}", OkxEnums.CHANNEL_POSITIONS, instId);
+ }
+
+ @Override
+ public void unsubscribe(WebSocketClient ws) {
+ JSONObject msg = new JSONObject();
+ msg.put("op", "unsubscribe");
+ JSONArray args = new JSONArray();
+ JSONObject arg = new JSONObject();
+ arg.put("channel", OkxEnums.CHANNEL_POSITIONS);
+ arg.put("instType", OkxEnums.INSTTYPE_SWAP);
+ arg.put("instId", instId);
+ args.add(arg);
+ msg.put("args", args);
+ ws.send(msg.toJSONString());
+ log.info("[{}] 取消订阅成功", OkxEnums.CHANNEL_POSITIONS);
+ }
+
+ @Override
+ public boolean handleMessage(JSONObject response) {
+ JSONObject argObj = response.getJSONObject("arg");
+ if (argObj == null) {
+ return false;
+ }
+ String channel = argObj.getString("channel");
+ if (!OkxEnums.CHANNEL_POSITIONS.equals(channel)) {
+ return false;
+ }
+ try {
+ JSONArray dataArray = response.getJSONArray("data");
+ if (dataArray == null || dataArray.isEmpty()) {
+ return true;
+ }
+ for (int i = 0; i < dataArray.size(); i++) {
+ JSONObject pos = dataArray.getJSONObject(i);
+ if (!instId.equals(pos.getString("instId"))) {
+ continue;
+ }
+ String posSide = pos.getString("posSide");
+ BigDecimal size = new BigDecimal(pos.getString("pos"));
+ BigDecimal avgPx = pos.containsKey("avgPx") && pos.getString("avgPx") != null
+ ? new BigDecimal(pos.getString("avgPx")) : BigDecimal.ZERO;
+
+ log.info("[{}] 持仓更新, 方向:{}, 数量:{}, 均价:{}, 未实现盈亏:{}, 保证金:{}",
+ OkxEnums.CHANNEL_POSITIONS, posSide, size, avgPx,
+ pos.get("upl"), pos.get("imr"));
+
+ if (gridTradeService != null) {
+ gridTradeService.onPositionUpdate(posSide, size, avgPx);
+ }
+ }
+ } catch (Exception e) {
+ log.error("[{}] 处理数据失败", OkxEnums.CHANNEL_POSITIONS, e);
+ }
+ return true;
+ }
+}
--
Gitblit v1.9.1