refactor(gateApi): 重构网格交易服务添加补仓重试机制

Administrator

4 days ago 01da1f754426a1285fc20b9cf1b80672b16d8814

 src/main/java/com/xcong/excoin/modules/gateApi/wsHandler/handler/CandlestickChannelHandler.java

@@ -11,8 +11,21 @@
import java.math.BigDecimal;

/**
 * K 线频道处理器。
 * 公开频道，无需认证。订阅 1m K 线，解析后回调 {@link GateGridTradeService#onKline}。
 * K 线（Candlestick）频道处理器。
 *
 * <h3>特点</h3>
 * 公开频道，无需 HMAC-SHA512 认证签名。订阅 1m 周期的 K 线数据。
 *
 * <h3>数据流</h3>
 * <pre>
 *   WebSocket 推送 update event
 *     → handleMessage() → 解析 OHLCV → log 打印 → gridTradeService.onKline(closePx)
 *       → 首次 K 线触发双开
 *       → 后续 K 线仅缓存 lastKlinePrice 供补仓参考
 * </pre>
 *
 * <h3>订阅格式</h3>
 * payload: {@code ["1m", contract]}
 *
 * @author Administrator
 */
@@ -31,73 +44,59 @@
    }

    @Override
    public String getChannelName() {
        return CHANNEL_NAME;
    }
    public String getChannelName() { return CHANNEL_NAME; }

    @Override
    public void subscribe(WebSocketClient ws) {
        JSONObject subscribeMsg = new JSONObject();
        subscribeMsg.put("time", System.currentTimeMillis() / 1000);
        subscribeMsg.put("channel", CHANNEL_NAME);
        subscribeMsg.put("event", "subscribe");
        JSONObject msg = new JSONObject();
        msg.put("time", System.currentTimeMillis() / 1000);
        msg.put("channel", CHANNEL_NAME);
        msg.put("event", "subscribe");
        JSONArray payload = new JSONArray();
        payload.add(INTERVAL);
        payload.add(contract);
        subscribeMsg.put("payload", payload);
        ws.send(subscribeMsg.toJSONString());
        log.info("[{}] 已发送订阅请求，合约: {}, 周期: {}", CHANNEL_NAME, contract, INTERVAL);
        msg.put("payload", payload);
        ws.send(msg.toJSONString());
        log.info("[{}] 订阅成功, 合约:{}, 周期:{}", CHANNEL_NAME, contract, INTERVAL);
    }

    @Override
    public void unsubscribe(WebSocketClient ws) {
        JSONObject unsubscribeMsg = new JSONObject();
        unsubscribeMsg.put("time", System.currentTimeMillis() / 1000);
        unsubscribeMsg.put("channel", CHANNEL_NAME);
        unsubscribeMsg.put("event", "unsubscribe");
        JSONObject msg = new JSONObject();
        msg.put("time", System.currentTimeMillis() / 1000);
        msg.put("channel", CHANNEL_NAME);
        msg.put("event", "unsubscribe");
        JSONArray payload = new JSONArray();
        payload.add(INTERVAL);
        payload.add(contract);
        unsubscribeMsg.put("payload", payload);
        ws.send(unsubscribeMsg.toJSONString());
        log.info("[{}] 已发送取消订阅请求，合约: {}, 周期: {}", CHANNEL_NAME, contract, INTERVAL);
        msg.put("payload", payload);
        ws.send(msg.toJSONString());
        log.info("[{}] 取消订阅成功", CHANNEL_NAME);
    }

    @Override
    public boolean handleMessage(JSONObject response) {
        String channel = response.getString("channel");
        if (!CHANNEL_NAME.equals(channel)) {
        if (!CHANNEL_NAME.equals(response.getString("channel"))) {
            return false;
        }
        try {
            JSONArray resultArray = response.getJSONArray("result");
            if (resultArray == null || resultArray.isEmpty()) {
                log.warn("[{}] 数据为空", CHANNEL_NAME);
                return true;
            }
            if (resultArray == null || resultArray.isEmpty()) { log.warn("[{}] 数据为空", CHANNEL_NAME); return true; }
            JSONObject data = resultArray.getJSONObject(0);
            BigDecimal closePx = new BigDecimal(data.getString("c"));
            long t = data.getLong("t");
            boolean windowClosed = data.getBooleanValue("w");

            log.info("========== Gate K线数据 ==========");
            log.info("名称(n): {}", data.getString("n"));
            log.info("时间  : {}", DateUtil.TimeStampToDateTime(t));
            log.info("开盘(o): {}", data.getString("o"));
            log.info("最高(h): {}", data.getString("h"));
            log.info("最低(l): {}", data.getString("l"));
            log.info("收盘(c): {}", data.getString("c"));
            log.info("成交量(v): {}", data.getString("v"));
            log.info("成交额(a): {}", data.getString("a"));
            log.info("K线完结(w): {}", windowClosed);
            log.info("名称: {} 时间: {}", data.getString("n"), DateUtil.TimeStampToDateTime(data.getLong("t")));
            log.info("开盘: {} 最高: {} 最低: {} 收盘: {} 成交量: {} 成交额: {} 已完结: {}",
                    data.getString("o"), data.getString("h"), data.getString("l"),
                    data.getString("c"), data.getString("v"), data.getString("a"),
                    data.getBooleanValue("w"));
            log.info("==================================");

            if (gridTradeService != null) {
                gridTradeService.onKline(closePx);
            }
        } catch (Exception e) {
            log.error("[{}] 处理数据失败", CHANNEL_NAME, e);
        }
        } catch (Exception e) { log.error("[{}] 处理数据失败", CHANNEL_NAME, e); }
        return true;
    }
}