Merge pull request #11171 from vegaprotocol/transfer_interval

feat: Add support for transfer interval

CHANGELOG.md

@@ -32,6 +32,7 @@
 - [11167](https://github.com/vegaprotocol/vega/issues/11167) - Add realised return reward metric.
 - [11165] (https://github.com/vegaprotocol/vega/issues/11165) - Include negative returns in relative returns reward metric.
 - [11151](https://github.com/vegaprotocol/vega/issues/11151) - Remove name field from the spot markets.
+- [11170](https://github.com/vegaprotocol/vega/issues/11170) - Add transfer interval support.
 - [11143](https://github.com/vegaprotocol/vega/issues/11143) - Add support for new asset proposal in batch governance proposal.
 - [11184](https://github.com/vegaprotocol/vega/issues/11184) - relax the transfer restriction of 'same transfer' to match the full dispatch strategy.
 - [11185](https://github.com/vegaprotocol/vega/issues/11185) - distinguish between zero and none metric for rewards. 

commands/transfer_funds.go

@@ -276,7 +276,7 @@ func validateDispatchStrategy(toAccountType vega.AccountType, dispatchStrategy *
 		errs.AddForProperty(prefix+".window_length", errors.New("should not be set for "+vega.AccountType_ACCOUNT_TYPE_REWARD_MARKET_PROPOSERS.String()))
 	}
 	if dispatchStrategy.WindowLength == 0 && toAccountType != vega.AccountType_ACCOUNT_TYPE_REWARD_MARKET_PROPOSERS {
-		errs.AddForProperty(prefix+".window_length", errors.New("must bet between 1 and 100"))
+		errs.AddForProperty(prefix+".window_length", errors.New("must be between 1 and 100"))
 	}
 	if dispatchStrategy.WindowLength > 100 {
 		errs.AddForProperty(prefix+".window_length", ErrMustBeAtMost100)
@@ -308,4 +308,8 @@ func validateDispatchStrategy(toAccountType vega.AccountType, dispatchStrategy *
 			}
 		}
 	}
+
+	if dispatchStrategy.TransferInterval != nil && (*dispatchStrategy.TransferInterval <= 0 || *dispatchStrategy.TransferInterval > 100) {
+		errs.AddForProperty(prefix+".transfer_interval", errors.New("must be between 1 and 100"))
+	}
 }

commands/transfer_funds_test.go

@@ -43,6 +43,10 @@ func TestTransferFunds(t *testing.T) {
 	capNegative := "-1"
 	capZero := "0"
 
+	tooLongTransferInterval := int32(101)
+	zeroTransferInterval := int32(0)
+	negativeTransferInterval := int32(-1)
+
 	cases := []struct {
 		transfer  commandspb.Transfer
 		errString string
@@ -1199,7 +1203,7 @@ func TestTransferFunds(t *testing.T) {
 				Amount:    "1",
 				Reference: "testing",
 			},
-			errString: "transfer.kind.dispatch_strategy.window_length (must bet between 1 and 100)",
+			errString: "transfer.kind.dispatch_strategy.window_length (must be between 1 and 100)",
 		},
 		{
 			transfer: commandspb.Transfer{
@@ -1299,6 +1303,84 @@ func TestTransferFunds(t *testing.T) {
 				Reference: "testing",
 			},
 		},
+		{
+			transfer: commandspb.Transfer{
+				FromAccountType: vega.AccountType_ACCOUNT_TYPE_GENERAL,
+				ToAccountType:   vega.AccountType_ACCOUNT_TYPE_REWARD_AVERAGE_POSITION,
+				Kind: &commandspb.Transfer_Recurring{
+					Recurring: &commandspb.RecurringTransfer{
+						StartEpoch: 10,
+						EndEpoch:   ptr.From(uint64(11)),
+						Factor:     "1",
+						DispatchStrategy: &vega.DispatchStrategy{
+							AssetForMetric:   "",
+							Metric:           vega.DispatchMetric_DISPATCH_METRIC_AVERAGE_POSITION,
+							EntityScope:      vega.EntityScope_ENTITY_SCOPE_INDIVIDUALS,
+							IndividualScope:  vega.IndividualScope_INDIVIDUAL_SCOPE_IN_TEAM,
+							WindowLength:     100,
+							TransferInterval: &tooLongTransferInterval,
+						},
+					},
+				},
+				To:        "84e2b15102a8d6c1c6b4bdf40af8a0dc21b040eaaa1c94cd10d17604b75fdc35",
+				Asset:     "080538b7cc2249de568cb4272a17f4d5e0b0a69a1a240acbf5119d816178daff",
+				Amount:    "1",
+				Reference: "testing",
+			},
+			errString: "transfer.kind.dispatch_strategy.transfer_interval (must be between 1 and 100)",
+		},
+		{
+			transfer: commandspb.Transfer{
+				FromAccountType: vega.AccountType_ACCOUNT_TYPE_GENERAL,
+				ToAccountType:   vega.AccountType_ACCOUNT_TYPE_REWARD_AVERAGE_POSITION,
+				Kind: &commandspb.Transfer_Recurring{
+					Recurring: &commandspb.RecurringTransfer{
+						StartEpoch: 10,
+						EndEpoch:   ptr.From(uint64(11)),
+						Factor:     "1",
+						DispatchStrategy: &vega.DispatchStrategy{
+							AssetForMetric:   "",
+							Metric:           vega.DispatchMetric_DISPATCH_METRIC_AVERAGE_POSITION,
+							EntityScope:      vega.EntityScope_ENTITY_SCOPE_INDIVIDUALS,
+							IndividualScope:  vega.IndividualScope_INDIVIDUAL_SCOPE_IN_TEAM,
+							WindowLength:     100,
+							TransferInterval: &zeroTransferInterval,
+						},
+					},
+				},
+				To:        "84e2b15102a8d6c1c6b4bdf40af8a0dc21b040eaaa1c94cd10d17604b75fdc35",
+				Asset:     "080538b7cc2249de568cb4272a17f4d5e0b0a69a1a240acbf5119d816178daff",
+				Amount:    "1",
+				Reference: "testing",
+			},
+			errString: "transfer.kind.dispatch_strategy.transfer_interval (must be between 1 and 100)",
+		},
+		{
+			transfer: commandspb.Transfer{
+				FromAccountType: vega.AccountType_ACCOUNT_TYPE_GENERAL,
+				ToAccountType:   vega.AccountType_ACCOUNT_TYPE_REWARD_AVERAGE_POSITION,
+				Kind: &commandspb.Transfer_Recurring{
+					Recurring: &commandspb.RecurringTransfer{
+						StartEpoch: 10,
+						EndEpoch:   ptr.From(uint64(11)),
+						Factor:     "1",
+						DispatchStrategy: &vega.DispatchStrategy{
+							AssetForMetric:   "",
+							Metric:           vega.DispatchMetric_DISPATCH_METRIC_AVERAGE_POSITION,
+							EntityScope:      vega.EntityScope_ENTITY_SCOPE_INDIVIDUALS,
+							IndividualScope:  vega.IndividualScope_INDIVIDUAL_SCOPE_IN_TEAM,
+							WindowLength:     100,
+							TransferInterval: &negativeTransferInterval,
+						},
+					},
+				},
+				To:        "84e2b15102a8d6c1c6b4bdf40af8a0dc21b040eaaa1c94cd10d17604b75fdc35",
+				Asset:     "080538b7cc2249de568cb4272a17f4d5e0b0a69a1a240acbf5119d816178daff",
+				Amount:    "1",
+				Reference: "testing",
+			},
+			errString: "transfer.kind.dispatch_strategy.transfer_interval (must be between 1 and 100)",
+		},
 	}
 
 	invalidAccountTypesForOneOff := []vega.AccountType{

core/banking/gov_transfers.go

@@ -92,6 +92,12 @@ func (e *Engine) distributeRecurringGovernanceTransfers(ctx context.Context) {
 			continue
 		}
 
+		if gTransfer.Config.RecurringTransferConfig.DispatchStrategy != nil && gTransfer.Config.RecurringTransferConfig.DispatchStrategy.TransferInterval != nil &&
+			(e.currentEpoch-gTransfer.Config.RecurringTransferConfig.StartEpoch+1) < uint64(*gTransfer.Config.RecurringTransferConfig.DispatchStrategy.TransferInterval) &&
+			(e.currentEpoch-gTransfer.Config.RecurringTransferConfig.StartEpoch+1)%uint64(*gTransfer.Config.RecurringTransferConfig.DispatchStrategy.TransferInterval) != 0 {
+			continue
+		}
+
 		amount, err := e.processGovernanceTransfer(ctx, gTransfer)
 		e.log.Info("processed transfer", logging.String("amount", amount.String()))
 

core/banking/recurring_transfers.go

@@ -220,6 +220,12 @@ func (e *Engine) distributeRecurringTransfers(ctx context.Context, newEpoch uint
 			continue
 		}
 
+		if v.DispatchStrategy != nil && v.DispatchStrategy.TransferInterval != nil &&
+			(newEpoch-v.StartEpoch+1) < uint64(*v.DispatchStrategy.TransferInterval) &&
+			(newEpoch-v.StartEpoch+1)%uint64(*v.DispatchStrategy.TransferInterval) != 0 {
+			continue
+		}
+
 		var (
 			startEpoch  = num.NewUint(v.StartEpoch).ToDecimal()
 			startAmount = v.Amount.ToDecimal()

core/execution/common/market_activity_tracker.go

@@ -111,11 +111,10 @@ type MarketActivityTracker struct {
 	partyContributionCache              map[string][]*types.PartyContributionScore
 	partyTakerNotionalVolume            map[string]*num.Uint
 	marketToPartyTakerNotionalVolume    map[string]map[string]*num.Uint
-	// transient map that is used and accessible only between the end of one epoch and the beginning of the next
-	// it's not needed for the snapshot because the switching between end of one epoch and the beginning of the new one is atommic.
-	takerFeesPaidInEpoch map[string]map[string]map[string]*num.Uint
-	ss                   *snapshotState
-	broker               Broker
+	takerFeesPaidInEpoch                []map[string]map[string]map[string]*num.Uint
+
+	ss     *snapshotState
+	broker Broker
 }
 
 // NewMarketActivityTracker instantiates the fees tracker.
@@ -129,7 +128,7 @@ func NewMarketActivityTracker(log *logging.Logger, teams Teams, balanceChecker A
 		partyTakerNotionalVolume:         map[string]*num.Uint{},
 		marketToPartyTakerNotionalVolume: map[string]map[string]*num.Uint{},
 		ss:                               &snapshotState{},
-		takerFeesPaidInEpoch:             map[string]map[string]map[string]*num.Uint{},
+		takerFeesPaidInEpoch:             []map[string]map[string]map[string]*num.Uint{},
 		broker:                           broker,
 	}
 
@@ -412,19 +411,23 @@ func (mat *MarketActivityTracker) OnEpochEvent(ctx context.Context, epoch types.
 		mat.partyContributionCache = map[string][]*types.PartyContributionScore{}
 		mat.clearDeletedMarkets()
 		mat.clearNotionalTakerVolume()
-		mat.takerFeesPaidInEpoch = map[string]map[string]map[string]*num.Uint{}
 	} else if epoch.Action == vega.EpochAction_EPOCH_ACTION_END {
+		m := map[string]map[string]map[string]*num.Uint{}
 		for asset, market := range mat.assetToMarketTrackers {
-			mat.takerFeesPaidInEpoch[asset] = map[string]map[string]*num.Uint{}
+			m[asset] = map[string]map[string]*num.Uint{}
 			for mkt, mt := range market {
-				mat.takerFeesPaidInEpoch[asset][mkt] = mt.aggregatedFees()
+				m[asset][mkt] = mt.aggregatedFees()
 				mt.processNotionalEndOfEpoch(epoch.StartTime, epoch.EndTime)
 				mt.processPositionEndOfEpoch(epoch.StartTime, epoch.EndTime)
 				mt.processM2MEndOfEpoch()
 				mt.processPartyRealisedReturnOfEpoch()
 				mt.clearFeeActivity()
 			}
 		}
+		if len(mat.takerFeesPaidInEpoch) == maxWindowSize {
+			mat.takerFeesPaidInEpoch = mat.takerFeesPaidInEpoch[1:]
+		}
+		mat.takerFeesPaidInEpoch = append(mat.takerFeesPaidInEpoch, m)
 	}
 	mat.currentEpoch = epoch.Seq
 }
@@ -1191,9 +1194,9 @@ func getTotalFees(totalFees []*num.Uint, windowSize int) num.Decimal {
 	return total.ToDecimal()
 }
 
-func (mat *MarketActivityTracker) GetLastEpochTakeFees(asset string, markets []string) map[string]*num.Uint {
+func (mat *MarketActivityTracker) getEpochTakeFees(asset string, markets []string, takerFeesPaidInEpoch map[string]map[string]map[string]*num.Uint) map[string]*num.Uint {
 	takerFees := map[string]*num.Uint{}
-	ast, ok := mat.takerFeesPaidInEpoch[asset]
+	ast, ok := takerFeesPaidInEpoch[asset]
 	if !ok {
 		return takerFees
 	}
@@ -1218,6 +1221,24 @@ func (mat *MarketActivityTracker) GetLastEpochTakeFees(asset string, markets []s
 	return takerFees
 }
 
+func (mat *MarketActivityTracker) GetLastEpochTakeFees(asset string, markets []string, epochs int32) map[string]*num.Uint {
+	takerFees := map[string]*num.Uint{}
+	for i := 0; i < int(epochs); i++ {
+		ind := len(mat.takerFeesPaidInEpoch) - i - 1
+		if ind < 0 {
+			break
+		}
+		m := mat.getEpochTakeFees(asset, markets, mat.takerFeesPaidInEpoch[ind])
+		for k, v := range m {
+			if _, ok := takerFees[k]; !ok {
+				takerFees[k] = num.UintZero()
+			}
+			takerFees[k].AddSum(v)
+		}
+	}
+	return takerFees
+}
+
 // calcTotalForWindowU returns the total relevant data from the given slice starting from the given dataIdx-1, going back <window_size> elements.
 func calcTotalForWindowU(party string, data []map[string]*num.Uint, windowSize int) (num.Decimal, bool) {
 	found := false

core/execution/common/market_activity_tracker_checkpoint.go

@@ -54,6 +54,7 @@ func (mat *MarketActivityTracker) Checkpoint() ([]byte, error) {
 		MarketActivity:                   marketTracker,
 		TakerNotionalVolume:              takerNotionalToProto(mat.partyTakerNotionalVolume),
 		MarketToPartyTakerNotionalVolume: marketToPartyTakerNotionalToProto(mat.marketToPartyTakerNotionalVolume),
+		EpochTakerFees:                   epochTakerFeesToProto(mat.takerFeesPaidInEpoch),
 	}
 	ret, err := proto.Marshal(msg)
 	if err != nil {
@@ -87,5 +88,22 @@ func (mat *MarketActivityTracker) Load(_ context.Context, data []byte) error {
 			}
 		}
 	}
+	if b.EpochTakerFees != nil {
+		for _, epochData := range b.EpochTakerFees {
+			epochMap := map[string]map[string]map[string]*num.Uint{}
+			for _, assetMarketParty := range epochData.EpochPartyTakerFeesPaid {
+				if _, ok := epochMap[assetMarketParty.Asset]; !ok {
+					epochMap[assetMarketParty.Asset] = map[string]map[string]*num.Uint{}
+				}
+				if _, ok := epochMap[assetMarketParty.Asset][assetMarketParty.Market]; !ok {
+					epochMap[assetMarketParty.Asset][assetMarketParty.Market] = map[string]*num.Uint{}
+				}
+				for _, tf := range assetMarketParty.TakerFees {
+					epochMap[assetMarketParty.Asset][assetMarketParty.Market][tf.Party] = num.UintFromBytes(tf.TakerFees)
+				}
+			}
+			mat.takerFeesPaidInEpoch = append(mat.takerFeesPaidInEpoch, epochMap)
+		}
+	}
 	return nil
 }