This commit is contained in:
tony 2026-07-08 23:34:24 +08:00
parent c6f90843f5
commit a448fe6a0a

View File

@ -23,8 +23,10 @@ import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
@ -86,6 +88,14 @@ public class AutoGridStrategy extends BaseStrategy {
* 正在平仓中的仓位,防止重复触发
*/
private final Map<String, Boolean> closingPositions = new ConcurrentHashMap<>();
/**
* 已处理过的买单成交 orderId,避免 fill / orders 双推重复开仓补单
*/
private final Set<String> processedBuyFillOrderIds = ConcurrentHashMap.newKeySet();
/**
* 已处理过的卖单成交 orderId
*/
private final Set<String> processedSellFillOrderIds = ConcurrentHashMap.newKeySet();
private final Object tradeLock = new Object();
@ -134,19 +144,39 @@ public class AutoGridStrategy extends BaseStrategy {
if (connectionId.equals(bitGetClient.getPublicConnectionId())) {
bitGetClient.subscribeTicker(getSymbol());
} else {
// 私有频道必须等 login 成功回调后再订阅,否则会报 30004
bitGetClient.login();
bitGetClient.subscribeFill();
bitGetClient.subscribeOrders();
}
}
@Override
public void onMessage(String connectionId, String message) {
logger.debug("WebSocket 收到消息:{}", message);
if ("pong".equals(message)) {
return;
}
try {
JSONObject json = JSONObject.parseObject(message);
// 登录成功后再订阅 fill / orders
if ("login".equals(json.getString("event"))) {
Integer code = json.getInteger("code");
if (code != null && code == 0) {
logger.info("私有 WebSocket 登录成功,开始订阅成交与订单");
bitGetClient.subscribeFill();
bitGetClient.subscribeOrders();
} else {
logger.error("私有 WebSocket 登录失败:{}", message);
submitAlarm("WebSocket 登录失败", message, AlarmLevel.ERROR);
}
return;
}
if ("error".equals(json.getString("event"))) {
logger.error("WebSocket 返回错误:{}", message);
return;
}
if (!json.containsKey("action")) {
return;
}
@ -165,10 +195,14 @@ public class AutoGridStrategy extends BaseStrategy {
handleTicker(data.getJSONObject(0));
} else if ("fill".equals(channel)
&& ("snapshot".equals(action) || "update".equals(action))) {
handleFill(data.getJSONObject(0));
for (int i = 0; i < data.size(); i++) {
handleFill(data.getJSONObject(i));
}
} else if ("orders".equals(channel)
&& ("snapshot".equals(action) || "update".equals(action))) {
handleOrderUpdate(data.getJSONObject(0));
for (int i = 0; i < data.size(); i++) {
handleOrderUpdate(data.getJSONObject(i));
}
}
} catch (Exception e) {
logger.error("WebSocket 消息处理异常", e);
@ -302,11 +336,6 @@ public class AutoGridStrategy extends BaseStrategy {
AccountVO account = getAccount();
bitGetClient.init(account.getApiKey(), account.getSecretKey(), account.getPassphrase(),
CommonConstant.httpClient, CommonConstant.webSocketManager);
// try {
// bitGetClient.setLeverage(getSymbol(), leverage);
// } catch (Exception e) {
// logger.warn("设置杠杆失败:{}", e.getMessage());
// }
try {
bitGetClient.createPublicWebsocket(webSocketMessageHandler);
bitGetClient.createPrivateWebsocket(webSocketMessageHandler);
@ -369,17 +398,6 @@ public class AutoGridStrategy extends BaseStrategy {
return AtrCalculator.calculate(klines, klinePeriod);
}
private String normalizeKlineInterval(String interval) {
if (interval == null) {
return "5m";
}
String normalized = interval.trim();
if (normalized.endsWith("m") || normalized.endsWith("H") || normalized.endsWith("D")) {
return normalized;
}
return normalized + "m";
}
// ======================== 价格边界 ========================
private void checkPriceBounds() {
@ -555,21 +573,22 @@ public class AutoGridStrategy extends BaseStrategy {
// ======================== WebSocket 事件处理 ========================
private void handleTicker(JSONObject ticker) {
String symbol = ticker.getString("symbol");
String symbol = resolveSymbol(ticker);
if (!getPlainSymbol().equals(symbol)) {
return;
}
BigDecimal lastPrice = ticker.getBigDecimal("lastPr");
if (lastPrice == null) {
lastPrice = ticker.getBigDecimal("last");
}
// BitGet ticker:优先 lastPr,兼容 last
BigDecimal lastPrice = firstPositive(
ticker.getBigDecimal("lastPr"),
ticker.getBigDecimal("last"),
ticker.getBigDecimal("markPrice"));
if (lastPrice != null) {
currentPrice = lastPrice;
}
}
private void handleFill(JSONObject fill) {
String symbol = fill.getString("symbol");
String symbol = resolveSymbol(fill);
if (!getPlainSymbol().equals(symbol)) {
return;
}
@ -579,60 +598,108 @@ public class AutoGridStrategy extends BaseStrategy {
BigDecimal price = fill.getBigDecimal("price");
BigDecimal volume = fill.getBigDecimal("baseVolume");
String orderId = fill.getString("orderId");
String clientOid = fill.getString("clientOid");
if ("buy".equals(side) && "open".equals(tradeSide)) {
onBuyFilled(price, volume, orderId);
} else if ("sell".equals(side) && "close".equals(tradeSide)) {
if (isOpenBuy(side, tradeSide)) {
onBuyFilled(price, volume, orderId, clientOid, price);
} else if (isCloseSell(side, tradeSide)) {
onSellFilled(price, volume, orderId);
}
}
/**
* BitGet orders 推送用 instId,fill 推送用 symbol。
* 成交以 status=filled 为可靠信号(fill 频道可能收不到)。
*/
private void handleOrderUpdate(JSONObject order) {
String symbol = order.getString("symbol");
String symbol = resolveSymbol(order);
if (!getPlainSymbol().equals(symbol)) {
return;
}
String status = order.getString("status");
if (!"cancelled".equals(status) && !"canceled".equals(status)) {
return;
}
String side = order.getString("side");
if (!"buy".equals(side)) {
String tradeSide = order.getString("tradeSide");
String orderId = order.getString("orderId");
String clientOid = order.getString("clientOid");
BigDecimal orderPrice = order.getBigDecimal("price");
if ("cancelled".equals(status) || "canceled".equals(status)) {
if ("buy".equals(side)) {
removePendingBuy(orderId, clientOid, orderPrice);
}
return;
}
BigDecimal price = order.getBigDecimal("price");
if (price != null) {
pendingBuyOrders.remove(priceKey(normalizePrice(price)));
if (!"filled".equals(status)) {
return;
}
BigDecimal fillPrice = firstPositive(
order.getBigDecimal("priceAvg"),
order.getBigDecimal("fillPrice"),
orderPrice);
BigDecimal volume = firstPositive(
order.getBigDecimal("accBaseVolume"),
order.getBigDecimal("baseVolume"),
order.getBigDecimal("size"));
if (isOpenBuy(side, tradeSide)) {
onBuyFilled(fillPrice, volume, orderId, clientOid, orderPrice);
} else if (isCloseSell(side, tradeSide)) {
onSellFilled(fillPrice, volume, orderId);
}
}
private void onBuyFilled(BigDecimal price, BigDecimal volume, String orderId) {
BigDecimal normalizedPrice = normalizePrice(price);
pendingBuyOrders.remove(priceKey(normalizedPrice));
private void onBuyFilled(BigDecimal fillPrice, BigDecimal volume, String orderId,
String clientOid, BigDecimal orderPrice) {
if (orderId != null && !processedBuyFillOrderIds.add(orderId)) {
return;
}
if (volume == null || volume.compareTo(BigDecimal.ZERO) <= 0) {
if (orderId != null) {
processedBuyFillOrderIds.remove(orderId);
}
logger.warn("买单成交数量无效,orderId={},volume={}", orderId, volume);
return;
}
String clientPositionId = "POS_" + normalizedPrice.toPlainString();
GridPosition position = new GridPosition(clientPositionId, normalizedPrice, volume);
GridBuyOrder pending = removePendingBuy(orderId, clientOid, orderPrice);
if (pending == null) {
if (orderId != null) {
processedBuyFillOrderIds.remove(orderId);
}
logger.debug("忽略非本策略买单成交,orderId={},clientOid={}", orderId, clientOid);
return;
}
BigDecimal gridPrice = pending.buyPrice;
String clientPositionId = "POS_" + gridPrice.toPlainString();
GridPosition position = new GridPosition(clientPositionId, gridPrice, volume);
openPositions.add(position);
BigDecimal tpPrice = position.takeProfitPrice(currentGridGap);
logger.info("买单成交:价格={},数量={},止盈目标={}", normalizedPrice, volume, tpPrice);
BigDecimal tpPrice = currentGridGap != null ? position.takeProfitPrice(currentGridGap) : null;
logger.info("买单成交:网格价={},成交价={},数量={},止盈目标={},orderId={}",
gridPrice, fillPrice, volume, tpPrice, orderId);
submitAlarm("网格买入成交",
String.format("买入价 %s,数量 %s,止盈目标 %s(间距 %s)",
formatPrice(normalizedPrice), volume.toPlainString(),
formatPrice(gridPrice), volume.toPlainString(),
formatPrice(tpPrice), formatPrice(currentGridGap)),
AlarmLevel.INFO);
submitOrderSync(orderId, clientPositionId, normalizedPrice, volume, OrderSide.BUY, OrderStatus.FILLED);
submitPositionSync(clientPositionId, normalizedPrice, volume, true);
submitOrderSync(orderId, clientPositionId, gridPrice, volume, OrderSide.BUY, OrderStatus.FILLED);
submitPositionSync(clientPositionId, gridPrice, volume, true);
// 成交后再挂下方一格,保证永远只有一笔待成交买单
placeNextBuyBelow(normalizedPrice);
placeNextBuyBelow(gridPrice);
}
private void onSellFilled(BigDecimal price, BigDecimal volume, String orderId) {
if (orderId != null && !processedSellFillOrderIds.add(orderId)) {
return;
}
GridPosition matched = null;
for (String closingId : closingPositions.keySet()) {
for (GridPosition position : openPositions) {
@ -645,7 +712,7 @@ public class AutoGridStrategy extends BaseStrategy {
break;
}
}
if (matched == null) {
if (matched == null && volume != null) {
for (GridPosition position : openPositions) {
if (position.quantity.compareTo(volume) == 0) {
matched = position;
@ -663,9 +730,10 @@ public class AutoGridStrategy extends BaseStrategy {
submitPositionSync(matched.clientPositionId, price, matched.quantity, false);
}
logger.info("卖单成交(止盈):价格={},数量={}", price, volume);
logger.info("卖单成交(止盈):价格={},数量={},orderId={}", price, volume, orderId);
submitAlarm("网格止盈成交",
String.format("卖出价 %s,数量 %s", formatPrice(price), volume.toPlainString()),
String.format("卖出价 %s,数量 %s", formatPrice(price),
volume != null ? volume.toPlainString() : "-"),
AlarmLevel.INFO);
submitOrderSync(orderId, matched != null ? matched.clientPositionId : "CLOSE", price, volume,
OrderSide.SELL, OrderStatus.FILLED);
@ -674,6 +742,77 @@ public class AutoGridStrategy extends BaseStrategy {
replenishBuyGrids();
}
private String resolveSymbol(JSONObject payload) {
String symbol = payload.getString("symbol");
if (symbol == null || symbol.isEmpty()) {
symbol = payload.getString("instId");
}
return symbol;
}
private boolean isOpenBuy(String side, String tradeSide) {
if (!"buy".equals(side)) {
return false;
}
return tradeSide == null
|| "open".equals(tradeSide)
|| "buy_single".equals(tradeSide);
}
private boolean isCloseSell(String side, String tradeSide) {
if (!"sell".equals(side)) {
return false;
}
return tradeSide == null
|| "close".equals(tradeSide)
|| "sell_single".equals(tradeSide)
|| "reduce_sell_single".equals(tradeSide);
}
private BigDecimal firstPositive(BigDecimal... values) {
if (values == null) {
return null;
}
for (BigDecimal value : values) {
if (value != null && value.compareTo(BigDecimal.ZERO) > 0) {
return value;
}
}
return null;
}
/**
* 优先按 exchangeOrderId / clientOid 移除,避免成交价与挂单价不一致导致残留。
*/
private GridBuyOrder removePendingBuy(String orderId, String clientOid, BigDecimal price) {
if (orderId != null) {
for (Iterator<Map.Entry<String, GridBuyOrder>> it = pendingBuyOrders.entrySet().iterator();
it.hasNext(); ) {
Map.Entry<String, GridBuyOrder> entry = it.next();
GridBuyOrder order = entry.getValue();
if (orderId.equals(order.exchangeOrderId)) {
it.remove();
return order;
}
}
}
if (clientOid != null) {
for (Iterator<Map.Entry<String, GridBuyOrder>> it = pendingBuyOrders.entrySet().iterator();
it.hasNext(); ) {
Map.Entry<String, GridBuyOrder> entry = it.next();
GridBuyOrder order = entry.getValue();
if (clientOid.equals(order.clientOrderId)) {
it.remove();
return order;
}
}
}
if (price != null) {
return pendingBuyOrders.remove(priceKey(normalizePrice(price)));
}
return null;
}
// ======================== 交易所同步 ========================
private void syncOpenOrdersFromExchange() {