feat: reward scaling

commands/transfer_funds.go

@@ -337,4 +337,18 @@ 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"))
 	}
+
+	if dispatchStrategy.TargetNotionalVolume != nil &&
+		((dispatchStrategy.Metric == vega.DispatchMetric_DISPATCH_METRIC_ELIGIBLE_ENTITIES && len(dispatchStrategy.AssetForMetric) == 0) ||
+			dispatchStrategy.Metric == vega.DispatchMetric_DISPATCH_METRIC_MARKET_VALUE ||
+			dispatchStrategy.Metric == vega.DispatchMetric_DISPATCH_METRIC_VALIDATOR_RANKING) {
+		errs.AddForProperty(prefix+".target_notional_volume", fmt.Errorf(fmt.Sprintf("not allowed for metric %s", dispatchStrategy.Metric)))
+	}
+	if dispatchStrategy.TargetNotionalVolume != nil && len(*dispatchStrategy.TargetNotionalVolume) > 0 {
+		_, overflow := num.UintFromString(*dispatchStrategy.TargetNotionalVolume, 10)
+		if overflow {
+			errs.AddForProperty(prefix+".target_notional_volume", ErrIsNotValidNumber)
+		}
+	}
+
 }

core/banking/engine.go

@@ -115,6 +115,7 @@ type MarketActivityTracker interface {
 	TeamStatsForMarkets(allMarketsForAssets, onlyTheseMarkets []string) map[string]map[string]*num.Uint
 	PublishGameMetric(ctx context.Context, dispatchStrategy []*vega.DispatchStrategy, now time.Time)
 	GameFinished(gameID string)
+	GetNotionalVolumeForAsset(asset string, markets []string, windowSize int) *num.Uint
 }
 
 type EthereumEventSource interface {

core/banking/gov_transfers.go

@@ -100,6 +100,7 @@ func (e *Engine) distributeRecurringGovernanceTransfers(ctx context.Context) {
 		}
 
 		amount, err := e.processGovernanceTransfer(ctx, gTransfer)
+		amount = e.scaleAmountByTargetNotional(gTransfer.Config.RecurringTransferConfig.DispatchStrategy, amount)
 		e.log.Info("processed transfer", logging.String("amount", amount.String()))
 
 		if err != nil {

core/banking/recurring_transfers.go

@@ -234,6 +234,23 @@ func (e *Engine) dispatchRequired(ctx context.Context, ds *vegapb.DispatchStrate
 	return required
 }
 
+func (e *Engine) scaleAmountByTargetNotional(ds *vegapb.DispatchStrategy, amount *num.Uint) *num.Uint {
+	if ds == nil {
+		return amount
+	}
+	if ds.TargetNotionalVolume == nil {
+		return amount
+	}
+	actualVolumeInWindow := e.marketActivityTracker.GetNotionalVolumeForAsset(ds.AssetForMetric, ds.Markets, int(ds.WindowLength))
+	if actualVolumeInWindow.IsZero() {
+		return num.UintZero()
+	}
+	targetNotional := num.MustUintFromString(*ds.TargetNotionalVolume, 10)
+	ratio := num.MinD(actualVolumeInWindow.ToDecimal().Div(targetNotional.ToDecimal()), num.DecimalOne())
+	amt, _ := num.UintFromDecimal(ratio.Mul(amount.ToDecimal()))
+	return amt
+}
+
 func (e *Engine) distributeRecurringTransfers(ctx context.Context, newEpoch uint64) {
 	var (
 		transfersDone = []events.Event{}
@@ -273,6 +290,9 @@ func (e *Engine) distributeRecurringTransfers(ctx context.Context, newEpoch uint
 			)
 		)
 
+		// scale transfer amount as necessary
+		amount = e.scaleAmountByTargetNotional(v.DispatchStrategy, amount)
+
 		// check if the amount is still enough
 		// ensure asset exists
 		a, err := e.assets.Get(v.Asset)

core/execution/common/market_activity_tracker.go

@@ -77,6 +77,8 @@ type marketTracker struct {
 	treasuryFeesPaid  map[string]*num.Uint
 	markPrice         *num.Uint
 
+	notionalVolumeForEpoch *num.Uint
+
 	totalMakerFeesReceived *num.Uint
 	totalMakerFeesPaid     *num.Uint
 	totalLpFees            *num.Uint
@@ -97,6 +99,7 @@ type marketTracker struct {
 	epochTimeWeightedNotional   []map[string]*num.Uint
 	epochPartyM2M               []map[string]num.Decimal
 	epochPartyRealisedReturn    []map[string]num.Decimal
+	epochNotionalVolume         []*num.Uint
 
 	valueTraded     *num.Uint
 	proposersPaid   map[string]struct{} // identifier of payout_asset : funder : markets_in_scope
@@ -203,6 +206,7 @@ func (mat *MarketActivityTracker) MarketProposed(asset, marketID, proposer strin
 		lpPaidFees:                  map[string]*num.Uint{},
 		buybackFeesPaid:             map[string]*num.Uint{},
 		treasuryFeesPaid:            map[string]*num.Uint{},
+		notionalVolumeForEpoch:      num.UintZero(),
 		totalMakerFeesReceived:      num.UintZero(),
 		totalMakerFeesPaid:          num.UintZero(),
 		totalLpFees:                 num.UintZero(),
@@ -219,6 +223,7 @@ func (mat *MarketActivityTracker) MarketProposed(asset, marketID, proposer strin
 		epochPartyM2M:               []map[string]num.Decimal{},
 		epochPartyRealisedReturn:    []map[string]decimal.Decimal{},
 		epochTimeWeightedPosition:   []map[string]uint64{},
+		epochNotionalVolume:         []*num.Uint{},
 		epochTimeWeightedNotional:   []map[string]*num.Uint{},
 		allPartiesCache:             map[string]struct{}{},
 		ammPartiesCache:             map[string]struct{}{},
@@ -543,6 +548,11 @@ func (mat *MarketActivityTracker) OnEpochEvent(ctx context.Context, epoch types.
 				mt.processM2MEndOfEpoch()
 				mt.processPartyRealisedReturnOfEpoch()
 				mt.clearFeeActivity()
+				if len(mt.epochNotionalVolume) == maxWindowSize {
+					mt.epochNotionalVolume = mt.epochNotionalVolume[1:]
+				}
+				mt.epochNotionalVolume = append(mt.epochNotionalVolume, mt.notionalVolumeForEpoch)
+				mt.notionalVolumeForEpoch = num.UintZero()
 			}
 		}
 		if len(mat.takerFeesPaidInEpoch) == maxWindowSize {
@@ -566,6 +576,33 @@ func (mat *MarketActivityTracker) clearDeletedMarkets() {
 	}
 }
 
+func (mat *MarketActivityTracker) GetNotionalVolumeForAsset(asset string, markets []string, windowSize int) *num.Uint {
+	total := num.UintZero()
+	trackers, ok := mat.assetToMarketTrackers[asset]
+	if !ok {
+		return total
+	}
+	marketsInScope := map[string]struct{}{}
+	for _, mkt := range markets {
+		marketsInScope[mkt] = struct{}{}
+	}
+	if len(markets) == 0 {
+		for mkt := range trackers {
+			marketsInScope[mkt] = struct{}{}
+		}
+	}
+	for mkt := range marketsInScope {
+		for i := 0; i < windowSize; i++ {
+			idx := len(trackers[mkt].epochNotionalVolume) - i - 1
+			if idx < 0 {
+				break
+			}
+			total.AddSum(trackers[mkt].epochNotionalVolume[idx])
+		}
+	}
+	return total
+}
+
 func (mat *MarketActivityTracker) CalculateTotalMakerContributionInQuantum(windowSize int) (map[string]*num.Uint, map[string]num.Decimal) {
 	m := map[string]*num.Uint{}
 	total := num.UintZero()
@@ -1355,6 +1392,7 @@ func (mt *marketTracker) recordNotional(party string, notional *num.Uint, price
 	n := mt.twNotional[party]
 	tn := int64(time.Sub(n.t).Seconds()) * scalingFactor
 	updateNotionalOnTrade(n, notional, price, t, tn, time)
+	mt.notionalVolumeForEpoch.AddSum(notional)
 }
 
 func (mt *marketTracker) processNotionalEndOfEpoch(epochStartTime time.Time, endEpochTime time.Time) {

datanode/gateway/graphql/schema.graphql

@@ -1619,6 +1619,9 @@ type DispatchStrategy {
 
   "Interval for the distribution of the transfer"
   transferInterval: Int
+
+  "The target notional to scaled the amount to take from the source account by"
+  targetNotionalVolume: String
 }
 
 type RankTable {

protos/sources/vega/vega.proto

@@ -1185,6 +1185,8 @@ message DispatchStrategy {
   optional string cap_reward_fee_multiple = 14;
   // Number of epochs between transfers, i.e. when 4, funds will be transferred every 4 epochs with the first transfer occurring 4 epochs after the transaction is processed.
   optional int32 transfer_interval = 15;
+  // If set, the amount to take from the source account and distribute as rewards is scaled.
+  optional string target_notional_volume = 16;
 }
 
 message Rank {