feat(gate): 添加策略远程控制和监控功能

Administrator

2026-08-13 c53a7cb56c5bc3e9343e7b74758742257bf8f38a

feat(gate): 添加策略远程控制和监控功能

- 新增指令确认消息模型 CmdAckMsg 用于响应 Station 指令
- 实现 CommandQueueInitializer 动态创建专属命令队列 QUEUE_GATE_CMD_{apiKeyMd5}
- 添加 GateCommand 模型定义 Station 下发的启停和参数更新指令
- 创建 GateCommandConsumer 消费远程指令并执行相应操作
- 扩展 GateConfigDTO 添加参数快照和默认值工厂方法
- 在 GateGridTradeService 中集成埋点功能发送策略事件统计
- 新增 GateStatsEvent 模型定义 JAR 上报的策略事件格式
- 更新 GateWebSocketClientManager 支持远程配置更新和重载启动
- 添加 HeartbeatMsg 模型和 HeartbeatScheduler 定时发送心跳
- 创建 InstanceInfo 模型存储 JAR 实例注册信息
- 配置 RabbitMQ 添加 Gate 管理交换机和相关队列绑定
- 实现 StatsEventProducer 统一消息发送器处理心跳和事件上报

 config/gate-config-7332fa3f94276d3dcc44fcd3bfff2379.json

New file
@@ -0,0 +1,28 @@
{
  "apiKey" : null,
  "contract" : null,
  "leverage" : null,
  "marginMode" : null,
  "positionMode" : null,
  "gridRate" : 0.005,
  "expectedProfit" : 0.15,
  "maxLoss" : 1.5,
  "baseQuantity" : "2",
  "quantity" : "2",
  "maxPositionSize" : 8,
  "stopLossCount" : 0,
  "takeProfitGridSpan" : 2,
  "restartGridSpan" : 0,
  "priceScale" : 0,
  "contractMultiplier" : null,
  "unrealizedPnlPriceMode" : null,
  "priceDriveEnabled" : true,
  "rounds" : 0,
  "stopLossCountMode" : "dual",
  "addPositionInterval" : 3,
  "addPositionQuantity" : 1,
  "maxPositionPerSide" : 0,
  "addPositionStartThreshold" : 1,
  "placeExcessTakeProfit" : false,
  "production" : false
}

 src/main/java/com/xcong/excoin/configurations/RabbitMqConfig.java

@@ -72,6 +72,18 @@
    // 平仓路由
    public static final String ROUTINGKEY_CLOSETRADE = "ROUTINGKEY_CLOSETRADE";

    // ==================== Gate 策略管理 ====================
    /** Gate 管理交换机 */
    public static final String EXCHANGE_GATE = "biue-exchange-gate";
    /** 心跳+事件+确认 队列（Station 独占消费） */
    public static final String QUEUE_GATE_HEARTBEAT = "QUEUE_GATE_HEARTBEAT";
    /** 心跳+事件+确认 路由键 */
    public static final String ROUTINGKEY_GATE_HEARTBEAT = "ROUTINGKEY_GATE_HEARTBEAT";
    /** 策略事件队列（Station 消费落库） */
    public static final String QUEUE_GATE_STATS = "QUEUE_GATE_STATS";
    /** 策略事件路由键 */
    public static final String ROUTINGKEY_GATE_STATS = "ROUTINGKEY_GATE_STATS";

    @Resource
    private ConnectionFactory connectionFactory;

@@ -286,4 +298,31 @@
        return BindingBuilder.bind(queueCloseTrade()).to(orderExchange()).with(RabbitMqConfig.ROUTINGKEY_CLOSETRADE);
    }

    // ==================== Gate 管理队列 ====================

    @Bean
    public DirectExchange gateExchange() {
        return new DirectExchange(EXCHANGE_GATE);
    }

    @Bean
    public Queue gateHeartbeatQueue() {
        return new Queue(QUEUE_GATE_HEARTBEAT, true);
    }

    @Bean
    public Binding bindingGateHeartbeat() {
        return BindingBuilder.bind(gateHeartbeatQueue()).to(gateExchange()).with(ROUTINGKEY_GATE_HEARTBEAT);
    }

    @Bean
    public Queue gateStatsQueue() {
        return new Queue(QUEUE_GATE_STATS, true);
    }

    @Bean
    public Binding bindingGateStats() {
        return BindingBuilder.bind(gateStatsQueue()).to(gateExchange()).with(ROUTINGKEY_GATE_STATS);
    }

}

 src/main/java/com/xcong/excoin/modules/gateApi/CommandQueueInitializer.java

New file
@@ -0,0 +1,70 @@
package com.xcong.excoin.modules.gateApi;

import com.xcong.excoin.configurations.RabbitMqConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.nio.charset.StandardCharsets;

/**
 * JAR 启动时动态创建专属命令队列 QUEUE_GATE_CMD_{apiKeyMd5}
 */
@Slf4j
@Component
@DependsOn("gateWebSocketClientManager")
public class CommandQueueInitializer {

    @Resource
    private AmqpAdmin amqpAdmin;

    @Resource
    private GateWebSocketClientManager manager;

    private volatile String queueName;

    @PostConstruct
    public void init() {
        try {
            // manager.config 此时已由 Manager 的 @PostConstruct 加载完成
            String apiKey = manager.getConfig().getApiKey();
            String apiKeyMd5 = md5(apiKey);
            String routingKey = "cmd." + apiKeyMd5;
            queueName = "QUEUE_GATE_CMD_" + apiKeyMd5;

            DirectExchange exchange = new DirectExchange(RabbitMqConfig.EXCHANGE_GATE);
            Queue queue = new Queue(queueName, true, false, true); // durable, non-exclusive, auto-delete
            Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey);

            amqpAdmin.declareQueue(queue);
            amqpAdmin.declareBinding(binding);

            log.info("[Gate] 命令队列已注册, queue={}, routingKey={}", queueName, routingKey);
        } catch (Exception e) {
            log.error("[Gate] 命令队列注册失败, 策略启停指令将无法接收", e);
            queueName = null;
        }
    }

    /** 返回队列名，供 @RabbitListener 引用 */
    public String getQueueName() {
        return queueName;
    }

    private static String md5(String input) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] digest = md.digest(input.getBytes(StandardCharsets.UTF_8));
            StringBuilder sb = new StringBuilder();
            for (byte b : digest) sb.append(String.format("%02x", b));
            return sb.toString();
        } catch (NoSuchAlgorithmException e) {
            return Integer.toHexString(input.hashCode());
        }
    }
}

 src/main/java/com/xcong/excoin/modules/gateApi/GateCommandConsumer.java

New file
@@ -0,0 +1,111 @@
package com.xcong.excoin.modules.gateApi;

import com.alibaba.fastjson.JSON;
import com.xcong.excoin.modules.station.model.CmdAckMsg;
import com.xcong.excoin.modules.station.model.GateCommand;
import com.xcong.excoin.modules.station.model.GateStatsEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;

/**
 * JAR 侧 — 消费 Station 下发的启停指令
 */
@Slf4j
@Component
public class GateCommandConsumer {

    @Resource
    private StatsEventProducer statsProducer;

    @Resource
    private GateWebSocketClientManager manager;

    /**
     * 监听专属命令队列 QUEUE_GATE_CMD_{apiKeyMd5}
     */
    @RabbitListener(queues = "#{commandQueueInitializer.queueName}")
    public void onCommand(String msg) {
        GateCommand cmd;
        try {
            cmd = JSON.parseObject(msg, GateCommand.class);
        } catch (Exception e) {
            log.error("[Gate] 指令解析失败: {}", msg, e);
            return;
        }

        String apiKeyMd5 = md5(manager.getConfig().getApiKey());
        boolean success = true;
        String newState = "";
        String message = "";

        GateGridTradeService strategy = manager.getGridTradeService();
        try {
            switch (cmd.getCommandType()) {
                case "START":
                    manager.reloadAndStart();
                    newState = "ACTIVE";
                    message = "策略已启动（已加载最新配置）";
                    log.info("[Gate] 远程启动, 已重载配置");
                    break;

                case "STOP":
                    strategy.stopGrid();
                    newState = strategy.getState().name();
                    message = "策略已停止";
                    log.info("[Gate] 远程停止, state={}", newState);
                    break;

                case "UPDATE_CONFIG":
                    if (!StringUtils.hasText(cmd.getPayload())) {
                        success = false;
                        message = "UPDATE_CONFIG 缺少 payload";
                        log.warn("[Gate] UPDATE_CONFIG 指令缺少 payload");
                        break;
                    }
                    GateConfigDTO dto = JSON.parseObject(cmd.getPayload(), GateConfigDTO.class);
                    manager.updateConfig(dto);
                    newState = "ACTIVE";
                    message = "参数已持久化，待下次启动生效";
                    log.info("[Gate] 远程参数已持久化");
                    break;

                default:
                    success = false;
                    message = "未知指令类型: " + cmd.getCommandType();
                    log.warn("[Gate] 未知指令: {}", cmd.getCommandType());
            }
        } catch (Exception e) {
            success = false;
            newState = "ERROR";
            message = e.getMessage();
            log.error("[Gate] 指令执行失败", e);
        }

        // 发送 ACK 到 Station
        CmdAckMsg ack = CmdAckMsg.builder()
                .commandId(cmd.getCommandId())
                .success(success)
                .newState(newState)
                .message(message)
                .build();
        GateStatsEvent event = statsProducer.newCmdAck(apiKeyMd5, ack);
        statsProducer.sendHeartbeat(event);
        log.debug("[Gate] ACK 已发送, cmdId={}, success={}", cmd.getCommandId(), success);
    }

    private static String md5(String input) {
        try {
            java.security.MessageDigest md = java.security.MessageDigest.getInstance("MD5");
            byte[] digest = md.digest(input.getBytes(java.nio.charset.StandardCharsets.UTF_8));
            StringBuilder sb = new StringBuilder();
            for (byte b : digest) sb.append(String.format("%02x", b));
            return sb.toString();
        } catch (Exception e) {
            return Integer.toHexString(input.hashCode());
        }
    }
}

 src/main/java/com/xcong/excoin/modules/gateApi/GateConfigDTO.java

@@ -6,6 +6,8 @@
import lombok.NoArgsConstructor;

import java.math.BigDecimal;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 * Gate 策略配置 DTO，用于 Web 控制面板参数传递。
@@ -104,4 +106,65 @@
                .placeExcessTakeProfit(config.isPlaceExcessTakeProfit())
                .build();
    }

    /**
     * 所有可调参数的扁平快照 — stats 埋点 + STRATEGY_START payload 用。
     * 新增参数只需在此方法加一行，无需修改 stats 消费者。
     */
    public Map<String, Object> toParamsMap() {
        Map<String, Object> m = new LinkedHashMap<>();
        m.put("contract", contract);
        m.put("leverage", leverage);
        m.put("gridRate", gridRate);
        m.put("expectedProfit", expectedProfit);
        m.put("maxLoss", maxLoss);
        m.put("baseQuantity", baseQuantity);
        m.put("quantity", quantity);
        m.put("maxPositionSize", maxPositionSize);
        m.put("stopLossCount", stopLossCount);
        m.put("takeProfitGridSpan", takeProfitGridSpan);
        m.put("priceDriveEnabled", priceDriveEnabled);
        m.put("rounds", rounds);
        m.put("stopLossCountMode", stopLossCountMode);
        m.put("addPositionInterval", addPositionInterval);
        m.put("addPositionQuantity", addPositionQuantity);
        m.put("maxPositionPerSide", maxPositionPerSide);
        m.put("addPositionStartThreshold", addPositionStartThreshold);
        m.put("placeExcessTakeProfit", placeExcessTakeProfit);
        return m;
    }

    /**
     * 带默认值的工厂方法 — 统一 Manager 和 HTML 的默认值入口。
     */
    public static GateConfigDTO defaultsFor(String apiKey) {
        return GateConfigDTO.builder()
                .apiKey(apiKey)
                .contract("ETH_USDT")
                .leverage("100")
                .marginMode("cross")
                .positionMode("dual")
                .gridRate(new BigDecimal("0.005"))
                .expectedProfit(new BigDecimal("0.15"))
                .maxLoss(new BigDecimal("1.5"))
                .baseQuantity("2")
                .quantity("2")
                .maxPositionSize(4)
                .stopLossCount(0)
                .takeProfitGridSpan(2)
                .restartGridSpan(0)
                .priceScale(1)
                .contractMultiplier(new BigDecimal("0.001"))
                .unrealizedPnlPriceMode("LAST_PRICE")
                .isProduction(true)
                .priceDriveEnabled(true)
                .rounds(0)
                .stopLossCountMode("dual")
                .addPositionInterval(3)
                .addPositionQuantity(1)
                .maxPositionPerSide(0)
                .addPositionStartThreshold(1)
                .placeExcessTakeProfit(false)
                .build();
    }
}

 src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java

@@ -103,6 +103,8 @@
    private static final String ORDER_TYPE_CLOSE_SHORT = "plan-close-short-position";

    private final GateConfig config;
    private final StatsEventProducer statsProducer;
    private String apiKeyMd5;
    private final GateTradeExecutor executor;
    private final FuturesApi futuresApi;
    private static final String SETTLE = "usdt";
@@ -157,10 +159,13 @@
    private volatile BigDecimal shortPositionSize = BigDecimal.ZERO;
    private Long userId;
    private volatile BigDecimal initialPrincipal = BigDecimal.ZERO;
    /** 上次 PNL 快照时间（毫秒），用于控制 PNL_SNAPSHOT 埋点频率 */
    private volatile long lastPnlSnapshotTime = 0;
    private volatile GateKlineWebSocketClient wsClient;

    public GateGridTradeService(GateConfig config) {
    public GateGridTradeService(GateConfig config, StatsEventProducer statsProducer) {
        this.config = config;
        this.statsProducer = statsProducer;
        ApiClient apiClient = new ApiClient();
        apiClient.setBasePath(config.getRestBasePath());
        apiClient.setApiKeySecret(config.getApiKey(), config.getApiSecret());
@@ -290,6 +295,44 @@
        }
    }

    // ---- 埋点 ----

    private String apiKeyMd5() {
        if (apiKeyMd5 == null) {
            try {
                java.security.MessageDigest md = java.security.MessageDigest.getInstance("MD5");
                byte[] digest = md.digest(config.getApiKey().getBytes(java.nio.charset.StandardCharsets.UTF_8));
                StringBuilder sb = new StringBuilder();
                for (byte b : digest) sb.append(String.format("%02x", b));
                apiKeyMd5 = sb.toString();
            } catch (Exception e) {
                apiKeyMd5 = Integer.toHexString(config.getApiKey().hashCode());
            }
        }
        return apiKeyMd5;
    }

    private void emitStats(String type, Object payload) {
        if (statsProducer == null) return;
        try {
            statsProducer.sendStats(statsProducer.newStats(type, apiKeyMd5(), payload));
        } catch (Exception e) {
            log.warn("[Gate] 埋点发送失败, type={}", type, e);
        }
    }

    /**
     * Java 8 兼容的 Map 构造工具（Map.of 为 Java 9 API，此处手动实现）。
     * 接受偶数个参数：key1, value1, key2, value2, ...
     */
    private static Map<String, Object> mapOf(Object... kv) {
        Map<String, Object> m = new LinkedHashMap<>();
        for (int i = 0; i < kv.length; i += 2) {
            m.put((String) kv[i], kv[i + 1]);
        }
        return m;
    }

    // ---- 启动/停止 ----

    /**
@@ -328,6 +371,13 @@
        currentRound = 0;

        log.info("[Gate] 网格策略已启动, 当前本金: {} USDT", initialPrincipal);

        // 埋点: STRATEGY_START — 附全量配置快照
        GateConfigDTO snapshot = GateConfigDTO.from(config);
        Map<String, Object> params = snapshot.toParamsMap();
        params.put("principal", initialPrincipal.toPlainString());
        params.put("contract", config.getContract());
        emitStats("STRATEGY_START", params);
    }

    /**
@@ -348,6 +398,14 @@
     */
    public void stopGrid() {
        state = StrategyState.STOPPED;

        // 埋点: STRATEGY_STOP
        emitStats("STRATEGY_STOP", mapOf(
                "reason", "manual",
                "rounds", currentRound,
                "pnl", cumulativePnl.toPlainString()
        ));

        executor.cancelAllPriceTriggeredOrders();
        closeExistingPositions();
        executor.shutdown();
@@ -451,8 +509,28 @@
                    .add(new BigDecimal(account.getUnrealisedPnl()))
                    .subtract(estimatedCloseFee);

            // 埋点: PNL_SNAPSHOT — 每60秒发射一次
            long now = System.currentTimeMillis();
            if (now - lastPnlSnapshotTime >= 60_000) {
                lastPnlSnapshotTime = now;
                BigDecimal total = new BigDecimal(account.getTotal());
                emitStats("PNL_SNAPSHOT", mapOf(
                        "cumulativePnl", cumulativePnl.toPlainString(),
                        "unrealizedPnl", new BigDecimal(account.getUnrealisedPnl()).toPlainString(),
                        "totalEquity", total.toPlainString(),
                        "markPrice", markPrice.toPlainString()
                ));
            }

            if (totalEquity.compareTo(target) > 0) {
                currentRound++;

                // 埋点: ROUND_COMPLETE
                emitStats("ROUND_COMPLETE", mapOf(
                        "roundNum", currentRound,
                        "totalEquity", totalEquity.toPlainString()
                ));

                int maxRounds = config.getRounds();
                log.info("[Gate] 盈亏达标(净权益{}→含手续费-{}=实际{}>目标{})，第{}轮完成",
                        new BigDecimal(account.getTotal()).add(new BigDecimal(account.getUnrealisedPnl())),
@@ -655,6 +733,13 @@
                int filledQty = Integer.parseInt(shortGridElement.getShortTraderParam().getQuantity());
                shortEntryTraderIdParam(shortGridElement, orderId, false);

                // 埋点: ENTRY_FILLED — 空仓加仓成交
                emitStats("ENTRY_FILLED", mapOf(
                        "direction", "short",
                        "gridId", shortGridElement.getId(),
                        "filledQty", filledQty
                ));

                // 防重入：同一网格存在多个入场单且相近时间成交时，只处理第一次 extend，
                // 后续成交打标 pendingReExtend，延展完成后自动用最新仓位重挂一次。
                if (shortGridElement.isExtendStopLossInProgress()) {
@@ -687,6 +772,13 @@

                int filledQty = Integer.parseInt(longGridElement.getLongTraderParam().getQuantity());
                longEntryTraderIdParam(longGridElement, orderId, false);

                // 埋点: ENTRY_FILLED — 多仓加仓成交
                emitStats("ENTRY_FILLED", mapOf(
                        "direction", "long",
                        "gridId", longGridElement.getId(),
                        "filledQty", filledQty
                ));

                // 防重入：同一网格存在多个入场单且相近时间成交时，只处理第一次 extend，
                // 后续成交打标 pendingReExtend，延展完成后自动用最新仓位重挂一次。
@@ -1347,6 +1439,13 @@
        accumulatedLongLossCount++;
        log.info("[Gate] 多仓止损触发 gridId:{}, 止损次数:{}{}, 开始追单",
                gridId, accumulatedLongLossCount, sameGrid ? "(同网格)" : "");

        // 埋点: STOP_LOSS_TRIGGERED
        emitStats("STOP_LOSS_TRIGGERED", mapOf(
                "direction", "long",
                "gridId", gridId,
                "lossCount", accumulatedLongLossCount
        ));
        int newEntryGridId = gridId + 1;

        GridElement newEntryGrid = GridElement.findById(newEntryGridId);
@@ -1392,6 +1491,13 @@
        accumulatedShortLossCount++;
        log.info("[Gate] 空仓止损触发 gridId:{}, 止损次数:{}{}, 开始追单",
                gridId, accumulatedShortLossCount, sameGrid ? "(同网格)" : "");

        // 埋点: STOP_LOSS_TRIGGERED
        emitStats("STOP_LOSS_TRIGGERED", mapOf(
                "direction", "short",
                "gridId", gridId,
                "lossCount", accumulatedShortLossCount
        ));
        int newEntryGridId = gridId - 1;

        GridElement newEntryGrid = GridElement.findById(newEntryGridId);
@@ -1960,8 +2066,12 @@
    public void setMarkPrice(BigDecimal markPrice) { this.markPrice = markPrice; }
    /** @return 策略是否处于活跃状态（非 STOPPED 且非 WAITING_KLINE） */
    public boolean isStrategyActive() { return state != StrategyState.STOPPED && state != StrategyState.WAITING_KLINE; }
    /** @return 当前已完成轮数 */
    public int getCurrentRound() { return currentRound; }
    /** @return 累计已实现盈亏（平仓推送驱动累加） */
    public BigDecimal getCumulativePnl() { return cumulativePnl; }
    /** @return 初始本金 */
    public BigDecimal getInitialPrincipal() { return initialPrincipal; }
    /** @return 当前未实现盈亏（每根 K 线实时计算） */
    public BigDecimal getUnrealizedPnl() { return unrealizedPnl; }
    /** @return Gate 用户 ID（用于私有频道订阅 payload） */

 src/main/java/com/xcong/excoin/modules/gateApi/GateWebSocketClientManager.java

@@ -66,6 +66,9 @@
    @Autowired
    private GateLogBuffer logBuffer;

    @Autowired
    private StatsEventProducer statsEventProducer;

    /** WebSocket 连接管理器 */
    private GateKlineWebSocketClient wsClient;
    /** 网格交易策略服务 */
@@ -84,20 +87,7 @@
                config = persistenceService.buildFromDTO(saved, DEFAULT_API_KEY);
                logBuffer.info("[管理器] 已加载持久化配置, apiKey=" + mask(DEFAULT_API_KEY));
            } else {
                // 构建默认配置 DTO（只有 7 项可调参数）
                GateConfigDTO defaults = GateConfigDTO.builder()
                        .apiKey(DEFAULT_API_KEY)
                        .gridRate(new BigDecimal("0.005"))
                        .expectedProfit(new BigDecimal("0.15"))
                        .maxLoss(new BigDecimal("1.5"))
                        .baseQuantity("2")
                        .quantity("2")
                        .maxPositionSize(4)
                        .stopLossCount(0)
                        .priceDriveEnabled(true)
                        .rounds(0)
                        .build();
                // 确保配置文件存在
                GateConfigDTO defaults = GateConfigDTO.defaultsFor(DEFAULT_API_KEY);
                persistenceService.ensureExists(DEFAULT_API_KEY, defaults);
                config = persistenceService.buildFromDTO(defaults, DEFAULT_API_KEY);
                logBuffer.info("[管理器] 已创建默认配置文件");
@@ -138,6 +128,36 @@
    public GateConfig getConfig() { return config; }

    /**
     * 远程更新策略参数 — 仅持久化到 JSON 文件，不自动重启。
     * 下次「启动」指令会从文件读取最新配置后重建策略。
     *
     * @param dto 从 Station 下发的可调参数（不含 apiSecret）
     */
    public void updateConfig(GateConfigDTO dto) {
        logBuffer.info("[管理器] 收到远程参数更新指令（仅持久化）");
        persistenceService.save(config.getApiKey(), dto);
        logBuffer.info("[管理器] 参数已保存，待下次启动生效");
    }

    /**
     * 从持久化文件重载配置并重启策略。
     * 用于远程「启动」指令：先读最新配置 → 重建服务 → 启动。
     */
    public void reloadAndStart() {
        logBuffer.info("[管理器] 从文件重载配置并启动...");
        GateConfigDTO saved = persistenceService.load(config.getApiKey());
        if (saved == null) {
            logBuffer.info("[管理器] 无持久化配置，使用当前内存配置启动");
            if (gridTradeService != null) {
                gridTradeService.startGrid();
            }
            return;
        }
        GateConfig newConfig = persistenceService.buildFromDTO(saved, config.getApiKey());
        restartWithConfig(newConfig);
    }

    /**
     * 使用新配置重启策略：停止旧策略 → 重建 WS → 重新 init + startGrid。
     */
    public void restartWithConfig(GateConfig newConfig) {
@@ -151,7 +171,7 @@
        }
        // 2. 使用新配置重建
        this.config = newConfig;
        this.gridTradeService = new GateGridTradeService(config);
        this.gridTradeService = new GateGridTradeService(config, statsEventProducer);
        gridTradeService.init();
        // 3. 重建 WS 客户端并重新注册 Handler
        this.wsClient = new GateKlineWebSocketClient(config.getWsUrl());

 src/main/java/com/xcong/excoin/modules/gateApi/HeartbeatScheduler.java

New file
@@ -0,0 +1,92 @@
package com.xcong.excoin.modules.gateApi;

import com.xcong.excoin.modules.station.model.GateStatsEvent;
import com.xcong.excoin.modules.station.model.HeartbeatMsg;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.security.MessageDigest;

/**
 * JAR 侧 — 心跳定时发送（每 30s）
 */
@Slf4j
@Component
@EnableScheduling
public class HeartbeatScheduler {

    @Resource
    private StatsEventProducer statsProducer;

    @Resource
    private GateWebSocketClientManager manager;

    private String apiKeyMd5;
    private String hostPort;

    // initialDelay = 0：启动后立即发第一次心跳，避免 Station 要等 30s 才能发现实例
    @Scheduled(fixedRate = 30_000, initialDelay = 1_000)
    public void heartbeat() {
        try {
            if (apiKeyMd5 == null) {
                apiKeyMd5 = md5(manager.getConfig().getApiKey());
                hostPort = resolveHostPort();
            }

            // 策略未启动（gridTradeService == null）时也发送心跳，
            // 否则 Station 无法发现该 JAR 实例，也就无法对它下发 START 指令（死锁）。
            GateGridTradeService strategy = manager.getGridTradeService();
            String state = (strategy != null && strategy.getState() != null)
                    ? strategy.getState().name() : "STOPPED";
            int currentRound = strategy != null ? strategy.getCurrentRound() : 0;
            String cumulativePnl = (strategy != null && strategy.getCumulativePnl() != null)
                    ? strategy.getCumulativePnl().toPlainString() : "0";
            String principal = (strategy != null && strategy.getInitialPrincipal() != null)
                    ? strategy.getInitialPrincipal().toPlainString() : "0";

            HeartbeatMsg hb = HeartbeatMsg.builder()
                    .contract(manager.getConfig().getContract())
                    .state(state)
                    .leverage(manager.getConfig().getLeverage())
                    .currentRound(currentRound)
                    .cumulativePnl(cumulativePnl)
                    .principal(principal)
                    .hostPort(hostPort)
                    .build();

            GateStatsEvent event = statsProducer.newHeartbeat(apiKeyMd5, hb);
            statsProducer.sendHeartbeat(event);

            log.info("[Gate] 心跳已发送, state={}, apiKeyMd5={}", state, apiKeyMd5);
        } catch (Exception e) {
            log.warn("[Gate] 心跳发送失败", e);
        }
    }

    private String resolveHostPort() {
        try {
            String host = InetAddress.getLocalHost().getHostAddress();
            // port 从 Spring 环境变量获取，默认 8888
            String port = System.getProperty("server.port", "8888");
            return host + ":" + port;
        } catch (Exception e) {
            return "unknown";
        }
    }

    private static String md5(String input) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] digest = md.digest(input.getBytes(java.nio.charset.StandardCharsets.UTF_8));
            StringBuilder sb = new StringBuilder();
            for (byte b : digest) sb.append(String.format("%02x", b));
            return sb.toString();
        } catch (Exception e) {
            return Integer.toHexString(input.hashCode());
        }
    }
}

 src/main/java/com/xcong/excoin/modules/gateApi/StatsEventProducer.java

New file
@@ -0,0 +1,72 @@
package com.xcong.excoin.modules.gateApi;

import com.alibaba.fastjson.JSON;
import com.xcong.excoin.configurations.RabbitMqConfig;
import com.xcong.excoin.modules.station.model.GateStatsEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * JAR 侧 — 统一消息发送器（心跳 / ACK / 策略事件 / Stats）
 */
@Slf4j
@Component
public class StatsEventProducer {

    private final RabbitTemplate rabbitTemplate;

    /** RabbitTemplate 是 prototype，必须用构造器注入（参考 OrderProducer） */
    @Autowired
    public StatsEventProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    /**
     * 发送心跳 / ACK 到 heartbeat 路由
     */
    public void sendHeartbeat(GateStatsEvent event) {
        send(RabbitMqConfig.EXCHANGE_GATE, RabbitMqConfig.ROUTINGKEY_GATE_HEARTBEAT, event);
    }

    /**
     * 发送策略事件到 stats 路由
     */
    public void sendStats(GateStatsEvent event) {
        send(RabbitMqConfig.EXCHANGE_GATE, RabbitMqConfig.ROUTINGKEY_GATE_STATS, event);
    }

    private void send(String exchange, String routingKey, GateStatsEvent event) {
        CorrelationData cd = new CorrelationData(event.getEventId());
        rabbitTemplate.convertAndSend(exchange, routingKey, JSON.toJSONString(event), cd);
        log.debug("[StatsProducer] 发送: type={}, apiKeyMd5={}", event.getType(), event.getApiKeyMd5());
    }

    // ==================== 便捷工厂方法 ====================

    public GateStatsEvent newHeartbeat(String apiKeyMd5, Object payload) {
        return build("HEARTBEAT", apiKeyMd5, payload);
    }

    public GateStatsEvent newCmdAck(String apiKeyMd5, Object payload) {
        return build("CMD_ACK", apiKeyMd5, payload);
    }

    public GateStatsEvent newStats(String type, String apiKeyMd5, Object payload) {
        return build(type, apiKeyMd5, payload);
    }

    private GateStatsEvent build(String type, String apiKeyMd5, Object payload) {
        return GateStatsEvent.builder()
                .eventId(UUID.randomUUID().toString())
                .type(type)
                .apiKeyMd5(apiKeyMd5)
                .timestamp(System.currentTimeMillis())
                .payload(JSON.toJSONString(payload))
                .build();
    }
}

 src/main/java/com/xcong/excoin/modules/station/model/CmdAckMsg.java

New file
@@ -0,0 +1,24 @@
package com.xcong.excoin.modules.station.model;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * 指令确认消息 payload
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class CmdAckMsg {
    /** 对应指令的 commandId */
    private String commandId;
    /** 是否成功 */
    private boolean success;
    /** 执行后的新状态 */
    private String newState;
    /** 描述信息 */
    private String message;
}

 src/main/java/com/xcong/excoin/modules/station/model/GateCommand.java

New file
@@ -0,0 +1,29 @@
package com.xcong.excoin.modules.station.model;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * Station 下发给 JAR 的启停 / 参数更新指令
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class GateCommand {
    /** 指令唯一 ID（用于 ACK 对应） */
    private String commandId;
    /** START / STOP / UPDATE_CONFIG */
    private String commandType;
    /** 目标 apiKey 的 MD5 */
    private String apiKeyMd5;
    /** 发令时间戳 */
    private Long timestamp;
    /**
     * 指令附加数据（JSON 字符串）。
     * UPDATE_CONFIG 时携带 GateConfigDTO 的 JSON，包含要更新的策略参数。
     */
    private String payload;
}

 src/main/java/com/xcong/excoin/modules/station/model/GateStatsEvent.java

New file
@@ -0,0 +1,26 @@
package com.xcong.excoin.modules.station.model;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * JAR 上报给 Station 的策略事件
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class GateStatsEvent {
    /** 事件唯一 ID */
    private String eventId;
    /** 事件类型：HEARTBEAT / CMD_ACK / STRATEGY_START / STRATEGY_STOP / ROUND_COMPLETE / STOP_LOSS_TRIGGERED / ENTRY_FILLED / PNL_SNAPSHOT */
    private String type;
    /** 目标 apiKey 的 MD5 */
    private String apiKeyMd5;
    /** 事件时间戳 */
    private Long timestamp;
    /** 纯字符串内容（JSON payload，Station 消费时反序列化） */
    private String payload;
}

 src/main/java/com/xcong/excoin/modules/station/model/HeartbeatMsg.java

New file
@@ -0,0 +1,30 @@
package com.xcong.excoin.modules.station.model;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * 心跳消息 payload
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class HeartbeatMsg {
    /** 合约名称 */
    private String contract;
    /** 策略状态 */
    private String state;
    /** 杠杆 */
    private String leverage;
    /** 当前轮次 */
    private int currentRound;
    /** 累计已实现盈亏 */
    private String cumulativePnl;
    /** 初始本金 */
    private String principal;
    /** host:port (用于 Station 直接 HTTP 调用，可选) */
    private String hostPort;
}

 src/main/java/com/xcong/excoin/modules/station/model/InstanceInfo.java

New file
@@ -0,0 +1,34 @@
package com.xcong.excoin.modules.station.model;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * JAR 实例注册信息（Station 内存 + Dashboard 展示）
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class InstanceInfo {
    /** apiKey 的 MD5 */
    private String apiKeyMd5;
    /** 合约名称 */
    private String contract;
    /** 策略状态 */
    private String state;
    /** 杠杆 */
    private String leverage;
    /** 当前轮次 */
    private int currentRound;
    /** 累计已实现盈亏 */
    private String cumulativePnl;
    /** 初始本金 */
    private String principal;
    /** host:port */
    private String hostPort;
    /** 最后心跳时间 */
    private long lastSeen;
}