Merge pull request #4835 from rizer1980/PR/bybit-stream

[bybit-stream] simple implemetation

pom.xml

@@ -188,6 +188,7 @@
     <module>xchange-stream-bitmex</module>
     <module>xchange-stream-bitstamp</module>
     <module>xchange-stream-btcmarkets</module>
+    <module>xchange-stream-bybit</module>
     <module>xchange-stream-cexio</module>
     <module>xchange-stream-coinbasepro</module>
     <module>xchange-stream-coinjar</module>

xchange-bybit/src/main/java/org/knowm/xchange/bybit/service/BybitDigest.java

@@ -28,7 +28,11 @@ public BybitDigest(String secretKeyBase64) {
   }
 
   public static ParamsDigest createInstance(String secretKeyBase64) {
-    return new BybitDigest(secretKeyBase64);
+    if (secretKeyBase64 != null) {
+      return new BybitDigest(secretKeyBase64);
+    } else {
+      return null;
+    }
   }
 
   @SneakyThrows

xchange-stream-bybit/pom.xml

@@ -0,0 +1,31 @@
+<?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.1.2-SNAPSHOT</version>
+  </parent>
+
+  <name>XChange Bybit Stream</name>
+  <artifactId>xchange-stream-bybit</artifactId>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.knowm.xchange</groupId>
+      <artifactId>xchange-stream-core</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.knowm.xchange</groupId>
+      <artifactId>xchange-bybit</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
+  </dependencies>
+
+
+</project>

xchange-stream-bybit/src/main/java/dto/BybitSubscribeMessage.java

@@ -0,0 +1,13 @@
+package dto;
+
+import java.util.List;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class BybitSubscribeMessage {
+    private final String op;
+    private final List<String> args;
+  }
+

xchange-stream-bybit/src/main/java/dto/marketdata/BybitOrderbook.java

@@ -0,0 +1,22 @@
+package dto.marketdata;
+
+import java.beans.ConstructorProperties;
+import lombok.Getter;
+
+@Getter
+public class BybitOrderbook {
+
+  private final String topic;
+  private final String dataType;
+  private final String ts;
+  private final BybitOrderbookData data;
+
+  @ConstructorProperties({"topic","type","ts","data"})
+  public BybitOrderbook(String topic, String dataType, String ts, BybitOrderbookData data) {
+    this.topic = topic;
+    this.dataType = dataType;
+    this.ts = ts;
+    this.data = data;
+  }
+
+}

xchange-stream-bybit/src/main/java/dto/marketdata/BybitOrderbookData.java

@@ -0,0 +1,33 @@
+package dto.marketdata;
+
+import java.beans.ConstructorProperties;
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class BybitOrderbookData {
+
+  private final String symbolName;
+  private final List<BybitPublicOrder> bid;
+  private final List<BybitPublicOrder> ask;
+  // Update ID. Is a sequence. Occasionally, you'll receive "u"=1, which is a snapshot data due to
+  // the restart of the service. So please overwrite your local orderbook
+  private final Integer u;
+  // Cross sequence
+  // You can use this field to compare different levels orderbook data, and for the smaller seq,
+  // then it means the data is generated earlier.
+  // in docs says than it is Integer, but in fact, we get in response bigger numbers
+  private final Long seq;
+
+  @ConstructorProperties({"s", "b", "a", "u", "seq"})
+  public BybitOrderbookData(String symbolName, List<BybitPublicOrder> bid,
+      List<BybitPublicOrder> ask,
+      Integer u,
+      Long seq) {
+    this.symbolName = symbolName;
+    this.bid = bid;
+    this.ask = ask;
+    this.u = u;
+    this.seq = seq;
+  }
+}

xchange-stream-bybit/src/main/java/dto/marketdata/BybitPublicOrder.java

@@ -0,0 +1,18 @@
+package dto.marketdata;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import lombok.Getter;
+
+@Getter
+public class BybitPublicOrder {
+
+  private final String price;
+  private final String size;
+
+  @JsonCreator
+  public BybitPublicOrder(String[] data) {
+    this.price = data[0];
+    this.size = data[1];
+  }
+}
+

xchange-stream-bybit/src/main/java/dto/trade/BybitTrade.java

@@ -0,0 +1,42 @@
+package dto.trade;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.math.BigDecimal;
+import java.util.Date;
+import lombok.Getter;
+import org.knowm.xchange.bybit.dto.trade.BybitSide;
+
+@Getter
+public class BybitTrade {
+
+  //https://bybit-exchange.github.io/docs/v5/websocket/public/trad
+  //The timestamp (ms) that the order is filled
+  private final Date timestamp;
+  //Symbol name
+  private final String instId;
+  //Side of taker. Buy,Sell
+  private final BybitSide side;
+  private final BigDecimal tradeSize;
+  private final BigDecimal tradePrice;
+  //L	string	Direction of price change. Unique field for future
+  private final String direction;
+  private final String tradeId;
+  //boolean	Whether it is a block trade order or not
+  private final boolean bT;
+
+
+  public BybitTrade(@JsonProperty("T") Date timestamp, @JsonProperty("s") String instId,
+      @JsonProperty("S") String side, @JsonProperty("v") BigDecimal tradeSize,
+      @JsonProperty("p") BigDecimal tradePrice, @JsonProperty("L") String direction,
+      @JsonProperty("i") String tradeId, @JsonProperty("BT") boolean bT) {
+    this.timestamp = timestamp;
+    this.instId = instId;
+    this.side = BybitSide.valueOf(side.toUpperCase());
+    this.tradeSize = tradeSize;
+    this.tradePrice = tradePrice;
+    this.direction = direction;
+    this.tradeId = tradeId;
+    this.bT = bT;
+  }
+}
+

xchange-stream-bybit/src/main/java/info/bitrich/xchangestream/bybit/BybitStreamAdapters.java

@@ -0,0 +1,62 @@
+package info.bitrich.xchangestream.bybit;
+
+import static org.knowm.xchange.bybit.BybitAdapters.getOrderType;
+
+import dto.marketdata.BybitPublicOrder;
+import dto.trade.BybitTrade;
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import dto.marketdata.BybitOrderbook;
+import org.knowm.xchange.dto.Order;
+import org.knowm.xchange.dto.Order.OrderType;
+import org.knowm.xchange.dto.marketdata.OrderBook;
+import org.knowm.xchange.dto.marketdata.Trade;
+import org.knowm.xchange.dto.marketdata.Trades;
+import org.knowm.xchange.dto.trade.LimitOrder;
+import org.knowm.xchange.instrument.Instrument;
+
+public class BybitStreamAdapters {
+  public static OrderBook adaptOrderBook(BybitOrderbook bybitOrderbooks, Instrument instrument) {
+    List<LimitOrder> asks = new ArrayList<>();
+    List<LimitOrder> bids = new ArrayList<>();
+    Date timestamp = new Date(Long.parseLong(bybitOrderbooks.getTs()));
+    bybitOrderbooks.getData().getAsk()
+        .forEach(bybitAsk -> asks.add(
+            adaptOrderBookOrder(bybitAsk, instrument, OrderType.ASK,timestamp)));
+
+    bybitOrderbooks.getData().getBid()
+        .forEach(bybitBid -> bids.add(
+            adaptOrderBookOrder(bybitBid, instrument, OrderType.BID,timestamp)));
+
+    return new OrderBook(timestamp,asks, bids);
+  }
+
+  public static Trades adaptTrades(List<BybitTrade> bybitTrades, Instrument instrument) {
+    List<Trade> trades = new ArrayList<>();
+
+    bybitTrades.forEach(
+        bybitTrade ->
+            trades.add(
+                new Trade.Builder()
+                    .id(bybitTrade.getTradeId())
+                    .instrument(instrument)
+                    .originalAmount(bybitTrade.getTradeSize())
+                    .price(bybitTrade.getTradePrice())
+                    .timestamp(bybitTrade.getTimestamp())
+                    .type(getOrderType(bybitTrade.getSide()))
+                    .build()));
+
+    return new Trades(trades);
+  }
+
+  public static LimitOrder adaptOrderBookOrder(BybitPublicOrder bybitPublicOrder,
+      Instrument instrument, Order.OrderType orderType, Date timestamp) {
+
+    return new LimitOrder(orderType, new BigDecimal(bybitPublicOrder.getSize()), instrument, "",
+        timestamp, new BigDecimal(bybitPublicOrder.getPrice()));
+  }
+
+
+}

xchange-stream-bybit/src/main/java/info/bitrich/xchangestream/bybit/BybitStreamingExchange.java

@@ -0,0 +1,76 @@
+package info.bitrich.xchangestream.bybit;
+
+import info.bitrich.xchangestream.core.ProductSubscription;
+import info.bitrich.xchangestream.core.StreamingExchange;
+import io.reactivex.Completable;
+import org.knowm.xchange.bybit.BybitExchange;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class BybitStreamingExchange extends BybitExchange implements StreamingExchange {
+
+  private final Logger LOG = LoggerFactory.getLogger(BybitStreamingExchange.class);
+
+  //https://bybit-exchange.github.io/docs/v5/ws/connect
+  public static final String URI = "wss://stream.bybit.com/v5/public";
+  public static final String TESTNET_URI = "wss://stream-testnet.bybit.com/v5/public";
+  public static final String AUTH_URI = "wss://stream.bybit.com/v5/private";
+  public static final String TESTNET_AUTH_URI = "wss://stream-testnet.bybit.com/v5/private";
+
+  //spot, linear, inverse or option
+  public static final String EXCHANGE_TYPE = "EXCHANGE_TYPE";
+
+  private BybitStreamingService streamingService;
+  private BybitStreamingMarketDataService streamingMarketDataService;
+
+  @Override
+  protected void initServices() {
+    super.initServices();
+    this.streamingService = new BybitStreamingService(getApiUrl(),
+        exchangeSpecification.getExchangeSpecificParametersItem(EXCHANGE_TYPE));
+    this.streamingMarketDataService = new BybitStreamingMarketDataService(streamingService);
+  }
+
+  private String getApiUrl() {
+    String apiUrl = null;
+    if (exchangeSpecification.getApiKey() == null) {
+      if (Boolean.TRUE.equals(
+          exchangeSpecification.getExchangeSpecificParametersItem(USE_SANDBOX))) {
+        apiUrl = TESTNET_URI;
+      } else {
+        apiUrl = URI;
+      }
+      apiUrl += "/" + exchangeSpecification.getExchangeSpecificParametersItem(EXCHANGE_TYPE);
+    }
+// TODO auth
+    return apiUrl;
+  }
+
+  @Override
+  public Completable connect(ProductSubscription... args) {
+    LOG.info("Connect to BybitStream");
+    return streamingService.connect();
+  }
+
+  @Override
+  public Completable disconnect() {
+    streamingService.pingPongDisconnectIfConnected();
+    return streamingService.disconnect();
+  }
+
+  @Override
+  public boolean isAlive() {
+    return streamingService != null && streamingService.isSocketOpen();
+  }
+
+  @Override
+  public void useCompressedMessages(boolean compressedMessages) {
+    streamingService.useCompressedMessages(compressedMessages);
+  }
+
+  @Override
+  public BybitStreamingMarketDataService getStreamingMarketDataService() {
+    return streamingMarketDataService;
+  }
+
+}