Merge pull request #1804 from c9s/c9s/xalign/improvements

README.md

@@ -392,10 +392,10 @@ persistence:
 
 Check out the strategy directory [strategy](pkg/strategy) for all built-in strategies:
 
-- `pricealert` strategy demonstrates how to use the notification system [pricealert](pkg/strategy/pricealert). See
+- `pricealert` strategy demonstrates how to use the notification system [pricealert](pkg/strategy/example/pricealert). See
   [document](./doc/strategy/pricealert.md).
 - `buyandhold` strategy demonstrates how to subscribe kline events and submit market
-  order [buyandhold](pkg/strategy/pricedrop)
+  order [buyandhold](pkg/strategy/example/pricedrop)
 - `bollgrid` strategy implements a basic grid strategy with the built-in bollinger
   indicator [bollgrid](pkg/strategy/bollgrid)
 - `grid` strategy implements the fixed price band grid strategy [grid](pkg/strategy/grid). See

README.zh_TW.md

@@ -370,8 +370,8 @@ persistence:
 
 查看策略目錄 [strategy](pkg/strategy) 以獲得所有內置策略：
 
-- `pricealert` 策略演示如何使用通知系統 [pricealert](pkg/strategy/pricealert)。參見[文件](./doc/strategy/pricealert.md).
-- `buyandhold` 策略演示如何訂閱 kline 事件並提交市場訂單 [buyandhold](pkg/strategy/pricedrop)
+- `pricealert` 策略演示如何使用通知系統 [pricealert](pkg/strategy/example/pricealert)。參見[文件](./doc/strategy/pricealert.md).
+- `buyandhold` 策略演示如何訂閱 kline 事件並提交市場訂單 [buyandhold](pkg/strategy/example/pricedrop)
 - `bollgrid` 策略實現了一個基本的網格策略，使用內置的布林通道指標 [bollgrid](pkg/strategy/bollgrid)
 - `grid` 策略實現了固定價格帶網格策略 [grid](pkg/strategy/grid)。參見[文件](./doc/strategy/grid.md).
 - `supertrend` 策略使用 Supertrend 指標作為趨勢，並使用 DEMA 指標作為噪聲

config/example/livenote.yaml

@@ -0,0 +1,16 @@
+---
+notifications:
+  slack:
+    defaultChannel: "dev-bbgo"
+    errorChannel: "dev-bbgo"
+
+sessions:
+  binance:
+    exchange: binance
+    envVarPrefix: binance
+
+exchangeStrategies:
+- on: binance
+  livenote:
+    symbol: BTCUSDT
+    interval: 5m

doc/strategy/pricealert.md

@@ -22,4 +22,4 @@ Setup Telegram/Slack notification before using Price Alert Strategy. See [Settin
 
 #### Examples
 
-See [pricealert.yaml](../../config/pricealert.yaml) and [pricealert-tg.yaml](../../config/pricealert-tg.yaml)
+See [pricealert.yaml](../../config/example/pricealert.yaml) and [pricealert-tg.yaml](../../config/pricealert-tg.yaml)

pkg/bbgo/livenote.go

@@ -0,0 +1,26 @@
+package bbgo
+
+import (
+	"github.com/sirupsen/logrus"
+
+	"github.com/c9s/bbgo/pkg/livenote"
+)
+
+// PostLiveNote posts a live note to slack or other services
+// The MessageID will be set after the message is posted if it's not set.
+func PostLiveNote(obj livenote.Object) {
+	if len(Notification.liveNotePosters) == 0 {
+		logrus.Warn("no live note poster is registered")
+		return
+	}
+
+	for _, poster := range Notification.liveNotePosters {
+		if err := poster.PostLiveNote(obj); err != nil {
+			logrus.WithError(err).Errorf("unable to post live note: %+v", obj)
+		}
+	}
+}
+
+type LiveNotePoster interface {
+	PostLiveNote(note livenote.Object) error
+}

pkg/bbgo/notification.go

@@ -49,6 +49,8 @@ func (n *NullNotifier) SendPhotoTo(channel string, buffer *bytes.Buffer) {}
 
 type Notifiability struct {
 	notifiers       []Notifier
+	liveNotePosters []LiveNotePoster
+
 	SessionChannelRouter *PatternChannelRouter `json:"-"`
 	SymbolChannelRouter  *PatternChannelRouter `json:"-"`
 	ObjectChannelRouter  *ObjectChannelRouter  `json:"-"`
@@ -81,6 +83,10 @@ func (m *Notifiability) RouteObject(obj interface{}) (channel string, ok bool) {
 // AddNotifier adds the notifier that implements the Notifier interface.
 func (m *Notifiability) AddNotifier(notifier Notifier) {
 	m.notifiers = append(m.notifiers, notifier)
+
+	if poster, ok := notifier.(LiveNotePoster); ok {
+		m.liveNotePosters = append(m.liveNotePosters, poster)
+	}
 }
 
 func (m *Notifiability) Notify(obj interface{}, args ...interface{}) {

pkg/cmd/strategy/builtin.go

@@ -17,6 +17,12 @@ import (
 	_ "github.com/c9s/bbgo/pkg/strategy/emastop"
 	_ "github.com/c9s/bbgo/pkg/strategy/etf"
 	_ "github.com/c9s/bbgo/pkg/strategy/ewoDgtrd"
+	_ "github.com/c9s/bbgo/pkg/strategy/example/kline"
+	_ "github.com/c9s/bbgo/pkg/strategy/example/livenote"
+	_ "github.com/c9s/bbgo/pkg/strategy/example/pricealert"
+	_ "github.com/c9s/bbgo/pkg/strategy/example/pricedrop"
+	_ "github.com/c9s/bbgo/pkg/strategy/example/rsicross"
+	_ "github.com/c9s/bbgo/pkg/strategy/example/skeleton"
 	_ "github.com/c9s/bbgo/pkg/strategy/factorzoo"
 	_ "github.com/c9s/bbgo/pkg/strategy/fixedmaker"
 	_ "github.com/c9s/bbgo/pkg/strategy/flashcrash"
@@ -25,20 +31,15 @@ import (
 	_ "github.com/c9s/bbgo/pkg/strategy/grid2"
 	_ "github.com/c9s/bbgo/pkg/strategy/harmonic"
 	_ "github.com/c9s/bbgo/pkg/strategy/irr"
-	_ "github.com/c9s/bbgo/pkg/strategy/kline"
 	_ "github.com/c9s/bbgo/pkg/strategy/linregmaker"
 	_ "github.com/c9s/bbgo/pkg/strategy/liquiditymaker"
 	_ "github.com/c9s/bbgo/pkg/strategy/marketcap"
 	_ "github.com/c9s/bbgo/pkg/strategy/pivotshort"
-	_ "github.com/c9s/bbgo/pkg/strategy/pricealert"
-	_ "github.com/c9s/bbgo/pkg/strategy/pricedrop"
 	_ "github.com/c9s/bbgo/pkg/strategy/random"
 	_ "github.com/c9s/bbgo/pkg/strategy/rebalance"
-	_ "github.com/c9s/bbgo/pkg/strategy/rsicross"
 	_ "github.com/c9s/bbgo/pkg/strategy/rsmaker"
 	_ "github.com/c9s/bbgo/pkg/strategy/schedule"
 	_ "github.com/c9s/bbgo/pkg/strategy/scmaker"
-	_ "github.com/c9s/bbgo/pkg/strategy/skeleton"
 	_ "github.com/c9s/bbgo/pkg/strategy/supertrend"
 	_ "github.com/c9s/bbgo/pkg/strategy/support"
 	_ "github.com/c9s/bbgo/pkg/strategy/swing"

pkg/livenote/livenote.go

@@ -0,0 +1,86 @@
+package livenote
+
+import "sync"
+
+type Object interface {
+	ObjectID() string
+}
+
+type LiveNote struct {
+	// MessageID is the unique identifier of the message
+	// for slack, it's the timestamp of the message
+	MessageID string `json:"messageId"`
+
+	ChannelID string `json:"channelId"`
+
+	Object Object
+
+	cachedObjID string
+}
+
+func NewLiveNote(object Object) *LiveNote {
+	return &LiveNote{
+		Object: object,
+	}
+}
+
+func (n *LiveNote) ObjectID() string {
+	if n.cachedObjID != "" {
+		return n.cachedObjID
+	}
+
+	n.cachedObjID = n.Object.ObjectID()
+	return n.cachedObjID
+}
+
+func (n *LiveNote) SetObject(object Object) {
+	n.Object = object
+}
+
+func (n *LiveNote) SetMessageID(messageID string) {
+	n.MessageID = messageID
+}
+
+func (n *LiveNote) SetChannelID(channelID string) {
+	n.ChannelID = channelID
+}
+
+type Pool struct {
+	notes map[string]*LiveNote
+	mu    sync.Mutex
+}
+
+func NewPool(size int64) *Pool {
+	return &Pool{
+		notes: make(map[string]*LiveNote, size),
+	}
+}
+
+func (p *Pool) Update(obj Object) *LiveNote {
+	objID := obj.ObjectID()
+
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	for _, note := range p.notes {
+		if note.ObjectID() == objID {
+			// update the object inside the note
+			note.SetObject(obj)
+			return note
+		}
+	}
+
+	note := NewLiveNote(obj)
+	p.add(objID, note)
+	return note
+}
+
+func (p *Pool) add(id string, note *LiveNote) {
+	p.notes[id] = note
+}
+
+func (p *Pool) Add(note *LiveNote) {
+	p.mu.Lock()
+	p.add(note.ObjectID(), note)
+	p.mu.Unlock()
+}

pkg/livenote/livenote_test.go

@@ -0,0 +1,44 @@
+package livenote
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+func TestLiveNotePool(t *testing.T) {
+	t.Run("same-kline", func(t *testing.T) {
+		pool := NewPool(100)
+		k := &types.KLine{
+			Symbol:    "BTCUSDT",
+			Interval:  types.Interval1m,
+			StartTime: types.Time(time.Date(2021, 1, 1, 0, 0, 0, 0, time.UTC)),
+		}
+
+		note := pool.Update(k)
+		note2 := pool.Update(k)
+		assert.Equal(t, note, note2, "the returned note object should be the same")
+	})
+
+	t.Run("different-kline", func(t *testing.T) {
+		pool := NewPool(100)
+		k := &types.KLine{
+			Symbol:    "BTCUSDT",
+			Interval:  types.Interval1m,
+			StartTime: types.Time(time.Date(2021, 1, 1, 0, 0, 0, 0, time.UTC)),
+		}
+
+		k2 := &types.KLine{
+			Symbol:    "BTCUSDT",
+			Interval:  types.Interval1m,
+			StartTime: types.Time(time.Date(2021, 1, 1, 0, 1, 0, 0, time.UTC)),
+		}
+
+		note := pool.Update(k)
+		note2 := pool.Update(k2)
+		assert.NotEqual(t, note, note2, "the returned note object should be different")
+	})
+}

pkg/notifier/slacknotifier/slack.go

@@ -8,6 +8,7 @@ import (
 
 	"golang.org/x/time/rate"
 
+	"github.com/c9s/bbgo/pkg/livenote"
 	"github.com/c9s/bbgo/pkg/types"
 
 	log "github.com/sirupsen/logrus"
@@ -26,6 +27,8 @@ type Notifier struct {
 	channel string
 
 	taskC chan notifyTask
+
+	liveNotePool *livenote.Pool
 }
 
 type NotifyOption func(notifier *Notifier)
@@ -35,6 +38,7 @@ func New(client *slack.Client, channel string, options ...NotifyOption) *Notifie
 		channel:      channel,
 		client:       client,
 		taskC:        make(chan notifyTask, 100),
+		liveNotePool: livenote.NewPool(100),
 	}
 
 	for _, o := range options {
@@ -54,7 +58,8 @@ func (n *Notifier) worker() {
 			return
 
 		case task := <-n.taskC:
-			limiter.Wait(ctx)
+			// ignore the wait error
+			_ = limiter.Wait(ctx)
 
 			_, _, err := n.client.PostMessageContext(ctx, task.Channel, task.Opts...)
 			if err != nil {
@@ -66,7 +71,50 @@ func (n *Notifier) worker() {
 	}
 }
 
+func (n *Notifier) PostLiveNote(obj livenote.Object) error {
+	note := n.liveNotePool.Update(obj)
+	ctx := context.Background()
+
+	channel := note.ChannelID
+	if channel == "" {
+		channel = n.channel
+	}
+
+	var attachment slack.Attachment
+	if creator, ok := note.Object.(types.SlackAttachmentCreator); ok {
+		attachment = creator.SlackAttachment()
+	} else {
+		return fmt.Errorf("livenote object does not support types.SlackAttachmentCreator interface")
+	}
+
+	opts := slack.MsgOptionAttachments(attachment)
+
+	if note.MessageID != "" {
+		// UpdateMessageContext returns channel, timestamp, text, err
+		_, _, _, err := n.client.UpdateMessageContext(ctx, channel, note.MessageID, opts)
+		if err != nil {
+			return err
+		}
+
+	} else {
+
+		respCh, respTs, err := n.client.PostMessageContext(ctx, channel, opts)
+		if err != nil {
+			log.WithError(err).
+				WithField("channel", n.channel).
+				Errorf("slack api error: %s", err.Error())
+			return err
+		}
+
+		note.SetChannelID(respCh)
+		note.SetMessageID(respTs)
+	}
+
+	return nil
+}
+
 func (n *Notifier) Notify(obj interface{}, args ...interface{}) {
+	// TODO: filter args for the channel option
 	n.NotifyTo(n.channel, obj, args...)
 }
 

pkg/strategy/example/livenote/strategy.go

@@ -0,0 +1,68 @@
+package livenote
+
+import (
+	"context"
+
+	"github.com/sirupsen/logrus"
+
+	"github.com/c9s/bbgo/pkg/bbgo"
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+// ID is the unique strategy ID, it needs to be in all lower case
+// For example, grid strategy uses "grid"
+const ID = "livenote"
+
+// log is a logrus.Entry that will be reused.
+// This line attaches the strategy field to the logger with our ID, so that the logs from this strategy will be tagged with our ID
+var log = logrus.WithField("strategy", ID)
+
+var ten = fixedpoint.NewFromInt(10)
+
+// init is a special function of golang, it will be called when the program is started
+// importing this package will trigger the init function call.
+func init() {
+	// Register our struct type to BBGO
+	// Note that you don't need to field the fields.
+	// BBGO uses reflect to parse your type information.
+	bbgo.RegisterStrategy(ID, &Strategy{})
+}
+
+// Strategy is a struct that contains the settings of your strategy.
+// These settings will be loaded from the BBGO YAML config file "bbgo.yaml" automatically.
+type Strategy struct {
+	Symbol string `json:"symbol"`
+}
+
+func (s *Strategy) ID() string {
+	return ID
+}
+
+func (s *Strategy) InstanceID() string {
+	return ID + "-" + s.Symbol
+}
+
+func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {
+	session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: "1m"})
+}
+
+// This strategy simply spent all available quote currency to buy the symbol whenever kline gets closed
+func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) error {
+	callback := func(k types.KLine) {
+		log.Info(k)
+		bbgo.PostLiveNote(&k)
+	}
+
+	// register our kline event handler
+	session.MarketDataStream.OnKLine(callback)
+	session.MarketDataStream.OnKLineClosed(callback)
+
+	// if you need to do something when the user data stream is ready
+	// note that you only receive order update, trade update, balance update when the user data stream is connect.
+	session.UserDataStream.OnStart(func() {
+		log.Infof("connected")
+	})
+
+	return nil
+}

pkg/strategy/xalign/strategy.go

@@ -22,8 +22,6 @@ const ID = "xalign"
 
 var log = logrus.WithField("strategy", ID)
 
-var activeTransferNotificationLimiter = rate.NewLimiter(rate.Every(5*time.Minute), 1)
-
 func init() {
 	bbgo.RegisterStrategy(ID, &Strategy{})
 }
@@ -63,6 +61,8 @@ type Strategy struct {
 	orderBooks map[string]*bbgo.ActiveOrderBook
 
 	orderStore *core.OrderStore
+
+	activeTransferNotificationLimiter *rate.Limiter
 }
 
 func (s *Strategy) ID() string {
@@ -92,6 +92,11 @@ func (s *Strategy) Defaults() error {
 	return nil
 }
 
+func (s *Strategy) Initialize() error {
+	s.activeTransferNotificationLimiter = rate.NewLimiter(rate.Every(5*time.Minute), 1)
+	return nil
+}
+
 func (s *Strategy) Validate() error {
 	if s.PreferredQuoteCurrencies == nil {
 		return errors.New("quoteCurrencies is not defined")
@@ -483,7 +488,7 @@ func (s *Strategy) align(ctx context.Context, sessions map[string]*bbgo.Exchange
 
 		s.resetFaultBalanceRecords(pendingWithdraw.Asset)
 
-		if activeTransferNotificationLimiter.Allow() {
+		if s.activeTransferNotificationLimiter.Allow() {
 			bbgo.Notify("Found active %s withdraw, skip balance align",
 				pendingWithdraw.Asset,
 				pendingWithdraw)
@@ -502,7 +507,7 @@ func (s *Strategy) align(ctx context.Context, sessions map[string]*bbgo.Exchange
 
 		s.resetFaultBalanceRecords(pendingDeposit.Asset)
 
-		if activeTransferNotificationLimiter.Allow() {
+		if s.activeTransferNotificationLimiter.Allow() {
 			bbgo.Notify("Found active %s deposit, skip balance align",
 				pendingDeposit.Asset,
 				pendingDeposit)

pkg/types/deposit.go

@@ -6,8 +6,9 @@ import (
 	"strings"
 	"time"
 
-	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/slack-go/slack"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
 )
 
 type DepositStatus string
@@ -64,6 +65,10 @@ func (d Deposit) EffectiveTime() time.Time {
 	return d.Time.Time()
 }
 
+func (d *Deposit) ObjectID() string {
+	return "deposit-" + d.Exchange.String() + "-" + d.Asset + "-" + d.Address + "-" + d.TransactionID
+}
+
 func (d Deposit) String() (o string) {
 	o = fmt.Sprintf("%s deposit %s %v <- ", d.Exchange, d.Asset, d.Amount)
 

pkg/types/kline.go

@@ -74,6 +74,10 @@ type KLine struct {
 	Closed         bool   `json:"closed" db:"closed"`
 }
 
+func (k *KLine) ObjectID() string {
+	return "kline-" + k.Symbol + k.Interval.String() + k.StartTime.Time().Format(time.RFC3339)
+}
+
 func (k *KLine) Set(o *KLine) {
 	k.GID = o.GID
 	k.Exchange = o.Exchange
@@ -280,8 +284,9 @@ func (k *KLine) SlackAttachment() slack.Attachment {
 				Short: true,
 			},
 		},
-		Footer:     "",
-		FooterIcon: "",
+
+		FooterIcon: ExchangeFooterIcon(k.Exchange),
+		Footer:     k.StartTime.String() + " ~ " + k.EndTime.String(),
 	}
 }
 

pkg/types/order.go

@@ -317,6 +317,10 @@ func (o Order) CsvRecords() [][]string {
 	}
 }
 
+func (o *Order) ObjectID() string {
+	return "order-" + o.Exchange.String() + "-" + o.Symbol + "-" + strconv.FormatUint(o.OrderID, 10)
+}
+
 // Backup backs up the current order quantity to a SubmitOrder object
 // so that we can post the order later when we want to restore the orders.
 func (o Order) Backup() SubmitOrder {

pkg/types/trade.go

@@ -208,7 +208,6 @@ func (trade Trade) SlackAttachment() slack.Attachment {
 
 	liquidity := trade.Liquidity()
 	text := templateutil.Render(slackTradeTextTemplate, trade)
-	footerIcon := ExchangeFooterIcon(trade.Exchange)
 
 	return slack.Attachment{
 		Text: text,
@@ -225,7 +224,7 @@ func (trade Trade) SlackAttachment() slack.Attachment {
 			{Title: "Liquidity", Value: liquidity, Short: true},
 			{Title: "Order ID", Value: strconv.FormatUint(trade.OrderID, 10), Short: true},
 		},
-		FooterIcon: footerIcon,
+		FooterIcon: ExchangeFooterIcon(trade.Exchange),
 		Footer:     strings.ToLower(trade.Exchange.String()) + templateutil.Render(" creation time {{ . }}", trade.Time.Time().Format(time.StampMilli)),
 	}
 }