package com.xcong.excoin.modules.okxNewPrice; import cn.hutool.core.collection.CollUtil; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import com.xcong.excoin.modules.okxNewPrice.celue.CaoZuoService; import com.xcong.excoin.modules.okxNewPrice.indicator.TradingStrategy; 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.param.Kline; import com.xcong.excoin.modules.okxNewPrice.okxWs.param.TradeRequestParam; import com.xcong.excoin.modules.okxNewPrice.okxWs.wanggeList.WangGeListEnum; import com.xcong.excoin.modules.okxNewPrice.okxWs.wanggeList.WangGeListService; import com.xcong.excoin.modules.okxNewPrice.okxpi.config.ExchangeInfoEnum; import com.xcong.excoin.modules.okxNewPrice.okxpi.config.ExchangeLoginService; import com.xcong.excoin.modules.okxNewPrice.utils.SSLConfig; import com.xcong.excoin.modules.okxNewPrice.utils.WsParamBuild; import com.xcong.excoin.utils.RedisUtils; import lombok.extern.slf4j.Slf4j; import org.java_websocket.client.WebSocketClient; import org.java_websocket.handshake.ServerHandshake; import java.math.BigDecimal; import java.net.URI; import java.net.URISyntaxException; import java.util.ArrayList; import java.util.Collection; import java.util.LinkedHashMap; import java.util.List; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; /** * OKX 新价格 WebSocket 客户端类，用于连接 OKX 的 WebSocket 接口， * 实时获取并处理标记价格（mark price）数据，并将价格信息存储到 Redis 中。 * 同时支持心跳检测、自动重连以及异常恢复机制。 * @author Administrator */ @Slf4j public class OkxKlineWebSocketClient { private final RedisUtils redisUtils; private final CaoZuoService caoZuoService; private final OkxWebSocketClientManager clientManager; private final WangGeListService wangGeListService; private WebSocketClient webSocketClient; private ScheduledExecutorService heartbeatExecutor; private volatile ScheduledFuture pongTimeoutFuture; private final AtomicReference lastMessageTime = new AtomicReference<>(System.currentTimeMillis()); // 连接状态标志 private final AtomicBoolean isConnected = new AtomicBoolean(false); private final AtomicBoolean isConnecting = new AtomicBoolean(false); private final AtomicBoolean isInitialized = new AtomicBoolean(false); private static final String CHANNEL = "candle5m"; // private static final String CHANNEL = "candle15m"; // 心跳超时时间(秒)，小于30秒 private static final int HEARTBEAT_TIMEOUT = 10; // 共享线程池用于重连等异步任务 private final ExecutorService sharedExecutor = Executors.newCachedThreadPool(r -> { Thread t = new Thread(r, "okx-ws-kline-worker"); t.setDaemon(true); return t; }); public OkxKlineWebSocketClient(RedisUtils redisUtils, CaoZuoService caoZuoService, OkxWebSocketClientManager clientManager, WangGeListService wangGeListService) { this.redisUtils = redisUtils; this.caoZuoService = caoZuoService; this.clientManager = clientManager; this.wangGeListService = wangGeListService; } /** * 初始化方法，创建并初始化WebSocket客户端实例 */ public void init() { if (!isInitialized.compareAndSet(false, true)) { log.warn("OkxKlineWebSocketClient 已经初始化过，跳过重复初始化"); return; } connect(); startHeartbeat(); } /** * 销毁方法，关闭WebSocket连接和相关资源 */ public void destroy() { log.info("开始销毁OkxKlineWebSocketClient"); // 设置关闭标志，避免重连 if (sharedExecutor != null && !sharedExecutor.isShutdown()) { sharedExecutor.shutdown(); } if (webSocketClient != null && webSocketClient.isOpen()) { try { webSocketClient.closeBlocking(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); log.warn("关闭WebSocket连接时被中断"); } } shutdownExecutorGracefully(heartbeatExecutor); if (pongTimeoutFuture != null) { pongTimeoutFuture.cancel(true); } shutdownExecutorGracefully(sharedExecutor); log.info("OkxKlineWebSocketClient销毁完成"); } private static final String WS_URL_MONIPAN = "wss://wspap.okx.com:8443/ws/v5/business"; private static final String WS_URL_SHIPAN = "wss://ws.okx.com:8443/ws/v5/business"; private static final boolean isAccountType = true; /** * 建立与 OKX WebSocket 服务器的连接。 * 设置回调函数以监听连接打开、接收消息、关闭和错误事件。 */ private void connect() { // 避免重复连接 if (isConnecting.get()) { log.info("连接已在进行中，跳过重复连接请求"); return; } if (!isConnecting.compareAndSet(false, true)) { log.info("连接已在进行中，跳过重复连接请求"); return; } try { SSLConfig.configureSSL(); System.setProperty("https.protocols", "TLSv1.2,TLSv1.3"); String WS_URL = WS_URL_MONIPAN; if (isAccountType){ WS_URL = WS_URL_SHIPAN; } URI uri = new URI(WS_URL); // 关闭之前的连接（如果存在） if (webSocketClient != null) { try { webSocketClient.closeBlocking(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); log.warn("关闭之前连接时被中断"); } } webSocketClient = new WebSocketClient(uri) { @Override public void onOpen(ServerHandshake handshake) { log.info("OKX kline WebSocket连接成功"); isConnected.set(true); isConnecting.set(false); // 检查应用是否正在关闭 if (sharedExecutor != null && !sharedExecutor.isShutdown()) { resetHeartbeatTimer(); subscribeChannels(); } else { log.warn("应用正在关闭，忽略WebSocket连接成功回调"); } } @Override public void onMessage(String message) { lastMessageTime.set(System.currentTimeMillis()); handleWebSocketMessage(message); resetHeartbeatTimer(); } @Override public void onClose(int code, String reason, boolean remote) { log.warn("OKX kline WebSocket连接关闭: code={}, reason={}", code, reason); isConnected.set(false); isConnecting.set(false); cancelPongTimeout(); if (sharedExecutor != null && !sharedExecutor.isShutdown() && !sharedExecutor.isTerminated()) { sharedExecutor.execute(() -> { try { reconnectWithBackoff(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); log.error("重连线程被中断", e); } catch (Exception e) { log.error("重连失败", e); } }); } else { log.warn("共享线程池已关闭，无法执行重连任务"); } } @Override public void onError(Exception ex) { log.error("OKX New Price WebSocket发生错误", ex); isConnected.set(false); } }; webSocketClient.connect(); } catch (URISyntaxException e) { log.error("WebSocket URI格式错误", e); isConnecting.set(false); } } /** * 订阅指定交易对的价格通道。 * 构造订阅请求并发送给服务端。 */ private void subscribeChannels() { JSONObject subscribeMsg = new JSONObject(); subscribeMsg.put("op", "subscribe"); JSONArray argsArray = new JSONArray(); JSONObject arg = new JSONObject(); arg.put("channel", CHANNEL); arg.put("instId", CoinEnums.HE_YUE.getCode()); argsArray.add(arg); subscribeMsg.put("args", argsArray); webSocketClient.send(subscribeMsg.toJSONString()); log.info("已发送 K线频道订阅请求，订阅通道数: {}", argsArray.size()); } /** * 处理从 WebSocket 收到的消息。 * 包括订阅确认、错误响应、心跳响应以及实际的数据推送。 * * @param message 来自 WebSocket 的原始字符串消息 */ private void handleWebSocketMessage(String message) { try { JSONObject response = JSON.parseObject(message); String event = response.getString("event"); if ("subscribe".equals(event)) { log.info(" K线频道订阅成功: {}", response.getJSONObject("arg")); } else if ("error".equals(event)) { log.error(" K线频道订阅错误: code={}, msg={}", response.getString("code"), response.getString("msg")); } else if ("pong".equals(event)) { log.debug("收到pong响应"); cancelPongTimeout(); } else { processPushData(response); } } catch (Exception e) { log.error("处理WebSocket消息失败: {}", message, e); } } /** * 解析并处理价格推送数据。 * 将最新的标记价格存入 Redis 并触发后续业务逻辑比较处理。 * 当价格变化时，调用CaoZuoService的caoZuo方法，触发所有账号的量化操作 * * @param response 包含价格数据的 JSON 对象 */ private void processPushData(JSONObject response) { try { /** * { * "arg": { * "channel": "candle1D", * "instId": "BTC-USDT" * }, * "data": [ * [ * "1629993600000", * "42500", * "48199.9", * "41006.1", * "41006.1", * "3587.41204591", * "166741046.22583129", * "166741046.22583129", * "0" * ] * ] * } */ 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; } String instId = arg.getString("instId"); if (instId == null) { log.warn("{}: 无效的推送数据，缺少 'instId' 字段", response); return; } if (CHANNEL.equals(channel) && CoinEnums.HE_YUE.getCode().equals(instId)) { JSONArray dataArray = response.getJSONArray("data"); if (dataArray == null || dataArray.isEmpty()) { log.warn("K线频道数据为空"); return; } JSONArray data = dataArray.getJSONArray(0); BigDecimal openPx = new BigDecimal(data.getString(1)); BigDecimal highPx = new BigDecimal(data.getString(2)); BigDecimal lowPx = new BigDecimal(data.getString(3)); BigDecimal closePx = new BigDecimal(data.getString(4)); BigDecimal vol = new BigDecimal(data.getString(5)); /** * K线状态 * 0：K线未完结 * 1：K线已完结 */ String confirm = data.getString(8); if ("1".equals(confirm)){ //调用策略 // 创建交易策略 TradingStrategy tradingStrategy = new TradingStrategy(); // 生成100个15分钟价格数据点 List kline15MinuteData = getKlineDataByInstIdAndBar(instId, "15m"); //stream流获取kline15MinuteData中的o数据的集合 List prices = kline15MinuteData.stream() .map(Kline::getO) .collect(Collectors.toList()); // 生成对应的高、低、收盘价数据 List high = kline15MinuteData.stream() .map(Kline::getH) .collect(Collectors.toList()); List low = kline15MinuteData.stream() .map(Kline::getL) .collect(Collectors.toList()); List close = prices; // 生成成交量数据 List volume = kline15MinuteData.stream() .map(Kline::getVol) .collect(Collectors.toList()); // 获取最新价格 BigDecimal currentPrice = closePx; // 生成多周期价格数据（5分钟、1小时、4小时） List kline5MinuteData = getKlineDataByInstIdAndBar(instId, "5m"); List fiveMinPrices = kline5MinuteData.stream() .map(Kline::getC) .collect(Collectors.toList()); List kline60MinuteData = getKlineDataByInstIdAndBar(instId, "1H"); List oneHourPrices = kline60MinuteData.stream() .map(Kline::getC) .collect(Collectors.toList()); List kline240MinuteData = getKlineDataByInstIdAndBar(instId, "4H"); List fourHourPrices = kline240MinuteData.stream() .map(Kline::getC) .collect(Collectors.toList()); // 其他参数 BigDecimal fundingRate = new BigDecimal("0.001"); // 正常资金费率 boolean hasLargeTransfer = false; // 无大额转账 boolean hasUpcomingEvent = false; // 无即将到来的重大事件 // 确定市场方向 TradingStrategy.Direction direction = tradingStrategy.getDirection(prices, high, low, close, currentPrice); System.out.println("市场方向（15分钟）: " + direction); if (direction == TradingStrategy.Direction.RANGING){ return; } /** * 获取当前网格信息 * 根据当前网格的持仓方向获取反方向是否存在持仓 * 如果持有，直接止损 */ Collection allClients = clientManager.getAllClients(); //如果为空，则直接返回 if (allClients.isEmpty()) { return; } // 获取所有OkxQuantWebSocketClient实例 for (OkxQuantWebSocketClient client : clientManager.getAllClients()) { String accountName = client.getAccountName(); if (accountName != null) { TradingStrategy.SignalType signal = TradingStrategy.SignalType.NONE; TradeRequestParam tradeRequestParam = new TradeRequestParam(); // 检查当前持仓状态 boolean hasLongPosition = false; // 示例：无当前做多持仓 boolean hasShortPosition = false; // 示例：无当前做空持仓 //先判断账户是否有持多仓 String positionLongAccountName = PositionsWs.initAccountName(accountName, CoinEnums.POSSIDE_LONG.getCode()); BigDecimal imrLong = PositionsWs.getAccountMap(positionLongAccountName).get("imr"); if (imrLong != null && imrLong.compareTo(BigDecimal.ZERO) > 0){ log.info("账户{}有持多仓", accountName); hasLongPosition = true; } //先判断账户是否有持空仓 String positionShortAccountName = PositionsWs.initAccountName(accountName, CoinEnums.POSSIDE_LONG.getCode()); BigDecimal imrShort = PositionsWs.getAccountMap(positionShortAccountName).get("imr"); if (imrShort != null && imrShort.compareTo(BigDecimal.ZERO) > 0){ log.info("账户{}有持空仓", accountName); hasShortPosition = true; } signal = tradingStrategy.generateSignal(prices, high, low, close, volume, currentPrice, hasLongPosition, hasShortPosition, fiveMinPrices, oneHourPrices, fourHourPrices, fundingRate, hasLargeTransfer, hasUpcomingEvent); log.info("账户{}交易信号: " + signal, accountName); if (TradingStrategy.SignalType.NONE == signal) { continue; }else if (TradingStrategy.SignalType.BUY == signal){ tradeRequestParam = caoZuoService.caoZuoStrategy(accountName, String.valueOf(currentPrice), CoinEnums.POSSIDE_LONG.getCode()); tradeRequestParam.setSide(CoinEnums.SIDE_BUY.getCode()); String clOrdId = WsParamBuild.getOrderNum(CoinEnums.SIDE_BUY.getCode()); tradeRequestParam.setClOrdId(clOrdId); String sz = InstrumentsWs.getAccountMap(accountName).get(CoinEnums.BUY_CNT_INIT.name()); tradeRequestParam.setSz(sz); TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParam); }else if (TradingStrategy.SignalType.SELL == signal){ tradeRequestParam = caoZuoService.caoZuoStrategy(accountName, String.valueOf(currentPrice), CoinEnums.POSSIDE_SHORT.getCode()); tradeRequestParam.setSide(CoinEnums.SIDE_SELL.getCode()); String clOrdId = WsParamBuild.getOrderNum(CoinEnums.SIDE_SELL.getCode()); tradeRequestParam.setClOrdId(clOrdId); String sz = InstrumentsWs.getAccountMap(accountName).get(CoinEnums.BUY_CNT_INIT.name()); tradeRequestParam.setSz(sz); TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParam); }else if (TradingStrategy.SignalType.CLOSE_BUY == signal){ tradeRequestParam = caoZuoService.caoZuoStrategy(accountName, String.valueOf(currentPrice), CoinEnums.POSSIDE_LONG.getCode()); tradeRequestParam.setSide(CoinEnums.SIDE_SELL.getCode()); String clOrdId = WsParamBuild.getOrderNum(CoinEnums.SIDE_SELL.getCode()); tradeRequestParam.setClOrdId(clOrdId); BigDecimal pos = PositionsWs.getAccountMap(PositionsWs.initAccountName(accountName, CoinEnums.POSSIDE_LONG.getCode())).get("pos"); if (BigDecimal.ZERO.compareTo( pos) >= 0) { tradeRequestParam.setTradeType(OrderParamEnums.TRADE_NO.getValue()); } tradeRequestParam.setSz(String.valueOf( pos)); TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParam); }else if (TradingStrategy.SignalType.CLOSE_SELL == signal){ tradeRequestParam = caoZuoService.caoZuoStrategy(accountName, String.valueOf(currentPrice), CoinEnums.POSSIDE_SHORT.getCode()); tradeRequestParam.setSide(CoinEnums.SIDE_BUY.getCode()); String clOrdId = WsParamBuild.getOrderNum(CoinEnums.SIDE_BUY.getCode()); tradeRequestParam.setClOrdId(clOrdId); BigDecimal pos = PositionsWs.getAccountMap(PositionsWs.initAccountName(accountName, CoinEnums.POSSIDE_SHORT.getCode())).get("pos"); if (BigDecimal.ZERO.compareTo( pos) >= 0) { tradeRequestParam.setTradeType(OrderParamEnums.TRADE_NO.getValue()); } tradeRequestParam.setSz(String.valueOf( pos)); TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParam); } } } } } } catch (Exception e) { log.error("处理 K线频道推送数据失败", e); } } /** * 触发所有账号的量化操作 * @param markPx 当前标记价格 */ private void triggerQuantOperations(String markPx) { try { // 1. 判断当前价格属于哪个网格 WangGeListEnum gridByPriceNew = WangGeListEnum.getGridByPrice(new BigDecimal(markPx)); if (gridByPriceNew == null) { log.error("当前 K线频道{}不在任何网格范围内，无法触发量化操作", markPx); return; } } catch (Exception e) { log.error("触发量化操作失败", e); } } private List getKlineDataByInstIdAndBar(String instId, String bar) { LinkedHashMap requestParam = new LinkedHashMap<>(); requestParam.put("instId",instId); requestParam.put("bar",bar); requestParam.put("limit","100"); String result = ExchangeLoginService.getInstance(ExchangeInfoEnum.OKX_UAT.name()).lineHistory(requestParam); log.info("加载OKX-KLINE，{}", result); JSONObject json = JSON.parseObject(result); String data = json.getString("data"); List klinesList = JSON.parseArray(data, String[].class); if(CollUtil.isEmpty(klinesList)){ return null; } ArrayList objects = new ArrayList<>(); for(String[] s : klinesList) { Kline kline = new Kline(); kline.setTs(s[0]); kline.setO(new BigDecimal(s[1])); kline.setH(new BigDecimal(s[2])); kline.setL(new BigDecimal(s[3])); kline.setC(new BigDecimal(s[4])); kline.setVol(new BigDecimal(s[5])); kline.setConfirm(s[8]); objects.add(kline); } return objects; } /** * 构建 Redis Key */ private String buildRedisKey(String instId) { return "PRICE_" + instId.replace("-", ""); } /** * 启动心跳检测任务。 * 使用 ScheduledExecutorService 定期检查是否需要发送 ping 请求来维持连接。 */ private void startHeartbeat() { if (heartbeatExecutor != null && !heartbeatExecutor.isTerminated()) { heartbeatExecutor.shutdownNow(); } heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(r -> { Thread t = new Thread(r, "okx-kline-heartbeat"); t.setDaemon(true); return t; }); heartbeatExecutor.scheduleWithFixedDelay(this::checkHeartbeatTimeout, 25, 25, TimeUnit.SECONDS); } /** * 重置心跳计时器。 * 当收到新消息或发送 ping 后取消当前超时任务并重新安排下一次超时检查。 */ private synchronized void resetHeartbeatTimer() { cancelPongTimeout(); if (heartbeatExecutor != null && !heartbeatExecutor.isShutdown()) { pongTimeoutFuture = heartbeatExecutor.schedule(this::checkHeartbeatTimeout, HEARTBEAT_TIMEOUT, TimeUnit.SECONDS); } } /** * 检查心跳超时情况。 * 若长时间未收到任何消息则主动发送 ping 请求保持连接活跃。 */ private void checkHeartbeatTimeout() { // 只有在连接状态下才检查心跳 if (!isConnected.get()) { return; } long currentTime = System.currentTimeMillis(); long lastTime = lastMessageTime.get(); if (currentTime - lastTime >= HEARTBEAT_TIMEOUT * 1000L) { sendPing(); } } /** * 发送 ping 请求至 WebSocket 服务端。 * 用于维持长连接有效性。 */ private void sendPing() { try { if (webSocketClient != null && webSocketClient.isOpen()) { JSONObject ping = new JSONObject(); ping.put("op", "ping"); webSocketClient.send(ping.toJSONString()); log.debug("发送ping请求"); } } catch (Exception e) { log.warn("发送ping失败", e); } } /** * 取消当前的心跳超时任务。 * 在收到 pong 或其他有效消息时调用此方法避免不必要的断开重连。 */ private synchronized void cancelPongTimeout() { if (pongTimeoutFuture != null && !pongTimeoutFuture.isDone()) { pongTimeoutFuture.cancel(true); } } /** * 执行 WebSocket 重连操作。 * 在连接意外中断后尝试重新建立连接。 */ private void reconnectWithBackoff() throws InterruptedException { int attempt = 0; int maxAttempts = 3; long delayMs = 5000; while (attempt < maxAttempts) { try { Thread.sleep(delayMs); connect(); return; } catch (Exception e) { log.warn("第{}次重连失败", attempt + 1, e); delayMs *= 2; attempt++; } } log.error("超过最大重试次数({})仍未连接成功", maxAttempts); } /** * 优雅关闭线程池 */ private void shutdownExecutorGracefully(ExecutorService executor) { if (executor == null || executor.isTerminated()) { return; } try { executor.shutdown(); if (!executor.awaitTermination(5, TimeUnit.SECONDS)) { executor.shutdownNow(); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); executor.shutdownNow(); } } }