Interface support for trade service (Implemented for bitmex and okcoin/okex)

xchange-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexAuthenticator.java

@@ -0,0 +1,38 @@
+package info.bitrich.xchangestream.bitmex;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.nio.charset.Charset;
+
+/**
+ * Created by heath on 2018/3/1.
+ */
+public class BitmexAuthenticator {
+
+    public static String getSHA256String(String str, String key) {
+
+        try {
+            Charset asciiCs = Charset.forName("US-ASCII");
+            SecretKeySpec signingKey = new SecretKeySpec(asciiCs.encode(key).array(), "HmacSHA256");
+            Mac sha256_HMAC = Mac.getInstance("HmacSHA256");
+            sha256_HMAC.init(signingKey);
+            byte[] mac_data = sha256_HMAC.doFinal(asciiCs.encode(str).array());
+            StringBuilder result = new StringBuilder();
+            for (final byte element : mac_data) {
+                result.append(Integer.toString((element & 0xff) + 0x100, 16).substring(1));
+            }
+            // System.out.println("SHA256String Result:[" + result + "]");
+            return result.toString().toUpperCase();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return null;
+    }
+
+    public static String generateSignature(String secret, String verb, String url, String nonce, String data) {
+        String message = verb + url + nonce + data;
+        // System.out.println(message);
+        return getSHA256String(message, secret);
+    }
+
+}

xchange-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingExchange.java

@@ -29,6 +29,7 @@ protected BitmexStreamingExchange(BitmexStreamingService streamingService) {
     protected void initServices() {
         super.initServices();
         streamingMarketDataService = new BitmexStreamingMarketDataService(streamingService);
+        streamingService.setExchangeSpecification(this.getExchangeSpecification());
     }
 
     @Override

xchange-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingMarketDataService.java

@@ -89,4 +89,5 @@ public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
             return trades;
         });
     }
+
 }

xchange-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingService.java

@@ -1,9 +1,12 @@
 package info.bitrich.xchangestream.bitmex;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.*;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import io.reactivex.Completable;
+import io.reactivex.CompletableSource;
+import org.knowm.xchange.ExchangeSpecification;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandler;
 import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler;
 import org.slf4j.Logger;
@@ -25,12 +28,52 @@ public class BitmexStreamingService extends JsonNettyStreamingService {
     private static final Logger LOG = LoggerFactory.getLogger(BitmexStreamingService.class);
     private final ObjectMapper mapper = new ObjectMapper();
 
+    protected ExchangeSpecification exchangeSpecification;
+
     public BitmexStreamingService(String apiUrl) {
         super(apiUrl, Integer.MAX_VALUE);
         mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
     }
 
-	 @Override
+    public void setExchangeSpecification(ExchangeSpecification exchangeSpecification) {
+        this.exchangeSpecification = exchangeSpecification;
+    }
+
+    private void login() throws JsonProcessingException {
+        long expires = System.currentTimeMillis() + 30;
+        String apiKey = this.exchangeSpecification.getApiKey();
+        String apiSecret = this.exchangeSpecification.getSecretKey();
+        String path = "/realtime";
+        String signature = BitmexAuthenticator.generateSignature(apiSecret,
+                "GET", path, String.valueOf(expires), "");
+
+        List<Object> args = Arrays.asList(apiKey, expires, signature);
+
+        Map<String, Object> cmd = new HashMap<>();
+        cmd.put("op", "authKey");
+        cmd.put("args", args);
+        this.sendMessage(mapper.writeValueAsString(cmd));
+    }
+
+    @Override
+    public Completable connect() {
+        // Note that we must override connect method in streaming service instead of streaming exchange, because of the auto reconnect feature of NettyStreamingService.
+        // We must ensure the authentication message is also resend when the connection is rebuilt.
+        Completable conn = super.connect();
+        if (this.exchangeSpecification.getApiKey() == null) {
+            return conn;
+        }
+        return conn.andThen((CompletableSource)(completable) -> {
+            try {
+                login();
+                completable.onComplete();
+            } catch (IOException e) {
+                completable.onError(e);
+            }
+        });
+    }
+
+    @Override
     protected void handleMessage(JsonNode message) {
         if (message.has("info") || message.has("success")) {
             return;

xchange-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingTradeService.java

@@ -0,0 +1,43 @@
+package info.bitrich.xchangestream.bitmex;
+
+
+import info.bitrich.xchangestream.bitmex.dto.BitmexOrder;
+import info.bitrich.xchangestream.core.StreamingTradeService;
+import io.reactivex.Observable;
+import org.knowm.xchange.currency.CurrencyPair;
+import org.knowm.xchange.dto.Order;
+import org.knowm.xchange.exceptions.NotYetImplementedForExchangeException;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+
+/**
+ * Created by Declan
+ */
+public class BitmexStreamingTradeService implements StreamingTradeService {
+
+    private final BitmexStreamingService streamingService;
+
+    public BitmexStreamingTradeService(BitmexStreamingService streamingService) {
+        this.streamingService = streamingService;
+    }
+
+    @Override
+    public Observable<Order> getOrders(CurrencyPair currencyPair, Object... args) {
+        String channelName = "order";
+        String instrument = currencyPair.base.toString() + currencyPair.counter.toString();
+        return streamingService.subscribeBitmexChannel(channelName).flatMapIterable(s -> {
+            BitmexOrder[] bitmexOrders = s.toBitmexOrders();
+            return Arrays.stream(bitmexOrders)
+                    .filter(bitmexOrder -> bitmexOrder.getSymbol() == instrument)
+                    .filter(BitmexOrder::isNotWorkingIndicator)
+                    .map(BitmexOrder::toOrder).collect(Collectors.toList());
+        });
+    }
+
+    @Override
+    public void submitOrder(Order order, CurrencyPair var1, Object... var2) {
+        throw new NotYetImplementedForExchangeException();
+    }
+}

xchange-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/dto/BitmexOrder.java

@@ -0,0 +1,125 @@
+package info.bitrich.xchangestream.bitmex.dto;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.knowm.xchange.bitmex.BitmexUtils;
+import org.knowm.xchange.dto.Order;
+import org.knowm.xchange.dto.trade.LimitOrder;
+import org.knowm.xchange.dto.trade.MarketOrder;
+
+
+import java.math.BigDecimal;
+
+
+public class BitmexOrder extends BitmexMarketDataEvent {
+
+    public enum OrderStatus {
+        NEW,
+        PARTIALLYFILLED,
+        FILLED,
+        TBD,
+        CANCELED,
+        REJECTED,
+        UNKNOW
+    }
+
+    private String orderID;
+
+    private int account;
+
+    private String side;
+
+    private BigDecimal price;
+
+    private BigDecimal avgPx;
+
+    private String ordType;
+
+    private OrderStatus ordStatus;
+
+    private String clOrdID;
+
+    private BigDecimal orderQty;
+
+    private BigDecimal cumQty;
+
+    public boolean isNotWorkingIndicator() {
+        return !workingIndicator;
+    }
+
+    private boolean workingIndicator;
+
+    @JsonCreator
+    public BitmexOrder(@JsonProperty("symbol") String symbol,
+                       @JsonProperty("timestamp") String timestamp,
+                       @JsonProperty("orderID") String orderID,
+                       @JsonProperty("account") int account,
+                       @JsonProperty("side") String side,
+                       @JsonProperty("price") BigDecimal price,
+                       @JsonProperty("avgPx") BigDecimal avgPx,
+                       @JsonProperty("ordType") String ordType,
+                       @JsonProperty("ordStatus") String ordStatus,
+                       @JsonProperty("clOrdID") String clOrdID,
+                       @JsonProperty("orderQty") BigDecimal orderQty,
+                       @JsonProperty("cumQty") BigDecimal cumQty,
+                       @JsonProperty("workingIndicator") boolean workingIndicator) {
+        super(symbol, timestamp);
+        this.orderID = orderID;
+        this.account = account;
+        this.side = side;
+        this.price = price;
+        this.avgPx = avgPx;
+        this.ordType = ordType;
+        try {
+            this.ordStatus = OrderStatus.valueOf(ordStatus.toUpperCase());
+        } catch (Exception e) {
+            this.ordStatus = OrderStatus.UNKNOW;
+        }
+        this.clOrdID = clOrdID;
+        this.orderQty = orderQty;
+        this.cumQty = cumQty;
+        this.workingIndicator = workingIndicator;
+    }
+
+    public Order toOrder() {
+        Order.Builder order;
+        if (ordType.equals("Market")) {
+            order = new MarketOrder.Builder(side.equals("Buy") ? Order.OrderType.BID : Order.OrderType.ASK, BitmexUtils.translateBitmexCurrencyPair(symbol));
+        } else {
+            order = new LimitOrder.Builder(side.equals("Buy") ? Order.OrderType.BID : Order.OrderType.ASK, BitmexUtils.translateBitmexCurrencyPair(symbol));
+            ((LimitOrder.Builder)order).limitPrice(price);
+        }
+        order.id(orderID)
+                .averagePrice(avgPx)
+                .originalAmount(orderQty)
+                .cumulativeAmount(cumQty);
+
+        switch (ordStatus) {
+            case NEW:
+                order.orderStatus(Order.OrderStatus.NEW);
+                break;
+            case PARTIALLYFILLED:
+                order.orderStatus(Order.OrderStatus.PARTIALLY_FILLED);
+                break;
+            case FILLED:
+                order.orderStatus(Order.OrderStatus.FILLED);
+                break;
+            case TBD:
+                order.orderStatus(Order.OrderStatus.PENDING_CANCEL);
+                break;
+            case CANCELED:
+                order.orderStatus(Order.OrderStatus.CANCELED);
+                break;
+            case REJECTED:
+                order.orderStatus(Order.OrderStatus.REJECTED);
+            default:
+                order.orderStatus(Order.OrderStatus.UNKNOWN);
+                break;
+        }
+        if (ordType.equals("Market")) {
+            return ((MarketOrder.Builder) order).build();
+        } else {
+            return ((LimitOrder.Builder) order).build();
+        }
+    }
+}

xchange-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/dto/BitmexWebSocketTransaction.java

@@ -68,6 +68,21 @@ public BitmexTrade[] toBitmexTrades() {
         return trades;
     }
 
+    public BitmexOrder[] toBitmexOrders() {
+        BitmexOrder[] orders = new BitmexOrder[this.data.size()];
+        for(int i = 0; i < this.data.size(); ++i) {
+            JsonNode jsonOrder = this.data.get(i);
+
+            try {
+                orders[i] = (BitmexOrder) this.mapper.readValue(jsonOrder.toString(), BitmexOrder.class);
+            } catch (IOException var5) {
+                var5.printStackTrace();
+            }
+        }
+
+        return orders;
+    }
+
     public String getTable() {
         return table;
     }
@@ -79,4 +94,6 @@ public String getAction() {
     public JsonNode getData() {
         return data;
     }
+
+
 }

xchange-okcoin/pom.xml

@@ -25,5 +25,15 @@
             <artifactId>xchange-okcoin</artifactId>
             <version>${xchange.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>16.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>fluent-hc</artifactId>
+            <version>4.5.5</version>
+        </dependency>
     </dependencies>
 </project>