Pools summary EP

.bsp/sbt.json

@@ -1 +1 @@
-{"name":"sbt","version":"1.5.5","bspVersion":"2.0.0-M5","languages":["scala"],"argv":["/Users/oskin/.sdkman/candidates/java/17.0.1-ms/bin/java","-Xms100m","-Xmx100m","-classpath","/Users/oskin/Library/Application Support/JetBrains/IntelliJIdea2021.2/plugins/Scala/launcher/sbt-launch.jar","xsbt.boot.Boot","-bsp","--sbt-launch-jar=/Users/oskin/Library/Application%20Support/JetBrains/IntelliJIdea2021.2/plugins/Scala/launcher/sbt-launch.jar"]}
+{"name":"sbt","version":"1.5.5","bspVersion":"2.0.0-M5","languages":["scala"],"argv":["/Library/Java/JavaVirtualMachines/jdk-10.0.2.jdk/Contents/Home/bin/java","-Xms100m","-Xmx100m","-classpath","/Users/oskin/Library/Application Support/JetBrains/Toolbox/apps/IDEA-C/ch-0/213.6777.52/IntelliJ IDEA CE.app.plugins/Scala/launcher/sbt-launch.jar","xsbt.boot.Boot","-bsp","--sbt-launch-jar=/Users/oskin/Library/Application%20Support/JetBrains/Toolbox/apps/IDEA-C/ch-0/213.6777.52/IntelliJ%20IDEA%20CE.app.plugins/Scala/launcher/sbt-launch.jar"]}

modules/markets-api/src/main/scala/org/ergoplatform/dex/markets/api/v1/endpoints/AmmStatsEndpoints.scala

@@ -14,7 +14,7 @@ final class AmmStatsEndpoints {
   val Group      = "ammStats"
 
   def endpoints: List[Endpoint[_, _, _, _]] =
-    getPoolLocks :: getPlatformStats :: getPoolStats :: getAvgPoolSlippage :: getPoolPriceChart :: getAmmMarkets :: convertToFiat :: Nil
+    getPoolLocks :: getPlatformStats :: getPoolStats :: getPoolsStats :: getAvgPoolSlippage :: getPoolPriceChart :: getAmmMarkets :: convertToFiat :: Nil
 
   def getPoolLocks: Endpoint[(PoolId, Int), HttpError, List[LiquidityLockInfo], Any] =
     baseEndpoint.get
@@ -34,6 +34,15 @@ final class AmmStatsEndpoints {
       .name("Pool stats")
       .description("Get statistics on the pool with the given ID")
 
+  def getPoolsStats: Endpoint[TimeWindow, HttpError, List[PoolSummary], Any] =
+    baseEndpoint.get
+      .in(PathPrefix / "pools" / "stats")
+      .in(timeWindow)
+      .out(jsonBody[List[PoolSummary]])
+      .tag(Group)
+      .name("Pools stats")
+      .description("Get statistics on all pools")
+
   def getPlatformStats: Endpoint[TimeWindow, HttpError, PlatformSummary, Any] =
     baseEndpoint.get
       .in(PathPrefix / "platform" / "stats")

modules/markets-api/src/main/scala/org/ergoplatform/dex/markets/api/v1/models/amm/types.scala

@@ -3,11 +3,17 @@ package org.ergoplatform.dex.markets.api.v1.models.amm
 import derevo.circe.{decoder, encoder}
 import derevo.derive
 import io.estatico.newtype.macros.newtype
+import org.ergoplatform.dex.markets.db.models.amm.{PoolFeesSnapshot, PoolInfo, PoolSnapshot, PoolVolumeSnapshot}
 import sttp.tapir.Schema
 import scala.math.BigDecimal.RoundingMode
 
 object types {
 
+  final case class PoolBundle(pools: List[PoolSnapshot],
+                              infos: List[PoolInfo],
+                              volumes: List[PoolVolumeSnapshot],
+                              fees: List[PoolFeesSnapshot])
+
   @derive(encoder, decoder)
   @newtype case class RealPrice(value: BigDecimal) {
     def setScale(scale: Int): RealPrice = RealPrice(value.setScale(scale, RoundingMode.HALF_UP))

modules/markets-api/src/main/scala/org/ergoplatform/dex/markets/api/v1/routes/AmmStatsRoutes.scala

@@ -21,7 +21,7 @@ final class AmmStatsRoutes[
   private val interpreter = Http4sServerInterpreter(opts)
 
   def routes: HttpRoutes[F] =
-    caching.middleware(getPoolLocksR <+> getPlatformStatsR <+> getPoolStatsR <+> getAvgPoolSlippageR <+> getPoolPriceChartR <+> getAmmMarketsR <+> convertToFiatR)
+    caching.middleware(getPoolLocksR <+> getPlatformStatsR <+> getPoolStatsR <+> getPoolsStatsR <+> getAvgPoolSlippageR <+> getPoolPriceChartR <+> getAmmMarketsR <+> convertToFiatR)
 
   def getPoolLocksR: HttpRoutes[F] = interpreter.toRoutes(getPoolLocks) { case (poolId, leastDeadline) =>
     locks.byPool(poolId, leastDeadline).adaptThrowable.value
@@ -31,6 +31,10 @@ final class AmmStatsRoutes[
     stats.getPoolSummary(poolId, tw).adaptThrowable.orNotFound(s"PoolStats{poolId=$poolId}").value
   }
 
+  def getPoolsStatsR: HttpRoutes[F] = interpreter.toRoutes(getPoolsStats) { tw =>
+    stats.getPoolsSummary(tw).adaptThrowable.value
+  }
+
   def getPlatformStatsR: HttpRoutes[F] = interpreter.toRoutes(getPlatformStats) { tw =>
     stats.getPlatformSummary(tw).adaptThrowable.value
   }

modules/markets-api/src/main/scala/org/ergoplatform/dex/markets/api/v1/services/AmmStats.scala

@@ -1,7 +1,7 @@
 package org.ergoplatform.dex.markets.api.v1.services
 
 import cats.Monad
-import cats.data.OptionT
+import cats.data.{NonEmptyList, OptionT}
 import cats.effect.Clock
 import mouse.anyf._
 import org.ergoplatform.common.models.TimeWindow
@@ -16,9 +16,13 @@ import org.ergoplatform.dex.markets.repositories.Pools
 import tofu.doobie.transactor.Txr
 import mouse.anyf._
 import cats.syntax.traverse._
-import org.ergoplatform.dex.markets.db.models.amm.{PoolSnapshot, PoolTrace, PoolVolumeSnapshot}
-import org.ergoplatform.dex.domain.{AssetClass, CryptoUnits, MarketId}
-import org.ergoplatform.dex.domain.FullAsset
+import org.ergoplatform.dex.markets.db.models.amm.{
+  PoolFeesSnapshot,
+  PoolSnapshot,
+  PoolTrace,
+  PoolVolumeSnapshot
+}
+import org.ergoplatform.dex.domain.{AssetClass, CryptoUnits, FullAsset, MarketId}
 import org.ergoplatform.dex.markets.modules.AmmStatsMath
 import org.ergoplatform.dex.markets.services.TokenFetcher
 import org.ergoplatform.ergo.TokenId
@@ -35,6 +39,8 @@ trait AmmStats[F[_]] {
 
   def getPoolSummary(poolId: PoolId, window: TimeWindow): F[Option[PoolSummary]]
 
+  def getPoolsSummary(window: TimeWindow): F[List[PoolSummary]]
+
   def getAvgPoolSlippage(poolId: PoolId, depth: Int): F[Option[PoolSlippage]]
 
   def getPoolPriceChart(poolId: PoolId, window: TimeWindow, resolution: Int): F[List[PricePoint]]
@@ -98,6 +104,67 @@ object AmmStats {
       } yield PlatformSummary(tvl, volume)
     }
 
+    private def calculateTVLsFor(pools: List[PoolSnapshot]): F[List[TotalValueLocked]] =
+      pools
+        .traverse { pool =>
+          fiatSolver
+            .convert(pool.lockedX, UsdUnits)
+            .flatMap(optX => fiatSolver.convert(pool.lockedY, UsdUnits).map(optY => (pool, optX, optY)))
+        }
+        .map { equivOpts =>
+          val equivs = equivOpts.filter { case (_, eqX, eqY) => eqX.isDefined && eqY.isDefined }
+          equivs.map { case (_, lx, ly) => TotalValueLocked(lx.get.value + ly.get.value, UsdUnits) }
+        }
+
+    private def convertVolumes(pools: List[PoolSnapshot], volumes: List[PoolVolumeSnapshot], window: TimeWindow) =
+      pools.map(p => volumes.find(_.poolId == p.id)).traverse {
+        case Some(vol) =>
+          for {
+            volX <- fiatSolver.convert(vol.volumeByX, UsdUnits)
+            volY <- fiatSolver.convert(vol.volumeByY, UsdUnits)
+          } yield
+            if (volX.isDefined && volY.isDefined) Volume(volX.get.value + volY.get.value, UsdUnits, window)
+            else Volume.empty(UsdUnits, window)
+        case None => Volume.empty(UsdUnits, window).pure[F]
+      }
+
+    private def convertFees(pools: List[PoolSnapshot], fees: List[PoolFeesSnapshot], window: TimeWindow) =
+      pools.map(p => fees.find(_.poolId == p.id)).traverse {
+        case Some(feesSnap) =>
+          for {
+            feesX <- fiatSolver.convert(feesSnap.feesByX, UsdUnits)
+            feesY <- fiatSolver.convert(feesSnap.feesByY, UsdUnits)
+          } yield
+            if (feesX.isDefined && feesY.isDefined) Fees(feesX.get.value + feesY.get.value, UsdUnits, window)
+            else Fees.empty(UsdUnits, window)
+        case None => Fees.empty(UsdUnits, window).pure[F]
+      }
+
+    def getPoolsSummary(window: TimeWindow): F[List[PoolSummary]] = {
+      val queryPoolsStats =
+        (for {
+          snaps   <- OptionT.liftF(pools.snapshots)
+          poolIds <- OptionT.fromOption[D](NonEmptyList.fromList(snaps.map(_.id)))
+          infos   <- OptionT.liftF(pools.infos(poolIds))
+          (allPools, info) =
+            snaps.filter(ps => infos.exists(_.poolId == ps.id)).map(p => (p, infos.find(_.poolId == p.id).get)).unzip
+          volumes <- OptionT.liftF(pools.volumes(window))
+          fees    <- OptionT.liftF(pools.fees(window))
+        } yield PoolBundle(allPools, info, volumes, fees)).value
+      (for {
+        poolBundle: PoolBundle <- OptionT(queryPoolsStats ||> txr.trans)
+        tvls                   <- OptionT.liftF(calculateTVLsFor(poolBundle.pools))
+        vols                   <- OptionT.liftF(convertVolumes(poolBundle.pools, poolBundle.volumes, window))
+        feeSnaps               <- OptionT.liftF(convertFees(poolBundle.pools, poolBundle.fees, window))
+        fullPoolInfos = poolBundle.pools zip poolBundle.infos zip (tvls, vols, feeSnaps).zipped.toList
+        res <- OptionT.liftF(fullPoolInfos.traverse { case ((pool, inf), (tvl, vol, fee)) =>
+                 ammMath
+                   .feePercentProjection(tvl, fee, inf, MillisInYear)
+                   .map(perc => PoolSummary(pool.id, pool.lockedX, pool.lockedY, tvl, vol, fee, perc))
+               })
+      } yield res).value.map(_.toList.flatten)
+    }
+
     def getPoolSummary(poolId: PoolId, window: TimeWindow): F[Option[PoolSummary]] = {
       val queryPoolStats =
         (for {

modules/markets-api/src/main/scala/org/ergoplatform/dex/markets/db/models/amm.scala

@@ -6,7 +6,7 @@ import org.ergoplatform.ergo.TokenId
 
 object amm {
 
-  final case class PoolInfo(confirmedAt: Long)
+  final case class PoolInfo(poolId: PoolId, confirmedAt: Long)
 
   final case class AssetInfo(
     id: TokenId,

modules/markets-api/src/main/scala/org/ergoplatform/dex/markets/db/sql/AnalyticsSql.scala

@@ -1,8 +1,9 @@
 package org.ergoplatform.dex.markets.db.sql
 
+import cats.data.NonEmptyList
 import doobie.implicits._
 import doobie.util.query.Query0
-import doobie.{Fragment, LogHandler}
+import doobie.{Fragment, Fragments, LogHandler}
 import org.ergoplatform.common.models.TimeWindow
 import org.ergoplatform.dex.domain.amm.PoolId
 import org.ergoplatform.dex.markets.db.models.amm._
@@ -12,9 +13,24 @@ final class AnalyticsSql(implicit lg: LogHandler) {
 
   def getInfo(id: PoolId): Query0[PoolInfo] =
     sql"""
-         |select s.timestamp from swaps s where s.pool_id = $id order by s.timestamp asc limit 1
+         |select s.pool_id, s.timestamp from swaps s where s.pool_id = $id order by s.timestamp asc limit 1
          """.stripMargin.query
 
+  def getInfos(ids: NonEmptyList[PoolId]): Query0[PoolInfo] = {
+    val q =
+    sql"""
+         |select s.pool_id, s.timestamp from swaps s
+         |left join (
+         |  select pool_id, min(timestamp) as ts
+         |  from swaps
+         |  group by pool_id
+         |) as sx on s.pool_id = sx.pool_id and s.timestamp = sx.ts
+         |where s.timestamp = sx.ts
+         """.stripMargin
+
+    (q ++ Fragments.in(fr"and s.pool_id", ids)).query
+  }
+
   def getPoolSnapshot(id: PoolId): Query0[PoolSnapshot] =
     sql"""
          |select p.pool_id, p.x_id, p.x_amount, ax.ticker, ax.decimals, p.y_id, p.y_amount, ay.ticker, ay.decimals, 0

modules/markets-api/src/main/scala/org/ergoplatform/dex/markets/repositories/Pools.scala

@@ -1,5 +1,6 @@
 package org.ergoplatform.dex.markets.repositories
 
+import cats.data.NonEmptyList
 import cats.tagless.syntax.functorK._
 import cats.{FlatMap, Functor}
 import derevo.derive
@@ -24,6 +25,10 @@ trait Pools[F[_]] {
     */
   def info(id: PoolId): F[Option[PoolInfo]]
 
+  /** Get general info about pools with given `ids`.
+    */
+  def infos(ids: NonEmptyList[PoolId]): F[List[PoolInfo]]
+
   /** Get snapshots of all pools.
     */
   def snapshots: F[List[PoolSnapshot]]
@@ -84,6 +89,9 @@ object Pools {
     def info(id: PoolId): ConnectionIO[Option[PoolInfo]] =
       sql.getInfo(id).option
 
+    def infos(ids: NonEmptyList[PoolId]): ConnectionIO[List[PoolInfo]] =
+      sql.getInfos(ids).to[List]
+
     def snapshots: ConnectionIO[List[PoolSnapshot]] =
       sql.getPoolSnapshots.to[List]
 
@@ -127,6 +135,13 @@ object Pools {
         _ <- trace"info(poolId=$poolId) -> ${r.size} info entities selected"
       } yield r
 
+    def infos(ids: NonEmptyList[PoolId]): Mid[F, List[PoolInfo]] =
+      for {
+        _ <- trace"info(poolId=$ids)"
+        r <- _
+        _ <- trace"info(poolId=$ids) -> ${r.size} info entities selected"
+      } yield r
+
     def snapshots: Mid[F, List[PoolSnapshot]] =
       for {
         _ <- trace"snapshots"