[bitget-stream] Add bitget streaming module

pom.xml

@@ -121,6 +121,7 @@
         <module>xchange-stream-binance</module>
         <module>xchange-stream-bitfinex</module>
         <module>xchange-stream-bitflyer</module>
+        <module>xchange-stream-bitget</module>
         <module>xchange-stream-bitmex</module>
         <module>xchange-stream-bitstamp</module>
         <module>xchange-stream-btcmarkets</module>

xchange-stream-bitget/.gitignore

@@ -0,0 +1,2 @@
+http-client.private.env.json
+integration-test.env.properties

xchange-stream-bitget/README.md

@@ -0,0 +1,21 @@
+## Using IntelliJ Idea HTTP client
+
+There are *.http files stored in `src/test/resources/rest` that can be used with IntelliJ Idea HTTP Client.
+
+Some requests need authorization, so the api credentials have to be stored in `http-client.private.env.json` in module's root. Sample content can be found in `example.http-client.private.env.json`
+
+> [!CAUTION]
+> Never commit your api credentials to the repository!
+
+
+[HTTP Client documentation](https://www.jetbrains.com/help/idea/http-client-in-product-code-editor.html)
+
+## Running integration tests that require API keys
+
+Integration tests that require API keys read them from environment variables. They can be defined in `integration-test.env.properties`. Sample content can be found in `example.integration-test.env.properties`.
+
+If no keys are provided the integration tests that need them are skipped.
+
+> [!CAUTION]
+> Never commit your api credentials to the repository!
+

xchange-stream-bitget/example.http-client.private.env.json

@@ -0,0 +1,7 @@
+{
+  "default": {
+    "api_key": "replace_me",
+    "api_secret": "replace_me",
+    "api_passphrase": "replace_me"
+  }
+}

xchange-stream-bitget/example.integration-test.env.properties

@@ -0,0 +1,4 @@
+apiKey=change_me
+secretKey=change_me
+passphrase=change_me
+

xchange-stream-bitget/http-client.env.json

@@ -0,0 +1,5 @@
+{
+  "default": {
+    "base_url": "wss://ws.bitget.com"
+  }
+}

xchange-stream-bitget/lombok.config

@@ -0,0 +1,2 @@
+lombok.equalsAndHashCode.callSuper = call
+lombok.tostring.callsuper = call

xchange-stream-bitget/pom.xml

@@ -0,0 +1,57 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.knowm.xchange</groupId>
+        <artifactId>xchange-parent</artifactId>
+        <version>5.2.1-SNAPSHOT</version>
+    </parent>
+    <artifactId>xchange-stream-bitget</artifactId>
+
+    <name>XChange Bitget Stream</name>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.knowm.xchange</groupId>
+            <artifactId>xchange-bitget</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.knowm.xchange</groupId>
+            <artifactId>xchange-stream-core</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+
+        <pluginManagement>
+            <plugins>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-failsafe-plugin</artifactId>
+                    <configuration>
+                        <systemPropertiesFile>integration-test.env.properties</systemPropertiesFile>
+                    </configuration>
+                </plugin>
+            </plugins>
+        </pluginManagement>
+
+    </build>
+
+</project>

xchange-stream-bitget/src/main/java/info/bitrich/xchangestream/bitget/BitgetPrivateStreamingService.java

@@ -0,0 +1,77 @@
+package info.bitrich.xchangestream.bitget;
+
+import info.bitrich.xchangestream.bitget.config.Config;
+import info.bitrich.xchangestream.bitget.dto.common.Operation;
+import info.bitrich.xchangestream.bitget.dto.request.BitgetLoginRequest;
+import info.bitrich.xchangestream.bitget.dto.request.BitgetLoginRequest.LoginPayload;
+import info.bitrich.xchangestream.bitget.dto.response.BitgetEventNotification;
+import info.bitrich.xchangestream.bitget.dto.response.BitgetEventNotification.Event;
+import info.bitrich.xchangestream.bitget.dto.response.BitgetWsNotification;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Map.Entry;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class BitgetPrivateStreamingService extends BitgetStreamingService {
+
+  private final String apiKey;
+  private final String apiSecret;
+  private final String apiPassword;
+
+  public BitgetPrivateStreamingService(
+      String apiUri, String apiKey, String apiSecret, String apiPassword) {
+    super(apiUri);
+    this.apiKey = apiKey;
+    this.apiSecret = apiSecret;
+    this.apiPassword = apiPassword;
+  }
+
+  /** Sends login message right after connecting */
+  @Override
+  public void resubscribeChannels() {
+    sendLoginMessage();
+  }
+
+  public void resubscribeChannelsAfterLogin() {
+    for (Entry<String, Subscription> entry : channels.entrySet()) {
+      try {
+        Subscription subscription = entry.getValue();
+        sendMessage(getSubscribeMessage(subscription.getChannelName(), subscription.getArgs()));
+      } catch (IOException e) {
+        log.error("Failed to reconnect channel: {}", entry.getKey());
+      }
+    }
+  }
+
+  @SneakyThrows
+  private void sendLoginMessage() {
+    Instant timestamp = Instant.now(Config.getInstance().getClock());
+    BitgetLoginRequest bitgetLoginRequest =
+        BitgetLoginRequest.builder()
+            .operation(Operation.LOGIN)
+            .payload(
+                LoginPayload.builder()
+                    .apiKey(apiKey)
+                    .passphrase(apiPassword)
+                    .timestamp(timestamp)
+                    .signature(BitgetStreamingAuthHelper.sign(timestamp, apiSecret))
+                    .build())
+            .build();
+    sendMessage(objectMapper.writeValueAsString(bitgetLoginRequest));
+  }
+
+  @Override
+  protected void handleMessage(BitgetWsNotification message) {
+    // subscribe to channels after sucessful login confirmation
+    if (message instanceof BitgetEventNotification) {
+      BitgetEventNotification eventNotification = (BitgetEventNotification) message;
+      if (eventNotification.getEvent() == Event.LOGIN && eventNotification.getCode() == 0) {
+        resubscribeChannelsAfterLogin();
+        return;
+      }
+    }
+    super.handleMessage(message);
+  }
+}

xchange-stream-bitget/src/main/java/info/bitrich/xchangestream/bitget/BitgetStreamingAdapters.java

@@ -0,0 +1,137 @@
+package info.bitrich.xchangestream.bitget;
+
+import info.bitrich.xchangestream.bitget.dto.common.BitgetChannel;
+import info.bitrich.xchangestream.bitget.dto.common.BitgetChannel.ChannelType;
+import info.bitrich.xchangestream.bitget.dto.common.BitgetChannel.MarketType;
+import info.bitrich.xchangestream.bitget.dto.response.BitgetTickerNotification;
+import info.bitrich.xchangestream.bitget.dto.response.BitgetTickerNotification.TickerData;
+import info.bitrich.xchangestream.bitget.dto.response.BitgetWsOrderBookSnapshotNotification;
+import info.bitrich.xchangestream.bitget.dto.response.BitgetWsOrderBookSnapshotNotification.OrderBookData;
+import info.bitrich.xchangestream.bitget.dto.response.BitgetWsUserTradeNotification;
+import info.bitrich.xchangestream.bitget.dto.response.BitgetWsUserTradeNotification.BitgetFillData;
+import info.bitrich.xchangestream.bitget.dto.response.BitgetWsUserTradeNotification.FeeDetail;
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import lombok.experimental.UtilityClass;
+import org.apache.commons.lang3.ArrayUtils;
+import org.knowm.xchange.bitget.BitgetAdapters;
+import org.knowm.xchange.currency.CurrencyPair;
+import org.knowm.xchange.dto.Order.OrderType;
+import org.knowm.xchange.dto.marketdata.OrderBook;
+import org.knowm.xchange.dto.marketdata.Ticker;
+import org.knowm.xchange.dto.trade.LimitOrder;
+import org.knowm.xchange.dto.trade.UserTrade;
+import org.knowm.xchange.instrument.Instrument;
+
+@UtilityClass
+public class BitgetStreamingAdapters {
+
+  public Ticker toTicker(BitgetTickerNotification notification) {
+    TickerData bitgetTickerDto = notification.getPayloadItems().get(0);
+
+    CurrencyPair currencyPair = BitgetAdapters.toCurrencyPair(bitgetTickerDto.getInstrument());
+    if (currencyPair == null) {
+      return null;
+    }
+
+    return new Ticker.Builder()
+        .instrument(currencyPair)
+        .open(bitgetTickerDto.getOpen24h())
+        .last(bitgetTickerDto.getLastPrice())
+        .bid(bitgetTickerDto.getBestBidPrice())
+        .ask(bitgetTickerDto.getBestAskPrice())
+        .high(bitgetTickerDto.getHigh24h())
+        .low(bitgetTickerDto.getLow24h())
+        .volume(bitgetTickerDto.getAssetVolume24h())
+        .quoteVolume(bitgetTickerDto.getQuoteVolume24h())
+        .timestamp(BitgetAdapters.toDate(bitgetTickerDto.getTimestamp()))
+        .bidSize(bitgetTickerDto.getBestBidSize())
+        .askSize(bitgetTickerDto.getBestAskSize())
+        .percentageChange(bitgetTickerDto.getChange24h())
+        .build();
+  }
+
+  /** Returns unique subscription id. Can be used as key for subscriptions caching */
+  public String toSubscriptionId(BitgetChannel bitgetChannel) {
+    return Stream.of(
+            bitgetChannel.getMarketType(),
+            bitgetChannel.getChannelType(),
+            bitgetChannel.getInstrumentId())
+        .map(String::valueOf)
+        .collect(Collectors.joining("_"));
+  }
+
+  /**
+   * Creates {@code BitgetChannel} from arguments
+   *
+   * @param args [{@code ChannelType}, {@code MarketType}, {@code Instrument}/{@code null}]
+   */
+  public BitgetChannel toBitgetChannel(Object... args) {
+    ChannelType channelType = (ChannelType) ArrayUtils.get(args, 0);
+    MarketType marketType = (MarketType) ArrayUtils.get(args, 1);
+    Instrument instrument = (Instrument) ArrayUtils.get(args, 2);
+
+    return BitgetChannel.builder()
+        .channelType(channelType)
+        .marketType(marketType)
+        .instrumentId(
+            Optional.ofNullable(instrument).map(BitgetAdapters::toString).orElse("default"))
+        .build();
+  }
+
+  public OrderBook toOrderBook(
+      BitgetWsOrderBookSnapshotNotification notification, Instrument instrument) {
+    OrderBookData orderBookData = notification.getPayloadItems().get(0);
+    List<LimitOrder> asks =
+        orderBookData.getAsks().stream()
+            .map(
+                priceSizeEntry ->
+                    new LimitOrder(
+                        OrderType.ASK,
+                        priceSizeEntry.getSize(),
+                        instrument,
+                        null,
+                        null,
+                        priceSizeEntry.getPrice()))
+            .collect(Collectors.toList());
+
+    List<LimitOrder> bids =
+        orderBookData.getBids().stream()
+            .map(
+                priceSizeEntry ->
+                    new LimitOrder(
+                        OrderType.BID,
+                        priceSizeEntry.getSize(),
+                        instrument,
+                        null,
+                        null,
+                        priceSizeEntry.getPrice()))
+            .collect(Collectors.toList());
+
+    return new OrderBook(BitgetAdapters.toDate(orderBookData.getTimestamp()), asks, bids);
+  }
+
+  public UserTrade toUserTrade(BitgetWsUserTradeNotification notification) {
+    BitgetFillData bitgetFillData = notification.getPayloadItems().get(0);
+    return new UserTrade(
+        bitgetFillData.getOrderSide(),
+        bitgetFillData.getAssetAmount(),
+        BitgetAdapters.toCurrencyPair(bitgetFillData.getSymbol()),
+        bitgetFillData.getPrice(),
+        BitgetAdapters.toDate(bitgetFillData.getUpdatedAt()),
+        bitgetFillData.getTradeId(),
+        bitgetFillData.getOrderId(),
+        bitgetFillData.getFeeDetails().stream()
+            .map(FeeDetail::getTotalFee)
+            .map(BigDecimal::abs)
+            .reduce(BigDecimal.ZERO, BigDecimal::add),
+        bitgetFillData.getFeeDetails().stream()
+            .map(FeeDetail::getCurrency)
+            .findFirst()
+            .orElse(null),
+        null);
+  }
+}

xchange-stream-bitget/src/main/java/info/bitrich/xchangestream/bitget/BitgetStreamingAuthHelper.java

@@ -0,0 +1,42 @@
+package info.bitrich.xchangestream.bitget;
+
+import java.nio.charset.StandardCharsets;
+import java.security.InvalidKeyException;
+import java.security.NoSuchAlgorithmException;
+import java.time.Instant;
+import java.util.Base64;
+import javax.crypto.Mac;
+import javax.crypto.SecretKey;
+import javax.crypto.spec.SecretKeySpec;
+import lombok.experimental.UtilityClass;
+import org.knowm.xchange.service.BaseParamsDigest;
+
+@UtilityClass
+public class BitgetStreamingAuthHelper {
+
+  /** Generates signature based on timestamp */
+  public String sign(Instant timestamp, String secretString) {
+    final SecretKey secretKey =
+        new SecretKeySpec(
+            secretString.getBytes(StandardCharsets.UTF_8), BaseParamsDigest.HMAC_SHA_256);
+    Mac mac = createMac(secretKey, secretKey.getAlgorithm());
+
+    String payloadToSign = String.format("%sGET/user/verify", timestamp.getEpochSecond());
+    mac.update(payloadToSign.getBytes(StandardCharsets.UTF_8));
+
+    return Base64.getEncoder().encodeToString(mac.doFinal());
+  }
+
+  private Mac createMac(SecretKey secretKey, String hmacString) {
+    try {
+      Mac mac = Mac.getInstance(hmacString);
+      mac.init(secretKey);
+      return mac;
+    } catch (InvalidKeyException e) {
+      throw new IllegalArgumentException("Invalid key for hmac initialization.", e);
+    } catch (NoSuchAlgorithmException e) {
+      throw new RuntimeException(
+          "Illegal algorithm for post body digest. Check the implementation.");
+    }
+  }
+}