add environment connect integration tests

README.md

@@ -118,30 +118,43 @@ import (
     "github.com/c9s/bbgo"
 )
 
-mysqlURL := viper.GetString("mysql-url")
+func main() {
-mysqlURL = fmt.Sprintf("%s?parseTime=true", mysqlURL)
+    mysqlURL := viper.GetString("mysql-url")
-db, err := sqlx.Connect("mysql", mysqlURL)
+    mysqlURL = fmt.Sprintf("%s?parseTime=true", mysqlURL)
-
+    db, err := sqlx.Connect("mysql", mysqlURL)
-if err != nil {
+    if err != nil {
         return err
-}
+    }
 
-t := bbgo.New(bbgo.Config{
+    environment := environ.New()
+    environment.AddExchange("binance", binance.New(viper.Getenv("binance-api-key"), viper.Getenv("binance-api-secret"))))
+    environment.AddExchange("max", max.New(viper.Getenv("max-key"), viper.Getenv("max-secret"))))
+
+    trader := bbgo.NewTrader(bbgo.Config{
+        Environment: environment,
         DB: db,
-})
+    })
-t.AddNotifier(slacknotifier.New(slackToken))
-t.AddLogHook(slacklog.NewLogHook(slackToken))
 
-t.AddExchange("binance", binance.New(viper.Getenv("binance-api-key"), viper.Getenv("binance-api-secret")))).
+    trader.AddNotifier(slacknotifier.New(slackToken))
-    Subscribe("binance", "btcusdt", "kline@5m", "book", "trade").
+    trader.AddLogHook(slacklog.NewLogHook(slackToken))
-    AddStrategy(bondtrade.New, bondtrade.New).
-    Symbols("btcusdt", "bnbusdt")
 
-t.AddExchange("max", max.New(viper.Getenv("max-key"), viper.Getenv("max-secret")))).
+    // when any trade execution happened
-    Subscribe("max", "btctwd", "kline@5m", "book", "trade").
+    trader.OnTrade(func(session string, exchange types.Exchange, trade types.Trade) {
-    AddStrategy(flashdrop.New, bondtrade.New)
+        notify(trade)
         
-t.AddCrossExchangeStrategy(hedgemaker.New(...))
+        notifyPnL()
+    })
+
+    // mount strategy on an exchange
+    trader.AddExchangeStrategy("binance",
+        bondtrade.New("btcusdt", "5m"),
+        bondtrade.New("ethusdt", "5m"))
+
+    // mount cross exchange strategy
+    trader.AddCrossExchangeStrategy(hedgemaker.New("max", "binance"))
+
+    t.Run(ctx)
+}
 ```
 
 ## Support

go.mod

@@ -3,6 +3,7 @@ module github.com/c9s/bbgo
 go 1.13
 
 require (
+	github.com/DATA-DOG/go-sqlmock v1.5.0
 	github.com/adshao/go-binance v0.0.0-20200604145522-bf563a35f17f
 	github.com/c9s/goose v0.0.0-20200415105707-8da682162a5b
 	github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239 // indirect

go.sum

@@ -14,6 +14,8 @@ dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
+github.com/DATA-DOG/go-sqlmock v1.5.0 h1:Shsta01QNfFxHCfpW6YH2STWB0MudeXXEWMr20OEh60=
+github.com/DATA-DOG/go-sqlmock v1.5.0/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/adshao/go-binance v0.0.0-20200604145522-bf563a35f17f h1:lVxx5HSt/imprfR8v577N3gCQmKmRgkGNz30FlHISO4=
 github.com/adshao/go-binance v0.0.0-20200604145522-bf563a35f17f/go.mod h1:XlIpE7brbCEQxp6VRouG/ZgjLjygQWE1xnc1DtQNp6I=

pkg/bbgo/account.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"sync"
 
-	"github.com/c9s/bbgo/pkg/exchange/binance"
 	"github.com/c9s/bbgo/pkg/types"
 	"github.com/c9s/bbgo/pkg/util"
 

pkg/bbgo/strategy_test.go

@@ -0,0 +1,68 @@
+package bbgo
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	time "time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/jmoiron/sqlx"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/c9s/bbgo/pkg/exchange/binance"
+	"github.com/c9s/bbgo/pkg/service"
+	"github.com/c9s/bbgo/pkg/types"
+
+)
+
+func TestTradeService(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	assert.NoError(t, err)
+	_ = mock
+
+	xdb := sqlx.NewDb(db, "mysql")
+	service.NewTradeService(xdb)
+	/*
+
+		stmt := mock.ExpectQuery(`SELECT \* FROM trades WHERE symbol = \? ORDER BY gid DESC LIMIT 1`)
+		stmt.WithArgs("BTCUSDT")
+		stmt.WillReturnRows(sqlmock.NewRows([]string{"gid", "id", "exchange", "symbol", "price", "quantity"}))
+
+		stmt2 := mock.ExpectQuery(`INSERT INTO trades (id, exchange, symbol, price, quantity, quote_quantity, side, is_buyer, is_maker, fee, fee_currency, traded_at)
+	        	            				VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
+		stmt2.WithArgs()
+	*/
+}
+
+func TestEnvironment_Connect(t *testing.T) {
+	mysqlURL := os.Getenv("MYSQL_URL")
+	if len(mysqlURL) == 0 {
+		t.Skip("require mysql url")
+	}
+	mysqlURL = fmt.Sprintf("%s?parseTime=true", mysqlURL)
+
+	key, secret := os.Getenv("BINANCE_API_KEY"), os.Getenv("BINANCE_API_SECRET")
+	if len(key) == 0 || len(secret) == 0 {
+		t.Skip("require key and secret")
+	}
+
+	exchange := binance.New(key, secret)
+	assert.NotNil(t, exchange)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	xdb, err := sqlx.Connect("mysql", mysqlURL)
+	assert.NoError(t, err)
+
+	environment := NewEnvironment(xdb)
+	environment.AddExchange("binance", exchange).
+		Subscribe(types.KLineChannel,"BTCUSDT", types.SubscribeOptions{ })
+
+	err = environment.Connect(ctx)
+	assert.NoError(t, err)
+
+	time.Sleep(5 * time.Second)
+}

pkg/bbgo/trader.go

@@ -8,13 +8,15 @@ import (
 
 	"github.com/fsnotify/fsnotify"
 	"github.com/jmoiron/sqlx"
+	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 
 	"github.com/c9s/bbgo/pkg/accounting"
 	"github.com/c9s/bbgo/pkg/bbgo/config"
-	"github.com/c9s/bbgo/pkg/exchange/binance"
 	"github.com/c9s/bbgo/pkg/service"
 	"github.com/c9s/bbgo/pkg/types"
+
+	_ "github.com/go-sql-driver/mysql"
 )
 
 // MarketStrategy represents the single Exchange strategy
@@ -24,28 +26,30 @@ type MarketStrategy interface {
 }
 
 type ExchangeSession struct {
+	// Session name
 	Name string
 
+	// The exchange account states
 	Account *Account
 
+	// Stream is the connection stream of the exchange
 	Stream types.Stream
 
 	Subscriptions []types.Subscription
 
-	Exchange *binance.Exchange
+	Exchange types.Exchange
-
-	Strategies []MarketStrategy
 
 	loadedSymbols map[string]struct{}
 
+	// Markets defines market configuration of a symbol
 	Markets map[string]types.Market
 
+	// Trades collects the executed trades from the exchange
+	// map: symbol -> []trade
 	Trades map[string][]types.Trade
 }
 
 func (session *ExchangeSession) Subscribe(channel types.Channel, symbol string, options types.SubscribeOptions) *ExchangeSession {
-	session.Symbols(symbol)
-
 	session.Subscriptions = append(session.Subscriptions, types.Subscription{
 		Channel: channel,
 		Symbol:  symbol,
@@ -55,31 +59,89 @@ func (session *ExchangeSession) Subscribe(channel types.Channel, symbol string,
 	return session
 }
 
-func (session *ExchangeSession) Symbols(symbols ...string) *ExchangeSession {
+// Environment presents the real exchange data layer
-	if session.loadedSymbols == nil {
+type Environment struct {
-		session.loadedSymbols = make(map[string]struct{})
+	TradeService *service.TradeService
-	}
+	TradeSync    *service.TradeSync
-
+
-	if session.Markets == nil {
+	ExchangeSessions map[string]*ExchangeSession
-		session.Markets = make(map[string]types.Market)
+}
-	}
+
-
+func NewEnvironment(db *sqlx.DB) *Environment {
-	for _, symbol := range symbols {
+	tradeService := &service.TradeService{DB: db}
-		session.loadedSymbols[symbol] = struct{}{}
+	return &Environment{
-
+		TradeService: tradeService,
-		if market, ok := types.FindMarket(symbol); ok {
+		TradeSync: &service.TradeSync{
-			session.Markets[symbol] = market
+			Service: tradeService,
-		} else {
+		},
-			log.Panicf("market of symbol %s not found", symbol)
+		ExchangeSessions: make(map[string]*ExchangeSession),
-		}
+	}
+}
+
+func (environ *Environment) AddExchange(name string, exchange types.Exchange) (session *ExchangeSession) {
+	session = &ExchangeSession{
+		Name:     name,
+		Exchange: exchange,
+		Markets:  make(map[string]types.Market),
+		Trades:   make(map[string][]types.Trade),
 	}
 
+	environ.ExchangeSessions[name] = session
 	return session
 }
 
-func (session *ExchangeSession) AddStrategy(strategy MarketStrategy) *ExchangeSession {
+func (environ *Environment) Connect(ctx context.Context) (err error) {
-	session.Strategies = append(session.Strategies, strategy)
+	startTime := time.Now().AddDate(0, 0, -7) // sync from 7 days ago
-	return session
+
+	for _, session := range environ.ExchangeSessions {
+		loadedSymbols := make(map[string]struct{})
+		for _, sub := range session.Subscriptions {
+			loadedSymbols[sub.Symbol] = struct{}{}
+		}
+
+		for symbol := range loadedSymbols {
+			if err := environ.TradeSync.Sync(ctx, session.Exchange, symbol, startTime); err != nil {
+				return err
+			}
+
+			var trades []types.Trade
+
+			tradingFeeCurrency := session.Exchange.PlatformFeeCurrency()
+			if strings.HasPrefix(symbol, tradingFeeCurrency) {
+				trades, err = environ.TradeService.QueryForTradingFeeCurrency(symbol, tradingFeeCurrency)
+			} else {
+				trades, err = environ.TradeService.Query(symbol)
+			}
+
+			if err != nil {
+				return err
+			}
+
+			log.Infof("symbol %s: %d trades loaded", symbol, len(trades))
+			session.Trades[symbol] = trades
+		}
+
+		session.Account, err = LoadAccount(ctx, session.Exchange)
+		if err != nil {
+			return err
+		}
+
+		session.Stream = session.Exchange.NewStream()
+		session.Stream.OnTrade(func(trade *types.Trade) {
+			// append trades
+			session.Trades[trade.Symbol] = append(session.Trades[trade.Symbol], *trade)
+
+			if err := environ.TradeService.Insert(*trade); err != nil {
+				log.WithError(err).Errorf("trade insert error: %+v", *trade)
+			}
+		})
+
+		if err := session.Stream.Connect(ctx); err != nil {
+			return err
+		}
+	}
+
+	return err
 }
 
 type Trader struct {
@@ -103,7 +165,7 @@ type Trader struct {
 	ExchangeSessions map[string]*ExchangeSession
 }
 
-func New(db *sqlx.DB, exchange types.Exchange, symbol string) *Trader {
+func NewTrader(db *sqlx.DB, exchange types.Exchange, symbol string) *Trader {
 	tradeService := &service.TradeService{DB: db}
 	return &Trader{
 		Symbol:       symbol,
@@ -119,27 +181,20 @@ func (trader *Trader) AddNotifier(notifier Notifier) {
 	trader.Notifiers = append(trader.Notifiers, notifier)
 }
 
-func (trader *Trader) AddExchange(name string, exchange *binance.Exchange) (session *ExchangeSession) {
-	session = &ExchangeSession{
-		Name:     name,
-		Exchange: exchange,
-	}
-
-	if trader.ExchangeSessions == nil {
-		trader.ExchangeSessions = make(map[string]*ExchangeSession)
-	}
-
-	trader.ExchangeSessions[name] = session
-	return session
-}
-
 func (trader *Trader) Connect(ctx context.Context) (err error) {
 	log.Info("syncing trades from exchange...")
 	startTime := time.Now().AddDate(0, 0, -7) // sync from 7 days ago
 
 	for _, session := range trader.ExchangeSessions {
-
 		for symbol := range session.loadedSymbols {
+			market, ok := types.FindMarket(symbol)
+			if !ok {
+				return errors.Errorf("market %s is not defined", symbol)
+			}
+
+			session.Markets[symbol] = market
+
+
 			if err := trader.TradeSync.Sync(ctx, session.Exchange, symbol, startTime); err != nil {
 				return err
 			}
@@ -158,9 +213,6 @@ func (trader *Trader) Connect(ctx context.Context) (err error) {
 			}
 
 			log.Infof("symbol %s: %d trades loaded", symbol, len(trades))
-			if session.Trades == nil {
-				session.Trades = make(map[string][]types.Trade)
-			}
 			session.Trades[symbol] = trades
 
 			stockManager := &StockDistribution{