Merge pull request #1808 from c9s/c9s/xalign/livenote

IMPROVE: [deposit2transfer] add livenote support

pkg/bbgo/config.go

@@ -65,6 +65,7 @@ func (m *ExchangeStrategyMount) Map() (map[string]interface{}, error) {
 type SlackNotification struct {
 	DefaultChannel string `json:"defaultChannel,omitempty"  yaml:"defaultChannel,omitempty"`
 	ErrorChannel   string `json:"errorChannel,omitempty"  yaml:"errorChannel,omitempty"`
+	QueueSize      int    `json:"queueSize,omitempty" yaml:"queueSize,omitempty"`
 }
 
 type SlackNotificationRouting struct {

pkg/bbgo/environment.go

@@ -862,7 +862,12 @@ func (environ *Environment) setupSlack(userConfig *Config, slackToken string, pe
 
 	var client = slack.New(slackToken, slackOpts...)
 
-	var notifier = slacknotifier.New(client, conf.DefaultChannel)
+	var notifierOpts []slacknotifier.NotifyOption
+	if conf.QueueSize > 0 {
+		notifierOpts = append(notifierOpts, slacknotifier.OptionQueueSize(conf.QueueSize))
+	}
+
+	var notifier = slacknotifier.New(client, conf.DefaultChannel, notifierOpts...)
 	Notification.AddNotifier(notifier)
 
 	// allocate a store, so that we can save the chatID for the owner

pkg/exchange/binance/exchange.go

@@ -622,12 +622,16 @@ func (e *Exchange) QueryDepositHistory(ctx context.Context, asset string, since,
 	}
 
 	for _, d := range records {
-		// 0(0:pending,6: credited but cannot withdraw, 1:success)
+		// 0(0:pending,6: credited but cannot withdraw, 7=Wrong Deposit,8=Waiting User confirm, 1:success)
 		// set the default status
 		status := types.DepositStatus(fmt.Sprintf("code: %d", d.Status))
 
 		// https://www.binance.com/en/support/faq/115003736451
 		switch d.Status {
+
+		case binanceapi.DepositStatusWrong:
+			status = types.DepositRejected
+
 		case binanceapi.DepositStatusPending:
 			status = types.DepositPending
 
@@ -647,8 +651,10 @@ func (e *Exchange) QueryDepositHistory(ctx context.Context, asset string, since,
 			AddressTag:    d.AddressTag,
 			TransactionID: d.TxId,
 			Status:        status,
+			RawStatus:     strconv.Itoa(int(d.Status)),
 			UnlockConfirm: d.UnlockConfirm,
 			Confirmation:  d.ConfirmTimes,
+			Network:       d.Network,
 		})
 	}
 

pkg/strategy/deposit2transfer/strategy.go

@@ -14,6 +14,7 @@ import (
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/exchange/retry"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/livenote"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -40,6 +41,12 @@ func init() {
 	bbgo.RegisterStrategy(ID, &Strategy{})
 }
 
+type SlackAlert struct {
+	Channel  string   `json:"channel"`
+	Mentions []string `json:"mentions"`
+	Pin      bool     `json:"pin"`
+}
+
 type Strategy struct {
 	Environment *bbgo.Environment
 
@@ -48,10 +55,13 @@ type Strategy struct {
 	Interval      types.Duration `json:"interval"`
 	TransferDelay types.Duration `json:"transferDelay"`
 
+	SlackAlert *SlackAlert `json:"slackAlert"`
+
 	marginTransferService marginTransferService
 	depositHistoryService types.ExchangeTransferService
 
-	session          *bbgo.ExchangeSession
+	session *bbgo.ExchangeSession
+
 	watchingDeposits map[string]types.Deposit
 	mu               sync.Mutex
 
@@ -68,7 +78,7 @@ func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {}
 
 func (s *Strategy) Defaults() error {
 	if s.Interval == 0 {
-		s.Interval = types.Duration(5 * time.Minute)
+		s.Interval = types.Duration(3 * time.Minute)
 	}
 
 	if s.TransferDelay == 0 {
@@ -137,7 +147,7 @@ func (s *Strategy) tickWatcher(ctx context.Context, interval time.Duration) {
 }
 
 func (s *Strategy) checkDeposits(ctx context.Context) {
-	accountLimiter := rate.NewLimiter(rate.Every(3*time.Second), 1)
+	accountLimiter := rate.NewLimiter(rate.Every(5*time.Second), 1)
 
 	for _, asset := range s.Assets {
 		logger := s.logger.WithField("asset", asset)
@@ -204,16 +214,43 @@ func (s *Strategy) checkDeposits(ctx context.Context) {
 				d.Amount.String(), d.Asset,
 				amount.String(), d.Asset)
 
+			if s.SlackAlert != nil {
+				bbgo.PostLiveNote(&d,
+					livenote.Channel(s.SlackAlert.Channel),
+					livenote.Comment(fmt.Sprintf("Transferring deposit asset %s %s into the margin account", amount.String(), d.Asset)),
+				)
+			}
+
 			err2 := retry.GeneralBackoff(ctx, func() error {
 				return s.marginTransferService.TransferMarginAccountAsset(ctx, d.Asset, amount, types.TransferIn)
 			})
 			if err2 != nil {
 				logger.WithError(err2).Errorf("unable to transfer deposit asset into the margin account")
+
+				if s.SlackAlert != nil {
+					bbgo.PostLiveNote(&d,
+						livenote.Channel(s.SlackAlert.Channel),
+						livenote.Comment(fmt.Sprintf("Margin account transfer error: %+v", err2)),
+					)
+				}
 			}
 		}
 	}
 }
 
+func (s *Strategy) addWatchingDeposit(deposit types.Deposit) {
+	s.watchingDeposits[deposit.TransactionID] = deposit
+
+	if s.SlackAlert != nil {
+		bbgo.PostLiveNote(&deposit,
+			livenote.Channel(s.SlackAlert.Channel),
+			livenote.Pin(s.SlackAlert.Pin),
+			livenote.CompareObject(true),
+			livenote.OneTimeMention(s.SlackAlert.Mentions...),
+		)
+	}
+}
+
 func (s *Strategy) scanDepositHistory(ctx context.Context, asset string, duration time.Duration) ([]types.Deposit, error) {
 	logger := s.logger.WithField("asset", asset)
 	logger.Debugf("scanning %s deposit history...", asset)
@@ -239,34 +276,39 @@ func (s *Strategy) scanDepositHistory(ctx context.Context, asset string, duratio
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	// update the watching deposits
 	for _, deposit := range deposits {
 		logger.Debugf("checking deposit: %+v", deposit)
 
 		if deposit.Asset != asset {
 			continue
 		}
 
+		// if the deposit record is already in the watch list, update it
 		if _, ok := s.watchingDeposits[deposit.TransactionID]; ok {
-			// if the deposit record is in the watch list, update it
-			s.watchingDeposits[deposit.TransactionID] = deposit
+			s.addWatchingDeposit(deposit)
 		} else {
+			// if the deposit record is not in the watch list, we need to check the status
+			// here the deposit is outside the watching list
 			switch deposit.Status {
 
 			case types.DepositSuccess:
+				// if the deposit is in success status, we need to check if it's newer than the latest deposit time
+				// this usually happens when the deposit is credited to the account very quickly
 				if depositTime, ok := s.lastAssetDepositTimes[asset]; ok {
 					// if it's newer than the latest deposit time, then we just add it the monitoring list
 					if deposit.Time.After(depositTime) {
-						logger.Infof("adding new success deposit: %s", deposit.TransactionID)
-						s.watchingDeposits[deposit.TransactionID] = deposit
+						logger.Infof("adding new succeedded deposit: %s", deposit.TransactionID)
+						s.addWatchingDeposit(deposit)
 					}
 				} else {
 					// ignore all initial deposits that are already in success status
-					logger.Infof("ignored succeess deposit: %s %+v", deposit.TransactionID, deposit)
+					logger.Infof("ignored expired succeedded deposit: %s %+v", deposit.TransactionID, deposit)
 				}
 
 			case types.DepositCredited, types.DepositPending:
 				logger.Infof("adding pending deposit: %s", deposit.TransactionID)
-				s.watchingDeposits[deposit.TransactionID] = deposit
+				s.addWatchingDeposit(deposit)
 			}
 		}
 	}
@@ -281,10 +323,12 @@ func (s *Strategy) scanDepositHistory(ctx context.Context, asset string, duratio
 	}
 
 	var succeededDeposits []types.Deposit
+
+	// find and move out succeeded deposits
 	for _, deposit := range s.watchingDeposits {
-		if deposit.Status == types.DepositSuccess {
+		switch deposit.Status {
+		case types.DepositSuccess:
 			logger.Infof("found pending -> success deposit: %+v", deposit)
-
 			current, required := deposit.GetCurrentConfirmation()
 			if required > 0 && deposit.UnlockConfirm > 0 && current < deposit.UnlockConfirm {
 				logger.Infof("deposit %s unlock confirm %d is not reached, current: %d, required: %d, skip this round", deposit.TransactionID, deposit.UnlockConfirm, current, required)

pkg/types/deposit.go

@@ -39,11 +39,15 @@ type Deposit struct {
 	TransactionID string           `json:"transactionID" db:"txn_id"`
 	Status        DepositStatus    `json:"status"`
 
+	RawStatus string `json:"rawStatus"`
+
 	// Required confirm for unlock balance
 	UnlockConfirm int `json:"unlockConfirm"`
 
 	// Confirmation format = "current/required", for example: "7/16"
 	Confirmation string `json:"confirmation"`
+
+	Network string `json:"network,omitempty"`
 }
 
 func (d Deposit) GetCurrentConfirmation() (current int, required int) {
@@ -107,19 +111,45 @@ func (d *Deposit) SlackAttachment() slack.Attachment {
 		})
 	}
 
+	if len(d.Confirmation) > 0 {
+		text := d.Confirmation
+		if d.UnlockConfirm > 0 {
+			text = fmt.Sprintf("%s (unlock %d)", d.Confirmation, d.UnlockConfirm)
+		}
+		fields = append(fields, slack.AttachmentField{
+			Title: "Confirmation",
+			Value: text,
+			Short: false,
+		})
+	}
+
+	if len(d.Network) > 0 {
+		fields = append(fields, slack.AttachmentField{
+			Title: "Network",
+			Value: d.Network,
+			Short: false,
+		})
+	}
+
+	fields = append(fields, slack.AttachmentField{
+		Title: "Amount",
+		Value: d.Amount.String() + " " + d.Asset,
+		Short: false,
+	})
+
 	return slack.Attachment{
 		Color: depositStatusSlackColor(d.Status),
-		Title: fmt.Sprintf("Deposit %s %s To %s", d.Amount.String(), d.Asset, d.Address),
+		Title: fmt.Sprintf("Deposit %s %s To %s (%s)", d.Amount.String(), d.Asset, d.Address, d.Exchange),
 		// TitleLink:   "",
 		Pretext: "",
 		Text:    "",
 		// ServiceName: "",
 		// ServiceIcon: "",
 		// FromURL:     "",
 		// OriginalURL: "",
-		Fields: fields,
-		Footer: fmt.Sprintf("Apply Time: %s", d.Time.Time().Format(time.RFC3339)),
-		// FooterIcon: "",
+		Fields:     fields,
+		Footer:     fmt.Sprintf("Apply Time: %s", d.Time.Time().Format(time.RFC3339)),
+		FooterIcon: ExchangeFooterIcon(d.Exchange),
 	}
 }