Merge pull request #1396 from c9s/chiahung/grid2/persistence-ttl

FEATURE: add ttl for position/grid2.profit_stats persistence

pkg/strategy/grid2/active_order_recover.go

@@ -66,22 +66,25 @@ func (s *Strategy) recoverActiveOrdersPeriodically(ctx context.Context) {
 		exchange:          s.session.Exchange,
 	}
 
+	var lastRecoverTime time.Time
+
 	for {
 		select {
-
 		case <-ctx.Done():
 			return
 
 		case <-ticker.C:
-			if err := syncActiveOrders(ctx, opts); err != nil {
-				log.WithError(err).Errorf("unable to sync active orders")
-			}
-
+			s.recoverC <- struct{}{}
 		case <-s.recoverC:
-			if err := syncActiveOrders(ctx, opts); err != nil {
-				log.WithError(err).Errorf("unable to sync active orders")
+			if !time.Now().After(lastRecoverTime.Add(10 * time.Minute)) {
+				continue
 			}
 
+			if err := syncActiveOrders(ctx, opts); err != nil {
+				log.WithError(err).Errorf("unable to sync active orders")
+			} else {
+				lastRecoverTime = time.Now()
+			}
 		}
 	}
 }

pkg/strategy/grid2/profit_stats.go

@@ -24,6 +24,9 @@ type GridProfitStats struct {
 	Market           types.Market                `json:"market,omitempty"`
 	Since            *time.Time                  `json:"since,omitempty"`
 	InitialOrderID   uint64                      `json:"initialOrderID"`
+
+	// ttl is the ttl to keep in persistence
+	ttl time.Duration
 }
 
 func newGridProfitStats(market types.Market) *GridProfitStats {
@@ -40,6 +43,17 @@ func newGridProfitStats(market types.Market) *GridProfitStats {
 	}
 }
 
+func (s *GridProfitStats) SetTTL(ttl time.Duration) {
+	if ttl.Nanoseconds() <= 0 {
+		return
+	}
+	s.ttl = ttl
+}
+
+func (s *GridProfitStats) Expiration() time.Duration {
+	return s.ttl
+}
+
 func (s *GridProfitStats) AddTrade(trade types.Trade) {
 	if s.TotalFee == nil {
 		s.TotalFee = make(map[string]fixedpoint.Value)

pkg/strategy/grid2/strategy.go

@@ -177,6 +177,7 @@ type Strategy struct {
 
 	GridProfitStats *GridProfitStats `persistence:"grid_profit_stats"`
 	Position        *types.Position  `persistence:"position"`
+	PersistenceTTL  types.Duration   `json:"persistenceTTL"`
 
 	// ExchangeSession is an injection field
 	ExchangeSession *bbgo.ExchangeSession
@@ -1835,13 +1836,17 @@ func (s *Strategy) Run(ctx context.Context, _ bbgo.OrderExecutor, session *bbgo.
 		s.ProfitSpread = s.Market.TruncatePrice(s.ProfitSpread)
 	}
 
+	s.logger.Infof("persistence ttl: %s", s.PersistenceTTL.Duration())
+
 	if s.GridProfitStats == nil {
 		s.GridProfitStats = newGridProfitStats(s.Market)
 	}
+	s.GridProfitStats.SetTTL(s.PersistenceTTL.Duration())
 
 	if s.Position == nil {
 		s.Position = types.NewPositionFromMarket(s.Market)
 	}
+	s.Position.SetTTL(s.PersistenceTTL.Duration())
 
 	// initialize and register prometheus metrics
 	if s.PrometheusLabels != nil {

pkg/types/position.go

@@ -65,6 +65,20 @@ type Position struct {
 
 	// Modify position callbacks
 	modifyCallbacks []func(baseQty fixedpoint.Value, quoteQty fixedpoint.Value, price fixedpoint.Value)
+
+	// ttl is the ttl to keep in persistence
+	ttl time.Duration
+}
+
+func (s *Position) SetTTL(ttl time.Duration) {
+	if ttl.Nanoseconds() <= 0 {
+		return
+	}
+	s.ttl = ttl
+}
+
+func (s *Position) Expiration() time.Duration {
+	return s.ttl
 }
 
 func (p *Position) CsvHeader() []string {