Merge branch 'offset_trades', closes #72

README.md

@@ -173,7 +173,7 @@ The following strategies are available **out of the box** with Kelp:
 
 - mirror ([source](plugins/mirrorStrategy.go)):
 
-    - **What:** mirrors an orderbook from another exchange by placing the same orders on Stellar after including a [spread][spread]. _Note: covering your trades on the backing exchange is not currently supported out-of-the-box_.
+    - **What:** mirrors an orderbook from another exchange by placing the same orders on Stellar after including a [spread][spread].
     - **Why:** To [hedge][hedge] your position on another exchange whenever a trade is executed to reduce inventory risk while keeping a spread
     - **Who:** Anyone who wants to reduce inventory risk and also has the capacity to take on a higher operational overhead in maintaining the bot system.
     - **Complexity:** Advanced

api/exchange.go

@@ -30,7 +30,9 @@ type TradesResult struct {
 }
 
 // TradeHistoryResult is the result of a GetTradeHistory call
+// this should be the same object as TradesResult but it's a separate object for backwards compatibility
 type TradeHistoryResult struct {
+	Cursor interface{}
 	Trades []model.Trade
 }
 
@@ -39,6 +41,32 @@ type TickerAPI interface {
 	GetTickerPrice(pairs []model.TradingPair) (map[model.TradingPair]Ticker, error)
 }
 
+// FillTracker knows how to track fills against open orders
+type FillTracker interface {
+	GetPair() (pair *model.TradingPair)
+	// TrackFills should be executed in a new thread
+	TrackFills() error
+	RegisterHandler(handler FillHandler)
+	NumHandlers() uint8
+}
+
+// FillHandler is invoked by the FillTracker (once registered) anytime an order is filled
+type FillHandler interface {
+	HandleFill(trade model.Trade) error
+}
+
+// TradeFetcher is the common method between FillTrackable and exchange
+// temporarily extracted out from TradeAPI so SDEX has the flexibility to only implement this rather than exchange and FillTrackable
+type TradeFetcher interface {
+	GetTradeHistory(maybeCursorStart interface{}, maybeCursorEnd interface{}) (*TradeHistoryResult, error)
+}
+
+// FillTrackable enables any implementing exchange to support fill tracking
+type FillTrackable interface {
+	TradeFetcher
+	GetLatestTradeCursor() (interface{}, error)
+}
+
 // TradeAPI is the interface we use as a generic API for trading on any crypto exchange
 type TradeAPI interface {
 	GetAssetConverter() *model.AssetConverter
@@ -47,7 +75,7 @@ type TradeAPI interface {
 
 	GetTrades(pair *model.TradingPair, maybeCursor interface{}) (*TradesResult, error)
 
-	GetTradeHistory(maybeCursorStart interface{}, maybeCursorEnd interface{}) (*TradeHistoryResult, error)
+	TradeFetcher
 
 	GetOpenOrders() (map[model.TradingPair][]model.OpenOrder, error)
 

api/level.go

@@ -11,4 +11,5 @@ type Level struct {
 // LevelProvider returns the levels for the given center price, which controls the spread and number of levels
 type LevelProvider interface {
 	GetLevels(maxAssetBase float64, maxAssetQuote float64) ([]Level, error)
+	GetFillHandlers() ([]FillHandler, error)
 }

api/strategy.go

@@ -12,6 +12,7 @@ type Strategy interface {
 	PreUpdate(maxAssetA float64, maxAssetB float64, trustA float64, trustB float64) error
 	UpdateWithOps(buyingAOffers []horizon.Offer, sellingAOffers []horizon.Offer) ([]build.TransactionMutator, error)
 	PostUpdate() error
+	GetFillHandlers() ([]FillHandler, error)
 }
 
 // SideStrategy represents a strategy on a single side of the orderbook
@@ -20,4 +21,5 @@ type SideStrategy interface {
 	PreUpdate(maxAssetA float64, maxAssetB float64, trustA float64, trustB float64) error
 	UpdateWithOps(offers []horizon.Offer) (ops []build.TransactionMutator, newTopOffer *model.Number, e error)
 	PostUpdate() error
+	GetFillHandlers() ([]FillHandler, error)
 }

cmd/exchanges.go

@@ -2,7 +2,6 @@ package cmd
 
 import (
 	"fmt"
-	"sort"
 
 	"github.com/interstellar/kelp/plugins"
 
@@ -16,20 +15,22 @@ var exchanagesCmd = &cobra.Command{
 
 func init() {
 	exchanagesCmd.Run = func(ccmd *cobra.Command, args []string) {
-		fmt.Printf("  Exchange\tDescription\n")
+		fmt.Printf("  Exchange\t\tSupports Trading\tDescription\n")
 		fmt.Printf("  --------------------------------------------------------------------------------\n")
 		exchanges := plugins.Exchanges()
 		for _, name := range sortedExchangeKeys(exchanges) {
-			fmt.Printf("  %-14s%s\n", name, exchanges[name])
+			fmt.Printf("  %-14s\t%v\t\t\t%s\n", name, exchanges[name].TradeEnabled, exchanges[name].Description)
 		}
 	}
 }
 
-func sortedExchangeKeys(m map[string]string) []string {
-	keys := []string{}
-	for name := range m {
-		keys = append(keys, name)
+func sortedExchangeKeys(m map[string]plugins.ExchangeContainer) []string {
+	keys := make([]string, len(m))
+	for k, v := range m {
+		if len(keys[v.SortOrder]) > 0 && keys[v.SortOrder] != k {
+			panic(fmt.Errorf("invalid sort order specified for strategies, SortOrder that was repeated: %d", v.SortOrder))
+		}
+		keys[v.SortOrder] = k
 	}
-	sort.Strings(keys)
 	return keys
 }

cmd/terminate.go

@@ -4,6 +4,7 @@ import (
 	"log"
 	"net/http"
 
+	"github.com/interstellar/kelp/model"
 	"github.com/interstellar/kelp/plugins"
 	"github.com/interstellar/kelp/support/utils"
 	"github.com/interstellar/kelp/terminator"
@@ -51,6 +52,8 @@ func init() {
 			-1, // not needed here
 			-1, // not needed here
 			false,
+			nil, // not needed here
+			map[model.Asset]horizon.Asset{},
 		)
 		terminator := terminator.MakeTerminator(client, sdex, *configFile.TradingAccount, configFile.TickIntervalSeconds, configFile.AllowInactiveMinutes)
 		// --- end initialization of objects ----

cmd/trade.go

@@ -137,6 +137,17 @@ func init() {
 		// --- start initialization of objects ----
 		threadTracker := multithreading.MakeThreadTracker()
 
+		assetBase := botConfig.AssetBase()
+		assetQuote := botConfig.AssetQuote()
+		tradingPair := &model.TradingPair{
+			Base:  model.Asset(utils.Asset2CodeString(assetBase)),
+			Quote: model.Asset(utils.Asset2CodeString(assetQuote)),
+		}
+
+		sdexAssetMap := map[model.Asset]horizon.Asset{
+			tradingPair.Base:  assetBase,
+			tradingPair.Quote: assetQuote,
+		}
 		sdex := plugins.MakeSDEX(
 			client,
 			botConfig.SourceSecretSeed,
@@ -148,12 +159,12 @@ func init() {
 			*operationalBuffer,
 			*operationalBufferNonNativePct,
 			*simMode,
+			tradingPair,
+			sdexAssetMap,
 		)
 
-		assetBase := botConfig.AssetBase()
-		assetQuote := botConfig.AssetQuote()
 		dataKey := model.MakeSortedBotKey(assetBase, assetQuote)
-		strat, e := plugins.MakeStrategy(sdex, &assetBase, &assetQuote, *strategy, *stratConfigPath)
+		strat, e := plugins.MakeStrategy(sdex, &assetBase, &assetQuote, *strategy, *stratConfigPath, *simMode)
 		if e != nil {
 			log.Println()
 			log.Println(e)
@@ -196,6 +207,34 @@ func init() {
 				}
 			}()
 		}
+
+		if botConfig.FillTrackerSleepMillis != 0 {
+			fillTracker := plugins.MakeFillTracker(tradingPair, threadTracker, sdex, botConfig.FillTrackerSleepMillis)
+			fillLogger := plugins.MakeFillLogger()
+			fillTracker.RegisterHandler(fillLogger)
+			strategyFillHandlers, e := strat.GetFillHandlers()
+			if e != nil {
+				log.Println()
+				log.Printf("problem encountered while instantiating the fill tracker: %s\n", e)
+				deleteAllOffersAndExit(botConfig, client, sdex)
+			}
+			if strategyFillHandlers != nil {
+				for _, h := range strategyFillHandlers {
+					fillTracker.RegisterHandler(h)
+				}
+			}
+
+			log.Printf("Starting fill tracker with %d handlers\n", fillTracker.NumHandlers())
+			go func() {
+				e := fillTracker.TrackFills()
+				if e != nil {
+					log.Println()
+					log.Printf("problem encountered while running the fill tracker: %s\n", e)
+					// we want to delete all the offers and exit here because we don't want the bot to run if fill tracking isn't working
+					deleteAllOffersAndExit(botConfig, client, sdex)
+				}
+			}()
+		}
 		// --- end initialization of services ---
 
 		log.Println("Starting the trader bot...")

examples/configs/trader/sample_mirror.cfg

@@ -35,3 +35,11 @@ VOLUME_DIVIDE_BY=500.0
 # spread % we should maintain per level between the mirrored exchange and SDEX (0 < spread < 1.0). This moves the price away from the center price on SDEX so we can cover the position on the external exchange, i.e. if this value is > 0 then the spread you provide on SDEX will be more than the spread on the exchange you are mirroring.
 # in this example the spread is 0.5%
 PER_LEVEL_SPREAD=0.005
+
+# set to true if you want the bot to offset your trades onto the backing exchange to realize the per_level_spread against each trade
+# requires you to specify the EXCHANGE_API_KEYS below
+#OFFSET_TRADES=true
+# you can use multiple API keys to overcome rate limit concerns
+#[[EXCHANGE_API_KEYS]]
+#KEY=""
+#SECRET=""

examples/configs/trader/sample_trader.cfg

@@ -26,6 +26,8 @@ TICK_INTERVAL_SECONDS=300
 # example: use 0 if you want to delete all offers on any error.
 # example: use 2 if you want to tolerate 2 continuous update cycle with errors, i.e. three continuous update cycles with errors will delete all offers.
 DELETE_CYCLES_THRESHOLD=0
+# how many milliseconds to sleep before checking for fills again, a value of 0 disables fill tracking
+FILL_TRACKER_SLEEP_MILLIS=0
 # the url for your horizon instance. If this url contains the string "test" then the bot assumes it is using the test network.
 HORIZON_URL="https://horizon-testnet.stellar.org"
 

glide.yaml

@@ -20,4 +20,4 @@ import:
 - package: github.com/lechengfan/googleauth
   version: 7595ba02fbce171759c10d69d96e4cd898d1fa93
 - package: github.com/nikhilsaraf/go-tools
-  version: bcfd18c97e1cdae3aa432337bb17d3927c3e4fc0
+  version: 19004f22be08c82a22e679726ca22853c65919ae

model/assets.go

@@ -67,7 +67,7 @@ func makeAssetConverter(asset2String map[Asset]string) *AssetConverter {
 func (c AssetConverter) ToString(a Asset) (string, error) {
 	s, ok := c.asset2String[a]
 	if !ok {
-		return "", errors.New("could not recognize Asset: " + string(a))
+		return fmt.Sprintf("missing[%s]", string(a)), nil
 	}
 	return s, nil
 }

model/dates.go

@@ -1,6 +1,9 @@
 package model
 
-import "fmt"
+import (
+	"fmt"
+	"time"
+)
 
 // Timestamp is millis since epoch
 type Timestamp int64
@@ -11,6 +14,11 @@ func MakeTimestamp(ts int64) *Timestamp {
 	return &timestamp
 }
 
+// MakeTimestampFromTime creates a new Timestamp
+func MakeTimestampFromTime(t time.Time) *Timestamp {
+	return MakeTimestamp(t.UnixNano() / int64(time.Millisecond))
+}
+
 func (t *Timestamp) String() string {
 	return fmt.Sprintf("%d", t.AsInt64())
 }

model/orderbook.go

@@ -25,6 +25,14 @@ func (a OrderAction) IsSell() bool {
 	return a == OrderActionSell
 }
 
+// Reverse returns the opposite action
+func (a OrderAction) Reverse() OrderAction {
+	if a.IsSell() {
+		return OrderActionBuy
+	}
+	return OrderActionSell
+}
+
 // String is the stringer function
 func (a OrderAction) String() string {
 	if a == OrderActionBuy {
@@ -96,13 +104,18 @@ type Order struct {
 
 // String is the stringer function
 func (o Order) String() string {
-	return fmt.Sprintf("Order[pair=%s, action=%s, type=%s, price=%s, vol=%s, ts=%d]",
+	tsString := "<nil>"
+	if o.Timestamp != nil {
+		tsString = fmt.Sprintf("%d", o.Timestamp.AsInt64())
+	}
+
+	return fmt.Sprintf("Order[pair=%s, action=%s, type=%s, price=%s, vol=%s, ts=%s]",
 		o.Pair,
 		o.OrderAction,
 		o.OrderType,
 		o.Price.AsString(),
 		o.Volume.AsString(),
-		o.Timestamp.AsInt64(),
+		tsString,
 	)
 }
 
@@ -202,7 +215,7 @@ type Trade struct {
 }
 
 func (t Trade) String() string {
-	return fmt.Sprintf("Trades[txid: %s, ts: %s, pair: %s, action: %s, type: %s, price: %s, volume: %s, cost: %s, fee: %s]",
+	return fmt.Sprintf("Trade[txid: %s, ts: %s, pair: %s, action: %s, type: %s, counterPrice: %s, baseVolume: %s, counterCost: %s, fee: %s]",
 		utils.CheckedString(t.TransactionID),
 		utils.CheckedString(t.Timestamp),
 		*t.Pair,

plugins/balancedLevelProvider.go

@@ -202,3 +202,8 @@ func (p *balancedLevelProvider) getLevel(maxAssetBase float64, maxAssetQuote flo
 	}
 	return level, nil
 }
+
+// GetFillHandlers impl
+func (p *balancedLevelProvider) GetFillHandlers() ([]api.FillHandler, error) {
+	return nil, nil
+}

plugins/ccxtExchange.go

@@ -18,10 +18,11 @@ type ccxtExchange struct {
 	delimiter      string
 	api            *sdk.Ccxt
 	precision      int8
+	simMode        bool
 }
 
 // makeCcxtExchange is a factory method to make an exchange using the CCXT interface
-func makeCcxtExchange(ccxtBaseURL string, exchangeName string) (api.Exchange, error) {
+func makeCcxtExchange(ccxtBaseURL string, exchangeName string, simMode bool) (api.Exchange, error) {
 	c, e := sdk.MakeInitializedCcxtExchange(ccxtBaseURL, exchangeName)
 	if e != nil {
 		return nil, fmt.Errorf("error making a ccxt exchange: %s", e)
@@ -32,6 +33,7 @@ func makeCcxtExchange(ccxtBaseURL string, exchangeName string) (api.Exchange, er
 		delimiter:      "/",
 		api:            c,
 		precision:      utils.SdexPrecision,
+		simMode:        simMode,
 	}, nil
 }
 

plugins/ccxtExchange_test.go

@@ -17,7 +17,7 @@ func TestGetTickerPrice_Ccxt(t *testing.T) {
 
 	for _, exchangeName := range supportedExchanges {
 		t.Run(exchangeName, func(t *testing.T) {
-			testCcxtExchange, e := makeCcxtExchange("http://localhost:3000", exchangeName)
+			testCcxtExchange, e := makeCcxtExchange("http://localhost:3000", exchangeName, false)
 			if !assert.NoError(t, e) {
 				return
 			}
@@ -44,7 +44,7 @@ func TestGetOrderBook_Ccxt(t *testing.T) {
 
 	for _, exchangeName := range supportedExchanges {
 		t.Run(exchangeName, func(t *testing.T) {
-			testCcxtExchange, e := makeCcxtExchange("http://localhost:3000", exchangeName)
+			testCcxtExchange, e := makeCcxtExchange("http://localhost:3000", exchangeName, false)
 			if !assert.NoError(t, e) {
 				return
 			}
@@ -77,7 +77,7 @@ func TestGetTrades_Ccxt(t *testing.T) {
 
 	for _, exchangeName := range supportedExchanges {
 		t.Run(exchangeName, func(t *testing.T) {
-			testCcxtExchange, e := makeCcxtExchange("http://localhost:3000", exchangeName)
+			testCcxtExchange, e := makeCcxtExchange("http://localhost:3000", exchangeName, false)
 			if !assert.NoError(t, e) {
 				return
 			}