liquidity: add peer-level liquidity rules to allow aggregate management

We add 'peer-level' rules to allow assessment of liquidity on a per-peer
level, rather than on an individual channel basis. No overlap is allowed
with the existing set of channel rules because this could lead to
contradictory rules.

liquidity/autoloop_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/btcsuite/btcutil"
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop"
 	"github.com/lightninglabs/loop/labels"
@@ -12,6 +13,7 @@ import (
 	"github.com/lightninglabs/loop/test"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/routing/route"
 )
 
 // TestAutoLoopDisabled tests the case where we need to perform a swap, but
@@ -266,6 +268,161 @@ func TestAutoLoopEnabled(t *testing.T) {
 	c.stop()
 }
 
+// TestCompositeRules tests the case where we have rules set on a per peer
+// and per channel basis, and perform swaps for both targets.
+func TestCompositeRules(t *testing.T) {
+	defer test.Guard(t)()
+
+	// Setup our channels so that we have two channels with peer 2, and
+	// a single channel with peer 1.
+	channel3 := lndclient.ChannelInfo{
+		ChannelID:     chanID3.ToUint64(),
+		PubKeyBytes:   peer2,
+		LocalBalance:  10000,
+		RemoteBalance: 0,
+		Capacity:      10000,
+	}
+
+	channels := []lndclient.ChannelInfo{
+		channel1, channel2, channel3,
+	}
+
+	// Create a set of parameters with autoloop enabled, set our budget to
+	// a value that will easily accommodate our two swaps.
+	params := Parameters{
+		Autoloop:                   true,
+		AutoFeeBudget:              100000,
+		AutoFeeStartDate:           testTime,
+		MaxAutoInFlight:            2,
+		FailureBackOff:             time.Hour,
+		SweepFeeRateLimit:          20000,
+		SweepConfTarget:            10,
+		MaximumPrepay:              20000,
+		MaximumSwapFeePPM:          1000,
+		MaximumRoutingFeePPM:       1000,
+		MaximumPrepayRoutingFeePPM: 1000,
+		MaximumMinerFee:            20000,
+		ChannelRules: map[lnwire.ShortChannelID]*ThresholdRule{
+			chanID1: chanRule,
+		},
+		PeerRules: map[route.Vertex]*ThresholdRule{
+			peer2: chanRule,
+		},
+	}
+
+	c := newAutoloopTestCtx(t, params, channels, testRestrictions)
+	c.start()
+
+	// Calculate our maximum allowed fees and create quotes that fall within
+	// our budget.
+	var (
+		// Create a quote for our peer level swap that is within
+		// our budget, with an amount which would balance the peer
+		/// across all of its channels.
+		peerAmount     = btcutil.Amount(15000)
+		maxPeerSwapFee = ppmToSat(peerAmount, params.MaximumSwapFeePPM)
+
+		peerSwapQuote = &loop.LoopOutQuote{
+			SwapFee:      maxPeerSwapFee,
+			PrepayAmount: params.MaximumPrepay - 20,
+		}
+
+		peerSwapQuoteRequest = &loop.LoopOutQuoteRequest{
+			Amount:          peerAmount,
+			SweepConfTarget: params.SweepConfTarget,
+		}
+
+		maxPeerRouteFee = ppmToSat(
+			peerAmount, params.MaximumRoutingFeePPM,
+		)
+
+		peerSwap = &loop.OutRequest{
+			Amount:            peerAmount,
+			MaxSwapRoutingFee: maxPeerRouteFee,
+			MaxPrepayRoutingFee: ppmToSat(
+				peerSwapQuote.PrepayAmount,
+				params.MaximumPrepayRoutingFeePPM,
+			),
+			MaxSwapFee:      peerSwapQuote.SwapFee,
+			MaxPrepayAmount: peerSwapQuote.PrepayAmount,
+			MaxMinerFee:     params.MaximumMinerFee,
+			SweepConfTarget: params.SweepConfTarget,
+			OutgoingChanSet: loopdb.ChannelSet{
+				chanID2.ToUint64(), chanID3.ToUint64(),
+			},
+			Label:     labels.AutoloopLabel(swap.TypeOut),
+			Initiator: autoloopSwapInitiator,
+		}
+		// Create a quote for our single channel swap that is within
+		// our budget.
+		chanAmount     = chan1Rec.Amount
+		maxChanSwapFee = ppmToSat(chanAmount, params.MaximumSwapFeePPM)
+
+		channelSwapQuote = &loop.LoopOutQuote{
+			SwapFee:      maxChanSwapFee,
+			PrepayAmount: params.MaximumPrepay - 10,
+		}
+
+		chanSwapQuoteRequest = &loop.LoopOutQuoteRequest{
+			Amount:          chanAmount,
+			SweepConfTarget: params.SweepConfTarget,
+		}
+
+		maxChanRouteFee = ppmToSat(
+			chanAmount, params.MaximumRoutingFeePPM,
+		)
+
+		chanSwap = &loop.OutRequest{
+			Amount:            chanAmount,
+			MaxSwapRoutingFee: maxChanRouteFee,
+			MaxPrepayRoutingFee: ppmToSat(
+				channelSwapQuote.PrepayAmount,
+				params.MaximumPrepayRoutingFeePPM,
+			),
+			MaxSwapFee:      channelSwapQuote.SwapFee,
+			MaxPrepayAmount: channelSwapQuote.PrepayAmount,
+			MaxMinerFee:     params.MaximumMinerFee,
+			SweepConfTarget: params.SweepConfTarget,
+			OutgoingChanSet: loopdb.ChannelSet{chanID1.ToUint64()},
+			Label:           labels.AutoloopLabel(swap.TypeOut),
+			Initiator:       autoloopSwapInitiator,
+		}
+		quotes = []quoteRequestResp{
+			{
+				request: peerSwapQuoteRequest,
+				quote:   peerSwapQuote,
+			},
+			{
+				request: chanSwapQuoteRequest,
+				quote:   channelSwapQuote,
+			},
+		}
+
+		loopOuts = []loopOutRequestResp{
+			{
+				request: peerSwap,
+				response: &loop.LoopOutSwapInfo{
+					SwapHash: lntypes.Hash{2},
+				},
+			},
+			{
+				request: chanSwap,
+				response: &loop.LoopOutSwapInfo{
+					SwapHash: lntypes.Hash{1},
+				},
+			},
+		}
+	)
+
+	// Tick our autolooper with no existing swaps, we expect a loop out
+	// swap to be dispatched for each of our rules. We set our server side
+	// maximum to be greater than the swap amount for our peer swap (which
+	// is the larger of the two swaps).
+	c.autoloop(1, peerAmount+1, nil, quotes, loopOuts)
+
+	c.stop()
+}
+
 // existingSwapFromRequest is a helper function which returns the db
 // representation of a loop out request with the event set provided.
 func existingSwapFromRequest(request *loop.OutRequest, initTime time.Time,

liquidity/balances.go

@@ -19,8 +19,10 @@ type balances struct {
 	// outgoing is the local balance of the channel.
 	outgoing btcutil.Amount
 
-	// channelID is the channel that has these balances.
-	channelID lnwire.ShortChannelID
+	// channels is the channel that has these balances represent. This may
+	// be more than one channel in the case where we are examining a peer's
+	// liquidity as a whole.
+	channels []lnwire.ShortChannelID
 
 	// pubkey is the public key of the peer we have this balances set with.
 	pubkey route.Vertex
@@ -29,10 +31,12 @@ type balances struct {
 // newBalances creates a balances struct from lndclient channel information.
 func newBalances(info lndclient.ChannelInfo) *balances {
 	return &balances{
-		capacity:  info.Capacity,
-		incoming:  info.RemoteBalance,
-		outgoing:  info.LocalBalance,
-		channelID: lnwire.NewShortChanIDFromInt(info.ChannelID),
-		pubkey:    info.PubKeyBytes,
+		capacity: info.Capacity,
+		incoming: info.RemoteBalance,
+		outgoing: info.LocalBalance,
+		channels: []lnwire.ShortChannelID{
+			lnwire.NewShortChanIDFromInt(info.ChannelID),
+		},
+		pubkey: info.PubKeyBytes,
 	}
 }

liquidity/liquidity.go

@@ -124,6 +124,7 @@ var (
 		AutoFeeBudget:              defaultBudget,
 		MaxAutoInFlight:            defaultMaxInFlight,
 		ChannelRules:               make(map[lnwire.ShortChannelID]*ThresholdRule),
+		PeerRules:                  make(map[route.Vertex]*ThresholdRule),
 		FailureBackOff:             defaultFailureBackoff,
 		SweepFeeRateLimit:          defaultSweepFeeRateLimit,
 		SweepConfTarget:            loop.DefaultSweepConfTarget,
@@ -181,6 +182,11 @@ var (
 
 	// ErrNoRules is returned when no rules are set for swap suggestions.
 	ErrNoRules = errors.New("no rules set for autoloop")
+
+	// ErrExclusiveRules is returned when a set of rules that may not be
+	// set together are specified.
+	ErrExclusiveRules = errors.New("channel and peer rules must be " +
+		"exclusive")
 )
 
 // Config contains the external functionality required to run the
@@ -289,27 +295,41 @@ type Parameters struct {
 	ClientRestrictions Restrictions
 
 	// ChannelRules maps a short channel ID to a rule that describes how we
-	// would like liquidity to be managed.
+	// would like liquidity to be managed. These rules and PeerRules are
+	// exclusively set to prevent overlap between peer and channel rules.
 	ChannelRules map[lnwire.ShortChannelID]*ThresholdRule
+
+	// PeerRules maps a peer's pubkey to a rule that applies to all the
+	// channels that we have with the peer collectively. These rules and
+	// ChannelRules are exclusively set to prevent overlap between peer
+	// and channel rules map to avoid ambiguity.
+	PeerRules map[route.Vertex]*ThresholdRule
 }
 
 // String returns the string representation of our parameters.
 func (p Parameters) String() string {
-	channelRules := make([]string, 0, len(p.ChannelRules))
+	ruleList := make([]string, 0, len(p.ChannelRules)+len(p.PeerRules))
 
 	for channel, rule := range p.ChannelRules {
-		channelRules = append(
-			channelRules, fmt.Sprintf("%v: %v", channel, rule),
+		ruleList = append(
+			ruleList, fmt.Sprintf("Channel: %v: %v", channel, rule),
 		)
 	}
 
-	return fmt.Sprintf("channel rules: %v, failure backoff: %v, sweep "+
+	for peer, rule := range p.PeerRules {
+		ruleList = append(
+			ruleList, fmt.Sprintf("Peer: %v: %v", peer, rule),
+		)
+
+	}
+
+	return fmt.Sprintf("rules: %v, failure backoff: %v, sweep "+
 		"fee rate limit: %v, sweep conf target: %v, maximum prepay: "+
 		"%v, maximum miner fee: %v, maximum swap fee ppm: %v, maximum "+
 		"routing fee ppm: %v, maximum prepay routing fee ppm: %v, "+
 		"auto budget: %v, budget start: %v, max auto in flight: %v, "+
 		"minimum swap size=%v, maximum swap size=%v",
-		strings.Join(channelRules, ","), p.FailureBackOff,
+		strings.Join(ruleList, ","), p.FailureBackOff,
 		p.SweepFeeRateLimit, p.SweepConfTarget, p.MaximumPrepay,
 		p.MaximumMinerFee, p.MaximumSwapFeePPM,
 		p.MaximumRoutingFeePPM, p.MaximumPrepayRoutingFeePPM,
@@ -317,9 +337,54 @@ func (p Parameters) String() string {
 		p.ClientRestrictions.Minimum, p.ClientRestrictions.Maximum)
 }
 
-// validate checks whether a set of parameters is valid. It takes the minimum
-// confirmations we allow for sweep confirmation target as a parameter.
-func (p Parameters) validate(minConfs int32, server *Restrictions) error {
+// haveRules returns a boolean indicating whether we have any rules configured.
+func (p Parameters) haveRules() bool {
+	if len(p.ChannelRules) != 0 {
+		return true
+	}
+
+	if len(p.PeerRules) != 0 {
+		return true
+	}
+
+	return false
+}
+
+// validate checks whether a set of parameters is valid. Our set of currently
+// open channels are required to check that there is no overlap between the
+// rules set on a per-peer level, and those set for specific channels. We can't
+// allow both, because then we're trying to cater for two separate liquidity
+// goals on the same channel. Since we use short channel ID, we don't need to
+// worry about pending channels (users would need to work very hard to get the
+// short channel ID for a pending channel). Likewise, we don't care about closed
+// channels, since there is no action that may occur on them, and we want to
+// allow peer-level rules to be set once a channel which had a specific rule
+// has been closed. It takes the minimum confirmations we allow for sweep
+// confirmation target as a parameter.
+// TODO(carla): prune channels that have been closed from rules.
+func (p Parameters) validate(minConfs int32, openChans []lndclient.ChannelInfo,
+	server *Restrictions) error {
+
+	// First, we check that the rules on a per peer and per channel do not
+	// overlap, since this could lead to contractions.
+	for _, channel := range openChans {
+		// If we don't have a rule for the peer, there's no way we have
+		// an overlap between this peer and the channel.
+		_, ok := p.PeerRules[channel.PubKeyBytes]
+		if !ok {
+			continue
+		}
+
+		shortID := lnwire.NewShortChanIDFromInt(channel.ChannelID)
+		_, ok = p.ChannelRules[shortID]
+		if ok {
+			log.Debugf("Rules for peer: %v and its channel: %v "+
+				"can't both be set", channel.PubKeyBytes, shortID)
+
+			return ErrExclusiveRules
+		}
+	}
+
 	for channel, rule := range p.ChannelRules {
 		if channel.ToUint64() == 0 {
 			return ErrZeroChannelID
@@ -331,6 +396,13 @@ func (p Parameters) validate(minConfs int32, server *Restrictions) error {
 		}
 	}
 
+	for peer, rule := range p.PeerRules {
+		if err := rule.validate(); err != nil {
+			return fmt.Errorf("peer: %v has invalid rule: %v",
+				peer, err)
+		}
+	}
+
 	// Check that our sweep limit is above our minimum fee rate. We use
 	// absolute fee floor rather than kw floor because we will allow users
 	// to specify fee rate is sat/vByte and want to allow 1 sat/vByte.
@@ -483,7 +555,12 @@ func (m *Manager) SetParameters(ctx context.Context, params Parameters) error {
 		return err
 	}
 
-	err = params.validate(m.cfg.MinimumConfirmations, restrictions)
+	channels, err := m.cfg.Lnd.Client.ListChannels(ctx)
+	if err != nil {
+		return err
+	}
+
+	err = params.validate(m.cfg.MinimumConfirmations, channels, restrictions)
 	if err != nil {
 		return err
 	}
@@ -510,6 +587,16 @@ func cloneParameters(params Parameters) Parameters {
 		paramCopy.ChannelRules[channel] = &ruleCopy
 	}
 
+	paramCopy.PeerRules = make(
+		map[route.Vertex]*ThresholdRule,
+		len(params.PeerRules),
+	)
+
+	for peer, rule := range params.PeerRules {
+		ruleCopy := *rule
+		paramCopy.PeerRules[peer] = &ruleCopy
+	}
+
 	return paramCopy
 }
 
@@ -566,11 +653,16 @@ type Suggestions struct {
 	// DisqualifiedChans maps the set of channels that we do not recommend
 	// swaps on to the reason that we did not recommend a swap.
 	DisqualifiedChans map[lnwire.ShortChannelID]Reason
+
+	// Disqualified peers maps the set of peers that we do not recommend
+	// swaps for to the reason that they were excluded.
+	DisqualifiedPeers map[route.Vertex]Reason
 }
 
 func newSuggestions() *Suggestions {
 	return &Suggestions{
 		DisqualifiedChans: make(map[lnwire.ShortChannelID]Reason),
+		DisqualifiedPeers: make(map[route.Vertex]Reason),
 	}
 }
 
@@ -595,6 +687,10 @@ func (m *Manager) singleReasonSuggestion(reason Reason) *Suggestions {
 		resp.DisqualifiedChans[id] = reason
 	}
 
+	for peer := range m.params.PeerRules {
+		resp.DisqualifiedPeers[peer] = reason
+	}
+
 	return resp
 }
 
@@ -612,7 +708,7 @@ func (m *Manager) SuggestSwaps(ctx context.Context, autoloop bool) (
 
 	// If we have no rules set, exit early to avoid unnecessary calls to
 	// lnd and the server.
-	if len(m.params.ChannelRules) == 0 {
+	if !m.params.haveRules() {
 		return nil, ErrNoRules
 	}
 
@@ -699,19 +795,59 @@ func (m *Manager) SuggestSwaps(ctx context.Context, autoloop bool) (
 		return nil, err
 	}
 
+	peerChannels := make(map[route.Vertex]*balances)
+	for _, channel := range channels {
+		bal, ok := peerChannels[channel.PubKeyBytes]
+		if !ok {
+			bal = &balances{}
+		}
+
+		chanID := lnwire.NewShortChanIDFromInt(channel.ChannelID)
+		bal.channels = append(bal.channels, chanID)
+		bal.capacity += channel.Capacity
+		bal.incoming += channel.RemoteBalance
+		bal.outgoing += channel.LocalBalance
+		bal.pubkey = channel.PubKeyBytes
+
+		peerChannels[channel.PubKeyBytes] = bal
+	}
+
 	// Get a summary of the channels and peers that are not eligible due
 	// to ongoing swaps.
 	traffic := m.currentSwapTraffic(loopOut, loopIn)
 
 	var (
-		suggestions  []swapSuggestion
-		disqualified = make(map[lnwire.ShortChannelID]Reason)
+		suggestions []swapSuggestion
+		resp        = newSuggestions()
 	)
 
+	for peer, balances := range peerChannels {
+		rule, haveRule := m.params.PeerRules[peer]
+		if !haveRule {
+			continue
+		}
+
+		suggestion, err := m.suggestSwap(
+			ctx, traffic, balances, rule, restrictions, autoloop,
+		)
+		var reasonErr *reasonError
+		if errors.As(err, &reasonErr) {
+			resp.DisqualifiedPeers[peer] = reasonErr.reason
+			continue
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		suggestions = append(suggestions, suggestion)
+	}
+
 	for _, channel := range channels {
 		balance := newBalances(channel)
 
-		rule, ok := m.params.ChannelRules[balance.channelID]
+		channelID := lnwire.NewShortChanIDFromInt(channel.ChannelID)
+		rule, ok := m.params.ChannelRules[channelID]
 		if !ok {
 			continue
 		}
@@ -722,7 +858,7 @@ func (m *Manager) SuggestSwaps(ctx context.Context, autoloop bool) (
 
 		var reasonErr *reasonError
 		if errors.As(err, &reasonErr) {
-			disqualified[balance.channelID] = reasonErr.reason
+			resp.DisqualifiedChans[channelID] = reasonErr.reason
 			continue
 		}
 
@@ -733,12 +869,6 @@ func (m *Manager) SuggestSwaps(ctx context.Context, autoloop bool) (
 		suggestions = append(suggestions, suggestion)
 	}
 
-	// Finally, run through all possible swaps, excluding swaps that are
-	// not feasible due to fee or budget restrictions.
-	resp := &Suggestions{
-		DisqualifiedChans: disqualified,
-	}
-
 	// If we have no swaps to execute after we have applied all of our
 	// limits, just return our set of disqualified swaps.
 	if len(suggestions) == 0 {
@@ -813,7 +943,7 @@ func (m *Manager) suggestSwap(ctx context.Context, traffic *swapTraffic,
 	autoloop bool) (swapSuggestion, error) {
 
 	// Check whether we can perform a swap.
-	err := traffic.maySwap(balance.pubkey, balance.channelID)
+	err := traffic.maySwap(balance.pubkey, balance.channels)
 	if err != nil {
 		return nil, err
 	}
@@ -930,11 +1060,14 @@ func (m *Manager) makeLoopOutRequest(ctx context.Context,
 
 	routeMaxFee := ppmToSat(amount, m.params.MaximumRoutingFeePPM)
 
+	var chanSet loopdb.ChannelSet
+	for _, channel := range balance.channels {
+		chanSet = append(chanSet, channel.ToUint64())
+	}
+
 	request := loop.OutRequest{
-		Amount: amount,
-		OutgoingChanSet: loopdb.ChannelSet{
-			balance.channelID.ToUint64(),
-		},
+		Amount:              amount,
+		OutgoingChanSet:     chanSet,
 		MaxPrepayRoutingFee: prepayMaxFee,
 		MaxSwapRoutingFee:   routeMaxFee,
 		MaxMinerFee:         m.params.MaximumMinerFee,
@@ -1143,21 +1276,23 @@ func newSwapTraffic() *swapTraffic {
 // maySwap returns a boolean that indicates whether we may perform a swap for a
 // peer and its set of channels.
 func (s *swapTraffic) maySwap(peer route.Vertex,
-	chanID lnwire.ShortChannelID) error {
+	channels []lnwire.ShortChannelID) error {
 
-	lastFail, recentFail := s.failedLoopOut[chanID]
-	if recentFail {
-		log.Debugf("Channel: %v not eligible for suggestions, was "+
-			"part of a failed swap at: %v", chanID, lastFail)
+	for _, chanID := range channels {
+		lastFail, recentFail := s.failedLoopOut[chanID]
+		if recentFail {
+			log.Debugf("Channel: %v not eligible for suggestions, was "+
+				"part of a failed swap at: %v", chanID, lastFail)
 
-		return newReasonError(ReasonFailureBackoff)
-	}
+			return newReasonError(ReasonFailureBackoff)
+		}
 
-	if s.ongoingLoopOut[chanID] {
-		log.Debugf("Channel: %v not eligible for suggestions, "+
-			"ongoing loop out utilizing channel", chanID)
+		if s.ongoingLoopOut[chanID] {
+			log.Debugf("Channel: %v not eligible for suggestions, "+
+				"ongoing loop out utilizing channel", chanID)
 
-		return newReasonError(ReasonLoopOut)
+			return newReasonError(ReasonLoopOut)
+		}
 	}
 
 	if s.ongoingLoopIn[peer] {

liquidity/liquidity_test.go

@@ -25,6 +25,7 @@ var (
 
 	chanID1 = lnwire.NewShortChanIDFromInt(1)
 	chanID2 = lnwire.NewShortChanIDFromInt(2)
+	chanID3 = lnwire.NewShortChanIDFromInt(3)
 
 	peer1 = route.Vertex{1}
 	peer2 = route.Vertex{2}
@@ -111,6 +112,10 @@ var (
 	// noneDisqualified can be used in tests where we don't have any
 	// disqualified channels so that we can use require.Equal.
 	noneDisqualified = make(map[lnwire.ShortChannelID]Reason)
+
+	// noPeersDisqualified can be used in tests where we don't have any
+	// disqualified peers so that we can use require.Equal.
+	noPeersDisqualified = make(map[route.Vertex]Reason)
 )
 
 // newTestConfig creates a default test config.
@@ -280,27 +285,36 @@ func TestRestrictedSuggestions(t *testing.T) {
 				defaultFailureBackoff * -3,
 			),
 		}
+
+		chanRules = map[lnwire.ShortChannelID]*ThresholdRule{
+			chanID1: chanRule,
+			chanID2: chanRule,
+		}
 	)
 
 	tests := []struct {
-		name     string
-		channels []lndclient.ChannelInfo
-		loopOut  []*loopdb.LoopOut
-		loopIn   []*loopdb.LoopIn
-		expected *Suggestions
+		name      string
+		channels  []lndclient.ChannelInfo
+		loopOut   []*loopdb.LoopOut
+		loopIn    []*loopdb.LoopIn
+		chanRules map[lnwire.ShortChannelID]*ThresholdRule
+		peerRules map[route.Vertex]*ThresholdRule
+		expected  *Suggestions
 	}{
 		{
 			name: "no existing swaps",
 			channels: []lndclient.ChannelInfo{
 				channel1,
 			},
-			loopOut: nil,
-			loopIn:  nil,
+			loopOut:   nil,
+			loopIn:    nil,
+			chanRules: chanRules,
 			expected: &Suggestions{
 				OutSwaps: []loop.OutRequest{
 					chan1Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -315,11 +329,13 @@ func TestRestrictedSuggestions(t *testing.T) {
 					},
 				},
 			},
+			chanRules: chanRules,
 			expected: &Suggestions{
 				OutSwaps: []loop.OutRequest{
 					chan1Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -334,11 +350,13 @@ func TestRestrictedSuggestions(t *testing.T) {
 					},
 				},
 			},
+			chanRules: chanRules,
 			expected: &Suggestions{
 				OutSwaps: []loop.OutRequest{
 					chan1Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -351,6 +369,7 @@ func TestRestrictedSuggestions(t *testing.T) {
 					Contract: chan1Out,
 				},
 			},
+			chanRules: chanRules,
 			expected: &Suggestions{
 				OutSwaps: []loop.OutRequest{
 					chan2Rec,
@@ -358,6 +377,7 @@ func TestRestrictedSuggestions(t *testing.T) {
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID1: ReasonLoopOut,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -372,6 +392,7 @@ func TestRestrictedSuggestions(t *testing.T) {
 					},
 				},
 			},
+			chanRules: chanRules,
 			expected: &Suggestions{
 				OutSwaps: []loop.OutRequest{
 					chan1Rec,
@@ -379,6 +400,7 @@ func TestRestrictedSuggestions(t *testing.T) {
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID2: ReasonLoopIn,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -396,10 +418,12 @@ func TestRestrictedSuggestions(t *testing.T) {
 					},
 				},
 			},
+			chanRules: chanRules,
 			expected: &Suggestions{
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID1: ReasonFailureBackoff,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -417,11 +441,13 @@ func TestRestrictedSuggestions(t *testing.T) {
 					},
 				},
 			},
+			chanRules: chanRules,
 			expected: &Suggestions{
 				OutSwaps: []loop.OutRequest{
 					chan1Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -439,10 +465,36 @@ func TestRestrictedSuggestions(t *testing.T) {
 					},
 				},
 			},
+			chanRules: chanRules,
 			expected: &Suggestions{
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID1: ReasonLoopOut,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
+			},
+		},
+		{
+			name: "existing on peer's channel",
+			channels: []lndclient.ChannelInfo{
+				channel1,
+				{
+					ChannelID:   chanID3.ToUint64(),
+					PubKeyBytes: peer1,
+				},
+			},
+			loopOut: []*loopdb.LoopOut{
+				{
+					Contract: chan1Out,
+				},
+			},
+			peerRules: map[route.Vertex]*ThresholdRule{
+				peer1: NewThresholdRule(0, 50),
+			},
+			expected: &Suggestions{
+				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: map[route.Vertex]Reason{
+					peer1: ReasonLoopOut,
+				},
 			},
 		},
 	}
@@ -464,9 +516,12 @@ func TestRestrictedSuggestions(t *testing.T) {
 			lnd.Channels = testCase.channels
 
 			params := defaultParameters
-			params.ChannelRules = map[lnwire.ShortChannelID]*ThresholdRule{
-				chanID1: chanRule,
-				chanID2: chanRule,
+			if testCase.chanRules != nil {
+				params.ChannelRules = testCase.chanRules
+			}
+
+			if testCase.peerRules != nil {
+				params.PeerRules = testCase.peerRules
 			}
 
 			testSuggestSwaps(
@@ -493,6 +548,7 @@ func TestSweepFeeLimit(t *testing.T) {
 					chan1Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -502,6 +558,7 @@ func TestSweepFeeLimit(t *testing.T) {
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID1: ReasonSweepFees,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 	}
@@ -537,19 +594,27 @@ func TestSweepFeeLimit(t *testing.T) {
 // TestSuggestSwaps tests getting of swap suggestions based on the rules set for
 // the liquidity manager and the current set of channel balances.
 func TestSuggestSwaps(t *testing.T) {
+	singleChannel := []lndclient.ChannelInfo{
+		channel1,
+	}
+
 	tests := []struct {
 		name        string
+		channels    []lndclient.ChannelInfo
 		rules       map[lnwire.ShortChannelID]*ThresholdRule
+		peerRules   map[route.Vertex]*ThresholdRule
 		suggestions *Suggestions
 		err         error
 	}{
 		{
-			name:  "no rules",
-			rules: map[lnwire.ShortChannelID]*ThresholdRule{},
-			err:   ErrNoRules,
+			name:     "no rules",
+			channels: singleChannel,
+			rules:    map[lnwire.ShortChannelID]*ThresholdRule{},
+			err:      ErrNoRules,
 		},
 		{
-			name: "loop out",
+			name:     "loop out",
+			channels: singleChannel,
 			rules: map[lnwire.ShortChannelID]*ThresholdRule{
 				chanID1: chanRule,
 			},
@@ -558,15 +623,76 @@ func TestSuggestSwaps(t *testing.T) {
 					chan1Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
-			name: "no rule for channel",
+			name:     "no rule for channel",
+			channels: singleChannel,
 			rules: map[lnwire.ShortChannelID]*ThresholdRule{
 				chanID2: NewThresholdRule(10, 10),
 			},
 			suggestions: &Suggestions{
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
+			},
+		},
+		{
+			name: "multiple peer rules",
+			channels: []lndclient.ChannelInfo{
+				{
+					PubKeyBytes:   peer1,
+					ChannelID:     chanID1.ToUint64(),
+					Capacity:      20000,
+					LocalBalance:  8000,
+					RemoteBalance: 12000,
+				},
+				{
+					PubKeyBytes:   peer1,
+					ChannelID:     chanID2.ToUint64(),
+					Capacity:      10000,
+					LocalBalance:  9000,
+					RemoteBalance: 1000,
+				},
+				{
+					PubKeyBytes:   peer2,
+					ChannelID:     chanID3.ToUint64(),
+					Capacity:      5000,
+					LocalBalance:  2000,
+					RemoteBalance: 3000,
+				},
+			},
+			peerRules: map[route.Vertex]*ThresholdRule{
+				peer1: NewThresholdRule(80, 0),
+				peer2: NewThresholdRule(40, 50),
+			},
+			suggestions: &Suggestions{
+				OutSwaps: []loop.OutRequest{
+					{
+						Amount: 10000,
+						OutgoingChanSet: loopdb.ChannelSet{
+							chanID1.ToUint64(),
+							chanID2.ToUint64(),
+						},
+						MaxPrepayRoutingFee: ppmToSat(
+							testQuote.PrepayAmount,
+							defaultPrepayRoutingFeePPM,
+						),
+						MaxSwapRoutingFee: ppmToSat(
+							10000,
+							defaultRoutingFeePPM,
+						),
+						MaxMinerFee:     defaultMaximumMinerFee,
+						MaxSwapFee:      testQuote.SwapFee,
+						MaxPrepayAmount: testQuote.PrepayAmount,
+						SweepConfTarget: loop.DefaultSweepConfTarget,
+						Initiator:       autoloopSwapInitiator,
+					},
+				},
+				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: map[route.Vertex]Reason{
+					peer2: ReasonLiquidityOk,
+				},
 			},
 		},
 	}
@@ -577,12 +703,16 @@ func TestSuggestSwaps(t *testing.T) {
 		t.Run(testCase.name, func(t *testing.T) {
 			cfg, lnd := newTestConfig()
 
-			lnd.Channels = []lndclient.ChannelInfo{
-				channel1,
-			}
+			lnd.Channels = testCase.channels
 
 			params := defaultParameters
-			params.ChannelRules = testCase.rules
+			if testCase.rules != nil {
+				params.ChannelRules = testCase.rules
+			}
+
+			if testCase.peerRules != nil {
+				params.PeerRules = testCase.peerRules
+			}
 
 			testSuggestSwaps(
 				t, newSuggestSwapsSetup(cfg, lnd, params),
@@ -607,6 +737,7 @@ func TestFeeLimits(t *testing.T) {
 					chan1Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -620,6 +751,7 @@ func TestFeeLimits(t *testing.T) {
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID1: ReasonPrepay,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -633,6 +765,7 @@ func TestFeeLimits(t *testing.T) {
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID1: ReasonMinerFee,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -647,6 +780,7 @@ func TestFeeLimits(t *testing.T) {
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID1: ReasonSwapFee,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 	}
@@ -719,6 +853,7 @@ func TestFeeBudget(t *testing.T) {
 					chan1Rec, chan2Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -734,6 +869,7 @@ func TestFeeBudget(t *testing.T) {
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID2: ReasonBudgetInsufficient,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -750,6 +886,7 @@ func TestFeeBudget(t *testing.T) {
 					chan1Rec, chan2Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -768,6 +905,7 @@ func TestFeeBudget(t *testing.T) {
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID2: ReasonBudgetInsufficient,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -782,6 +920,7 @@ func TestFeeBudget(t *testing.T) {
 					chanID1: ReasonBudgetElapsed,
 					chanID2: ReasonBudgetElapsed,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 	}
@@ -875,6 +1014,7 @@ func TestInFlightLimit(t *testing.T) {
 					chan1Rec, chan2Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -885,6 +1025,7 @@ func TestInFlightLimit(t *testing.T) {
 					chan1Rec, chan2Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -902,6 +1043,7 @@ func TestInFlightLimit(t *testing.T) {
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID2: ReasonInFlight,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -917,6 +1059,7 @@ func TestInFlightLimit(t *testing.T) {
 					chanID1: ReasonInFlight,
 					chanID2: ReasonInFlight,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -935,6 +1078,7 @@ func TestInFlightLimit(t *testing.T) {
 					chanID1: ReasonInFlight,
 					chanID2: ReasonInFlight,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 	}
@@ -1026,6 +1170,7 @@ func TestSizeRestrictions(t *testing.T) {
 					chan1Rec,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -1040,6 +1185,7 @@ func TestSizeRestrictions(t *testing.T) {
 				DisqualifiedChans: map[lnwire.ShortChannelID]Reason{
 					chanID1: ReasonLiquidityOk,
 				},
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{
@@ -1055,6 +1201,7 @@ func TestSizeRestrictions(t *testing.T) {
 					outSwap,
 				},
 				DisqualifiedChans: noneDisqualified,
+				DisqualifiedPeers: noPeersDisqualified,
 			},
 		},
 		{