Update Historical Price Data Integration: Remove Figure dlob, Add Osm…

…osis (#518) * add osmosis api call and a test for running call * add source column to token historical daily, add date calculation for query to osmosis * refactor async service to use osmosis * dynamically determine time frames for osmosis api * Add tests * change println to debug statement * remove unused imports * fix lints * reorder imports * round to thrid decimal place * move third decimal calls * remove old dlob calls * add change log --------- Co-authored-by: Matt Witkowski <[email protected]>
provenance-io · Jun 6, 2024 · b9972b0 · b9972b0
1 parent 4fdfa73
commit b9972b0
Show file tree

Hide file tree

Showing 10 changed files with 253 additions and 52 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -33,6 +33,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 
 ## Unreleased
 
+* Update historical price data integration by removing figure's dlob and adding osmosis datasource [#519](https://github.com/provenance-io/explorer-service/issues/519)
 * Dynamic loading of proto descriptors [#520](https://github.com/provenance-io/explorer-service/issues/520)
 
 ## [v5.8.0](https://github.com/provenance-io/explorer-service/releases/tag/v5.8.0) - 2024-03-05

diff --git a/...se/src/main/resources/db/migration/V1_90__Add_source_column_to_token_historical_daily.sql b/...se/src/main/resources/db/migration/V1_90__Add_source_column_to_token_historical_daily.sql
@@ -0,0 +1,6 @@
+SELECT 'Add source column to token historical daily' AS comment;
+ALTER TABLE token_historical_daily
+    ADD COLUMN source TEXT;
+
+UPDATE token_historical_daily
+SET source = 'dlob';
diff --git a/service/src/main/kotlin/io/provenance/explorer/domain/entities/ExplorerCache.kt b/service/src/main/kotlin/io/provenance/explorer/domain/entities/ExplorerCache.kt
@@ -241,16 +241,18 @@ object TokenHistoricalDailyTable : IdTable<DateTime>(name = "token_historical_da
     val timestamp = datetime("historical_timestamp")
     override val id = timestamp.entityId()
     val data = jsonb<TokenHistoricalDailyTable, CmcHistoricalQuote>("data", OBJECT_MAPPER)
+    val dataSource = text("source")
 }
 
 class TokenHistoricalDailyRecord(id: EntityID<DateTime>) : Entity<DateTime>(id) {
     companion object : EntityClass<DateTime, TokenHistoricalDailyRecord>(TokenHistoricalDailyTable) {
 
-        fun save(date: DateTime, data: CmcHistoricalQuote) =
+        fun save(date: DateTime, data: CmcHistoricalQuote, source: String) =
             transaction {
                 TokenHistoricalDailyTable.insertIgnore {
                     it[this.timestamp] = date
                     it[this.data] = data
+                    it[this.dataSource] = source
                 }
             }
 

diff --git a/service/src/main/kotlin/io/provenance/explorer/domain/extensions/CoinExtensions.kt b/service/src/main/kotlin/io/provenance/explorer/domain/extensions/CoinExtensions.kt
@@ -103,6 +103,10 @@ fun List<CoinStr>.diff(newList: List<CoinStr>) =
 
 fun BigDecimal.roundWhole() = this.setScale(0, RoundingMode.HALF_EVEN)
 
+fun BigDecimal.toThirdDecimal(): BigDecimal {
+    return this.setScale(3, RoundingMode.DOWN)
+}
+
 fun List<CoinStr>.mapToProtoCoin() =
     this.groupBy({ it.denom }) { it.amount.toBigDecimal() }
         .mapValues { (_, v) -> v.sumOf { it } }

diff --git a/service/src/main/kotlin/io/provenance/explorer/domain/models/explorer/OsmosisModels.kt b/service/src/main/kotlin/io/provenance/explorer/domain/models/explorer/OsmosisModels.kt
@@ -0,0 +1,24 @@
+package io.provenance.explorer.domain.models
+
+import java.math.BigDecimal
+
+data class OsmosisHistoricalPrice(
+    val time: Long,
+    val high: BigDecimal,
+    val low: BigDecimal,
+    val close: BigDecimal,
+    val open: BigDecimal,
+    val volume: BigDecimal
+)
+
+data class OsmosisApiResponse(
+    val result: OsmosisResult
+)
+
+data class OsmosisResult(
+    val data: OsmosisData
+)
+
+data class OsmosisData(
+    val json: List<OsmosisHistoricalPrice>
+)
diff --git a/service/src/main/kotlin/io/provenance/explorer/domain/models/explorer/TokenModels.kt b/service/src/main/kotlin/io/provenance/explorer/domain/models/explorer/TokenModels.kt
@@ -43,7 +43,7 @@ data class TokenHistoricalDataRequest(
             )
         )
 
-    private val tokenHistoricalCsvBaseHeaders: MutableList<String> =
+    val tokenHistoricalCsvBaseHeaders: MutableList<String> =
         mutableListOf("Date", "Open", "High", "Low", "Close", "Volume - USD")
 
     fun datesValidation() =

diff --git a/service/src/main/kotlin/io/provenance/explorer/service/TokenService.kt b/service/src/main/kotlin/io/provenance/explorer/service/TokenService.kt
@@ -6,7 +6,8 @@ import io.ktor.client.call.body
 import io.ktor.client.plugins.ResponseException
 import io.ktor.client.request.accept
 import io.ktor.client.request.get
-import io.ktor.client.request.parameter
+import io.ktor.client.statement.HttpResponse
+import io.ktor.client.statement.bodyAsText
 import io.ktor.http.ContentType
 import io.provenance.explorer.KTOR_CLIENT_JAVA
 import io.provenance.explorer.VANILLA_MAPPER
@@ -24,13 +25,15 @@ import io.provenance.explorer.domain.entities.TokenHistoricalDailyRecord
 import io.provenance.explorer.domain.entities.addressList
 import io.provenance.explorer.domain.entities.vestingAccountTypes
 import io.provenance.explorer.domain.exceptions.validate
+import io.provenance.explorer.domain.extensions.CsvData
 import io.provenance.explorer.domain.extensions.pageCountOfResults
 import io.provenance.explorer.domain.extensions.roundWhole
 import io.provenance.explorer.domain.extensions.startOfDay
 import io.provenance.explorer.domain.extensions.toCoinStr
 import io.provenance.explorer.domain.extensions.toOffset
 import io.provenance.explorer.domain.extensions.toPercentage
-import io.provenance.explorer.domain.models.explorer.DlobHistBase
+import io.provenance.explorer.domain.models.OsmosisApiResponse
+import io.provenance.explorer.domain.models.OsmosisHistoricalPrice
 import io.provenance.explorer.domain.models.explorer.TokenHistoricalDataRequest
 import io.provenance.explorer.grpc.v1.AccountGrpcClient
 import io.provenance.explorer.model.AssetHolder
@@ -49,10 +52,12 @@ import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.jetbrains.exposed.sql.transactions.transaction
 import org.joda.time.DateTime
-import org.joda.time.format.DateTimeFormat
+import org.joda.time.DateTimeZone
+import org.joda.time.Duration
 import org.springframework.stereotype.Service
 import java.math.BigDecimal
 import java.math.RoundingMode
+import java.net.URLEncoder
 import java.util.zip.ZipEntry
 import java.util.zip.ZipOutputStream
 import javax.servlet.ServletOutputStream
@@ -220,58 +225,124 @@ class TokenService(private val accountClient: AccountGrpcClient) {
                 )
             }
     }
+    fun getTokenHistorical(fromDate: DateTime?, toDate: DateTime?) =
+        TokenHistoricalDailyRecord.findForDates(fromDate?.startOfDay(), toDate?.startOfDay())
+
+    fun getTokenLatest() = CacheUpdateRecord.fetchCacheByKey(CacheKeys.UTILITY_TOKEN_LATEST.key)?.cacheValue?.let {
+        VANILLA_MAPPER.readValue<CmcLatestDataAbbrev>(it)
+    }
 
-    fun getHistoricalFromDlob(startTime: DateTime, tickerId: String): DlobHistBase? = runBlocking {
+    fun fetchOsmosisData(fromDate: DateTime?): List<OsmosisHistoricalPrice> = runBlocking {
+        val input = buildInputQuery(fromDate, determineTimeFrame(fromDate))
         try {
-            KTOR_CLIENT_JAVA.get("https://www.dlob.io:443/gecko/external/api/v1/exchange/historical_trades") {
-                parameter("ticker_id", tickerId)
-                parameter("type", "buy")
-                parameter("start_time", DateTimeFormat.forPattern("dd-MM-yyyy").print(startTime))
+            val url = """https://app.osmosis.zone/api/edge-trpc-assets/assets.getAssetHistoricalPrice?input=$input"""
+            val response: HttpResponse = KTOR_CLIENT_JAVA.get(url) {
                 accept(ContentType.Application.Json)
-            }.body()
+            }
+
+            val rawResponse: String = response.bodyAsText()
+            logger.debug("Osmosis GET: $url Raw Response: $rawResponse")
+
+            val osmosisApiResponse: OsmosisApiResponse = response.body()
+            osmosisApiResponse.result.data.json
         } catch (e: ResponseException) {
-            return@runBlocking null.also { logger.error("Error fetching from Dlob: ${e.response}") }
+            logger.error("Error fetching from Osmosis API: ${e.response}")
+            emptyList()
         } catch (e: Exception) {
-            return@runBlocking null.also { logger.error("Error fetching from Dlob: ${e.message}") }
-        } catch (e: Throwable) {
-            return@runBlocking null.also { logger.error("Error fetching from Dlob: ${e.message}") }
+            logger.error("Error fetching from Osmosis API: ${e.message}")
+            emptyList()
         }
     }
 
-    fun getHistoricalFromDlob(startTime: DateTime): DlobHistBase? {
-        val tickerIds = listOf("HASH_USD", "HASH_USDOMNI")
+    enum class TimeFrame(val minutes: Int) {
+        FIVE_MINUTES(5),
+        TWO_HOURS(120),
+        ONE_DAY(1440)
+    }
 
-        val dlobHistorical = tickerIds
-            .flatMap { getHistoricalFromDlob(startTime, it)?.buy.orEmpty() }
+    /**
+     * Determines the appropriate TimeFrame based on the fromDate.
+     *
+     * @param fromDate The starting date to determine the time frame.
+     * @return The appropriate TimeFrame enum value.
+     */
+    fun determineTimeFrame(fromDate: DateTime?): TimeFrame {
+        val now = DateTime.now(DateTimeZone.UTC)
+        val duration = Duration(fromDate, now)
 
-        return if (dlobHistorical.isNotEmpty()) DlobHistBase(dlobHistorical) else null
+        return when {
+            duration.standardDays <= 14 -> TimeFrame.FIVE_MINUTES
+            duration.standardDays <= 60 -> TimeFrame.TWO_HOURS
+            else -> TimeFrame.ONE_DAY
+        }
     }
 
-    fun getTokenHistorical(fromDate: DateTime?, toDate: DateTime?) =
-        TokenHistoricalDailyRecord.findForDates(fromDate?.startOfDay(), toDate?.startOfDay())
-
-    fun getTokenLatest() = CacheUpdateRecord.fetchCacheByKey(CacheKeys.UTILITY_TOKEN_LATEST.key)?.cacheValue?.let {
-        VANILLA_MAPPER.readValue<CmcLatestDataAbbrev>(it)
+    /**
+     * Builds the input query parameter for fetching historical data.
+     *
+     * This function constructs a URL-encoded JSON query parameter for fetching historical data based on the given
+     * `fromDate` and `timeFrame`. The `timeFrame` represents the number of minutes between updates. The allowed values
+     * for `timeFrame` are defined in the `TimeFrame` enum:
+     * - FIVE_MINUTES: data goes back 2 weeks.
+     * - TWO_HOURS: data goes back 2 months.
+     * - ONE_DAY: data goes back to the beginning of time.
+     *
+     * The function calculates the total number of frames (`numRecentFrames`) from the `fromDate` to the current time,
+     * based on the specified `timeFrame`.
+     *
+     * @param fromDate The starting date from which to calculate the number of frames.
+     * @param timeFrame The time interval between updates, specified as a `TimeFrame` enum value.
+     * @return A URL-encoded JSON string to be used as a query parameter for fetching historical data.
+     */
+    fun buildInputQuery(fromDate: DateTime?, timeFrame: TimeFrame): String {
+        val coinDenom = "ibc/CE5BFF1D9BADA03BB5CCA5F56939392A761B53A10FBD03B37506669C3218D3B2"
+        val now = DateTime.now(DateTimeZone.UTC)
+        val duration = Duration(fromDate, now)
+        val numRecentFrames = (duration.standardMinutes / timeFrame.minutes).toInt()
+        return URLEncoder.encode(
+            """{"json":{"coinDenom":"$coinDenom","timeFrame":{"custom":{"timeFrame":${timeFrame.minutes},"numRecentFrames":$numRecentFrames}}}}""",
+            "UTF-8"
+        )
     }
 
     fun getHashPricingDataDownload(filters: TokenHistoricalDataRequest, resp: ServletOutputStream): ZipOutputStream {
         validate(filters.datesValidation())
         val baseFileName = filters.getFileNameBase()
-        val fileList = filters.getFileList()
+
+        val fileList = runBlocking {
+            val data = fetchOsmosisData(filters.fromDate)
+            listOf(
+                CsvData(
+                    "TokenHistoricalData",
+                    filters.tokenHistoricalCsvBaseHeaders,
+                    data.map { it.toCsv() }
+                )
+            )
+        }
 
         val zos = ZipOutputStream(resp)
         fileList.forEach { file ->
             zos.putNextEntry(ZipEntry("$baseFileName - ${file.fileName}.csv"))
             zos.write(file.writeCsvEntry())
             zos.closeEntry()
         }
-        // Adding in a txt file with the applied filters
         zos.putNextEntry(ZipEntry("$baseFileName - FILTERS.txt"))
         zos.write(filters.writeFilters())
         zos.closeEntry()
         zos.close()
         return zos
     }
+
+    private fun OsmosisHistoricalPrice.toCsv(): List<String> {
+        return listOf(
+            time.toString(),
+            open.toString(),
+            high.toString(),
+            low.toString(),
+            close.toString(),
+            volume.toString()
+        )
+    }
 }
 
 fun BigDecimal.asPercentOf(divisor: BigDecimal): BigDecimal = this.divide(divisor, 20, RoundingMode.CEILING)
diff --git a/service/src/main/kotlin/io/provenance/explorer/service/async/AsyncService.kt b/service/src/main/kotlin/io/provenance/explorer/service/async/AsyncService.kt
@@ -40,7 +40,8 @@ import io.provenance.explorer.domain.extensions.monthToQuarter
 import io.provenance.explorer.domain.extensions.percentChange
 import io.provenance.explorer.domain.extensions.startOfDay
 import io.provenance.explorer.domain.extensions.toDateTime
-import io.provenance.explorer.domain.models.explorer.DlobHistorical
+import io.provenance.explorer.domain.extensions.toThirdDecimal
+import io.provenance.explorer.domain.models.OsmosisHistoricalPrice
 import io.provenance.explorer.grpc.extensions.getMsgSubTypes
 import io.provenance.explorer.grpc.extensions.getMsgType
 import io.provenance.explorer.model.CmcHistoricalQuote
@@ -298,71 +299,71 @@ class AsyncService(
         if (latest != null) {
             startDate = latest.timestamp.minusDays(1).startOfDay()
         }
-        val dlobRes = tokenService.getHistoricalFromDlob(startDate) ?: return
-        logger.info("Updating token historical data starting from $startDate with ${dlobRes.buy.size} buy records for roll-up.")
+        val dlobRes = tokenService.fetchOsmosisData(startDate) ?: return
+        logger.info("Updating token historical data starting from $startDate with ${dlobRes.size} buy records for roll-up.")
 
         val baseMap = Interval(startDate, today)
             .let { int -> generateSequence(int.start) { dt -> dt.plusDays(1) }.takeWhile { dt -> dt < int.end } }
-            .map { it to emptyList<DlobHistorical>() }.toMap().toMutableMap()
+            .map { it to emptyList<OsmosisHistoricalPrice>() }.toMap().toMutableMap()
         var prevPrice = TokenHistoricalDailyRecord.lastKnownPriceForDate(startDate)
 
         baseMap.putAll(
-            dlobRes.buy
-                .filter { DateTime(it.trade_timestamp * 1000).startOfDay() != today }
-                .groupBy { DateTime(it.trade_timestamp * 1000).startOfDay() }
+            dlobRes
+                .filter { DateTime(it.time * 1000).startOfDay() != today }
+                .groupBy { DateTime(it.time * 1000).startOfDay() }
         )
         baseMap.forEach { (k, v) ->
-            val high = v.maxByOrNull { it.price }
-            val low = v.minByOrNull { it.price }
-            val open = v.minByOrNull { DateTime(it.trade_timestamp * 1000) }?.price ?: prevPrice
-            val close = v.maxByOrNull { DateTime(it.trade_timestamp * 1000) }?.price ?: prevPrice
+            val high = v.maxByOrNull { it.high.toThirdDecimal() }
+            val low = v.minByOrNull { it.low.toThirdDecimal() }
+            val open = v.minByOrNull { DateTime(it.time * 1000) }?.open ?: prevPrice
+            val close = v.maxByOrNull { DateTime(it.time * 1000) }?.close ?: prevPrice
             val closeDate = k.plusDays(1).minusMillis(1)
-            val usdVolume = v.sumOf { it.target_volume }.stripTrailingZeros()
+            val usdVolume = v.sumOf { it.volume.toThirdDecimal() }.stripTrailingZeros()
             val record = CmcHistoricalQuote(
                 time_open = k,
                 time_close = closeDate,
-                time_high = if (high != null) DateTime(high.trade_timestamp * 1000) else k,
-                time_low = if (low != null) DateTime(low.trade_timestamp * 1000) else k,
+                time_high = if (high != null) DateTime(high.time * 1000) else k,
+                time_low = if (low != null) DateTime(low.time * 1000) else k,
                 quote = mapOf(
                     USD_UPPER to
                         CmcQuote(
                             open = open,
-                            high = high?.price ?: prevPrice,
-                            low = low?.price ?: prevPrice,
+                            high = high?.high ?: prevPrice,
+                            low = low?.low ?: prevPrice,
                             close = close,
                             volume = usdVolume,
                             market_cap = close.multiply(
                                 tokenService.totalSupply().divide(UTILITY_TOKEN_BASE_MULTIPLIER)
-                            ),
+                            ).toThirdDecimal(),
                             timestamp = closeDate
                         )
                 )
             ).also { prevPrice = close }
-            TokenHistoricalDailyRecord.save(record.time_open.startOfDay(), record)
+            TokenHistoricalDailyRecord.save(record.time_open.startOfDay(), record, "osmosis")
         }
     }
 
     @Scheduled(cron = "0 0/5 * * * ?") // Every 5 minutes
     fun updateTokenLatest() {
         val today = DateTime.now().withZone(DateTimeZone.UTC)
         val startDate = today.minusDays(7)
-        tokenService.getHistoricalFromDlob(startDate)?.buy
-            ?.sortedBy { it.trade_timestamp }
+        tokenService.fetchOsmosisData(startDate)
+            ?.sortedBy { it.time }
             ?.let { list ->
-                val prevRecIdx = list.indexOfLast { DateTime(it.trade_timestamp * 1000).isBefore(today.minusDays(1)) }
+                val prevRecIdx = list.indexOfLast { DateTime(it.time * 1000).isBefore(today.minusDays(1)) }
                 val prevRecord = list[prevRecIdx]
-                val price = list.last().price
+                val price = list.last().close.toThirdDecimal()
                 val percentChg = if (prevRecIdx == list.lastIndex) {
                     BigDecimal.ZERO
                 } else {
-                    price.percentChange(prevRecord.price)
+                    price.percentChange(prevRecord.close.toThirdDecimal())
                 }
                 val vol24Hr = if (prevRecIdx == list.lastIndex) {
                     BigDecimal.ZERO
                 } else {
-                    list.subList(prevRecIdx + 1, list.lastIndex + 1).sumOf { it.target_volume }.stripTrailingZeros()
+                    list.subList(prevRecIdx + 1, list.lastIndex + 1).sumOf { it.volume.toThirdDecimal() }.stripTrailingZeros()
                 }
-                val marketCap = price.multiply(tokenService.totalSupply().divide(UTILITY_TOKEN_BASE_MULTIPLIER))
+                val marketCap = price.multiply(tokenService.totalSupply().divide(UTILITY_TOKEN_BASE_MULTIPLIER)).toThirdDecimal()
                 val rec = CmcLatestDataAbbrev(
                     today,
                     mapOf(USD_UPPER to CmcLatestQuoteAbbrev(price, percentChg, vol24Hr, marketCap, today))