fix(okxNewPrice): 解决交易订单为空时的空指针异常

Administrator

2025-12-26 4884ebffeed775d63dbf7bb7d0ece8182f10c344

 src/main/java/com/xcong/excoin/modules/okxNewPrice/OkxKlineWebSocketClient.java

@@ -7,6 +7,7 @@
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.indicator.macdAndMatrategy.MacdMaStrategy;
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;
@@ -256,10 +257,141 @@
                log.debug("收到pong响应");
                cancelPongTimeout();
            } else {
                processPushData(response);
//                processPushData(response);
                processPushDataV2(response);
            }
        } catch (Exception e) {
            log.error("处理WebSocket消息失败: {}", message, e);
        }
    }

    /**
     * 解析并处理价格推送数据。
     * 将最新的标记价格存入 Redis 并触发后续业务逻辑比较处理。
     * 当价格变化时，调用CaoZuoService的caoZuo方法，触发所有账号的量化操作
     *
     * @param response 包含价格数据的 JSON 对象
     */
    private void processPushDataV2(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)){
                    //调用策略
                    // 创建策略实例
                    MacdMaStrategy strategy = new MacdMaStrategy();

                    // 生成100个15分钟价格数据点
                    List<Kline> kline15MinuteData = getKlineDataByInstIdAndBar(instId, "15m");
                    List<BigDecimal> historicalPrices = kline15MinuteData.stream()
                            .map(Kline::getC)
                            .collect(Collectors.toList());
                    log.info("生成100个15分钟价格数据点成功！");
                    // 使用策略分析最新价格数据
                    MacdMaStrategy.TradingOrder tradingOrder = strategy.generateTradingOrder(historicalPrices);
                    if (tradingOrder == null){
                        return;
                    }
                    Collection<OkxQuantWebSocketClient> allClients = clientManager.getAllClients();
                    //如果为空，则直接返回
                    if (allClients.isEmpty()) {
                        return;
                    }
                    // 获取所有OkxQuantWebSocketClient实例
                    for (OkxQuantWebSocketClient client : clientManager.getAllClients()) {
                        String accountName = client.getAccountName();
                        if (accountName != null) {
                            // 根据信号执行交易操作
                            TradeRequestParam tradeRequestParam = new TradeRequestParam();
                            String posSide = tradingOrder.getPosSide();
                            String side = tradingOrder.getSide();
                            String currentPrice = String.valueOf(closePx);
                            tradeRequestParam = caoZuoService.caoZuoStrategy(accountName, currentPrice, posSide);
                            String clOrdId = WsParamBuild.getOrderNum(side);
                            tradeRequestParam.setClOrdId(clOrdId);
                            String sz = null;
                            if (posSide == CoinEnums.POSSIDE_LONG.getCode() && side == CoinEnums.SIDE_BUY.getCode()){
                                sz = InstrumentsWs.getAccountMap(accountName).get(CoinEnums.BUY_CNT_INIT.name());
                                tradeRequestParam.setSz(sz);
                                TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParam);

                                BigDecimal pos = PositionsWs.getAccountMap(PositionsWs.initAccountName(accountName, CoinEnums.POSSIDE_SHORT.getCode())).get("pos");
                                if (BigDecimal.ZERO.compareTo( pos) >= 0) {
                                    TradeRequestParam tradeRequestParamOld = caoZuoService.caoZuoZhiSunEvent(accountName, String.valueOf(closePx), CoinEnums.POSSIDE_SHORT.getCode());
                                    TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParamOld);
                                }
                            }else if (posSide == CoinEnums.POSSIDE_SHORT.getCode() && side == CoinEnums.SIDE_SELL.getCode()){
                                sz = InstrumentsWs.getAccountMap(accountName).get(CoinEnums.BUY_CNT_INIT.name());
                                tradeRequestParam.setSz(sz);
                                TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParam);

                                BigDecimal pos = PositionsWs.getAccountMap(PositionsWs.initAccountName(accountName, CoinEnums.POSSIDE_LONG.getCode())).get("pos");
                                if (BigDecimal.ZERO.compareTo( pos) >= 0) {
                                    TradeRequestParam tradeRequestParamOld = caoZuoService.caoZuoZhiSunEvent(accountName, String.valueOf(closePx), CoinEnums.POSSIDE_LONG.getCode());
                                    TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParamOld);
                                }
                            }
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("处理 K线频道推送数据失败", e);
        }
    }

@@ -498,7 +630,7 @@
            LinkedHashMap<String, Object> requestParam = new LinkedHashMap<>();
            requestParam.put("instId", instId);
            requestParam.put("bar", bar);
            requestParam.put("limit", "100");
            requestParam.put("limit", "200");
            String result = ExchangeLoginService.getInstance(ExchangeInfoEnum.OKX_UAT.name()).lineHistory(requestParam);
            log.info("加载OKX-KLINE，{}", result);