volume filter should allow specifying multiple markets for aggregate …

…limits (closes #348) (#351)

* 1 - change marketID in volumeFilter to marketIDs

* 2 - parse market_ids from volume filter config string

* 3 - fix comment in sample_trader referencing filterMode

* 4 - add example of market_ids to volumeFilter in sample_trader

* 5 - add missing params to Stringer method of VolumeFilterConfig

* 6 - specify case when we don't have any market_ids

* 7 - fix typo in sample_trader comment

* 8 - fixed typo: markets_ids -> market_ids

* 9 - better parseMarketIdsArray function

* 10 - fix sql query for daily values with IN operator

* 11 - add single quote to marketIDs in sql query

* Revert "11 - add single quote to marketIDs in sql query"

This reverts commit 8e77b78.

* 12 - use a query template to fill in the IN clause

* 13 - fix TestParseMarketIdsArray

* 14 - reword explanation of filter in sample_trader.cfg

database/schema.go

@@ -41,8 +41,8 @@ const SqlQueryMarketsById = "SELECT market_id, exchange_name, base, quote FROM m
 // sqlQueryDbVersion queries the db_version table
 const sqlQueryDbVersion = "SELECT version FROM db_version ORDER BY version desc LIMIT 1"
 
-// SqlQueryDailyValues queries the trades table to get the values for a given day
-const SqlQueryDailyValues = "SELECT SUM(base_volume) as total_base_volume, SUM(counter_cost) as total_counter_volume FROM trades WHERE market_id = $1 AND DATE(date_utc) = $2 and action = $3 group by DATE(date_utc)"
+// SqlQueryDailyValuesTemplate queries the trades table to get the values for a given day
+const SqlQueryDailyValuesTemplate = "SELECT SUM(base_volume) as total_base_volume, SUM(counter_cost) as total_counter_volume FROM trades WHERE market_id IN (%s) AND DATE(date_utc) = $1 and action = $2 group by DATE(date_utc)"
 
 /*
 	query helper functions

examples/configs/trader/sample_trader.cfg

@@ -98,7 +98,7 @@ HORIZON_URL="https://horizon-testnet.stellar.org"
 # the best way to use these filters is to uncomment the one you want to use and update the price (last param) accordingly.
 #FILTERS = [
 #    # limit the amount of the base asset that is sold every day, denominated in units of the base asset (needs POSTGRES_DB)
-#    # The fifth param can be either "exact" or "ignore" ("exact" is recommended):
+#    # The sixth param can be either "exact" or "ignore" ("exact" is recommended):
 #    #     - "exact" indicates that the volume filter should modify the amount of the offer that will cause the capacity limit
 #    #        to be exceeded (when daily sold amounts are close to the limit). This will result in the exact number of units of
 #    #        the asset to be sold for the given day.
@@ -108,7 +108,7 @@ HORIZON_URL="https://horizon-testnet.stellar.org"
 #    "volume/daily/sell/base/3500.0/exact",
 #
 #    # limit the amount of the base asset that is sold every day, denominated in units of the quote asset (needs POSTGRES_DB)
-#    # The fifth param can be either "exact" or "ignore" ("exact" is recommended):
+#    # The sixth param can be either "exact" or "ignore" ("exact" is recommended):
 #    #     - "exact" indicates that the volume filter should modify the amount of the offer that will cause the capacity limit
 #    #        to be exceeded (when daily sold amounts are close to the limit). This will result in the exact number of units of
 #    #        the asset to be sold for the given day.
@@ -117,6 +117,13 @@ HORIZON_URL="https://horizon-testnet.stellar.org"
 #    #        of the asset to be sold for the given day.
 #    "volume/daily/sell/quote/1000.0/ignore",
 #
+#    # include additional markets in the filter.
+#    # market_ids is an array whose values are market_ids from the postgres database.
+#    # in the example below, we will consider the daily volume from the markets 4c19915f47 and db4531d586, in addition to the local
+#    # market, and limit the sum of the total across these markets to the limit specified in the filter string.
+#    # It's the user's responsibility to ensure that each market_id corresponds to the asset pair and exchange they want included.
+#    "volume/daily:market_ids=[4c19915f47,db4531d586]/sell/base/3500.0/exact",
+#
 #    # limit offers based on a minimim price requirement
 #    "price/min/0.04",
 #

plugins/filterFactory.go

@@ -3,13 +3,24 @@ package plugins
 import (
 	"database/sql"
 	"fmt"
+	"regexp"
 	"strconv"
 	"strings"
 
 	hProtocol "github.com/stellar/go/protocols/horizon"
 	"github.com/stellar/kelp/model"
 )
 
+var marketIDRegex *regexp.Regexp
+
+func init() {
+	midRxp, e := regexp.Compile("^[a-zA-Z0-9]{10}$")
+	if e != nil {
+		panic("unable to compile marketID regexp")
+	}
+	marketIDRegex = midRxp
+}
+
 var filterMap = map[string]func(f *FilterFactory, configInput string) (SubmitFilter, error){
 	"volume":    filterVolume,
 	"price":     filterPrice,
@@ -53,9 +64,31 @@ func filterVolume(f *FilterFactory, configInput string) (SubmitFilter, error) {
 		return nil, fmt.Errorf("could not parse volume filter mode from input (%s): %s", configInput, e)
 	}
 	config := &VolumeFilterConfig{mode: mode}
-	if parts[1] != "daily" {
-		return nil, fmt.Errorf("invalid input (%s), the second part needs to be \"daily\"", configInput)
+
+	limitWindowParts := strings.Split(parts[1], ":")
+	if limitWindowParts[0] != "daily" {
+		return nil, fmt.Errorf("invalid input (%s), the second part needs to equal or start with \"daily\"", configInput)
+	} else if len(limitWindowParts) == 2 {
+		errInvalid := fmt.Errorf("invalid input (%s), the modifier for \"daily\" can only be \"market_ids\" like so 'daily:market_ids=[4c19915f47,db4531d586]'", configInput)
+		if !strings.HasPrefix(limitWindowParts[1], "market_ids=") {
+			return nil, fmt.Errorf("%s: invalid modifier prefix in '%s'", errInvalid, limitWindowParts[1])
+		}
+
+		modifierParts := strings.Split(limitWindowParts[1], "=")
+		if len(modifierParts) != 2 {
+			return nil, fmt.Errorf("%s: invalid parts for modifier with length %d, should have been 2", errInvalid, len(modifierParts))
+		}
+
+		marketIds, e := parseMarketIdsArray(modifierParts[1])
+		if e != nil {
+			return nil, fmt.Errorf("%s: %s", errInvalid, e)
+		}
+
+		config.additionalMarketIDs = marketIds
+	} else if len(limitWindowParts) != 1 {
+		return nil, fmt.Errorf("invalid input (%s), the second part needs to be \"daily\" and can have only one modifier \"market_ids\" like so 'daily:market_ids=[4c19915f47,db4531d586]'", configInput)
 	}
+
 	if parts[2] != "sell" {
 		return nil, fmt.Errorf("invalid input (%s), the third part needs to be \"sell\"", configInput)
 	}
@@ -85,6 +118,32 @@ func filterVolume(f *FilterFactory, configInput string) (SubmitFilter, error) {
 	)
 }
 
+func parseMarketIdsArray(marketIdsArrayString string) ([]string, error) {
+	if !strings.HasPrefix(marketIdsArrayString, "[") {
+		return nil, fmt.Errorf("market_ids array should begin with '['")
+	}
+
+	if !strings.HasSuffix(marketIdsArrayString, "]") {
+		return nil, fmt.Errorf("market_ids array should end with ']'")
+	}
+
+	arrayStringCleaned := marketIdsArrayString[:len(marketIdsArrayString)-1][1:]
+	marketIds := strings.Split(arrayStringCleaned, ",")
+	if len(marketIds) == 0 {
+		return nil, fmt.Errorf("market_ids array length should be greater than 0")
+	}
+
+	marketIdsTrimmed := []string{}
+	for _, mid := range marketIds {
+		trimmedMid := strings.TrimSpace(mid)
+		if !marketIDRegex.MatchString(trimmedMid) {
+			return nil, fmt.Errorf("invalid market_id entry '%s'", trimmedMid)
+		}
+		marketIdsTrimmed = append(marketIdsTrimmed, trimmedMid)
+	}
+	return marketIdsTrimmed, nil
+}
+
 func filterPrice(f *FilterFactory, configInput string) (SubmitFilter, error) {
 	parts := strings.Split(configInput, "/")
 	if len(parts) != 3 {

plugins/filterFactory_test.go

@@ -0,0 +1,36 @@
+package plugins
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseMarketIdsArray(t *testing.T) {
+	testCases := []struct {
+		marketIdsArrayString string
+		want                 []string
+	}{
+		{
+			marketIdsArrayString: "[abcde1234Z,01234gFHij]",
+			want:                 []string{"abcde1234Z", "01234gFHij"},
+		}, {
+			marketIdsArrayString: "[abcde1234Z, 01234gFHij]",
+			want:                 []string{"abcde1234Z", "01234gFHij"},
+		}, {
+			marketIdsArrayString: "[abcde1234Z]",
+			want:                 []string{"abcde1234Z"},
+		},
+	}
+
+	for _, kase := range testCases {
+		t.Run(kase.marketIdsArrayString, func(t *testing.T) {
+			output, e := parseMarketIdsArray(kase.marketIdsArrayString)
+			if !assert.NoError(t, e) {
+				return
+			}
+
+			assert.Equal(t, kase.want, output)
+		})
+	}
+}

plugins/volumeFilter.go

@@ -38,17 +38,20 @@ type VolumeFilterConfig struct {
 	SellBaseAssetCapInBaseUnits  *float64
 	SellBaseAssetCapInQuoteUnits *float64
 	mode                         volumeFilterMode
+	additionalMarketIDs          []string
 	// buyBaseAssetCapInBaseUnits   *float64
 	// buyBaseAssetCapInQuoteUnits  *float64
 }
 
 type volumeFilter struct {
-	name       string
-	baseAsset  hProtocol.Asset
-	quoteAsset hProtocol.Asset
-	marketID   string
-	config     *VolumeFilterConfig
-	db         *sql.DB
+	name                string
+	baseAsset           hProtocol.Asset
+	quoteAsset          hProtocol.Asset
+	sqlQueryDailyValues string
+	marketIDs           []string
+	action              string
+	config              *VolumeFilterConfig
+	db                  *sql.DB
 }
 
 // makeFilterVolume makes a submit filter that limits orders placed based on the daily volume traded
@@ -75,17 +78,32 @@ func makeFilterVolume(
 		return nil, fmt.Errorf("could not convert quote asset (%s) from trading pair via the passed in assetDisplayFn: %s", string(tradingPair.Quote), e)
 	}
 	marketID := makeMarketID(exchangeName, baseAssetString, quoteAssetString)
+	marketIDs := utils.Dedupe(append([]string{marketID}, config.additionalMarketIDs...))
+	sqlQueryDailyValues := makeSqlQueryDailyValues(marketIDs)
 
 	return &volumeFilter{
-		name:       "volumeFilter",
-		baseAsset:  baseAsset,
-		quoteAsset: quoteAsset,
-		marketID:   marketID,
-		config:     config,
-		db:         db,
+		name:                "volumeFilter",
+		baseAsset:           baseAsset,
+		quoteAsset:          quoteAsset,
+		sqlQueryDailyValues: sqlQueryDailyValues,
+		marketIDs:           marketIDs,
+		action:              "sell",
+		config:              config,
+		db:                  db,
 	}, nil
 }
 
+func makeSqlQueryDailyValues(marketIDs []string) string {
+	inClauseParts := []string{}
+	for _, mid := range marketIDs {
+		inValue := fmt.Sprintf("'%s'", mid)
+		inClauseParts = append(inClauseParts, inValue)
+	}
+	inClause := strings.Join(inClauseParts, ", ")
+
+	return fmt.Sprintf(database.SqlQueryDailyValuesTemplate, inClause)
+}
+
 var _ SubmitFilter = &volumeFilter{}
 
 // Validate ensures validity
@@ -98,14 +116,14 @@ func (c *VolumeFilterConfig) Validate() error {
 
 // String is the stringer method
 func (c *VolumeFilterConfig) String() string {
-	return fmt.Sprintf("VolumeFilterConfig[SellBaseAssetCapInBaseUnits=%s, SellBaseAssetCapInQuoteUnits=%s]",
-		utils.CheckedFloatPtr(c.SellBaseAssetCapInBaseUnits), utils.CheckedFloatPtr(c.SellBaseAssetCapInQuoteUnits))
+	return fmt.Sprintf("VolumeFilterConfig[SellBaseAssetCapInBaseUnits=%s, SellBaseAssetCapInQuoteUnits=%s, mode=%s, additionalMarketIDs=%v]",
+		utils.CheckedFloatPtr(c.SellBaseAssetCapInBaseUnits), utils.CheckedFloatPtr(c.SellBaseAssetCapInQuoteUnits), c.mode, c.additionalMarketIDs)
 }
 
 func (f *volumeFilter) Apply(ops []txnbuild.Operation, sellingOffers []hProtocol.Offer, buyingOffers []hProtocol.Offer) ([]txnbuild.Operation, error) {
 	dateString := time.Now().UTC().Format(postgresdb.DateFormatString)
-	// TODO do for buying base and also for a flipped marketID
-	dailyValuesBaseSold, e := f.dailyValuesByDate(f.marketID, dateString, "sell")
+	// TODO do for buying base and also for flipped marketIDs
+	dailyValuesBaseSold, e := f.dailyValuesByDate(dateString)
 	if e != nil {
 		return nil, fmt.Errorf("could not load dailyValuesByDate for today (%s): %s", dateString, e)
 	}
@@ -229,8 +247,8 @@ type dailyValues struct {
 	quoteVol float64
 }
 
-func (f *volumeFilter) dailyValuesByDate(marketID string, dateUTC string, action string) (*dailyValues, error) {
-	row := f.db.QueryRow(database.SqlQueryDailyValues, marketID, dateUTC, action)
+func (f *volumeFilter) dailyValuesByDate(dateUTC string) (*dailyValues, error) {
+	row := f.db.QueryRow(f.sqlQueryDailyValues, dateUTC, f.action)
 
 	var baseVol sql.NullFloat64
 	var quoteVol sql.NullFloat64