From e70e04de37946bbb516b40aca0b1c086528398cd Mon Sep 17 00:00:00 2001
From: Administrator <15274802129@163.com>
Date: Mon, 24 Aug 2026 14:04:37 +0800
Subject: [PATCH] feat(gate): 添加策略远程控制和监控功能
---
src/main/java/com/xcong/excoin/configurations/RabbitMqConfig.java | 39 +++
src/main/java/com/xcong/excoin/modules/gateApi/GateWebSocketClientManager.java | 50 ++-
src/main/java/com/xcong/excoin/modules/gateApi/GateCommandConsumer.java | 111 ++++++++
src/main/java/com/xcong/excoin/modules/station/model/CmdAckMsg.java | 24 +
config/gate-config-7332fa3f94276d3dcc44fcd3bfff2379.json | 28 ++
src/main/java/com/xcong/excoin/modules/station/model/InstanceInfo.java | 34 ++
src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java | 112 ++++++++
src/main/java/com/xcong/excoin/modules/gateApi/StatsEventProducer.java | 72 +++++
src/main/java/com/xcong/excoin/modules/station/model/HeartbeatMsg.java | 30 ++
src/main/java/com/xcong/excoin/modules/gateApi/HeartbeatScheduler.java | 92 +++++++
src/main/java/com/xcong/excoin/modules/gateApi/GateConfigDTO.java | 63 ++++
src/main/java/com/xcong/excoin/modules/gateApi/CommandQueueInitializer.java | 70 +++++
src/main/java/com/xcong/excoin/modules/station/model/GateStatsEvent.java | 26 ++
src/main/java/com/xcong/excoin/modules/station/model/GateCommand.java | 29 ++
14 files changed, 764 insertions(+), 16 deletions(-)
diff --git a/config/gate-config-7332fa3f94276d3dcc44fcd3bfff2379.json b/config/gate-config-7332fa3f94276d3dcc44fcd3bfff2379.json
new file mode 100644
index 0000000..34db0fd
--- /dev/null
+++ b/config/gate-config-7332fa3f94276d3dcc44fcd3bfff2379.json
@@ -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
+}
\ No newline at end of file
diff --git a/src/main/java/com/xcong/excoin/configurations/RabbitMqConfig.java b/src/main/java/com/xcong/excoin/configurations/RabbitMqConfig.java
index 0a791bf..0c12c52 100644
--- a/src/main/java/com/xcong/excoin/configurations/RabbitMqConfig.java
+++ b/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);
+ }
+
}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/CommandQueueInitializer.java b/src/main/java/com/xcong/excoin/modules/gateApi/CommandQueueInitializer.java
new file mode 100644
index 0000000..b310d5d
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/CommandQueueInitializer.java
@@ -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());
+ }
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/GateCommandConsumer.java b/src/main/java/com/xcong/excoin/modules/gateApi/GateCommandConsumer.java
new file mode 100644
index 0000000..89907d8
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/GateCommandConsumer.java
@@ -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());
+ }
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/GateConfigDTO.java b/src/main/java/com/xcong/excoin/modules/gateApi/GateConfigDTO.java
index 851b7dc..97be386 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/GateConfigDTO.java
+++ b/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();
+ }
}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java b/src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java
index 8c90d42..e262510 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/GateGridTradeService.java
+++ b/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） */
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/GateWebSocketClientManager.java b/src/main/java/com/xcong/excoin/modules/gateApi/GateWebSocketClientManager.java
index 199c466..6f96e47 100644
--- a/src/main/java/com/xcong/excoin/modules/gateApi/GateWebSocketClientManager.java
+++ b/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());
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/HeartbeatScheduler.java b/src/main/java/com/xcong/excoin/modules/gateApi/HeartbeatScheduler.java
new file mode 100644
index 0000000..26cbd5c
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/HeartbeatScheduler.java
@@ -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());
+ }
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/gateApi/StatsEventProducer.java b/src/main/java/com/xcong/excoin/modules/gateApi/StatsEventProducer.java
new file mode 100644
index 0000000..6d38665
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/gateApi/StatsEventProducer.java
@@ -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();
+ }
+}
diff --git a/src/main/java/com/xcong/excoin/modules/station/model/CmdAckMsg.java b/src/main/java/com/xcong/excoin/modules/station/model/CmdAckMsg.java
new file mode 100644
index 0000000..0a8ed23
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/station/model/CmdAckMsg.java
@@ -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;
+}
diff --git a/src/main/java/com/xcong/excoin/modules/station/model/GateCommand.java b/src/main/java/com/xcong/excoin/modules/station/model/GateCommand.java
new file mode 100644
index 0000000..59f795a
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/station/model/GateCommand.java
@@ -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;
+}
diff --git a/src/main/java/com/xcong/excoin/modules/station/model/GateStatsEvent.java b/src/main/java/com/xcong/excoin/modules/station/model/GateStatsEvent.java
new file mode 100644
index 0000000..67886d3
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/station/model/GateStatsEvent.java
@@ -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;
+}
diff --git a/src/main/java/com/xcong/excoin/modules/station/model/HeartbeatMsg.java b/src/main/java/com/xcong/excoin/modules/station/model/HeartbeatMsg.java
new file mode 100644
index 0000000..897c5cc
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/station/model/HeartbeatMsg.java
@@ -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;
+}
diff --git a/src/main/java/com/xcong/excoin/modules/station/model/InstanceInfo.java b/src/main/java/com/xcong/excoin/modules/station/model/InstanceInfo.java
new file mode 100644
index 0000000..371faac
--- /dev/null
+++ b/src/main/java/com/xcong/excoin/modules/station/model/InstanceInfo.java
@@ -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;
+}
--
Gitblit v1.9.1