feat: change hubi websocket endpoint

namjug-kim · namjug-kim · commit c1b4de074f23 · 2021-01-15T18:31:37.000+09:00
diff --git a/reactive-crypto-hubi/src/main/kotlin/com/njkim/reactivecrypto/hubi/HubiJsonObjectMapper.kt b/reactive-crypto-hubi/src/main/kotlin/com/njkim/reactivecrypto/hubi/HubiJsonObjectMapper.kt
@@ -23,19 +23,26 @@ import com.fasterxml.jackson.databind.JsonDeserializer
 import com.njkim.reactivecrypto.core.ExchangeJsonObjectMapper
 import com.njkim.reactivecrypto.core.common.model.currency.CurrencyPair
 import com.njkim.reactivecrypto.core.common.model.order.TradeSideType
+import com.njkim.reactivecrypto.core.common.util.CurrencyPairUtil
 import java.io.IOException
 import java.math.BigDecimal
-import java.time.Instant
-import java.time.ZoneId
+import java.time.ZoneOffset
 import java.time.ZonedDateTime
+import java.time.format.DateTimeFormatterBuilder
+import java.util.*
 
 class HubiJsonObjectMapper : ExchangeJsonObjectMapper {
 
     override fun zonedDateTimeDeserializer(): JsonDeserializer<ZonedDateTime>? {
+        val dateTimeFormatter = DateTimeFormatterBuilder()
+            .parseCaseInsensitive()
+            .appendPattern("MMM dd, yyyy hh:mm:ss a")
+            .toFormatter(Locale.ENGLISH)
+
         return object : JsonDeserializer<ZonedDateTime>() {
             @Throws(IOException::class, JsonProcessingException::class)
             override fun deserialize(p: JsonParser, ctxt: DeserializationContext): ZonedDateTime {
-                return Instant.ofEpochMilli(p.valueAsLong).atZone(ZoneId.systemDefault())
+                return ZonedDateTime.parse(p.valueAsString, dateTimeFormatter.withZone(ZoneOffset.UTC))
             }
         }
     }
@@ -44,8 +51,8 @@ class HubiJsonObjectMapper : ExchangeJsonObjectMapper {
         return object : JsonDeserializer<CurrencyPair>() {
             @Throws(IOException::class, JsonProcessingException::class)
             override fun deserialize(p: JsonParser, ctxt: DeserializationContext): CurrencyPair {
-                val splitCurrencyPair = p.valueAsString.split("_")
-                return CurrencyPair.parse(splitCurrencyPair[0], splitCurrencyPair[1])
+                val rawCurrencyPair = p.valueAsString
+                return CurrencyPairUtil.parse(rawCurrencyPair) ?: error("can't find currencyPair : $rawCurrencyPair")
             }
         }
     }
diff --git a/reactive-crypto-hubi/src/main/kotlin/com/njkim/reactivecrypto/hubi/HubiWebsocketClient.kt b/reactive-crypto-hubi/src/main/kotlin/com/njkim/reactivecrypto/hubi/HubiWebsocketClient.kt
@@ -27,21 +27,20 @@ import com.njkim.reactivecrypto.core.common.model.order.TickData
 import com.njkim.reactivecrypto.core.common.model.order.TradeSideType
 import com.njkim.reactivecrypto.core.common.util.toEpochMilli
 import com.njkim.reactivecrypto.core.websocket.AbstractExchangeWebsocketClient
-import com.njkim.reactivecrypto.hubi.model.HubiMessageFrame
-import com.njkim.reactivecrypto.hubi.model.HubiOrderBook
-import com.njkim.reactivecrypto.hubi.model.HubiTickDataWrapper
+import com.njkim.reactivecrypto.hubi.model.HubiDepthResponse
 import mu.KotlinLogging
 import reactor.core.publisher.Flux
 import reactor.kotlin.core.publisher.toFlux
 import reactor.netty.http.client.HttpClient
+import java.math.BigDecimal
 import java.time.ZonedDateTime
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicLong
 
 class HubiWebsocketClient : AbstractExchangeWebsocketClient() {
     private val log = KotlinLogging.logger {}
 
-    private val baseUri = "wss://api.hubi.com/ws/connect/v1"
+    private val baseUri = "wss://api.hubi.com/ws/futures/public/market"
 
     private val objectMapper: ObjectMapper = createJsonObjectMapper().objectMapper()
 
@@ -50,12 +49,12 @@ class HubiWebsocketClient : AbstractExchangeWebsocketClient() {
     }
 
     override fun createDepthSnapshot(subscribeTargets: List<CurrencyPair>): Flux<OrderBook> {
-        val subscribeRequests = subscribeTargets
-            .map { "${it.baseCurrency.symbol}${it.quoteCurrency.symbol}".toLowerCase() }
+        val currentOrderBookMap: MutableMap<CurrencyPair, OrderBook> = ConcurrentHashMap()
+
+        val subscribeRequests = subscribeTargets.asSequence()
+            .map { "${it.baseCurrency.symbol}${it.quoteCurrency.symbol}".toUpperCase() }
             .map { symbol ->
-                """
-                    {"channel":"depth_all","symbol":"$symbol"}
-                """.trimIndent()
+                """{"op":"subscribe", "channel":"/api/depth/depth", "key":"$symbol"}"""
             }
             .toFlux()
 
@@ -68,66 +67,119 @@ class HubiWebsocketClient : AbstractExchangeWebsocketClient() {
                     .then()
                     .thenMany(inbound.aggregateFrames().receive().asString())
             }
-            .filter { it.contains("\"dataType\":\"depth_all\"") }
-            .map { objectMapper.readValue<HubiMessageFrame<HubiOrderBook>>(it) }
+            .filter { it.contains(""""event":"/api/depth/depth"""") }
+            .map { objectMapper.readValue<HubiDepthResponse>(it) }
             .map { messageFrame ->
                 val eventTime = ZonedDateTime.now()
                 OrderBook(
-                    "${messageFrame.symbol}${eventTime.toEpochMilli()}",
-                    messageFrame.symbol,
+                    "${messageFrame.key}${eventTime.toEpochMilli()}",
+                    messageFrame.key,
                     eventTime,
                     ExchangeVendor.HUBI,
-                    messageFrame.data.bids.map { OrderBookUnit(it.price, it.amount, TradeSideType.BUY, null) },
-                    messageFrame.data.asks.map { OrderBookUnit(it.price, it.amount, TradeSideType.SELL, null) }.sortedBy { it.price }
+                    messageFrame.buyDepth.map { OrderBookUnit(it.price, it.qty, TradeSideType.BUY, it.count) },
+                    messageFrame.sellDepth.map { OrderBookUnit(it.price, it.qty, TradeSideType.SELL, it.count) }
+                        .sortedBy { it.price }
                 )
             }
+            .map { orderBook ->
+                if (!currentOrderBookMap.containsKey(orderBook.currencyPair)) {
+                    val filteredOrderBook = orderBook.copy(
+                        bids = orderBook.bids.filter { it.quantity > BigDecimal.ZERO },
+                        asks = orderBook.asks.filter { it.quantity > BigDecimal.ZERO }
+                    )
+                    currentOrderBookMap[orderBook.currencyPair] = filteredOrderBook
+                    return@map filteredOrderBook
+                }
+
+                val prevOrderBook = currentOrderBookMap[orderBook.currencyPair]!!
+
+                val askMap: MutableMap<BigDecimal, OrderBookUnit> = prevOrderBook.asks
+                    .associateBy { it.price.stripTrailingZeros() }
+                    .toMutableMap()
+
+                orderBook.asks.forEach { updatedAsk ->
+                    askMap.compute(updatedAsk.price.stripTrailingZeros()) { _, oldValue ->
+                        when {
+                            updatedAsk.quantity <= BigDecimal.ZERO -> null
+                            oldValue == null -> updatedAsk
+                            else -> oldValue.copy(
+                                quantity = updatedAsk.quantity,
+                                orderNumbers = updatedAsk.orderNumbers
+                            )
+                        }
+                    }
+                }
+
+                val bidMap: MutableMap<BigDecimal, OrderBookUnit> = prevOrderBook.bids
+                    .associateBy { it.price.stripTrailingZeros() }
+                    .toMutableMap()
+
+                orderBook.bids.forEach { updatedBid ->
+                    bidMap.compute(updatedBid.price.stripTrailingZeros()) { _, oldValue ->
+                        when {
+                            updatedBid.quantity <= BigDecimal.ZERO -> null
+                            oldValue == null -> updatedBid
+                            else -> oldValue.copy(
+                                quantity = updatedBid.quantity,
+                                orderNumbers = updatedBid.orderNumbers
+                            )
+                        }
+                    }
+                }
+
+                val currentOrderBook = prevOrderBook.copy(
+                    eventTime = orderBook.eventTime,
+                    asks = askMap.values.sortedBy { orderBookUnit -> orderBookUnit.price },
+                    bids = bidMap.values.sortedByDescending { orderBookUnit -> orderBookUnit.price }
+                )
+                currentOrderBookMap[currentOrderBook.currencyPair] = currentOrderBook
+                currentOrderBook
+            }
+            .doFinally { currentOrderBookMap.clear() } // cleanup memory limit orderBook when disconnected
     }
 
     override fun createTradeWebsocket(subscribeTargets: List<CurrencyPair>): Flux<TickData> {
         val lastPublishedTimestamp: MutableMap<CurrencyPair, AtomicLong> = ConcurrentHashMap()
 
-        val subscribeRequests = subscribeTargets
-            .map { "${it.baseCurrency.symbol}${it.quoteCurrency.symbol}".toLowerCase() }
+        val subscribeRequests = subscribeTargets.asSequence()
+            .map { "${it.baseCurrency.symbol}${it.quoteCurrency.symbol}".toUpperCase() }
             .map { symbol ->
-                """
-                    {"channel":"trade_history","symbol":"$symbol"}
-                """.trimIndent()
+                """{"op":"subscribe", "channel":"/api/depth/depth", "key":"$symbol"}"""
             }
             .toFlux()
 
         return HttpClient.create()
             .wiretap(log.isDebugEnabled)
-            .websocket(65536)
+            .websocket(262144)
             .uri(baseUri)
             .handle { inbound, outbound ->
                 outbound.sendString(subscribeRequests)
                     .then()
-                    .thenMany(inbound.aggregateFrames(65536).receive().asString())
+                    .thenMany(inbound.aggregateFrames().receive().asString())
             }
-            .filter { it.contains("\"dataType\":\"trade_history\"") }
-            .map { objectMapper.readValue<HubiMessageFrame<HubiTickDataWrapper>>(it) }
-            .map { it.data }
+            .filter { it.contains(""""event":"/api/depth/depth"""") }
+            .map { objectMapper.readValue<HubiDepthResponse>(it) }
             .flatMapIterable {
                 it.trades
                     .takeWhile { hubiTickData ->
                         // hubi trade history response contain history data
                         val lastTradeEpochMilli =
                             lastPublishedTimestamp.computeIfAbsent(hubiTickData.symbol) { AtomicLong() }
-                        val isNew = hubiTickData.time.toEpochMilli() > lastTradeEpochMilli.toLong()
+                        val isNew = hubiTickData.timestamp.toEpochMilli() > lastTradeEpochMilli.toLong()
                         if (isNew) {
-                            lastTradeEpochMilli.set(hubiTickData.time.toEpochMilli())
+                            lastTradeEpochMilli.set(hubiTickData.timestamp.toEpochMilli())
                         }
                         isNew
                     }
                     .map { hubiTickData ->
                         TickData(
-                            "${hubiTickData.symbol}${hubiTickData.time}",
-                            hubiTickData.time,
+                            "${hubiTickData.symbol}${hubiTickData.timestamp}",
+                            hubiTickData.timestamp,
                             hubiTickData.price,
-                            hubiTickData.amount,
+                            hubiTickData.qty,
                             hubiTickData.symbol,
                             ExchangeVendor.HUBI,
-                            hubiTickData.type
+                            if (hubiTickData.buyActive) TradeSideType.BUY else TradeSideType.SELL
                         )
                     }
                     .reversed()
diff --git a/reactive-crypto-hubi/src/main/kotlin/com/njkim/reactivecrypto/hubi/model/HubiDepthResponse.kt b/reactive-crypto-hubi/src/main/kotlin/com/njkim/reactivecrypto/hubi/model/HubiDepthResponse.kt
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2019 namjug-kim
+ *
+ * LINE Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.njkim.reactivecrypto.hubi.model
+
+import com.fasterxml.jackson.annotation.JsonProperty
+import com.njkim.reactivecrypto.core.common.model.currency.CurrencyPair
+import java.math.BigDecimal
+import java.time.ZonedDateTime
+
+data class HubiDepthResponse(
+    @get:JsonProperty("buyDepth")
+    val buyDepth: List<HubiDepthUnit>,
+
+    @get:JsonProperty("sellDepth")
+    val sellDepth: List<HubiDepthUnit>,
+
+    @get:JsonProperty("trades")
+    val trades: List<HubiTradeUnit>,
+
+    @get:JsonProperty("key")
+    val key: CurrencyPair,
+
+    @get:JsonProperty("event")
+    val event: String
+)
+
+data class HubiDepthUnit(
+    @get:JsonProperty("price")
+    val price: BigDecimal,
+
+    @get:JsonProperty("qty")
+    val qty: BigDecimal,
+
+    @get:JsonProperty("count")
+    val count: Int,
+
+    @get:JsonProperty("iceCount")
+    val iceCount: Int
+)
+
+data class HubiTradeUnit(
+    @get:JsonProperty("id")
+    val id: String,
+
+    @get:JsonProperty("symbol")
+    val symbol: CurrencyPair,
+
+    @get:JsonProperty("price")
+    val price: BigDecimal,
+
+    @get:JsonProperty("qty")
+    val qty: BigDecimal,
+
+    @get:JsonProperty("buyActive")
+    val buyActive: Boolean,
+
+    @get:JsonProperty("timestamp")
+    val timestamp: ZonedDateTime
+)
diff --git a/reactive-crypto-hubi/src/main/kotlin/com/njkim/reactivecrypto/hubi/model/HubiMessageFrame.kt b/reactive-crypto-hubi/src/main/kotlin/com/njkim/reactivecrypto/hubi/model/HubiMessageFrame.kt
@@ -24,12 +24,12 @@ data class HubiMessageFrame<T>(
     @get:JsonProperty("timestamp")
     val timestamp: ZonedDateTime?,
 
-    @get:JsonProperty("symbol")
-    val symbol: CurrencyPair,
+    @get:JsonProperty("key")
+    val key: CurrencyPair,
 
-    @get:JsonProperty("dataType")
-    val dataType: String,
+    @get:JsonProperty("event")
+    val event: String,
 
     @get:JsonProperty("data")
     val data: T
-)
+)
diff --git a/reactive-crypto-hubi/src/main/kotlin/com/njkim/reactivecrypto/hubi/model/HubiOrderBook.kt b/reactive-crypto-hubi/src/main/kotlin/com/njkim/reactivecrypto/hubi/model/HubiOrderBook.kt
diff --git a/reactive-crypto-hubi/src/test/java/com/njkim/reactivecrypto/hubi/HubiWebsocketClientJavaTest.java b/reactive-crypto-hubi/src/test/java/com/njkim/reactivecrypto/hubi/HubiWebsocketClientJavaTest.java
@@ -17,7 +17,7 @@ public class HubiWebsocketClientJavaTest {
     @Test
     public void hubi_tick_data_subscribe() {
         // given
-        CurrencyPair targetCurrencyPair = CurrencyPair.parse("BTC", "USDT");
+        CurrencyPair targetCurrencyPair = CurrencyPair.parse("XBTC", "USD");
         Flux<TickData> tickDataFlux = new HubiWebsocketClient()
                 .createTradeWebsocket(Collections.singletonList(targetCurrencyPair));
 
@@ -52,12 +52,14 @@ public void hubi_tick_data_subscribe() {
     @Test
     public void hubi_orderBook_subscribe() {
         // given
-        CurrencyPair targetCurrencyPair = CurrencyPair.parse("BTC", "USDT");
+        CurrencyPair targetCurrencyPair = CurrencyPair.parse("XBTC", "USD");
         Flux<OrderBook> orderBookFlux = new HubiWebsocketClient()
                 .createDepthSnapshot(Collections.singletonList(targetCurrencyPair));
 
         // when
-        StepVerifier.create(orderBookFlux.limitRequest(2))
+        StepVerifier.create(orderBookFlux.limitRequest(5))
+                // skip first 3 request
+                .expectNextCount(3)
                 // then
                 .assertNext(orderBook -> {
                     assertThat(orderBook).isNotNull();
@@ -115,4 +117,4 @@ public void hubi_orderBook_subscribe() {
                 })
                 .verifyComplete();
     }
-}
+}
diff --git a/reactive-crypto-hubi/src/test/kotlin/com/njkim/reactivecrypto/hubi/HubiWebsocketClientTest.kt b/reactive-crypto-hubi/src/test/kotlin/com/njkim/reactivecrypto/hubi/HubiWebsocketClientTest.kt
