Merge pull request #1160 from c9s/fix/profit-fixer

FIX: [grid2] add profit fixer and options

pkg/bbgo/persistence.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"os"
 	"reflect"
+	"sync"
 
 	"github.com/codingconcepts/env"
 	"github.com/pkg/errors"
@@ -28,6 +29,13 @@ func Sync(ctx context.Context, obj interface{}) {
 	isolation := GetIsolationFromContext(ctx)
 
 	ps := isolation.persistenceServiceFacade.Get()
+
+	locker, ok := obj.(sync.Locker)
+	if ok {
+		locker.Lock()
+		defer locker.Unlock()
+	}
+
 	err := storePersistenceFields(obj, id, ps)
 	if err != nil {
 		log.WithError(err).Errorf("persistence sync failed")

pkg/strategy/grid2/profit_fixer.go

@@ -0,0 +1,78 @@
+package grid2
+
+import (
+	"context"
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/c9s/bbgo/pkg/exchange/batch"
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type ProfitFixer struct {
+	symbol         string
+	grid           *Grid
+	historyService types.ExchangeTradeHistoryService
+}
+
+func newProfitFixer(grid *Grid, symbol string, historyService types.ExchangeTradeHistoryService) *ProfitFixer {
+	return &ProfitFixer{
+		symbol:         symbol,
+		grid:           grid,
+		historyService: historyService,
+	}
+}
+
+// Fix fixes the total quote profit of the given grid
+func (f *ProfitFixer) Fix(ctx context.Context, since, until time.Time, initialOrderID uint64, profitStats *GridProfitStats) error {
+	profitStats.TotalQuoteProfit = fixedpoint.Zero
+	profitStats.ArbitrageCount = 0
+
+	q := &batch.ClosedOrderBatchQuery{ExchangeTradeHistoryService: f.historyService}
+	orderC, errC := q.Query(ctx, f.symbol, since, until, initialOrderID)
+
+	for {
+		select {
+		case <-ctx.Done():
+			if errors.Is(ctx.Err(), context.Canceled) {
+				return nil
+			}
+
+			return ctx.Err()
+
+		case order, ok := <-orderC:
+			if !ok {
+				return <-errC
+			}
+
+			if !f.grid.HasPrice(order.Price) {
+				continue
+			}
+
+			if profitStats.InitialOrderID == 0 {
+				profitStats.InitialOrderID = order.OrderID
+			}
+
+			if order.Status != types.OrderStatusFilled {
+				continue
+			}
+
+			if order.Type != types.OrderTypeLimit {
+				continue
+			}
+
+			if order.Side != types.SideTypeSell {
+				continue
+			}
+
+			quoteProfit := order.Quantity.Mul(f.grid.Spread)
+			profitStats.TotalQuoteProfit = profitStats.TotalQuoteProfit.Add(quoteProfit)
+			profitStats.ArbitrageCount++
+
+			log.Debugf("profitFixer: filledSellOrder=%#v", order)
+			log.Debugf("profitFixer: profitStats=%#v", profitStats)
+		}
+	}
+}

pkg/strategy/grid2/profit_fixer_test.go

@@ -0,0 +1,102 @@
+package grid2
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+	"github.com/c9s/bbgo/pkg/types/mocks"
+)
+
+func mustNewTime(v string) time.Time {
+	t, err := time.Parse(time.RFC3339, v)
+	if err != nil {
+		panic(err)
+	}
+
+	return t
+}
+
+var testClosedOrderID = uint64(0)
+
+func newClosedLimitOrder(symbol string, side types.SideType, price, quantity fixedpoint.Value, ta ...time.Time) types.Order {
+	testClosedOrderID++
+	creationTime := time.Now()
+	updateTime := creationTime
+
+	if len(ta) > 0 {
+		creationTime = ta[0]
+		if len(ta) > 1 {
+			updateTime = ta[1]
+		}
+	}
+
+	return types.Order{
+		SubmitOrder: types.SubmitOrder{
+			Symbol:   symbol,
+			Side:     side,
+			Type:     types.OrderTypeLimit,
+			Quantity: quantity,
+			Price:    price,
+		},
+		Exchange:         types.ExchangeBinance,
+		OrderID:          testClosedOrderID,
+		Status:           types.OrderStatusFilled,
+		ExecutedQuantity: quantity,
+		CreationTime:     types.Time(creationTime),
+		UpdateTime:       types.Time(updateTime),
+	}
+}
+
+func TestProfitFixer(t *testing.T) {
+	testClosedOrderID = 0
+
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	ctx := context.Background()
+	mockHistoryService := mocks.NewMockExchangeTradeHistoryService(mockCtrl)
+
+	mockHistoryService.EXPECT().QueryClosedOrders(ctx, "ETHUSDT", mustNewTime("2022-01-01T00:00:00Z"), mustNewTime("2022-01-07T00:00:00Z"), uint64(0)).
+		Return([]types.Order{
+			newClosedLimitOrder("ETHUSDT", types.SideTypeBuy, number(1800.0), number(0.1), mustNewTime("2022-01-01T00:01:00Z")),
+			newClosedLimitOrder("ETHUSDT", types.SideTypeBuy, number(1700.0), number(0.1), mustNewTime("2022-01-01T00:01:00Z")),
+			newClosedLimitOrder("ETHUSDT", types.SideTypeSell, number(1800.0), number(0.1), mustNewTime("2022-01-01T00:01:00Z")),
+			newClosedLimitOrder("ETHUSDT", types.SideTypeSell, number(1900.0), number(0.1), mustNewTime("2022-01-01T00:03:00Z")),
+			newClosedLimitOrder("ETHUSDT", types.SideTypeSell, number(1905.0), number(0.1), mustNewTime("2022-01-01T00:03:00Z")),
+		}, nil)
+
+	mockHistoryService.EXPECT().QueryClosedOrders(ctx, "ETHUSDT", mustNewTime("2022-01-01T00:03:00Z"), mustNewTime("2022-01-07T00:00:00Z"), uint64(5)).
+		Return([]types.Order{
+			newClosedLimitOrder("ETHUSDT", types.SideTypeBuy, number(1900.0), number(0.1), mustNewTime("2022-01-01T00:04:00Z")),
+			newClosedLimitOrder("ETHUSDT", types.SideTypeBuy, number(1800.0), number(0.1), mustNewTime("2022-01-01T00:04:00Z")),
+			newClosedLimitOrder("ETHUSDT", types.SideTypeBuy, number(1700.0), number(0.1), mustNewTime("2022-01-01T00:04:00Z")),
+			newClosedLimitOrder("ETHUSDT", types.SideTypeSell, number(1800.0), number(0.1), mustNewTime("2022-01-01T00:04:00Z")),
+			newClosedLimitOrder("ETHUSDT", types.SideTypeSell, number(1900.0), number(0.1), mustNewTime("2022-01-01T00:08:00Z")),
+		}, nil)
+
+	mockHistoryService.EXPECT().QueryClosedOrders(ctx, "ETHUSDT", mustNewTime("2022-01-01T00:08:00Z"), mustNewTime("2022-01-07T00:00:00Z"), uint64(10)).
+		Return([]types.Order{}, nil)
+
+	grid := NewGrid(number(1000.0), number(2000.0), number(11), number(0.01))
+	grid.CalculateArithmeticPins()
+
+	since, err := time.Parse(time.RFC3339, "2022-01-01T00:00:00Z")
+	assert.NoError(t, err)
+
+	until, err := time.Parse(time.RFC3339, "2022-01-07T00:00:00Z")
+	assert.NoError(t, err)
+
+	stats := &GridProfitStats{}
+	fixer := newProfitFixer(grid, "ETHUSDT", mockHistoryService)
+	err = fixer.Fix(ctx, since, until, 0, stats)
+	assert.NoError(t, err)
+
+	assert.Equal(t, "40", stats.TotalQuoteProfit.String())
+	assert.Equal(t, 4, stats.ArbitrageCount)
+}

pkg/strategy/grid2/recover.go

@@ -63,12 +63,27 @@ func (s *Strategy) recoverByScanningTrades(ctx context.Context, session *bbgo.Ex
 		return errors.Wrap(err, "grid recover error")
 	}
 
+	// emit ready after recover
+	s.EmitGridReady()
+
 	// debug and send metrics
 	// wait for the reverse order to be placed
 	time.Sleep(2 * time.Second)
 	debugGrid(s.logger, s.grid, s.orderExecutor.ActiveMakerOrders())
-	// emit ready after recover
-	s.EmitGridReady()
+
+	defer bbgo.Sync(ctx, s)
+
+	if s.EnableProfitFixer {
+		until := time.Now()
+		since := until.Add(-7 * 24 * time.Hour)
+		if s.FixProfitSince != nil {
+			since = s.FixProfitSince.Time()
+		}
+
+		fixer := newProfitFixer(s.grid, s.Symbol, historyService)
+		// set initial order ID = 0 instead of s.GridProfitStats.InitialOrderID because the order ID could be incorrect
+		return fixer.Fix(ctx, since, until, 0, s.GridProfitStats)
+	}
 
 	return nil
 }

pkg/strategy/grid2/strategy.go

@@ -165,6 +165,9 @@ type Strategy struct {
 	SkipSpreadCheck             bool `json:"skipSpreadCheck"`
 	RecoverGridByScanningTrades bool `json:"recoverGridByScanningTrades"`
 
+	EnableProfitFixer bool        `json:"enableProfitFixer"`
+	FixProfitSince    *types.Time `json:"fixProfitSince"`
+
 	// Debug enables the debug mode
 	Debug bool `json:"debug"`
 
@@ -196,6 +199,9 @@ type Strategy struct {
 
 	tradingCtx, writeCtx context.Context
 	cancelWrite          context.CancelFunc
+
+	// this ensures that bbgo.Sync to lock the object
+	sync.Mutex
 }
 
 func (s *Strategy) ID() string {

pkg/types/exchange.go

@@ -108,6 +108,7 @@ type ExchangeAmountFeeProtect interface {
 	SetModifyOrderAmountForFee(ExchangeFee)
 }
 
+//go:generate mockgen -destination=mocks/mock_exchange_trade_history.go -package=mocks . ExchangeTradeHistoryService
 type ExchangeTradeHistoryService interface {
 	QueryTrades(ctx context.Context, symbol string, options *TradeQueryOptions) ([]Trade, error)
 	QueryClosedOrders(ctx context.Context, symbol string, since, until time.Time, lastOrderID uint64) (orders []Order, err error)

pkg/types/mocks/mock_exchange_trade_history.go

@@ -0,0 +1,67 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: github.com/c9s/bbgo/pkg/types (interfaces: ExchangeTradeHistoryService)
+
+// Package mocks is a generated GoMock package.
+package mocks
+
+import (
+	context "context"
+	reflect "reflect"
+	time "time"
+
+	types "github.com/c9s/bbgo/pkg/types"
+	gomock "github.com/golang/mock/gomock"
+)
+
+// MockExchangeTradeHistoryService is a mock of ExchangeTradeHistoryService interface.
+type MockExchangeTradeHistoryService struct {
+	ctrl     *gomock.Controller
+	recorder *MockExchangeTradeHistoryServiceMockRecorder
+}
+
+// MockExchangeTradeHistoryServiceMockRecorder is the mock recorder for MockExchangeTradeHistoryService.
+type MockExchangeTradeHistoryServiceMockRecorder struct {
+	mock *MockExchangeTradeHistoryService
+}
+
+// NewMockExchangeTradeHistoryService creates a new mock instance.
+func NewMockExchangeTradeHistoryService(ctrl *gomock.Controller) *MockExchangeTradeHistoryService {
+	mock := &MockExchangeTradeHistoryService{ctrl: ctrl}
+	mock.recorder = &MockExchangeTradeHistoryServiceMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockExchangeTradeHistoryService) EXPECT() *MockExchangeTradeHistoryServiceMockRecorder {
+	return m.recorder
+}
+
+// QueryClosedOrders mocks base method.
+func (m *MockExchangeTradeHistoryService) QueryClosedOrders(arg0 context.Context, arg1 string, arg2, arg3 time.Time, arg4 uint64) ([]types.Order, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "QueryClosedOrders", arg0, arg1, arg2, arg3, arg4)
+	ret0, _ := ret[0].([]types.Order)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// QueryClosedOrders indicates an expected call of QueryClosedOrders.
+func (mr *MockExchangeTradeHistoryServiceMockRecorder) QueryClosedOrders(arg0, arg1, arg2, arg3, arg4 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "QueryClosedOrders", reflect.TypeOf((*MockExchangeTradeHistoryService)(nil).QueryClosedOrders), arg0, arg1, arg2, arg3, arg4)
+}
+
+// QueryTrades mocks base method.
+func (m *MockExchangeTradeHistoryService) QueryTrades(arg0 context.Context, arg1 string, arg2 *types.TradeQueryOptions) ([]types.Trade, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "QueryTrades", arg0, arg1, arg2)
+	ret0, _ := ret[0].([]types.Trade)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// QueryTrades indicates an expected call of QueryTrades.
+func (mr *MockExchangeTradeHistoryServiceMockRecorder) QueryTrades(arg0, arg1, arg2 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "QueryTrades", reflect.TypeOf((*MockExchangeTradeHistoryService)(nil).QueryTrades), arg0, arg1, arg2)
+}