New filter system along with new filters (closes #321) (#323)

* 1 - inject submitFilters

* 2 - inject db instance

* 3 - extract filtering infrastructure logic from makerModeFilter for reuse

* 4 - separate out TimestampFormatString and DateFormatString

* 5 - first draft of volumeFilter.go (untested, many TODOs)

* 6 - fetch marketID in volumeFilter

* 7 - better index for daily trades table query

* 8 - take in volumeFilter config values

* 9 - update sample_trader.cfg to include sample for volumeFilterConfig

* 10 - StructString displays values in structs in a cleaner way + automatic pointer unwrapping

* 11 - treat values as 0 when there are no recorded trades in the db for the day

* 12 - fix initialization of dailyTBB to avoid a null pointer dereference

* 13 - fix bug caused by variable shadowing in volumeFilter.go

* 14 - limit check should be lte, not lt check

* 15 - update filterOps() to run existing offers through filter function

converts existing offers to the equivalent operation so it can be run through the filter function

* 16 - added TODO for simplification of filterOps

* 17 - update sample config to denote that volumeFilter only works with the sell strategy and requires POSTGRES_DB config

* 18 - minPrice filter

* 19 - add filterName to filterOps log line

* 20 - MAX_PRICE filter

* 21 - new format of filters in config file with corresponding filterFactory.go

* 22 - force fail if any new filters are specified without sell or delete strategies

* 23 - extract logic that never drops delete operations into filterOps()

* 24 - included additional explanation in FILTERS section of sample config file

* 25 - fix Errorf statement

cmd/trade.go

@@ -1,6 +1,7 @@
 package cmd
 
 import (
+	"database/sql"
 	"fmt"
 	"io"
 	"log"
@@ -329,7 +330,9 @@ func makeBot(
 	exchangeShim api.ExchangeShim,
 	ieif *plugins.IEIF,
 	tradingPair *model.TradingPair,
+	db *sql.DB,
 	strategy api.Strategy,
+	assetDisplayFn model.AssetDisplayFn,
 	threadTracker *multithreading.ThreadTracker,
 	options inputs,
 ) *trader.Trader {
@@ -344,30 +347,67 @@ func makeBot(
 		// we want to delete all the offers and exit here since there is something wrong with our setup
 		deleteAllOffersAndExit(l, botConfig, client, sdex, exchangeShim, threadTracker)
 	}
-	dataKey := model.MakeSortedBotKey(botConfig.AssetBase(), botConfig.AssetQuote())
+
+	assetBase := botConfig.AssetBase()
+	assetQuote := botConfig.AssetQuote()
+	dataKey := model.MakeSortedBotKey(assetBase, assetQuote)
 	alert, e := monitoring.MakeAlert(botConfig.AlertType, botConfig.AlertAPIKey)
 	if e != nil {
 		l.Infof("Unable to set up monitoring for alert type '%s' with the given API key\n", botConfig.AlertType)
 	}
-	bot := trader.MakeBot(
+
+	// start make filters
+	submitFilters := []plugins.SubmitFilter{
+		plugins.MakeFilterOrderConstraints(exchangeShim.GetOrderConstraints(tradingPair), assetBase, assetQuote),
+	}
+	if submitMode == api.SubmitModeMakerOnly {
+		submitFilters = append(submitFilters,
+			plugins.MakeFilterMakerMode(exchangeShim, sdex, tradingPair),
+		)
+	}
+	if len(botConfig.Filters) > 0 && *options.strategy != "sell" && *options.strategy != "delete" {
+		log.Println()
+		utils.PrintErrorHintf("FILTERS currently only supported on 'sell' and 'delete' strategies, remove FILTERS from the trader config file")
+		// we want to delete all the offers and exit here since there is something wrong with our setup
+		deleteAllOffersAndExit(l, botConfig, client, sdex, exchangeShim, threadTracker)
+	}
+	filterFactory := plugins.FilterFactory{
+		ExchangeName:   botConfig.TradingExchangeName(),
+		TradingPair:    tradingPair,
+		AssetDisplayFn: assetDisplayFn,
+		BaseAsset:      assetBase,
+		QuoteAsset:     assetQuote,
+		DB:             db,
+	}
+	for _, filterString := range botConfig.Filters {
+		filter, e := filterFactory.MakeFilter(filterString)
+		if e != nil {
+			log.Println()
+			log.Println(e)
+			// we want to delete all the offers and exit here since there is something wrong with our setup
+			deleteAllOffersAndExit(l, botConfig, client, sdex, exchangeShim, threadTracker)
+		}
+		submitFilters = append(submitFilters, filter)
+	}
+	// end make filters
+
+	return trader.MakeTrader(
 		client,
 		ieif,
-		botConfig.AssetBase(),
-		botConfig.AssetQuote(),
-		tradingPair,
+		assetBase,
+		assetQuote,
 		botConfig.TradingAccount(),
 		sdex,
 		exchangeShim,
 		strategy,
 		timeController,
 		botConfig.DeleteCyclesThreshold,
-		submitMode,
+		submitFilters,
 		threadTracker,
 		options.fixedIterations,
 		dataKey,
 		alert,
 	)
-	return bot
 }
 
 func convertDeprecatedBotConfigValues(l logger.Logger, botConfig trader.BotConfig) trader.BotConfig {
@@ -434,6 +474,20 @@ func runTradeCmd(options inputs) {
 		tradingPair.Base:  botConfig.AssetBase(),
 		tradingPair.Quote: botConfig.AssetQuote(),
 	}
+	assetDisplayFn := model.MakePassthroughAssetDisplayFn()
+	if botConfig.IsTradingSdex() {
+		assetDisplayFn = model.MakeSdexMappedAssetDisplayFn(sdexAssetMap)
+	}
+
+	var db *sql.DB
+	if botConfig.PostgresDbConfig != nil {
+		var e error
+		db, e = database.ConnectInitializedDatabase(botConfig.PostgresDbConfig)
+		if e != nil {
+			logger.Fatal(l, fmt.Errorf("problem encountered while initializing the db: %s", e))
+		}
+		log.Printf("made db instance with config: %s\n", botConfig.PostgresDbConfig.MakeConnectString())
+	}
 	exchangeShim, sdex := makeExchangeShimSdex(
 		l,
 		botConfig,
@@ -467,7 +521,9 @@ func runTradeCmd(options inputs) {
 		exchangeShim,
 		ieif,
 		tradingPair,
+		db,
 		strategy,
+		assetDisplayFn,
 		threadTracker,
 		options,
 	)
@@ -496,7 +552,8 @@ func runTradeCmd(options inputs) {
 		sdex,
 		exchangeShim,
 		tradingPair,
-		sdexAssetMap,
+		assetDisplayFn,
+		db,
 		threadTracker,
 	)
 	startQueryServer(
@@ -565,7 +622,8 @@ func startFillTracking(
 	sdex *plugins.SDEX,
 	exchangeShim api.ExchangeShim,
 	tradingPair *model.TradingPair,
-	sdexAssetMap map[model.Asset]hProtocol.Asset,
+	assetDisplayFn model.AssetDisplayFn,
+	db *sql.DB,
 	threadTracker *multithreading.ThreadTracker,
 ) {
 	strategyFillHandlers, e := strategy.GetFillHandlers()
@@ -580,19 +638,7 @@ func startFillTracking(
 		fillTracker := plugins.MakeFillTracker(tradingPair, threadTracker, exchangeShim, botConfig.FillTrackerSleepMillis, botConfig.FillTrackerDeleteCyclesThreshold)
 		fillLogger := plugins.MakeFillLogger()
 		fillTracker.RegisterHandler(fillLogger)
-		if botConfig.PostgresDbConfig != nil {
-			db, e := database.ConnectInitializedDatabase(botConfig.PostgresDbConfig)
-			if e != nil {
-				l.Info("")
-				l.Errorf("problem encountered while initializing the db: %s", e)
-				deleteAllOffersAndExit(l, botConfig, client, sdex, exchangeShim, threadTracker)
-			}
-			log.Printf("made db instance with config: %s\n", botConfig.PostgresDbConfig.MakeConnectString())
-
-			assetDisplayFn := model.MakePassthroughAssetDisplayFn()
-			if botConfig.IsTradingSdex() {
-				assetDisplayFn = model.MakeSdexMappedAssetDisplayFn(sdexAssetMap)
-			}
+		if db != nil {
 			fillDBWriter := plugins.MakeFillDBWriter(db, assetDisplayFn, botConfig.TradingExchangeName())
 			fillTracker.RegisterHandler(fillDBWriter)
 		}

database/schema.go

@@ -17,6 +17,8 @@ const sqlTradesTableCreate = "CREATE TABLE IF NOT EXISTS trades (market_id TEXT
 	indexes
 */
 const sqlTradesIndexCreate = "CREATE INDEX IF NOT EXISTS date ON trades (market_id, date_utc)"
+const sqlTradesIndexDrop = "DROP INDEX IF EXISTS date"
+const sqlTradesIndexCreate2 = "CREATE INDEX IF NOT EXISTS trades_mdd ON trades (market_id, DATE(date_utc), date_utc)"
 
 /*
 	insert statements
@@ -39,6 +41,9 @@ 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)"
+
 /*
 	query helper functions
 */

database/upgrade.go

@@ -18,6 +18,10 @@ var upgradeScripts = []*UpgradeScript{
 		sqlTradesTableCreate,
 		sqlTradesIndexCreate,
 	),
+	makeUpgradeScript(3,
+		sqlTradesIndexDrop,
+		sqlTradesIndexCreate2,
+	),
 }
 
 // UpgradeScript encapsulates a script to be run to upgrade the database from one version to the next
@@ -104,7 +108,7 @@ func runUpgradeScripts(db *sql.DB, scripts []*UpgradeScript) error {
 		// add entry to db_version table
 		sqlInsertDbVersion := fmt.Sprintf(sqlDbVersionTableInsertTemplate,
 			script.version,
-			startTime.Format(postgresdb.DateFormatString),
+			startTime.Format(postgresdb.TimestampFormatString),
 			len(script.commands),
 			elapsedMillis,
 		)

examples/configs/trader/sample_trader.cfg

@@ -51,15 +51,6 @@ HORIZON_URL="https://horizon-testnet.stellar.org"
 # the URL to use for your CCXT-rest instance. Defaults to http://localhost:3000 if unset
 #CCXT_REST_URL="http://localhost:3000"
 
-# specify parameters for how we compute the operation fee from the /fee_stats endpoint
-[FEE]
-# trigger when "ledger_capacity_usage" in /fee_stats is >= this value
-CAPACITY_TRIGGER=0.8
-# percentile computation to use from /fee_stats (10, 20, ..., 90, 95, 99)
-PERCENTILE=90
-# max fee in stroops per operation to use
-MAX_OP_FEE_STROOPS=5000
-
 # uncomment below to add support for monitoring.
 # type of alerting system to use, currently only "PagerDuty" is supported.
 #ALERT_TYPE="PagerDuty"
@@ -99,6 +90,33 @@ MAX_OP_FEE_STROOPS=5000
 # (optional) minimum volume of quote units needed to place an order on the non-sdex (centralized) exchange
 #CENTRALIZED_MIN_QUOTE_VOLUME_OVERRIDE=10.0
 
+# uncomment to include these filters in order (these filters only work with sell strategy for now)
+# these are the only four filters available for now via this new filtration method and any new filters added will include a
+# corresponding sample entry with an explanation.
+# 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)
+#    "volume/sell/base/3500.0",
+#
+#    # limit the amount of the base asset that is sold every day, denominated in units of the quote asset (needs POSTGRES_DB)
+#    "volume/sell/quote/1000.0",
+#
+#    # limit offers based on a minimim price requirement
+#    "price/min/0.04",
+#
+#    # limit offers based on a maximum price requirement
+#    "price/max/1.00",
+#]
+
+# specify parameters for how we compute the operation fee from the /fee_stats endpoint
+[FEE]
+# trigger when "ledger_capacity_usage" in /fee_stats is >= this value
+CAPACITY_TRIGGER=0.8
+# percentile computation to use from /fee_stats (10, 20, ..., 90, 95, 99)
+PERCENTILE=90
+# max fee in stroops per operation to use
+MAX_OP_FEE_STROOPS=5000
+
 # uncomment if you want to track fills in a postgres db
 #[POSTGRES_DB]
 #HOST="localhost"

plugins/balancedStrategy.go

@@ -26,7 +26,7 @@ type balancedConfig struct {
 
 // String impl.
 func (c balancedConfig) String() string {
-	return utils.StructString(c, nil)
+	return utils.StructString(c, 0, nil)
 }
 
 // makeBalancedStrategy is a factory method for balancedStrategy

plugins/buysellStrategy.go

@@ -55,7 +55,7 @@ func MakeBuysellConfig(
 
 // String impl.
 func (c BuySellConfig) String() string {
-	return utils.StructString(c, nil)
+	return utils.StructString(c, 0, nil)
 }
 
 // makeBuySellStrategy is a factory method

plugins/fillDBWriter.go

@@ -46,14 +46,17 @@ type FillDBWriter struct {
 	market *tradingMarket
 }
 
-// makeTradingMarket makes a market along with the ID field
-func makeTradingMarket(exchangeName string, baseAsset string, quoteAsset string) *tradingMarket {
+func makeMarketID(exchangeName string, baseAsset string, quoteAsset string) string {
 	idString := fmt.Sprintf("%s_%s_%s", exchangeName, baseAsset, quoteAsset)
 	h := sha256.New()
 	h.Write([]byte(idString))
 	sha256Hash := fmt.Sprintf("%x", h.Sum(nil))
-	sha256HashPrefix := sha256Hash[0:marketIdHashLength]
+	return sha256Hash[0:marketIdHashLength]
+}
 
+// makeTradingMarket makes a market along with the ID field
+func makeTradingMarket(exchangeName string, baseAsset string, quoteAsset string) *tradingMarket {
+	sha256HashPrefix := makeMarketID(exchangeName, baseAsset, quoteAsset)
 	return &tradingMarket{
 		ID:           sha256HashPrefix,
 		ExchangeName: exchangeName,
@@ -156,7 +159,7 @@ func (f *FillDBWriter) HandleFill(trade model.Trade) error {
 	txid := utils.CheckedString(trade.TransactionID)
 	timeSeconds := trade.Timestamp.AsInt64() / 1000
 	date := time.Unix(timeSeconds, 0).UTC()
-	dateString := date.Format(postgresdb.DateFormatString)
+	dateString := date.Format(postgresdb.TimestampFormatString)
 
 	market, e := f.fetchOrRegisterMarket(trade)
 	if e != nil {