exchange.git - Gitblit

			@@ -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,156 @@
			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, "1D");
			List<BigDecimal> fiveMinPrices = kline15MinuteData.stream()
			.map(Kline::getC)
			.collect(Collectors.toList());
			try {
			// 执行策略
			strategy.execute(fiveMinPrices);
			boolean skip = strategy.getSkip();
			if (skip){
			log.info("跳过");
			return;
			}

			System.out.println("策略初始化成功！");
			} catch (Exception e) {
			System.err.println("策略初始化失败：" + e.getMessage());
			e.printStackTrace();
			}

			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();
			tradeRequestParam = strategy.getOrderParamOpen(tradeRequestParam);
			String posSide = tradeRequestParam.getPosSide();
			String side = tradeRequestParam.getSide();
			BigDecimal posHold = PositionsWs.getAccountMap(PositionsWs.initAccountName(accountName, posSide)).get("pos");
			if (posHold.compareTo(BigDecimal.ZERO) > 0){
			tradeRequestParam = strategy.getOrderParamClose(tradeRequestParam);
			}
			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()){
			if (side == CoinEnums.SIDE_BUY.getCode()){
			sz = InstrumentsWs.getAccountMap(accountName).get(CoinEnums.BUY_CNT_INIT.name());
			}else if (side == CoinEnums.SIDE_SELL.getCode()){
			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());
			}
			sz = String.valueOf( pos);
			}
			}else if (posSide == CoinEnums.POSSIDE_SHORT.getCode()){
			if (side == CoinEnums.SIDE_BUY.getCode()){
			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());
			}
			sz = String.valueOf( pos);
			}else if (side == CoinEnums.SIDE_SELL.getCode()){
			sz = InstrumentsWs.getAccountMap(accountName).get(CoinEnums.BUY_CNT_INIT.name());
			}
			}
			tradeRequestParam.setSz(sz);
			TradeOrderWs.orderEvent(client.getWebSocketClient(), tradeRequestParam);
			}
			}
			}
			}
			} catch (Exception e) {
			log.error("处理 K线频道推送数据失败", e);
			}
			}

			@@ -498,7 +645,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);