Add funding payments processing

build.gradle.kts

@@ -53,7 +53,7 @@ allprojects {
 }
 
 group = "exchange.dydx.abacus"
-version = "1.14.6"
+version = "1.14.7"
 
 repositories {
     google()

docs/SubAccount/SubAccountFundingPayment.md

@@ -6,7 +6,7 @@ data class SubaccountFundingPayment(
  val rate: Double,  
  val positionSize: Double,  
  val price: Double?,  
- val effectiveAtMilliSeconds: Double  
+ val createdAtMilliseconds: Double  
 )
 
 ## marketId
@@ -29,6 +29,6 @@ Position size at the moment of payment
 
 price
 
-## effectiveAtMilliSeconds
+## createdAtMilliseconds
 
 Timestamp of the funding payment

src/commonMain/kotlin/exchange.dydx.abacus/output/account/SubaccountFundingPayment.kt

@@ -1,8 +1,6 @@
 package exchange.dydx.abacus.output.account
 
-import exchange.dydx.abacus.protocols.ParserProtocol
 import exchange.dydx.abacus.utils.IList
-import exchange.dydx.abacus.utils.Logger
 import exchange.dydx.abacus.utils.ParsingHelper
 import kollections.JsExport
 import kollections.toIList
@@ -16,67 +14,22 @@ data class SubaccountFundingPayment(
     val rate: Double,
     val positionSize: Double,
     val price: Double?,
-    val effectiveAtMilliSeconds: Double,
+    val createdAtMilliseconds: Double,
+    val side: PositionSide,
 ) {
     companion object {
-        internal fun create(
-            existing: SubaccountFundingPayment?,
-            parser: ParserProtocol,
-            data: Map<*, *>?,
-        ): SubaccountFundingPayment? {
-            Logger.d { "creating Account Funding Payment\n" }
-
-            data?.let {
-                val marketId = parser.asString(data["marketId"])
-                val payment = parser.asDouble(data["payment"])
-                val rate = parser.asDouble(data["rate"])
-                val positionSize = parser.asDouble(data["positionSize"])
-                val price = parser.asDouble(data["price"])
-                val effectiveAtMilliSeconds =
-                    parser.asDatetime(data["effectiveAt"])?.toEpochMilliseconds()?.toDouble()
-                if (marketId != null && payment != null && rate != null && positionSize != null && effectiveAtMilliSeconds != null) {
-                    return if (existing?.marketId != marketId ||
-                        existing.payment != payment ||
-                        existing.rate != rate ||
-                        existing.positionSize != positionSize ||
-                        existing.price != price ||
-                        existing.effectiveAtMilliSeconds != effectiveAtMilliSeconds
-                    ) {
-                        SubaccountFundingPayment(
-                            marketId,
-                            payment,
-                            rate,
-                            positionSize,
-                            price,
-                            effectiveAtMilliSeconds,
-                        )
-                    } else {
-                        existing
-                    }
-                }
-            }
-            Logger.d { "Account Funding Payment not valid" }
-            return null
-        }
-
-        fun create(
+        internal fun merge(
             existing: IList<SubaccountFundingPayment>?,
-            parser: ParserProtocol,
-            data: List<Map<String, Any>>?,
-        ): IList<SubaccountFundingPayment>? {
-            return ParsingHelper.merge(parser, existing, data, { obj, itemData ->
-                val time1 = (obj as SubaccountFundingPayment).effectiveAtMilliSeconds
-                val time2 =
-                    parser.asDatetime(itemData["effectiveAt"])?.toEpochMilliseconds()
-                        ?.toDouble()
-                ParsingHelper.compare(time1, time2 ?: 0.0, true)
-            }, { _, obj, itemData ->
-                obj ?: SubaccountFundingPayment.create(
-                    null,
-                    parser,
-                    parser.asMap(itemData),
-                )
-            })?.toIList()
+            new: IList<SubaccountFundingPayment>?,
+        ): IList<SubaccountFundingPayment> {
+            return ParsingHelper.merge(
+                existing = existing,
+                new = new,
+                comparison = { obj, newItem ->
+                    ParsingHelper.compare(obj.createdAtMilliseconds, newItem.createdAtMilliseconds, true)
+                },
+                syncItems = true,
+            ).toIList()
         }
     }
 }

src/commonMain/kotlin/exchange.dydx.abacus/processor/wallet/WalletProcessor.kt

@@ -11,6 +11,7 @@ import exchange.dydx.abacus.state.manager.BlockAndTime
 import exchange.dydx.abacus.state.manager.HistoricalTradingRewardsPeriod
 import exchange.dydx.abacus.utils.mutable
 import exchange.dydx.abacus.utils.safeSet
+import indexer.codegen.IndexerFundingPaymentResponseObject
 import indexer.codegen.IndexerHistoricalTradingRewardAggregation
 import indexer.codegen.IndexerPnlTicksResponseObject
 import indexer.codegen.IndexerTransferResponseObject
@@ -181,6 +182,19 @@ internal class WalletProcessor(
         return existing
     }
 
+    internal fun processFundingPayments(
+        existing: InternalWalletState,
+        payload: List<IndexerFundingPaymentResponseObject>?,
+        subaccountNumber: Int,
+    ): InternalWalletState {
+        existing.account = v4accountProcessor.processFundingPayments(
+            existing = existing.account,
+            payload = payload,
+            subaccountNumber = subaccountNumber,
+        )
+        return existing
+    }
+
     internal fun processFills(
         existing: InternalWalletState,
         payload: List<IndexerCompositeFillObject>?,

src/commonMain/kotlin/exchange.dydx.abacus/processor/wallet/account/AccountProcessor.kt

@@ -15,6 +15,7 @@ import exchange.dydx.abacus.state.InternalSubaccountState
 import exchange.dydx.abacus.state.manager.BlockAndTime
 import exchange.dydx.abacus.state.manager.HistoricalTradingRewardsPeriod
 import exchange.dydx.abacus.utils.safeSet
+import indexer.codegen.IndexerFundingPaymentResponseObject
 import indexer.codegen.IndexerHistoricalBlockTradingReward
 import indexer.codegen.IndexerHistoricalTradingRewardAggregation
 import indexer.codegen.IndexerPnlTicksResponseObject
@@ -218,6 +219,17 @@ internal class V4AccountProcessor(
         return existing
     }
 
+    internal fun processFundingPayments(
+        existing: InternalAccountState,
+        payload: List<IndexerFundingPaymentResponseObject>?,
+        subaccountNumber: Int,
+    ): InternalAccountState {
+        val subaccount = existing.subaccounts[subaccountNumber] ?: InternalSubaccountState(subaccountNumber = subaccountNumber)
+        val newSubaccount = subaccountsProcessor.processFundingPayments(subaccount, payload)
+        existing.subaccounts[subaccountNumber] = newSubaccount
+        return existing
+    }
+
     internal fun processFills(
         existing: InternalAccountState,
         payload: List<IndexerCompositeFillObject>?,
@@ -229,7 +241,7 @@ internal class V4AccountProcessor(
         return existing
     }
 
-    fun processTransfers(
+    internal fun processTransfers(
         existing: InternalAccountState,
         payload: List<IndexerTransferResponseObject>?,
         subaccountNumber: Int,

src/commonMain/kotlin/exchange.dydx.abacus/processor/wallet/account/FundingPaymentProcessor.kt

@@ -0,0 +1,42 @@
+package exchange.dydx.abacus.processor.wallet.account
+
+import exchange.dydx.abacus.output.account.PositionSide
+import exchange.dydx.abacus.output.account.SubaccountFundingPayment
+import exchange.dydx.abacus.processor.base.BaseProcessor
+import exchange.dydx.abacus.protocols.ParserProtocol
+import indexer.codegen.IndexerFundingPaymentResponseObject
+
+internal interface FundingPaymentProcessorProtocol {
+    fun process(
+        existing: SubaccountFundingPayment?,
+        payload: IndexerFundingPaymentResponseObject,
+    ): SubaccountFundingPayment?
+}
+
+internal class FundingPaymentProcessor(
+    parser: ParserProtocol
+) : BaseProcessor(parser), FundingPaymentProcessorProtocol {
+
+    override fun process(
+        existing: SubaccountFundingPayment?,
+        payload: IndexerFundingPaymentResponseObject
+    ): SubaccountFundingPayment? {
+        val createdAt = parser.asDatetime(payload.createdAt) ?: return null
+        val ticker = parser.asString(payload.ticker) ?: return null
+        val oraclePrice = parser.asDouble(payload.oraclePrice) ?: return null
+        val size = parser.asDouble(payload.size) ?: return null
+        val side = PositionSide.invoke(payload.side) ?: return null
+        val rate = parser.asDouble(payload.rate) ?: return null
+        val payment = parser.asDouble(payload.payment) ?: return null
+
+        return SubaccountFundingPayment(
+            marketId = ticker,
+            payment = payment,
+            rate = rate,
+            positionSize = size,
+            price = oraclePrice,
+            createdAtMilliseconds = createdAt.toEpochMilliseconds().toDouble(),
+            side = side,
+        )
+    }
+}

src/commonMain/kotlin/exchange.dydx.abacus/processor/wallet/account/FundingPaymentsProcessor.kt

@@ -0,0 +1,35 @@
+package exchange.dydx.abacus.processor.wallet.account
+
+import exchange.dydx.abacus.output.account.SubaccountFundingPayment
+import exchange.dydx.abacus.processor.base.BaseProcessor
+import exchange.dydx.abacus.protocols.ParserProtocol
+import indexer.codegen.IndexerFundingPaymentResponseObject
+import kotlinx.datetime.Instant
+
+internal class FundingPaymentsProcessor(
+    parser: ParserProtocol,
+    private val paymentProcessor: FundingPaymentProcessor = FundingPaymentProcessor(parser = parser)
+) : BaseProcessor(parser) {
+    fun process(
+        existing: List<SubaccountFundingPayment>?,
+        payload: List<IndexerFundingPaymentResponseObject>,
+    ): List<SubaccountFundingPayment>? {
+        val new = payload.mapNotNull { eachPayload ->
+            paymentProcessor.process(
+                existing = null,
+                payload = eachPayload,
+            )
+        }
+        return merge(
+            parser = parser,
+            existing = existing,
+            incoming = new,
+            timeField = { item ->
+                item?.createdAtMilliseconds?.toLong()?.let {
+                    Instant.fromEpochMilliseconds(it)
+                }
+            },
+            ascending = true,
+        )
+    }
+}

src/commonMain/kotlin/exchange.dydx.abacus/processor/wallet/account/SubaccountProcessor.kt

@@ -11,6 +11,7 @@ import exchange.dydx.abacus.state.InternalSubaccountCalculated
 import exchange.dydx.abacus.state.InternalSubaccountState
 import exchange.dydx.abacus.state.manager.BlockAndTime
 import indexer.codegen.IndexerAssetPositionResponseObject
+import indexer.codegen.IndexerFundingPaymentResponseObject
 import indexer.codegen.IndexerPerpetualPositionResponseObject
 import indexer.codegen.IndexerPerpetualPositionStatus
 import indexer.codegen.IndexerPnlTicksResponseObject
@@ -30,6 +31,7 @@ internal open class SubaccountProcessor(
     private val fillsProcessor = FillsProcessor(parser, localizer)
     private val transfersProcessor = TransfersProcessor(parser, localizer)
     private val historicalPNLsProcessor = HistoricalPNLsProcessor(parser)
+    private val fundingPaymentProcessor = FundingPaymentsProcessor(parser)
     private val subaccountCalculator = SubaccountCalculator(parser)
 
     internal fun processSubscribed(
@@ -258,6 +260,20 @@ internal open class SubaccountProcessor(
         return existing
     }
 
+    internal fun processFundingPayments(
+        existing: InternalSubaccountState,
+        payload: List<IndexerFundingPaymentResponseObject>?,
+    ): InternalSubaccountState {
+        val newFundingPayments = fundingPaymentProcessor.process(
+            existing = existing.fundingPayments,
+            payload = payload ?: emptyList(),
+        )
+        if (existing.fundingPayments != newFundingPayments) {
+            existing.fundingPayments = newFundingPayments
+        }
+        return existing
+    }
+
     internal fun processFills(
         existing: InternalSubaccountState,
         payload: List<IndexerCompositeFillObject>?,

src/commonMain/kotlin/exchange.dydx.abacus/state/InternalState.kt

@@ -18,6 +18,7 @@ import exchange.dydx.abacus.output.WithdrawalGating
 import exchange.dydx.abacus.output.account.PositionSide
 import exchange.dydx.abacus.output.account.StakingRewards
 import exchange.dydx.abacus.output.account.SubaccountFill
+import exchange.dydx.abacus.output.account.SubaccountFundingPayment
 import exchange.dydx.abacus.output.account.SubaccountHistoricalPNL
 import exchange.dydx.abacus.output.account.SubaccountOrder
 import exchange.dydx.abacus.output.account.SubaccountPositionResources
@@ -316,6 +317,7 @@ internal data class InternalSubaccountState(
     var orders: List<SubaccountOrder>? = null,
     var transfers: List<SubaccountTransfer>? = null,
     var historicalPNLs: List<SubaccountHistoricalPNL>? = null,
+    var fundingPayments: List<SubaccountFundingPayment>? = null,
     var positions: Map<String, InternalPerpetualPosition>? = null,
     var assetPositions: Map<String, InternalAssetPositionState>? = null,
     var subaccountNumber: Int,
@@ -341,6 +343,7 @@ internal data class InternalSubaccountState(
             orders = orders?.map { it.copy() },
             transfers = transfers?.map { it.copy() },
             historicalPNLs = historicalPNLs?.map { it.copy() },
+            fundingPayments = fundingPayments?.map { it.copy() },
             positions = positions?.map { it.key to it.value.copy() }?.toMap(),
             assetPositions = assetPositions?.map { it.key to it.value.copy() }?.toMap(),
             subaccountNumber = subaccountNumber,

src/commonMain/kotlin/exchange.dydx.abacus/state/machine/TradingStateMachine+Candles.kt

@@ -9,7 +9,6 @@ import indexer.codegen.IndexerCandleResponseObject
 import kollections.iListOf
 import kollections.toIList
 
-// Called in test code only
 internal fun TradingStateMachine.candles(payload: String): StateChanges {
     val json = parser.decodeJsonObject(payload)
     return if (json != null) {

src/commonMain/kotlin/exchange.dydx.abacus/state/machine/TradingStateMachine+FundingPayments.kt

@@ -0,0 +1,21 @@
+package exchange.dydx.abacus.state.machine
+
+import exchange.dydx.abacus.protocols.asTypedObject
+import exchange.dydx.abacus.state.Changes
+import exchange.dydx.abacus.state.StateChanges
+import indexer.codegen.IndexerFundingPaymentResponse
+import kollections.iListOf
+
+internal fun TradingStateMachine.fundingPayments(payload: String, subaccountNumber: Int): StateChanges {
+    val response = parser.asTypedObject<IndexerFundingPaymentResponse>(payload)
+    if (response != null && response.fundingPayments.isNullOrEmpty().not()) {
+        walletProcessor.processFundingPayments(
+            existing = internalState.wallet,
+            payload = response.fundingPayments?.toList(),
+            subaccountNumber = subaccountNumber,
+        )
+        return StateChanges(iListOf(Changes.fundingPayments), null, iListOf(subaccountNumber))
+    } else {
+        return StateChanges(iListOf<Changes>())
+    }
+}

src/commonMain/kotlin/exchange.dydx.abacus/state/machine/TradingStateMachine+Wallet.kt

@@ -147,15 +147,6 @@ internal fun TradingStateMachine.onChainUserStats(payload: String): StateChanges
     }
 }
 
-internal fun TradingStateMachine.fills(payload: String, subaccountNumber: Int): StateChanges {
-    val json = parser.decodeJsonObject(payload)
-    return if (json != null) {
-        receivedFills(json, subaccountNumber)
-    } else {
-        StateChanges.noChange
-    }
-}
-
 internal fun TradingStateMachine.receivedFills(
     payload: Map<String, Any>,
     subaccountNumber: Int,
@@ -176,15 +167,6 @@ internal fun TradingStateMachine.receivedFills(
     }
 }
 
-internal fun TradingStateMachine.transfers(payload: String, subaccountNumber: Int): StateChanges {
-    val json = parser.decodeJsonObject(payload)
-    return if (json != null) {
-        receivedTransfers(json, subaccountNumber)
-    } else {
-        StateChanges.noChange
-    }
-}
-
 internal fun TradingStateMachine.receivedTransfers(
     payload: Map<String, Any>,
     subaccountNumber: Int,

src/commonMain/kotlin/exchange.dydx.abacus/state/machine/TradingStateMachine.kt

@@ -796,10 +796,11 @@ open class TradingStateMachine(
             if (changes.changes.contains(Changes.fundingPayments)) {
                 val modifiedFundingPayments = fundingPayments?.toIMutableMap() ?: mutableMapOf()
                 var subaccountFundingPayments = fundingPayments?.get(subaccountText)
-                subaccountFundingPayments = SubaccountFundingPayment.create(
-                    subaccountFundingPayments,
-                    parser,
-                    subaccountFundingPayments(subaccountNumber) as? IList<Map<String, Any>>,
+                val newPayments =
+                    internalState.wallet.account.subaccounts[subaccountNumber]?.fundingPayments?.toIList()
+                subaccountFundingPayments = SubaccountFundingPayment.merge(
+                    existing = subaccountFundingPayments,
+                    new = newPayments,
                 )
                 modifiedFundingPayments.typedSafeSet(subaccountText, subaccountFundingPayments)
                 fundingPayments = modifiedFundingPayments