add TradeSlice with sync

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/c9s/rockhopper v1.2.1-0.20210115022144-cc77e66fc34f
 	github.com/codingconcepts/env v0.0.0-20200821220118-a8fbf8d84482
 	github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239 // indirect
-	github.com/gin-gonic/gin v1.6.3 // indirect
+	github.com/gin-gonic/gin v1.6.3
 	github.com/go-playground/validator/v10 v10.4.1 // indirect
 	github.com/go-redis/redis/v8 v8.4.0
 	github.com/go-sql-driver/mysql v1.5.0

pkg/bbgo/environment.go

@@ -195,14 +195,7 @@ func (environ *Environment) Init(ctx context.Context) (err error) {
 				return fmt.Errorf("market %s is not defined", symbol)
 			}
 
-			position := &Position{
-				Symbol:        symbol,
-				BaseCurrency:  market.BaseCurrency,
-				QuoteCurrency: market.QuoteCurrency,
-			}
-
 			var trades []types.Trade
-
 			if environ.TradeSync != nil {
 				log.Infof("syncing trades from %s for symbol %s...", session.Exchange.Name(), symbol)
 				if err := environ.TradeSync.SyncTrades(ctx, session.Exchange, symbol, environ.tradeScanTime); err != nil {
@@ -221,16 +214,24 @@ func (environ *Environment) Init(ctx context.Context) (err error) {
 				}
 
 				log.Infof("symbol %s: %d trades loaded", symbol, len(trades))
-
-				position.AddTrades(trades)
 			}
 
-			session.positions[symbol] = position
-			position.BindStream(session.Stream)
-
 			session.Trades[symbol] = trades
+			session.Stream.OnTradeUpdate(func(trade types.Trade) {
+				session.Trades[trade.Symbol] = append(session.Trades[trade.Symbol], trade)
+			})
+
 			session.lastPrices[symbol] = 0.0
 
+			position := &Position{
+				Symbol:        symbol,
+				BaseCurrency:  market.BaseCurrency,
+				QuoteCurrency: market.QuoteCurrency,
+			}
+			position.AddTrades(trades)
+			position.BindStream(session.Stream)
+			session.positions[symbol] = position
+
 			orderStore := NewOrderStore(symbol)
 			orderStore.BindStream(session.Stream)
 			session.orderStores[symbol] = orderStore
@@ -270,9 +271,6 @@ func (environ *Environment) Init(ctx context.Context) (err error) {
 			session.lastPrices[kline.Symbol] = kline.Close
 		})
 
-		session.Stream.OnTradeUpdate(func(trade types.Trade) {
-			session.Trades[trade.Symbol] = append(session.Trades[trade.Symbol], trade)
-		})
 
 		// feed klines into the market data store
 		if environ.startTime == emptyTime {

pkg/bbgo/redis_persistence_test.go

@@ -1,7 +1,6 @@
 package bbgo
 
 import (
-	"os"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -26,7 +25,7 @@ func TestRedisPersistentService(t *testing.T) {
 	var fp fixedpoint.Value
 	err = store.Load(fp)
 	assert.Error(t, err)
-	assert.EqualError(t, os.ErrNotExist, err.Error())
+	assert.EqualError(t, ErrPersistenceNotExists, err.Error())
 
 	fp = fixedpoint.NewFromFloat(3.1415)
 	err = store.Save(&fp)

pkg/types/trade.go

@@ -2,6 +2,7 @@ package types
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/slack-go/slack"
@@ -15,6 +16,26 @@ func init() {
 	_ = PlainText(&Trade{})
 }
 
+type TradeSlice struct {
+	mu    sync.Mutex
+	Items []Trade
+}
+
+func (s *TradeSlice) Slice() []Trade {
+	s.mu.Lock()
+	slice := make([]Trade, len(s.Items), len(s.Items))
+	copy(slice, s.Items)
+	s.mu.Unlock()
+
+	return slice
+}
+
+func (s *TradeSlice) Append(t Trade) {
+	s.mu.Lock()
+	s.Items = append(s.Items, t)
+	s.mu.Unlock()
+}
+
 type Trade struct {
 	// GID is the global ID
 	GID int64 `json:"gid" db:"gid"`