From 1cb6ffd630a5ba5fe04f725b7699de4c6f654dc7 Mon Sep 17 00:00:00 2001
From: Administrator <15274802129@163.com>
Date: Wed, 17 Dec 2025 13:50:44 +0800
Subject: [PATCH] feat(okxNewPrice): 优化网格交易策略逻辑
---
src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxQuantWebSocketClient.java | 112 ++++++++++++++++++++++++++++++++++++-------------------
1 files changed, 73 insertions(+), 39 deletions(-)
diff --git a/src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxQuantWebSocketClient.java b/src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxQuantWebSocketClient.java
index 93ec73b..cdb9117 100644
--- a/src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxQuantWebSocketClient.java
+++ b/src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxQuantWebSocketClient.java
@@ -30,15 +30,11 @@
* @author Administrator
*/
@Slf4j
-@Component
-@ConditionalOnProperty(prefix = "app", name = "quant", havingValue = "true")
public class OkxQuantWebSocketClient {
- @Autowired
- private WangGeService wangGeService;
- @Autowired
- private CaoZuoService caoZuoService;
- @Autowired
- private RedisUtils redisUtils;
+ private final WangGeService wangGeService;
+ private final CaoZuoService caoZuoService;
+ private final RedisUtils redisUtils;
+ private final ExchangeInfoEnum account;
private WebSocketClient webSocketClient;
private ScheduledExecutorService heartbeatExecutor;
@@ -48,6 +44,14 @@
// 连接状态标志
private final AtomicBoolean isConnected = new AtomicBoolean(false);
private final AtomicBoolean isConnecting = new AtomicBoolean(false);
+
+ public OkxQuantWebSocketClient(ExchangeInfoEnum account, WangGeService wangGeService,
+ CaoZuoService caoZuoService, RedisUtils redisUtils) {
+ this.account = account;
+ this.wangGeService = wangGeService;
+ this.caoZuoService = caoZuoService;
+ this.redisUtils = redisUtils;
+ }
private static final String WS_URL_MONIPAN = "wss://wspap.okx.com:8443/ws/v5/private";
private static final String WS_URL_SHIPAN = "wss://ws.okx.com:8443/ws/v5/private";
@@ -91,21 +95,50 @@
* 销毁方法，在 Spring Bean 销毁前执行。
* 关闭 WebSocket 连接、停止心跳定时器及相关的线程资源。
*/
+// @PreDestroy
+// public void destroy() {
+// if (webSocketClient != null && webSocketClient.isOpen()) {
+// subscribeAccountChannel(UNSUBSCRIBE);
+// subscribePositionChannel(UNSUBSCRIBE);
+// subscribeOrderInfoChannel(UNSUBSCRIBE);
+// webSocketClient.close();
+// }
+// shutdownExecutorGracefully(heartbeatExecutor);
+// if (pongTimeoutFuture != null) {
+// pongTimeoutFuture.cancel(true);
+// }
+// shutdownExecutorGracefully(sharedExecutor);
+//
+// // 移除了 reconnectScheduler 的关闭操作
+// }
@PreDestroy
public void destroy() {
- if (webSocketClient != null && webSocketClient.isOpen()) {
- subscribeAccountChannel(UNSUBSCRIBE);
- subscribePositionChannel(UNSUBSCRIBE);
- subscribeOrderInfoChannel(UNSUBSCRIBE);
- webSocketClient.close();
+ log.info("开始销毁OkxQuantWebSocketClient");
+
+ // 设置关闭标志，避免重连
+ if (sharedExecutor != null && !sharedExecutor.isShutdown()) {
+ sharedExecutor.shutdown();
}
+
+ if (webSocketClient != null && webSocketClient.isOpen()) {
+ try {
+ subscribeAccountChannel(UNSUBSCRIBE);
+ subscribePositionChannel(UNSUBSCRIBE);
+ subscribeOrderInfoChannel(UNSUBSCRIBE);
+ webSocketClient.closeBlocking();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("关闭WebSocket连接时被中断");
+ }
+ }
+
shutdownExecutorGracefully(heartbeatExecutor);
if (pongTimeoutFuture != null) {
pongTimeoutFuture.cancel(true);
}
shutdownExecutorGracefully(sharedExecutor);
- // 移除了 reconnectScheduler 的关闭操作
+ log.info("OkxQuantWebSocketClient销毁完成");
}
private void shutdownExecutorGracefully(ExecutorService executor) {
@@ -140,12 +173,11 @@
}
try {
- InstrumentsWs.handleEvent();
- wangGeService.initWangGe();
+ InstrumentsWs.handleEvent(account.name());
SSLConfig.configureSSL();
System.setProperty("https.protocols", "TLSv1.2,TLSv1.3");
String WS_URL = WS_URL_MONIPAN;
- if (ExchangeInfoEnum.OKX_UAT.isAccountType()){
+ if (account.isAccountType()){
WS_URL = WS_URL_SHIPAN;
}
URI uri = new URI(WS_URL);
@@ -170,7 +202,7 @@
// 棜查应用是否正在关闭
if (!sharedExecutor.isShutdown()) {
resetHeartbeatTimer();
- websocketLogin();
+ websocketLogin(account);
} else {
log.warn("应用正在关闭，忽略WebSocket连接成功回调");
}
@@ -220,8 +252,8 @@
}
}
- private void websocketLogin() {
- LoginWs.websocketLogin(webSocketClient);
+ private void websocketLogin(ExchangeInfoEnum account) {
+ LoginWs.websocketLogin(webSocketClient, account);
}
private void subscribeBalanceAndPositionChannel(String option) {
@@ -249,7 +281,7 @@
private void handleWebSocketMessage(String message) {
try {
if ("pong".equals(message)) {
- log.debug("收到心跳响应");
+ log.debug("{}: 收到心跳响应", account.name());
cancelPongTimeout();
return;
}
@@ -260,26 +292,26 @@
String code = response.getString("code");
if ("0".equals(code)) {
String connId = response.getString("connId");
- log.info("WebSocket登录成功, connId: {}", connId);
+ log.info("{}: WebSocket登录成功, connId: {}", account.name(), connId);
subscribeAccountChannel(SUBSCRIBE);
subscribeOrderInfoChannel(SUBSCRIBE);
subscribePositionChannel(SUBSCRIBE);
} else {
- log.error("WebSocket登录失败, code: {}, msg: {}", code, response.getString("msg"));
+ log.error("{}: WebSocket登录失败, code: {}, msg: {}", account.name(), code, response.getString("msg"));
}
} else if ("subscribe".equals(event)) {
subscribeEvent(response);
} else if ("error".equals(event)) {
- log.error("订阅错误: code={}, msg={}",
- response.getString("code"), response.getString("msg"));
+ log.error("{}: 订阅错误: code={}, msg={}",
+ account.name(), response.getString("code"), response.getString("msg"));
} else if ("channel-conn-count".equals(event)) {
- log.info("连接限制更新: channel={}, connCount={}",
- response.getString("channel"), response.getString("connCount"));
+ log.info("{}: 连接限制更新: channel={}, connCount={}",
+ account.name(), response.getString("channel"), response.getString("connCount"));
} else {
processPushData(response);
}
} catch (Exception e) {
- log.error("处理WebSocket消息失败: {}", message, e);
+ log.error("{}: 处理WebSocket消息失败: {}", account.name(), message, e);
}
}
@@ -296,13 +328,13 @@
return;
}
if (OrderInfoWs.ORDERINFOWS_CHANNEL.equals(channel)) {
- OrderInfoWs.initEvent(response);
+ OrderInfoWs.initEvent(response, account.name());
}
if (AccountWs.ACCOUNTWS_CHANNEL.equals(channel)) {
- AccountWs.initEvent(response);
+ AccountWs.initEvent(response, account.name());
}
if (PositionsWs.POSITIONSWS_CHANNEL.equals(channel)) {
- PositionsWs.initEvent(response);
+ PositionsWs.initEvent(response, account.name());
}
}
@@ -318,30 +350,32 @@
if (TradeOrderWs.ORDERWS_CHANNEL.equals(op)) {
// 直接使用Object类型接收，避免强制类型转换
Object data = response.get("data");
- log.info("收到下单推送结果: {}", JSON.toJSONString(data));
+ log.info("{}: 收到下单推送结果: {}", account.name(), JSON.toJSONString(data));
return;
}
}
JSONObject arg = response.getJSONObject("arg");
if (arg == null) {
- log.warn("无效的推送数据，缺少 'arg' 字段 :{}",response);
+ log.warn("{}: 无效的推送数据，缺少 'arg' 字段 :{}", account.name(), response);
return;
}
String channel = arg.getString("channel");
if (channel == null) {
- log.warn("无效的推送数据，缺少 'channel' 字段{}",response);
+ log.warn("{}: 无效的推送数据，缺少 'channel' 字段{}", account.name(), response);
return;
}
+ // 注意：当前实现中，OrderInfoWs等类使用静态Map存储数据
+ // 这会导致多账号之间的数据冲突。需要进一步修改这些类的设计，让数据存储与特定账号关联
if (OrderInfoWs.ORDERINFOWS_CHANNEL.equals(channel)) {
- OrderInfoWs.handleEvent(response, redisUtils);
+ OrderInfoWs.handleEvent(response, redisUtils, account.name());
}else if (AccountWs.ACCOUNTWS_CHANNEL.equals(channel)) {
- AccountWs.handleEvent(response);
- String side = caoZuoService.caoZuo();
- TradeOrderWs.orderEvent(webSocketClient, side);
+ AccountWs.handleEvent(response, account.name());
+ String side = caoZuoService.caoZuo(account.name());
+ TradeOrderWs.orderEvent(webSocketClient, side, account.name());
} else if (PositionsWs.POSITIONSWS_CHANNEL.equals(channel)) {
- PositionsWs.handleEvent(response);
+ PositionsWs.handleEvent(response, account.name());
} else if (BalanceAndPositionWs.CHANNEL_NAME.equals(channel)) {
BalanceAndPositionWs.handleEvent(response);
}
--
Gitblit v1.9.1