Merge pull request #1197 from c9s/strategy/scmaker

FEATURE: [strategy] add stable coin market maker

config/scmaker.yaml

@@ -0,0 +1,55 @@
+sessions:
+  binance:
+    exchange: max
+    envVarPrefix: max
+    makerFeeRate: 0%
+    takerFeeRate: 0.025%
+
+exchangeStrategies:
+- on: max
+  scmaker:
+    symbol: USDCUSDT
+
+    ## adjustmentUpdateInterval is the interval for adjusting position
+    adjustmentUpdateInterval: 1m
+
+    ## liquidityUpdateInterval is the interval for updating liquidity orders
+    liquidityUpdateInterval: 1h
+
+    midPriceEMA:
+      interval: 1h
+      window: 99
+
+    ## priceRangeBollinger is used for the liquidity price range
+    priceRangeBollinger:
+      interval: 1h
+      window: 10
+      k: 1.0
+
+    numOfLiquidityLayers: 10
+
+    liquidityLayerTickSize: 0.0001
+
+    strengthInterval: 1m
+
+    minProfit: 0.01%
+
+    liquidityScale:
+      exp:
+        domain: [0, 9]
+        range: [1, 4]
+
+backtest:
+  sessions:
+    - max
+  startTime: "2023-05-20"
+  endTime: "2023-06-01"
+  symbols:
+  - USDCUSDT
+  account:
+    max:
+      makerFeeRate: 0.0%
+      takerFeeRate: 0.025%
+      balances:
+        USDC: 5000
+        USDT: 5000

go.sum

@@ -671,6 +671,7 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.7.4 h1:wZRexSlwd7ZXfKINDLsO4r7WBt3gTKONc6K/VesHvHM=
 github.com/stretchr/testify v1.7.4/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/subosito/gotenv v1.2.0 h1:Slr1R9HxAlEKefgq5jn9U+DnETlIUa6HfgEzj0g5d7s=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=

pkg/backtest/exchange.go

@@ -12,18 +12,18 @@ for each kline, the backtest engine:
 
 There are 2 ways that a strategy could work with backtest engine:
 
-1. the strategy receives kline from the market data stream, and then it submits the order by the given market data to the backtest engine.
+ 1. the strategy receives kline from the market data stream, and then it submits the order by the given market data to the backtest engine.
-   backtest engine receives the order and then pushes the trade and order updates to the user data stream.
+    backtest engine receives the order and then pushes the trade and order updates to the user data stream.
 
-   the strategy receives the trade and update its position.
+    the strategy receives the trade and update its position.
 
-2. the strategy places the orders when it starts. (like grid) the strategy then receives the order updates and then submit a new order
+ 2. the strategy places the orders when it starts. (like grid) the strategy then receives the order updates and then submit a new order
-   by its order update message.
+    by its order update message.
 
 We need to ensure that:
 
-1. if the strategy submits the order from the market data stream, since it's a separate goroutine, the strategy should block the backtest engine
+ 1. if the strategy submits the order from the market data stream, since it's a separate goroutine, the strategy should block the backtest engine
-   to process the trades before the next kline is published.
+    to process the trades before the next kline is published.
 */
 package backtest
 
@@ -270,8 +270,8 @@ func (e *Exchange) QueryTicker(ctx context.Context, symbol string) (*types.Ticke
 		Open:   kline.Open,
 		High:   kline.High,
 		Low:    kline.Low,
-		Buy:    kline.Close,
+		Buy:    kline.Close.Sub(matching.Market.TickSize),
-		Sell:   kline.Close,
+		Sell:   kline.Close.Add(matching.Market.TickSize),
 	}, nil
 }
 

pkg/bbgo/scale.go

@@ -12,6 +12,7 @@ type Scale interface {
 	Formula() string
 	FormulaOf(x float64) string
 	Call(x float64) (y float64)
+	Sum(step float64) float64
 }
 
 func init() {
@@ -21,6 +22,7 @@ func init() {
 	_ = Scale(&QuadraticScale{})
 }
 
+// f(x) := ab^x
 // y := ab^x
 // shift xs[0] to 0 (x - h)
 // a = y1
@@ -56,6 +58,14 @@ func (s *ExponentialScale) Solve() error {
 	return nil
 }
 
+func (s *ExponentialScale) Sum(step float64) float64 {
+	sum := 0.0
+	for x := s.Domain[0]; x <= s.Domain[1]; x += step {
+		sum += s.Call(x)
+	}
+	return sum
+}
+
 func (s *ExponentialScale) String() string {
 	return s.Formula()
 }
@@ -100,6 +110,14 @@ func (s *LogarithmicScale) Call(x float64) (y float64) {
 	return y
 }
 
+func (s *LogarithmicScale) Sum(step float64) float64 {
+	sum := 0.0
+	for x := s.Domain[0]; x <= s.Domain[1]; x += step {
+		sum += s.Call(x)
+	}
+	return sum
+}
+
 func (s *LogarithmicScale) String() string {
 	return s.Formula()
 }
@@ -158,6 +176,14 @@ func (s *LinearScale) Call(x float64) (y float64) {
 	return y
 }
 
+func (s *LinearScale) Sum(step float64) float64 {
+	sum := 0.0
+	for x := s.Domain[0]; x <= s.Domain[1]; x += step {
+		sum += s.Call(x)
+	}
+	return sum
+}
+
 func (s *LinearScale) String() string {
 	return s.Formula()
 }
@@ -201,6 +227,14 @@ func (s *QuadraticScale) Call(x float64) (y float64) {
 	return y
 }
 
+func (s *QuadraticScale) Sum(step float64) float64 {
+	sum := 0.0
+	for x := s.Domain[0]; x <= s.Domain[1]; x += step {
+		sum += s.Call(x)
+	}
+	return sum
+}
+
 func (s *QuadraticScale) String() string {
 	return s.Formula()
 }
@@ -266,18 +300,20 @@ func (rule *SlideRule) Scale() (Scale, error) {
 // LayerScale defines the scale DSL for maker layers, e.g.,
 //
 // quantityScale:
-//   byLayer:
+//
-//     exp:
+//	byLayer:
-//       domain: [1, 5]
+//	  exp:
-//       range: [0.01, 1.0]
+//	    domain: [1, 5]
+//	    range: [0.01, 1.0]
 //
 // and
 //
 // quantityScale:
-//   byLayer:
+//
-//     linear:
+//	byLayer:
-//       domain: [1, 3]
+//	  linear:
-//       range: [0.01, 1.0]
+//	    domain: [1, 3]
+//	    range: [0.01, 1.0]
 type LayerScale struct {
 	LayerRule *SlideRule `json:"byLayer"`
 }
@@ -303,18 +339,20 @@ func (s *LayerScale) Scale(layer int) (quantity float64, err error) {
 // PriceVolumeScale defines the scale DSL for strategy, e.g.,
 //
 // quantityScale:
-//   byPrice:
+//
-//     exp:
+//	byPrice:
-//       domain: [10_000, 50_000]
+//	  exp:
-//       range: [0.01, 1.0]
+//	    domain: [10_000, 50_000]
+//	    range: [0.01, 1.0]
 //
 // and
 //
 // quantityScale:
-//   byVolume:
+//
-//     linear:
+//	byVolume:
-//       domain: [10_000, 50_000]
+//	  linear:
-//       range: [0.01, 1.0]
+//	    domain: [10_000, 50_000]
+//	    range: [0.01, 1.0]
 type PriceVolumeScale struct {
 	ByPriceRule  *SlideRule `json:"byPrice"`
 	ByVolumeRule *SlideRule `json:"byVolume"`

pkg/cmd/strategy/builtin.go

@@ -29,6 +29,7 @@ import (
 	_ "github.com/c9s/bbgo/pkg/strategy/rebalance"
 	_ "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"

pkg/exchange/binance/convert.go

@@ -46,11 +46,11 @@ func toGlobalMarket(symbol binance.Symbol) types.Market {
 	}
 
 	if market.MinNotional.IsZero() {
-		log.Warn("binance market %s minNotional is zero", market.Symbol)
+		log.Warnf("binance market %s minNotional is zero", market.Symbol)
 	}
 
 	if market.MinQuantity.IsZero() {
-		log.Warn("binance market %s minQuantity is zero", market.Symbol)
+		log.Warnf("binance market %s minQuantity is zero", market.Symbol)
 	}
 
 	return market

pkg/indicator/float64series.go

@@ -11,8 +11,8 @@ type Float64Series struct {
 	slice floats.Slice
 }
 
-func NewFloat64Series(v ...float64) Float64Series {
+func NewFloat64Series(v ...float64) *Float64Series {
-	s := Float64Series{}
+	s := &Float64Series{}
 	s.slice = v
 	s.SeriesBase.Series = s.slice
 	return s

pkg/indicator/v2_atrp.go

@@ -1,11 +1,13 @@
 package indicator
 
 type ATRPStream struct {
-	Float64Series
+	*Float64Series
 }
 
 func ATRP2(source KLineSubscription, window int) *ATRPStream {
-	s := &ATRPStream{}
+	s := &ATRPStream{
+		Float64Series: NewFloat64Series(),
+	}
 	tr := TR2(source)
 	atr := RMA2(tr, window, true)
 	atr.OnUpdate(func(x float64) {

pkg/indicator/v2_boll.go

@@ -0,0 +1,52 @@
+package indicator
+
+type BOLLStream struct {
+	// the band series
+	*Float64Series
+
+	UpBand, DownBand *Float64Series
+
+	window int
+	k      float64
+
+	SMA    *SMAStream
+	StdDev *StdDevStream
+}
+
+// BOOL2 is bollinger indicator
+// the data flow:
+//
+// priceSource ->
+//
+//	-> calculate SMA
+//	-> calculate stdDev -> calculate bandWidth -> get latest SMA -> upBand, downBand
+func BOLL2(source Float64Source, window int, k float64) *BOLLStream {
+	// bind these indicators before our main calculator
+	sma := SMA2(source, window)
+	stdDev := StdDev2(source, window)
+
+	s := &BOLLStream{
+		Float64Series: NewFloat64Series(),
+		UpBand:        NewFloat64Series(),
+		DownBand:      NewFloat64Series(),
+		window:        window,
+		k:             k,
+		SMA:           sma,
+		StdDev:        stdDev,
+	}
+	s.Bind(source, s)
+
+	// on band update
+	s.Float64Series.OnUpdate(func(band float64) {
+		mid := s.SMA.Last(0)
+		s.UpBand.PushAndEmit(mid + band)
+		s.DownBand.PushAndEmit(mid - band)
+	})
+	return s
+}
+
+func (s *BOLLStream) Calculate(v float64) float64 {
+	stdDev := s.StdDev.Last(0)
+	band := stdDev * s.k
+	return band
+}

pkg/indicator/v2_cma.go

@@ -1,7 +1,7 @@
 package indicator
 
 type CMAStream struct {
-	Float64Series
+	*Float64Series
 }
 
 func CMA2(source Float64Source) *CMAStream {

pkg/indicator/v2_cross.go

@@ -13,7 +13,7 @@ const (
 
 // CrossStream subscribes 2 upstreams, and calculate the cross signal
 type CrossStream struct {
-	Float64Series
+	*Float64Series
 
 	a, b floats.Slice
 }

pkg/indicator/v2_ewma.go

@@ -1,7 +1,7 @@
 package indicator
 
 type EWMAStream struct {
-	Float64Series
+	*Float64Series
 
 	window     int
 	multiplier float64
@@ -19,6 +19,10 @@ func EWMA2(source Float64Source, window int) *EWMAStream {
 
 func (s *EWMAStream) Calculate(v float64) float64 {
 	last := s.slice.Last(0)
+	if last == 0.0 {
+		return v
+	}
+
 	m := s.multiplier
 	return (1.0-m)*last + m*v
 }

pkg/indicator/v2_macd_test.go

@@ -35,13 +35,13 @@ func Test_MACD2(t *testing.T) {
 		{
 			name:   "random_case",
 			kLines: buildKLines(input),
-			want:   0.7967670223776384,
+			want:   0.7740187187598249,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			prices := &PriceStream{}
+			prices := ClosePrices(nil)
 			macd := MACD2(prices, 12, 26, 9)
 			for _, k := range tt.kLines {
 				prices.EmitUpdate(k.Close.Float64())

pkg/indicator/v2_multiply.go

@@ -3,7 +3,7 @@ package indicator
 import "github.com/c9s/bbgo/pkg/datatype/floats"
 
 type MultiplyStream struct {
-	Float64Series
+	*Float64Series
 	a, b floats.Slice
 }
 

pkg/indicator/v2_pivothigh.go

@@ -5,7 +5,7 @@ import (
 )
 
 type PivotHighStream struct {
-	Float64Series
+	*Float64Series
 	rawValues           floats.Slice
 	window, rightWindow int
 }

pkg/indicator/v2_pivotlow.go

@@ -5,7 +5,7 @@ import (
 )
 
 type PivotLowStream struct {
-	Float64Series
+	*Float64Series
 	rawValues           floats.Slice
 	window, rightWindow int
 }

pkg/indicator/v2_price.go

@@ -11,22 +11,23 @@ type KLineSubscription interface {
 }
 
 type PriceStream struct {
-	Float64Series
+	*Float64Series
 
 	mapper KLineValueMapper
 }
 
 func Price(source KLineSubscription, mapper KLineValueMapper) *PriceStream {
 	s := &PriceStream{
-		mapper: mapper,
+		Float64Series: NewFloat64Series(),
+		mapper:        mapper,
 	}
 
-	s.SeriesBase.Series = s.slice
+	if source != nil {
-
+		source.AddSubscriber(func(k types.KLine) {
-	source.AddSubscriber(func(k types.KLine) {
+			v := s.mapper(k)
-		v := s.mapper(k)
+			s.PushAndEmit(v)
-		s.PushAndEmit(v)
+		})
-	})
+	}
 	return s
 }
 

pkg/indicator/v2_rma.go

@@ -2,7 +2,7 @@ package indicator
 
 type RMAStream struct {
 	// embedded structs
-	Float64Series
+	*Float64Series
 
 	// config fields
 	Adjust bool

pkg/indicator/v2_rsi.go

@@ -2,7 +2,7 @@ package indicator
 
 type RSIStream struct {
 	// embedded structs
-	Float64Series
+	*Float64Series
 
 	// config fields
 	window int

pkg/indicator/v2_rsi_test.go

@@ -67,7 +67,7 @@ func Test_RSI2(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			// RSI2()
-			prices := &PriceStream{}
+			prices := ClosePrices(nil)
 			rsi := RSI2(prices, tt.window)
 
 			t.Logf("data length: %d", len(tt.values))

pkg/indicator/v2_sma.go

@@ -3,7 +3,7 @@ package indicator
 import "github.com/c9s/bbgo/pkg/types"
 
 type SMAStream struct {
-	Float64Series
+	*Float64Series
 	window    int
 	rawValues *types.Queue
 }

pkg/indicator/v2_stddev.go

@@ -3,7 +3,7 @@ package indicator
 import "github.com/c9s/bbgo/pkg/types"
 
 type StdDevStream struct {
-	Float64Series
+	*Float64Series
 
 	rawValues *types.Queue
 

pkg/indicator/v2_subtract.go

@@ -6,7 +6,7 @@ import (
 
 // SubtractStream subscribes 2 upstream data, and then subtract these 2 values
 type SubtractStream struct {
-	Float64Series
+	*Float64Series
 
 	a, b floats.Slice
 	i    int

pkg/indicator/v2_tr.go

@@ -9,7 +9,7 @@ import (
 // This TRStream calculates the ATR first
 type TRStream struct {
 	// embedded struct
-	Float64Series
+	*Float64Series
 
 	// private states
 	previousClose float64

pkg/strategy/scmaker/intensity.go

@@ -0,0 +1,44 @@
+package scmaker
+
+import (
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/indicator"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type IntensityStream struct {
+	*indicator.Float64Series
+
+	Buy, Sell *indicator.RMAStream
+	window    int
+}
+
+func Intensity(source indicator.KLineSubscription, window int) *IntensityStream {
+	s := &IntensityStream{
+		Float64Series: indicator.NewFloat64Series(),
+		window:        window,
+
+		Buy:  indicator.RMA2(indicator.NewFloat64Series(), window, false),
+		Sell: indicator.RMA2(indicator.NewFloat64Series(), window, false),
+	}
+
+	threshold := fixedpoint.NewFromFloat(100.0)
+	source.AddSubscriber(func(k types.KLine) {
+		volume := k.Volume.Float64()
+
+		// ignore zero volume events or <= 10usd events
+		if volume == 0.0 || k.Close.Mul(k.Volume).Compare(threshold) <= 0 {
+			return
+		}
+
+		c := k.Close.Compare(k.Open)
+		if c > 0 {
+			s.Buy.PushAndEmit(volume)
+		} else if c < 0 {
+			s.Sell.PushAndEmit(volume)
+		}
+		s.Float64Series.PushAndEmit(k.High.Sub(k.Low).Float64())
+	})
+
+	return s
+}

pkg/strategy/scmaker/strategy.go

@@ -0,0 +1,435 @@
+package scmaker
+
+import (
+	"context"
+	"fmt"
+	"math"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/c9s/bbgo/pkg/bbgo"
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/indicator"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+const ID = "scmaker"
+
+var ten = fixedpoint.NewFromInt(10)
+
+type BollingerConfig struct {
+	Interval types.Interval `json:"interval"`
+	Window   int            `json:"window"`
+	K        float64        `json:"k"`
+}
+
+func init() {
+	bbgo.RegisterStrategy(ID, &Strategy{})
+}
+
+// Strategy scmaker is a stable coin market maker
+type Strategy struct {
+	Environment *bbgo.Environment
+	Market      types.Market
+
+	Symbol string `json:"symbol"`
+
+	NumOfLiquidityLayers int `json:"numOfLiquidityLayers"`
+
+	LiquidityUpdateInterval types.Interval   `json:"liquidityUpdateInterval"`
+	PriceRangeBollinger     *BollingerConfig `json:"priceRangeBollinger"`
+	StrengthInterval        types.Interval   `json:"strengthInterval"`
+
+	AdjustmentUpdateInterval types.Interval `json:"adjustmentUpdateInterval"`
+
+	MidPriceEMA            *types.IntervalWindow `json:"midPriceEMA"`
+	LiquiditySlideRule     *bbgo.SlideRule       `json:"liquidityScale"`
+	LiquidityLayerTickSize fixedpoint.Value      `json:"liquidityLayerTickSize"`
+
+	MinProfit fixedpoint.Value `json:"minProfit"`
+
+	Position    *types.Position    `json:"position,omitempty" persistence:"position"`
+	ProfitStats *types.ProfitStats `json:"profitStats,omitempty" persistence:"profit_stats"`
+
+	session                                 *bbgo.ExchangeSession
+	orderExecutor                           *bbgo.GeneralOrderExecutor
+	liquidityOrderBook, adjustmentOrderBook *bbgo.ActiveOrderBook
+	book                                    *types.StreamOrderBook
+
+	liquidityScale bbgo.Scale
+
+	// indicators
+	ewma      *indicator.EWMAStream
+	boll      *indicator.BOLLStream
+	intensity *IntensityStream
+}
+
+func (s *Strategy) ID() string {
+	return ID
+}
+
+func (s *Strategy) InstanceID() string {
+	return fmt.Sprintf("%s:%s", ID, s.Symbol)
+}
+
+func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {
+	session.Subscribe(types.BookChannel, s.Symbol, types.SubscribeOptions{})
+	session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: s.AdjustmentUpdateInterval})
+	session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: s.LiquidityUpdateInterval})
+
+	if s.MidPriceEMA != nil {
+		session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: s.MidPriceEMA.Interval})
+	}
+}
+
+func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) error {
+	instanceID := s.InstanceID()
+
+	s.session = session
+	s.book = types.NewStreamBook(s.Symbol)
+	s.book.BindStream(session.UserDataStream)
+
+	s.liquidityOrderBook = bbgo.NewActiveOrderBook(s.Symbol)
+	s.liquidityOrderBook.BindStream(session.UserDataStream)
+
+	s.adjustmentOrderBook = bbgo.NewActiveOrderBook(s.Symbol)
+	s.adjustmentOrderBook.BindStream(session.UserDataStream)
+
+	// If position is nil, we need to allocate a new position for calculation
+	if s.Position == nil {
+		s.Position = types.NewPositionFromMarket(s.Market)
+	}
+
+	// Always update the position fields
+	s.Position.Strategy = ID
+	s.Position.StrategyInstanceID = instanceID
+
+	// if anyone of the fee rate is defined, this assumes that both are defined.
+	// so that zero maker fee could be applied
+	if s.session.MakerFeeRate.Sign() > 0 || s.session.TakerFeeRate.Sign() > 0 {
+		s.Position.SetExchangeFeeRate(s.session.ExchangeName, types.ExchangeFee{
+			MakerFeeRate: s.session.MakerFeeRate,
+			TakerFeeRate: s.session.TakerFeeRate,
+		})
+	}
+
+	if s.ProfitStats == nil {
+		s.ProfitStats = types.NewProfitStats(s.Market)
+	}
+
+	scale, err := s.LiquiditySlideRule.Scale()
+	if err != nil {
+		return err
+	}
+
+	if err := scale.Solve(); err != nil {
+		return err
+	}
+
+	s.liquidityScale = scale
+
+	s.orderExecutor = bbgo.NewGeneralOrderExecutor(session, s.Symbol, ID, instanceID, s.Position)
+	s.orderExecutor.BindEnvironment(s.Environment)
+	s.orderExecutor.BindProfitStats(s.ProfitStats)
+	s.orderExecutor.Bind()
+	s.orderExecutor.TradeCollector().OnPositionUpdate(func(position *types.Position) {
+		bbgo.Sync(ctx, s)
+	})
+
+	s.initializeMidPriceEMA(session)
+	s.initializePriceRangeBollinger(session)
+	s.initializeIntensityIndicator(session)
+
+	session.MarketDataStream.OnKLineClosed(func(k types.KLine) {
+		if k.Interval == s.AdjustmentUpdateInterval {
+			s.placeAdjustmentOrders(ctx)
+		}
+
+		if k.Interval == s.LiquidityUpdateInterval {
+			s.placeLiquidityOrders(ctx)
+		}
+	})
+
+	return nil
+}
+
+func (s *Strategy) initializeMidPriceEMA(session *bbgo.ExchangeSession) {
+	kLines := indicator.KLines(session.MarketDataStream, s.Symbol, s.MidPriceEMA.Interval)
+	s.ewma = indicator.EWMA2(indicator.ClosePrices(kLines), s.MidPriceEMA.Window)
+}
+
+func (s *Strategy) initializeIntensityIndicator(session *bbgo.ExchangeSession) {
+	kLines := indicator.KLines(session.MarketDataStream, s.Symbol, s.StrengthInterval)
+	s.intensity = Intensity(kLines, 10)
+}
+
+func (s *Strategy) initializePriceRangeBollinger(session *bbgo.ExchangeSession) {
+	kLines := indicator.KLines(session.MarketDataStream, s.Symbol, s.PriceRangeBollinger.Interval)
+	closePrices := indicator.ClosePrices(kLines)
+	s.boll = indicator.BOLL2(closePrices, s.PriceRangeBollinger.Window, s.PriceRangeBollinger.K)
+}
+
+func (s *Strategy) placeAdjustmentOrders(ctx context.Context) {
+	_ = s.adjustmentOrderBook.GracefulCancel(ctx, s.session.Exchange)
+
+	if s.Position.IsDust() {
+		return
+	}
+
+	ticker, err := s.session.Exchange.QueryTicker(ctx, s.Symbol)
+	if logErr(err, "unable to query ticker") {
+		return
+	}
+
+	if _, err := s.session.UpdateAccount(ctx); err != nil {
+		logErr(err, "unable to update account")
+		return
+	}
+
+	baseBal, _ := s.session.Account.Balance(s.Market.BaseCurrency)
+	quoteBal, _ := s.session.Account.Balance(s.Market.QuoteCurrency)
+
+	var adjOrders []types.SubmitOrder
+
+	posSize := s.Position.Base.Abs()
+	tickSize := s.Market.TickSize
+
+	if s.Position.IsShort() {
+		price := profitProtectedPrice(types.SideTypeBuy, s.Position.AverageCost, ticker.Sell.Add(tickSize.Neg()), s.session.MakerFeeRate, s.MinProfit)
+		quoteQuantity := fixedpoint.Min(price.Mul(posSize), quoteBal.Available)
+		bidQuantity := quoteQuantity.Div(price)
+
+		if s.Market.IsDustQuantity(bidQuantity, price) {
+			return
+		}
+
+		adjOrders = append(adjOrders, types.SubmitOrder{
+			Symbol:      s.Symbol,
+			Type:        types.OrderTypeLimitMaker,
+			Side:        types.SideTypeBuy,
+			Price:       price,
+			Quantity:    bidQuantity,
+			Market:      s.Market,
+			TimeInForce: types.TimeInForceGTC,
+		})
+	} else if s.Position.IsLong() {
+		price := profitProtectedPrice(types.SideTypeSell, s.Position.AverageCost, ticker.Buy.Add(tickSize), s.session.MakerFeeRate, s.MinProfit)
+		askQuantity := fixedpoint.Min(posSize, baseBal.Available)
+
+		if s.Market.IsDustQuantity(askQuantity, price) {
+			return
+		}
+
+		adjOrders = append(adjOrders, types.SubmitOrder{
+			Symbol:      s.Symbol,
+			Type:        types.OrderTypeLimitMaker,
+			Side:        types.SideTypeSell,
+			Price:       price,
+			Quantity:    askQuantity,
+			Market:      s.Market,
+			TimeInForce: types.TimeInForceGTC,
+		})
+	}
+
+	createdOrders, err := s.orderExecutor.SubmitOrders(ctx, adjOrders...)
+	if logErr(err, "unable to place liquidity orders") {
+		return
+	}
+
+	s.adjustmentOrderBook.Add(createdOrders...)
+}
+
+func (s *Strategy) placeLiquidityOrders(ctx context.Context) {
+	err := s.liquidityOrderBook.GracefulCancel(ctx, s.session.Exchange)
+	if logErr(err, "unable to cancel orders") {
+		return
+	}
+
+	ticker, err := s.session.Exchange.QueryTicker(ctx, s.Symbol)
+	if logErr(err, "unable to query ticker") {
+		return
+	}
+
+	if _, err := s.session.UpdateAccount(ctx); err != nil {
+		logErr(err, "unable to update account")
+		return
+	}
+
+	baseBal, _ := s.session.Account.Balance(s.Market.BaseCurrency)
+	quoteBal, _ := s.session.Account.Balance(s.Market.QuoteCurrency)
+
+	spread := ticker.Sell.Sub(ticker.Buy)
+	tickSize := fixedpoint.Max(s.LiquidityLayerTickSize, s.Market.TickSize)
+
+	midPriceEMA := s.ewma.Last(0)
+	midPrice := fixedpoint.NewFromFloat(midPriceEMA)
+
+	makerQuota := &bbgo.QuotaTransaction{}
+	makerQuota.QuoteAsset.Add(quoteBal.Available)
+	makerQuota.BaseAsset.Add(baseBal.Available)
+
+	bandWidth := s.boll.Last(0)
+
+	log.Infof("spread: %f mid price ema: %f boll band width: %f", spread.Float64(), midPriceEMA, bandWidth)
+
+	n := s.liquidityScale.Sum(1.0)
+
+	var bidPrices []fixedpoint.Value
+	var askPrices []fixedpoint.Value
+
+	// calculate and collect prices
+	for i := 0; i <= s.NumOfLiquidityLayers; i++ {
+		fi := fixedpoint.NewFromInt(int64(i))
+
+		bidPrice := ticker.Buy
+		askPrice := ticker.Sell
+
+		if i == s.NumOfLiquidityLayers {
+			bwf := fixedpoint.NewFromFloat(bandWidth)
+			bidPrice = midPrice.Add(bwf.Neg())
+			askPrice = midPrice.Add(bwf)
+		} else if i > 0 {
+			sp := tickSize.Mul(fi)
+			bidPrice = midPrice.Sub(sp)
+			askPrice = midPrice.Add(sp)
+
+			if bidPrice.Compare(ticker.Buy) < 0 {
+				bidPrice = ticker.Buy.Sub(sp)
+			}
+
+			if askPrice.Compare(ticker.Sell) > 0 {
+				askPrice = ticker.Sell.Add(sp)
+			}
+		}
+
+		bidPrice = s.Market.TruncatePrice(bidPrice)
+		askPrice = s.Market.TruncatePrice(askPrice)
+
+		bidPrices = append(bidPrices, bidPrice)
+		askPrices = append(askPrices, askPrice)
+	}
+
+	availableBase := baseBal.Available
+	availableQuote := quoteBal.Available
+
+	log.Infof("balances before liq orders: %s, %s",
+		baseBal.String(),
+		quoteBal.String())
+
+	if !s.Position.IsDust() {
+		if s.Position.IsLong() {
+			availableBase = availableBase.Sub(s.Position.Base)
+			availableBase = s.Market.RoundDownQuantityByPrecision(availableBase)
+		} else if s.Position.IsShort() {
+			posSizeInQuote := s.Position.Base.Mul(ticker.Sell)
+			availableQuote = availableQuote.Sub(posSizeInQuote)
+		}
+	}
+
+	askX := availableBase.Float64() / n
+	bidX := availableQuote.Float64() / (n * (fixedpoint.Sum(bidPrices).Float64()))
+
+	askX = math.Trunc(askX*1e8) / 1e8
+	bidX = math.Trunc(bidX*1e8) / 1e8
+
+	var liqOrders []types.SubmitOrder
+	for i := 0; i <= s.NumOfLiquidityLayers; i++ {
+		bidQuantity := fixedpoint.NewFromFloat(s.liquidityScale.Call(float64(i)) * bidX)
+		askQuantity := fixedpoint.NewFromFloat(s.liquidityScale.Call(float64(i)) * askX)
+		bidPrice := bidPrices[i]
+		askPrice := askPrices[i]
+
+		log.Infof("liqudity layer #%d %f/%f = %f/%f", i, askPrice.Float64(), bidPrice.Float64(), askQuantity.Float64(), bidQuantity.Float64())
+
+		placeBuy := true
+		placeSell := true
+		averageCost := s.Position.AverageCost
+		// when long position, do not place sell orders below the average cost
+		if !s.Position.IsDust() {
+			if s.Position.IsLong() && askPrice.Compare(averageCost) < 0 {
+				placeSell = false
+			}
+
+			if s.Position.IsShort() && bidPrice.Compare(averageCost) > 0 {
+				placeBuy = false
+			}
+		}
+
+		quoteQuantity := bidQuantity.Mul(bidPrice)
+
+		if s.Market.IsDustQuantity(bidQuantity, bidPrice) || !makerQuota.QuoteAsset.Lock(quoteQuantity) {
+			placeBuy = false
+		}
+
+		if s.Market.IsDustQuantity(askQuantity, askPrice) || !makerQuota.BaseAsset.Lock(askQuantity) {
+			placeSell = false
+		}
+
+		if placeBuy {
+			liqOrders = append(liqOrders, types.SubmitOrder{
+				Symbol:      s.Symbol,
+				Side:        types.SideTypeBuy,
+				Type:        types.OrderTypeLimitMaker,
+				Quantity:    bidQuantity,
+				Price:       bidPrice,
+				Market:      s.Market,
+				TimeInForce: types.TimeInForceGTC,
+			})
+		}
+
+		if placeSell {
+			liqOrders = append(liqOrders, types.SubmitOrder{
+				Symbol:      s.Symbol,
+				Side:        types.SideTypeSell,
+				Type:        types.OrderTypeLimitMaker,
+				Quantity:    askQuantity,
+				Price:       askPrice,
+				Market:      s.Market,
+				TimeInForce: types.TimeInForceGTC,
+			})
+		}
+	}
+
+	makerQuota.Commit()
+
+	createdOrders, err := s.orderExecutor.SubmitOrders(ctx, liqOrders...)
+	if logErr(err, "unable to place liquidity orders") {
+		return
+	}
+
+	s.liquidityOrderBook.Add(createdOrders...)
+}
+
+func profitProtectedPrice(side types.SideType, averageCost, price, feeRate, minProfit fixedpoint.Value) fixedpoint.Value {
+	switch side {
+	case types.SideTypeSell:
+		minProfitPrice := averageCost.Add(
+			averageCost.Mul(feeRate.Add(minProfit)))
+		return fixedpoint.Max(minProfitPrice, price)
+
+	case types.SideTypeBuy:
+		minProfitPrice := averageCost.Sub(
+			averageCost.Mul(feeRate.Add(minProfit)))
+		return fixedpoint.Min(minProfitPrice, price)
+
+	}
+	return price
+}
+
+func logErr(err error, msgAndArgs ...interface{}) bool {
+	if err == nil {
+		return false
+	}
+
+	if len(msgAndArgs) == 0 {
+		log.WithError(err).Error(err.Error())
+	} else if len(msgAndArgs) == 1 {
+		msg := msgAndArgs[0].(string)
+		log.WithError(err).Error(msg)
+	} else if len(msgAndArgs) > 1 {
+		msg := msgAndArgs[0].(string)
+		log.WithError(err).Errorf(msg, msgAndArgs[1:]...)
+	}
+
+	return true
+}

pkg/strategy/xalign/strategy.go

@@ -236,9 +236,9 @@ func (s *Strategy) CrossRun(ctx context.Context, _ bbgo.OrderExecutionRouter, se
 
 		orderBook := bbgo.NewActiveOrderBook("")
 		orderBook.BindStream(session.UserDataStream)
+		s.orderBooks[sessionName] = orderBook
 
 		s.sessions[sessionName] = session
-		s.orderBooks[sessionName] = orderBook
 	}
 
 	bbgo.OnShutdown(ctx, func(ctx context.Context, wg *sync.WaitGroup) {
@@ -313,6 +313,7 @@ func (s *Strategy) align(ctx context.Context, sessions map[string]*bbgo.Exchange
 				} else {
 					log.Errorf("orderbook %s not found", selectedSession.Name)
 				}
+				s.orderBooks[selectedSession.Name].Add(*createdOrder)
 			}
 		}
 	}

pkg/types/position.go

@@ -170,7 +170,12 @@ func (p *Position) NewMarketCloseOrder(percentage fixedpoint.Value) *SubmitOrder
 	}
 }
 
-func (p *Position) IsDust(price fixedpoint.Value) bool {
+func (p *Position) IsDust(a ...fixedpoint.Value) bool {
+	price := p.AverageCost
+	if len(a) > 0 {
+		price = a[0]
+	}
+
 	base := p.Base.Abs()
 	return p.Market.IsDustQuantity(base, price)
 }