exchange.git - Gitblit

			@@ -1,13 +1,10 @@
			package com.xcong.excoin.modules.okxNewPrice;

			import cn.hutool.core.util.ObjectUtil;
			import cn.hutool.core.util.StrUtil;
			import com.alibaba.fastjson.JSON;
			import com.alibaba.fastjson.JSONObject;
			import com.xcong.excoin.modules.okxNewPrice.celue.CaoZuoService;
			import com.xcong.excoin.modules.okxNewPrice.okxWs.*;
			import com.xcong.excoin.modules.okxNewPrice.okxWs.enums.CoinEnums;
			import com.xcong.excoin.modules.okxNewPrice.okxWs.enums.OrderParamEnums;
			import com.xcong.excoin.modules.okxNewPrice.okxWs.enums.ExchangeInfoEnum;
			import com.xcong.excoin.modules.okxNewPrice.utils.SSLConfig;
			import com.xcong.excoin.modules.okxNewPrice.wangge.WangGeService;
			import com.xcong.excoin.utils.RedisUtils;
			@@ -20,10 +17,10 @@

			import javax.annotation.PostConstruct;
			import javax.annotation.PreDestroy;
			import java.math.BigDecimal;
			import java.net.URI;
			import java.net.URISyntaxException;
			import java.util.concurrent.*;
			import java.util.concurrent.atomic.AtomicBoolean;
			import java.util.concurrent.atomic.AtomicReference;

			/**
			@@ -47,10 +44,13 @@
			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 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";
			private static final boolean INTERNET = false;

			/**
			* 订阅频道指令
			@@ -76,6 +76,8 @@
			public void init() {
			connect();
			startHeartbeat();

			// 移除了每小时重连的定时任务
			}

			/**
			@@ -95,6 +97,8 @@
			pongTimeoutFuture.cancel(true);
			}
			shutdownExecutorGracefully(sharedExecutor);

			// 移除了 reconnectScheduler 的关闭操作
			}

			private void shutdownExecutorGracefully(ExecutorService executor) {
			@@ -117,21 +121,46 @@
			* 设置回调函数以监听连接打开、接收消息、关闭和错误事件。
			*/
			private void connect() {
			// 避免重复连接
			if (isConnecting.get()) {
			log.info("连接已在进行中，跳过重复连接请求");
			return;
			}

			if (!isConnecting.compareAndSet(false, true)) {
			log.info("连接已在进行中，跳过重复连接请求");
			return;
			}

			try {
			InstrumentsWs.handleEvent(redisUtils);
			InstrumentsWs.handleEvent();
			wangGeService.initWangGe();
			SSLConfig.configureSSL();
			System.setProperty("https.protocols", "TLSv1.2,TLSv1.3");
			String WS_URL = WS_URL_MONIPAN;
			if (INTERNET){
			if (ExchangeInfoEnum.OKX_UAT.isAccountType()){
			WS_URL = WS_URL_SHIPAN;
			}
			URI uri = new URI(WS_URL);

			// 关闭之前的连接（如果存在）
			if (webSocketClient != null) {
			try {
			webSocketClient.closeBlocking();
			} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			log.warn("关闭之前连接时被中断");
			}
			}

			webSocketClient = new WebSocketClient(uri) {
			@Override
			public void onOpen(ServerHandshake handshake) {
			log.info("OKX account-order WebSocket连接成功");
			// 检查应用是否正在关闭
			isConnected.set(true);
			isConnecting.set(false);

			// 棜查应用是否正在关闭
			if (!sharedExecutor.isShutdown()) {
			resetHeartbeatTimer();
			websocketLogin();
			@@ -150,6 +179,8 @@
			@Override
			public void onClose(int code, String reason, boolean remote) {
			log.warn("OKX account-order WebSocket连接关闭: code={}, reason={}", code, reason);
			isConnected.set(false);
			isConnecting.set(false);
			cancelPongTimeout();

			if (sharedExecutor != null && !sharedExecutor.isShutdown() && !sharedExecutor.isTerminated()) {
			@@ -171,12 +202,14 @@
			@Override
			public void onError(Exception ex) {
			log.error("OKX account-order WebSocket发生错误", ex);
			isConnected.set(false);
			}
			};

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

			@@ -209,7 +242,7 @@
			private void handleWebSocketMessage(String message) {
			try {
			if ("pong".equals(message)) {
			log.info("收到心跳响应");
			log.debug("收到心跳响应");
			cancelPongTimeout();
			return;
			}
			@@ -228,7 +261,7 @@
			log.error("WebSocket登录失败, code: {}, msg: {}", code, response.getString("msg"));
			}
			} else if ("subscribe".equals(event)) {
			log.info("订阅成功: {}", response.getJSONObject("arg"));
			subscribeEvent(response);
			} else if ("error".equals(event)) {
			log.error("订阅错误: code={}, msg={}",
			response.getString("code"), response.getString("msg"));
			@@ -243,6 +276,29 @@
			}
			}

			private void subscribeEvent(JSONObject response) {
			JSONObject arg = response.getJSONObject("arg");
			if (arg == null) {
			log.warn("无效的推送数据，缺少 'arg' 字段 :{}",response);
			return;
			}

			String channel = arg.getString("channel");
			if (channel == null) {
			log.warn("无效的推送数据，缺少 'channel' 字段{}",response);
			return;
			}
			if (OrderInfoWs.ORDERINFOWS_CHANNEL.equals(channel)) {
			OrderInfoWs.initEvent(response);
			}
			if (AccountWs.ACCOUNTWS_CHANNEL.equals(channel)) {
			AccountWs.initEvent(response);
			}
			if (PositionsWs.POSITIONSWS_CHANNEL.equals(channel)) {
			PositionsWs.initEvent(response);
			}
			}

			/**
			* 解析并处理价格推送数据。
			* 将最新的标记价格存入 Redis 并触发后续业务逻辑比较处理。
			@@ -250,62 +306,38 @@
			* @param response 包含价格数据的 JSON 对象
			*/
			private void processPushData(JSONObject response) {
			String op = response.getString("op");
			if (op != null){
			if (TradeOrderWs.ORDERWS_CHANNEL.equals(op)) {
			// 直接使用Object类型接收，避免强制类型转换
			Object data = response.get("data");
			log.info("收到下单推送结果: {}", JSON.toJSONString(data));
			return;
			}
			}
			JSONObject arg = response.getJSONObject("arg");
			if (arg == null) {
			log.warn("无效的推送数据，缺少 'arg' 字段");
			log.warn("无效的推送数据，缺少 'arg' 字段 :{}",response);
			return;
			}

			String channel = arg.getString("channel");
			if (channel == null) {
			log.warn("无效的推送数据，缺少 'channel' 字段");
			log.warn("无效的推送数据，缺少 'channel' 字段{}",response);
			return;
			}

			if (OrderInfoWs.ORDERINFOWS_CHANNEL.equals(channel)) {
			OrderInfoWs.handleEvent(response, redisUtils);
			}else if (AccountWs.ACCOUNTWS_CHANNEL.equals(channel)) {
			AccountWs.handleEvent(response, redisUtils);
			AccountWs.handleEvent(response);
			} else if (PositionsWs.POSITIONSWS_CHANNEL.equals(channel)) {
			PositionsWs.handleEvent(response, redisUtils);

			String posKey = PositionsWs.POSITIONSWS_CHANNEL + ":" + CoinEnums.HE_YUE.getCode() + ":pos";
			String pos = (String) redisUtils.get(posKey);
			if (StrUtil.isBlank(pos)) {
			log.error("未获取到持仓数量");
			TradeOrderWs.orderEvent(webSocketClient, redisUtils, OrderParamEnums.INIT.getValue());
			return;
			}

			String state = (String) redisUtils.get(InstrumentsWs.INSTRUMENTSWS_CHANNEL + ":" + CoinEnums.HE_YUE.getCode() + ":state");
			if (OrderParamEnums.STATE_3.getValue().equals(state)){
			log.error("持仓盈亏超过下单总保证金，止损冷静一天......");
			TradeOrderWs.orderEvent(webSocketClient, redisUtils, OrderParamEnums.OUT.getValue());
			return;
			}
			String uplKey = PositionsWs.POSITIONSWS_CHANNEL + ":" + CoinEnums.HE_YUE.getCode() + ":upl";
			String upl = (String) redisUtils.get(uplKey);
			if (StrUtil.isBlank(upl)){
			upl = "0";
			}

			String totalOrderUsdtKey = AccountWs.ACCOUNTWS_CHANNEL + ":" + CoinEnums.USDT.getCode() + ":totalOrderUsdt";
			String totalOrderUsdt = (String) redisUtils.get(totalOrderUsdtKey);
			BigDecimal multiply = new BigDecimal(upl).multiply(new BigDecimal("-1"));

			if (new BigDecimal(totalOrderUsdt).compareTo(multiply) < 0) {
			log.error("持仓盈亏超过下单总保证金，止损冷静一天......");
			TradeOrderWs.orderEvent(webSocketClient, redisUtils, OrderParamEnums.OUT.getValue());
			return;
			}

			String side = caoZuoService.caoZuo();
			if (StrUtil.isNotBlank(pos)) {
			TradeOrderWs.orderEvent(webSocketClient, redisUtils, side);
			}
			PositionsWs.handleEvent(response);
			} else if (BalanceAndPositionWs.CHANNEL_NAME.equals(channel)) {
			BalanceAndPositionWs.handleEvent(response);
			}
			String side = caoZuoService.caoZuo();
			TradeOrderWs.orderEvent(webSocketClient, side);
			}

			/**
			@@ -323,8 +355,11 @@
			return t;
			});

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

			// 移除了 schedulePeriodicReconnect 方法

			/**
			* 重置心跳计时器。
			@@ -340,11 +375,18 @@
			}
			}

			// 移除了 performScheduledReconnect 方法

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

			long currentTime = System.currentTimeMillis();
			long lastTime = lastMessageTime.get();

			@@ -383,15 +425,30 @@
			* 在连接意外中断后尝试重新建立连接。
			*/
			private void reconnectWithBackoff() throws InterruptedException {
			// 如果正在连接，则不重复发起重连
			if (isConnecting.get()) {
			log.info("连接已在进行中，跳过重连请求");
			return;
			}

			int attempt = 0;
			int maxAttempts = 5;
			long delayMs = 1000;

			while (attempt < maxAttempts) {
			while (attempt < maxAttempts && !isConnected.get()) {
			try {
			Thread.sleep(delayMs);
			connect();
			return;

			// 等待连接建立
			for (int i = 0; i < 10 && isConnecting.get(); i++) {
			Thread.sleep(500);
			}

			if (isConnected.get()) {
			log.info("重连成功");
			return;
			}
			} catch (Exception e) {
			log.warn("第{}次重连失败", attempt + 1, e);
			delayMs *= 2;
			@@ -401,4 +458,4 @@

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