feat(okxNewPrice): 切换WebSocket数据源为K线频道并优化数据处理逻辑

Administrator

2025-12-29 3bec00212427e5e03fc386c2b28a0071359eb003

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

@@ -57,8 +57,8 @@
    private final AtomicBoolean isConnecting = new AtomicBoolean(false);
    private final AtomicBoolean isInitialized = new AtomicBoolean(false);

//    private static final String CHANNEL = "mark-price";
    private static final String CHANNEL = "candle1m";
//    private static final String CHANNEL = "candle5m";
//    private static final String CHANNEL = "candle15m";

    // 心跳超时时间(秒)，小于30秒
@@ -123,7 +123,7 @@

    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;
    private static final boolean isAccountType = false;

    /**
     * 建立与 OKX WebSocket 服务器的连接。
@@ -343,11 +343,10 @@
                    List<BigDecimal> historicalPrices = kline15MinuteData.stream()
                            .map(Kline::getC)
                            .collect(Collectors.toList());
                    log.info("生成100个15分钟价格数据点成功！");
                    log.info("生成100个1分钟价格数据点成功！");
                    // 使用策略分析最新价格数据
                    MacdMaStrategy.TradingOrder tradingOrderOpen = strategy.generateTradingOrder(historicalPrices,MacdMaStrategy.OperationType.open.name());
                    MacdMaStrategy.TradingOrder tradingOrderClose = strategy.generateTradingOrder(historicalPrices,MacdMaStrategy.OperationType.close.name());
                    if (tradingOrderOpen == null && tradingOrderClose == null){
                    if (tradingOrderOpen == null ){
                        return;
                    }
                    Collection<OkxQuantWebSocketClient> allClients = clientManager.getAllClients();
@@ -378,18 +377,6 @@
                                tradeRequestParam.setSz(sz);
                                TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParam);
                            }
                            if (ObjectUtil.isNotEmpty(tradingOrderClose)){
                                // 根据信号执行交易操作
                                TradeRequestParam tradeRequestParam = new TradeRequestParam();

                                String posSide = tradingOrderClose.getPosSide();
                                tradeRequestParam.setPosSide(posSide);
                                String currentPrice = String.valueOf(closePx);
                                tradeRequestParam = caoZuoService.caoZuoZhiSunEvent(accountName, currentPrice, posSide);

                                TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParam);
                            }

                        }
                    }
                }
@@ -471,7 +458,7 @@
                    TradingStrategy tradingStrategy = new TradingStrategy();

                    // 生成100个15分钟价格数据点
                    List<Kline> kline15MinuteData = getKlineDataByInstIdAndBar(instId, "15m");
                    List<Kline> kline15MinuteData = getKlineDataByInstIdAndBar(instId, "1m");
                    //stream流获取kline15MinuteData中的o数据的集合
                    List<BigDecimal> prices = kline15MinuteData.stream()
                            .map(Kline::getC)