fix: rate settings in telegram, make elliottwave draw async

config/elliottwave.yaml

@@ -24,17 +24,17 @@ exchangeStrategies:
 - on: binance
   elliottwave:
     minInterval: 1s
-    symbol: BNBBUSD
+    symbol: BTCUSDT
     limitOrder: true
-    quantity: 0.16
+    #quantity: 0.16
     # kline interval for indicators
-    interval: 1m
+    interval: 1s
     stoploss: 0.01%
     windowATR: 14
-    windowQuick: 5
+    windowQuick: 4
-    windowSlow: 9
+    windowSlow: 155
     source: hl2
-    pendingMinutes: 10
+    pendingMinInterval: 5
     useHeikinAshi: true
 
     drawGraph: true
@@ -47,12 +47,12 @@ exchangeStrategies:
     # when farest price from entry goes over that ratio, start using the callback ratio accordingly to do trailingstop
     #trailingActivationRatio: [0.01, 0.016, 0.05]
     #trailingActivationRatio: [0.001, 0.0081, 0.022]
-    trailingActivationRatio: [0.0017, 0.01, 0.015]
+    #trailingActivationRatio: [0.0017, 0.01, 0.015]
-    #trailingActivationRatio: []
+    trailingActivationRatio: []
-    #trailingCallbackRate: []
+    trailingCallbackRate: []
     #trailingCallbackRate: [0.002, 0.01, 0.1]
     #trailingCallbackRate: [0.0004, 0.0009, 0.018]
-    trailingCallbackRate: [0.0006, 0.0019, 0.006]
+    #trailingCallbackRate: [0.0006, 0.0019, 0.006]
 
     #exits:
     #   - roiStopLoss:
@@ -108,13 +108,13 @@ sync:
   sessions:
     - binance
   symbols:
-    - BNBBUSD
+    - BTCUSDT
 
 backtest:
-  startTime: "2022-09-01"
+  startTime: "2022-10-15"
-  endTime: "2022-09-30"
+  endTime: "2022-10-19"
   symbols:
-    - BNBBUSD
+    - BTCUSDT
   sessions: [binance]
   syncSecKLines: true
   accounts:
@@ -122,5 +122,5 @@ backtest:
       makerFeeRate: 0.000
       takerFeeRate: 0.000
       balances:
-        BNB: 0
+        BTC: 0
-        BUSD: 100
+        USDT: 100

pkg/bbgo/serialmarketdatastore.go

@@ -46,7 +46,7 @@ func (store *SerialMarketDataStore) Subscribe(interval types.Interval) {
 func (store *SerialMarketDataStore) BindStream(ctx context.Context, stream types.Stream) {
 	if store.UseAggTrade {
 		if IsBackTesting {
-			log.Errorf("Right now in backtesting, aggTrade event is not yet supported. Use OnKLineClosed instead.")
+			log.Errorf("right now in backtesting, aggTrade event is not yet supported. Use OnKLineClosed instead.")
 			stream.OnKLineClosed(store.handleKLineClosed)
 			return
 		}
@@ -85,6 +85,10 @@ func (store *SerialMarketDataStore) handleMarketTrade(trade types.Trade) {
 
 func (store *SerialMarketDataStore) tickerProcessor(ctx context.Context) {
 	duration := store.MinInterval.Duration()
+	relativeTime := time.Now().UnixNano() % int64(duration)
+	waitTime := int64(duration) - relativeTime
+	ch := time.After(time.Duration(waitTime))
+	<-ch
 	intervalCloseTicker := time.NewTicker(duration)
 	defer intervalCloseTicker.Stop()
 

pkg/bbgo/trader.go

@@ -166,52 +166,6 @@ func (trader *Trader) SetRiskControls(riskControls *RiskControls) {
 	trader.riskControls = riskControls
 }
 
-func (trader *Trader) Subscribe() {
-	// pre-subscribe the data
-	for sessionName, strategies := range trader.exchangeStrategies {
-		session := trader.environment.sessions[sessionName]
-		for _, strategy := range strategies {
-			if defaulter, ok := strategy.(StrategyDefaulter); ok {
-				if err := defaulter.Defaults(); err != nil {
-					panic(err)
-				}
-			}
-
-			if initializer, ok := strategy.(StrategyInitializer); ok {
-				if err := initializer.Initialize(); err != nil {
-					panic(err)
-				}
-			}
-
-			if subscriber, ok := strategy.(ExchangeSessionSubscriber); ok {
-				subscriber.Subscribe(session)
-			} else {
-				log.Errorf("strategy %s does not implement ExchangeSessionSubscriber", strategy.ID())
-			}
-		}
-	}
-
-	for _, strategy := range trader.crossExchangeStrategies {
-		if defaulter, ok := strategy.(StrategyDefaulter); ok {
-			if err := defaulter.Defaults(); err != nil {
-				panic(err)
-			}
-		}
-
-		if initializer, ok := strategy.(StrategyInitializer); ok {
-			if err := initializer.Initialize(); err != nil {
-				panic(err)
-			}
-		}
-
-		if subscriber, ok := strategy.(CrossExchangeSessionSubscriber); ok {
-			subscriber.CrossSubscribe(trader.environment.sessions)
-		} else {
-			log.Errorf("strategy %s does not implement CrossExchangeSessionSubscriber", strategy.ID())
-		}
-	}
-}
-
 func (trader *Trader) RunSingleExchangeStrategy(ctx context.Context, strategy SingleExchangeStrategy, session *ExchangeSession, orderExecutor OrderExecutor) error {
 	if v, ok := strategy.(StrategyValidator); ok {
 		if err := v.Validate(); err != nil {
@@ -262,7 +216,7 @@ func (trader *Trader) RunAllSingleExchangeStrategy(ctx context.Context) error {
 	return nil
 }
 
-func (trader *Trader) injectFields(ctx context.Context) error {
+func (trader *Trader) injectFieldsAndSubscribe(ctx context.Context) error {
 	// load and run Session strategies
 	for sessionName, strategies := range trader.exchangeStrategies {
 		var session = trader.environment.sessions[sessionName]
@@ -350,6 +304,24 @@ func (trader *Trader) injectFields(ctx context.Context) error {
 		if err := trader.injectCommonServices(strategy); err != nil {
 			return err
 		}
+
+		if defaulter, ok := strategy.(StrategyDefaulter); ok {
+			if err := defaulter.Defaults(); err != nil {
+				return err
+			}
+		}
+
+		if initializer, ok := strategy.(StrategyInitializer); ok {
+			if err := initializer.Initialize(); err != nil {
+				return err
+			}
+		}
+
+		if subscriber, ok := strategy.(CrossExchangeSessionSubscriber); ok {
+			subscriber.CrossSubscribe(trader.environment.sessions)
+		} else {
+			log.Errorf("strategy %s does not implement CrossExchangeSessionSubscriber", strategy.ID())
+		}
 	}
 
 	return nil
@@ -361,7 +333,7 @@ func (trader *Trader) Run(ctx context.Context) error {
 	// trader.environment.Connect will call interact.Start
 	interact.AddCustomInteraction(NewCoreInteraction(trader.environment, trader))
 
-	if err := trader.injectFields(ctx); err != nil {
+	if err := trader.injectFieldsAndSubscribe(ctx); err != nil {
 		return err
 	}
 

pkg/notifier/telegramnotifier/telegram.go

@@ -15,7 +15,7 @@ import (
 	"github.com/c9s/bbgo/pkg/types"
 )
 
-var apiLimiter = rate.NewLimiter(rate.Every(1*time.Second), 1)
+var apiLimiter = rate.NewLimiter(rate.Every(50*time.Millisecond), 20)
 
 var log = logrus.WithField("service", "telegram")
 

pkg/strategy/drift/strategy.go

@@ -98,7 +98,7 @@ type Strategy struct {
 	SmootherWindow            int              `json:"smootherWindow"`
 	FisherTransformWindow     int              `json:"fisherTransformWindow"`
 	ATRWindow                 int              `json:"atrWindow"`
-	PendingMinutes            int              `json:"pendingMinutes" modifiable:"true"`  // if order not be traded for pendingMinutes of time, cancel it.
+	PendingMinInterval        int              `json:"pendingMinInterval" modifiable:"true"` // if order not be traded for pendingMinInterval of time, cancel it.
 	NoRebalance               bool             `json:"noRebalance" modifiable:"true"`        // disable rebalance
 	TrendWindow               int              `json:"trendWindow"`                          // trendLine is used for rebalancing the position. When trendLine goes up, hold base, otherwise hold quote
 	RebalanceFilter           float64          `json:"rebalanceFilter" modifiable:"true"`    // beta filter on the Linear Regression of trendLine
@@ -276,7 +276,7 @@ func (s *Strategy) smartCancel(ctx context.Context, pricef, atr float64) (int, e
 				continue
 			}
 			log.Warnf("%v | counter: %d, system: %d", order, s.orderPendingCounter[order.OrderID], s.counter)
-			if s.counter-s.orderPendingCounter[order.OrderID] > s.PendingMinutes {
+			if s.counter-s.orderPendingCounter[order.OrderID] > s.PendingMinInterval {
 				toCancel = true
 			} else if order.Side == types.SideTypeBuy {
 				// 75% of the probability

pkg/strategy/elliottwave/draw.go

@@ -13,6 +13,7 @@ import (
 
 func (s *Strategy) InitDrawCommands(store *bbgo.SerialMarketDataStore, profit, cumProfit types.Series) {
 	bbgo.RegisterCommand("/draw", "Draw Indicators", func(reply interact.Reply) {
+		go func() {
 			canvas := s.DrawIndicators(store)
 			if canvas == nil {
 				reply.Message("cannot render indicators")
@@ -25,35 +26,36 @@ func (s *Strategy) InitDrawCommands(store *bbgo.SerialMarketDataStore, profit, c
 				return
 			}
 			bbgo.SendPhoto(&buffer)
+		}()
 	})
 	bbgo.RegisterCommand("/pnl", "Draw PNL(%) per trade", func(reply interact.Reply) {
+		go func() {
 			canvas := s.DrawPNL(profit)
 			var buffer bytes.Buffer
 			if err := canvas.Render(chart.PNG, &buffer); err != nil {
-			log.WithError(err).Errorf("cannot render pnl in drift")
+				log.WithError(err).Errorf("cannot render pnl in ewo")
 				reply.Message(fmt.Sprintf("[error] cannot render pnl in ewo: %v", err))
 				return
 			}
 			bbgo.SendPhoto(&buffer)
+		}()
 	})
 	bbgo.RegisterCommand("/cumpnl", "Draw Cummulative PNL(Quote)", func(reply interact.Reply) {
+		go func() {
 			canvas := s.DrawCumPNL(cumProfit)
 			var buffer bytes.Buffer
 			if err := canvas.Render(chart.PNG, &buffer); err != nil {
-			log.WithError(err).Errorf("cannot render cumpnl in drift")
+				log.WithError(err).Errorf("cannot render cumpnl in ewo")
-			reply.Message(fmt.Sprintf("[error] canot render cumpnl in drift: %v", err))
+				reply.Message(fmt.Sprintf("[error] canot render cumpnl in ewo: %v", err))
 				return
 			}
 			bbgo.SendPhoto(&buffer)
+		}()
 	})
 }
 
 func (s *Strategy) DrawIndicators(store *bbgo.SerialMarketDataStore) *types.Canvas {
-	klines, ok := store.KLinesOfInterval(types.Interval1m)
+	time := types.Time(s.startTime)
-	if !ok {
-		return nil
-	}
-	time := (*klines)[len(*klines)-1].StartTime
 	canvas := types.NewCanvas(s.InstanceID(), s.Interval)
 	Length := s.priceLines.Length()
 	if Length > 300 {
@@ -109,10 +111,10 @@ func (s *Strategy) Draw(store *bbgo.SerialMarketDataStore, profit, cumProfit typ
 		log.WithError(err).Errorf("cannot create on path " + s.GraphIndicatorPath)
 		return
 	}
-	defer f.Close()
 	if err = canvas.Render(chart.PNG, f); err != nil {
 		log.WithError(err).Errorf("cannot render elliottwave")
 	}
+	f.Close()
 
 	canvas = s.DrawPNL(profit)
 	f, err = os.Create(s.GraphPNLPath)
@@ -120,19 +122,19 @@ func (s *Strategy) Draw(store *bbgo.SerialMarketDataStore, profit, cumProfit typ
 		log.WithError(err).Errorf("cannot create on path " + s.GraphPNLPath)
 		return
 	}
-	defer f.Close()
 	if err = canvas.Render(chart.PNG, f); err != nil {
 		log.WithError(err).Errorf("cannot render pnl")
 		return
 	}
+	f.Close()
 	canvas = s.DrawCumPNL(cumProfit)
 	f, err = os.Create(s.GraphCumPNLPath)
 	if err != nil {
 		log.WithError(err).Errorf("cannot create on path " + s.GraphCumPNLPath)
 		return
 	}
-	defer f.Close()
 	if err = canvas.Render(chart.PNG, f); err != nil {
 		log.WithError(err).Errorf("cannot render cumpnl")
 	}
+	f.Close()
 }

pkg/strategy/elliottwave/strategy.go

@@ -49,7 +49,7 @@ type Strategy struct {
 	WindowATR          int              `json:"windowATR"`
 	WindowQuick        int              `json:"windowQuick"`
 	WindowSlow         int              `json:"windowSlow"`
-	PendingMinutes int              `json:"pendingMinutes" modifiable:"true"`
+	PendingMinInterval int              `json:"pendingMinInterval" modifiable:"true"`
 	UseHeikinAshi      bool             `json:"useHeikinAshi"`
 
 	// whether to draw graph or not by the end of backtest
@@ -196,7 +196,7 @@ func (s *Strategy) smartCancel(ctx context.Context, pricef float64) int {
 			}
 			log.Warnf("%v | counter: %d, system: %d", order, s.orderPendingCounter[order.OrderID], s.counter)
 			toCancel := false
-			if s.counter-s.orderPendingCounter[order.OrderID] >= s.PendingMinutes {
+			if s.counter-s.orderPendingCounter[order.OrderID] >= s.PendingMinInterval {
 				toCancel = true
 			} else if order.Side == types.SideTypeBuy {
 				if order.Price.Float64()+s.atr.Last()*2 <= pricef {
@@ -211,7 +211,7 @@ func (s *Strategy) smartCancel(ctx context.Context, pricef float64) int {
 				panic("not supported side for the order")
 			}
 			if toCancel {
-				err := s.GeneralOrderExecutor.GracefulCancel(ctx, order)
+				err := s.GeneralOrderExecutor.CancelNoWait(ctx, order)
 				if err == nil {
 					delete(s.orderPendingCounter, order.OrderID)
 				} else {
@@ -235,6 +235,9 @@ func (s *Strategy) trailingCheck(price float64, direction string) bool {
 		s.lowestPrice = price
 	}
 	isShort := direction == "short"
+	if isShort && s.sellPrice == 0 || !isShort && s.buyPrice == 0 {
+		return false
+	}
 	for i := len(s.TrailingCallbackRate) - 1; i >= 0; i-- {
 		trailingCallbackRate := s.TrailingCallbackRate[i]
 		trailingActivationRatio := s.TrailingActivationRatio[i]
@@ -244,7 +247,7 @@ func (s *Strategy) trailingCheck(price float64, direction string) bool {
 			}
 		} else {
 			if (s.highestPrice-s.buyPrice)/s.buyPrice > trailingActivationRatio {
-				return (s.highestPrice-price)/price > trailingCallbackRate
+				return (s.highestPrice-price)/s.buyPrice > trailingCallbackRate
 			}
 		}
 	}
@@ -351,15 +354,19 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 			s.highestPrice = 0
 			s.lowestPrice = 0
 		} else if s.Position.IsLong() {
-			s.buyPrice = price
+			s.buyPrice = s.Position.ApproximateAverageCost.Float64()
 			s.sellPrice = 0
-			s.highestPrice = s.buyPrice
+			s.highestPrice = math.Max(s.buyPrice, s.highestPrice)
 			s.lowestPrice = 0
 		} else {
-			s.sellPrice = price
+			s.sellPrice = s.Position.ApproximateAverageCost.Float64()
 			s.buyPrice = 0
 			s.highestPrice = 0
+			if s.lowestPrice == 0 {
 				s.lowestPrice = s.sellPrice
+			} else {
+				s.lowestPrice = math.Min(s.lowestPrice, s.sellPrice)
+			}
 		}
 	})
 	s.initTickerFunctions()
@@ -477,6 +484,8 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
 	bull := kline.Close.Compare(kline.Open) > 0
 
 	balances := s.GeneralOrderExecutor.Session().GetAccount().Balances()
+	startTime := kline.StartTime.Time()
+	if startTime.Round(time.Second) == startTime.Round(time.Minute) {
 		bbgo.Notify("source: %.4f, price: %.4f lowest: %.4f highest: %.4f sp %.4f bp %.4f", sourcef, pricef, s.lowestPrice, s.highestPrice, s.sellPrice, s.buyPrice)
 		bbgo.Notify("balances: [Total] %v %s [Base] %s(%v %s) [Quote] %s",
 			s.CalcAssetValue(price),
@@ -486,6 +495,7 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
 			s.Market.QuoteCurrency,
 			balances[s.Market.QuoteCurrency].String(),
 		)
+	}
 
 	shortCondition := ewo[0] < ewo[1] && ewo[1] >= ewo[2] && (ewo[1] <= ewo[2] || ewo[2] >= ewo[3]) || s.sellPrice == 0 && ewo[0] < ewo[1] && ewo[1] < ewo[2]
 	longCondition := ewo[0] > ewo[1] && ewo[1] <= ewo[2] && (ewo[1] >= ewo[2] || ewo[2] <= ewo[3]) || s.buyPrice == 0 && ewo[0] > ewo[1] && ewo[1] > ewo[2]
@@ -493,18 +503,19 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
 	exitShortCondition := s.sellPrice > 0 && !shortCondition && s.sellPrice*(1.+stoploss) <= highf || s.sellPrice+atr <= highf || s.trailingCheck(highf, "short")
 	exitLongCondition := s.buyPrice > 0 && !longCondition && s.buyPrice*(1.-stoploss) >= lowf || s.buyPrice-atr >= lowf || s.trailingCheck(lowf, "long")
 
-	if exitShortCondition || exitLongCondition {
+	if exitShortCondition || exitLongCondition || (longCondition && bull) || (shortCondition && !bull) {
-		if err := s.GeneralOrderExecutor.GracefulCancel(ctx); err != nil {
+		if hold := s.smartCancel(ctx, pricef); hold > 0 {
-			log.WithError(err).Errorf("cannot cancel orders")
 			return
 		}
+	} else {
+		s.smartCancel(ctx, pricef)
+		return
+	}
+	if exitShortCondition || exitLongCondition {
 		s.ClosePosition(ctx, fixedpoint.One)
 	}
+
 	if longCondition && bull {
-		if err := s.GeneralOrderExecutor.GracefulCancel(ctx); err != nil {
-			log.WithError(err).Errorf("cannot cancel orders")
-			return
-		}
 		if source.Compare(price) > 0 {
 			source = price
 		}
@@ -527,10 +538,6 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
 		return
 	}
 	if shortCondition && !bull {
-		if err := s.GeneralOrderExecutor.GracefulCancel(ctx); err != nil {
-			log.WithError(err).Errorf("cannot cancel orders")
-			return
-		}
 		if source.Compare(price) < 0 {
 			source = price
 		}