bbgo_origin/pkg/strategy/xgap/strategy.go

377 lines
11 KiB
Go
Raw Normal View History

2021-12-26 07:13:51 +00:00
package xgap
2021-03-22 01:54:38 +00:00
import (
"context"
"fmt"
"math"
2022-01-14 03:59:40 +00:00
"math/rand"
2021-03-22 01:54:38 +00:00
"sync"
"time"
"github.com/sirupsen/logrus"
"github.com/c9s/bbgo/pkg/bbgo"
2021-03-22 09:26:53 +00:00
"github.com/c9s/bbgo/pkg/exchange/max"
2021-03-22 01:54:38 +00:00
"github.com/c9s/bbgo/pkg/fixedpoint"
2021-03-22 09:26:53 +00:00
"github.com/c9s/bbgo/pkg/service"
2021-03-22 01:54:38 +00:00
"github.com/c9s/bbgo/pkg/types"
2022-01-14 04:03:29 +00:00
"github.com/c9s/bbgo/pkg/util"
2021-03-22 01:54:38 +00:00
)
2021-12-26 07:13:51 +00:00
const ID = "xgap"
2021-03-22 01:54:38 +00:00
const stateKey = "state-v1"
var log = logrus.WithField("strategy", ID)
func init() {
bbgo.RegisterStrategy(ID, &Strategy{})
}
func (s *Strategy) ID() string {
return ID
}
2021-03-22 07:50:47 +00:00
type State struct {
2021-03-22 10:48:18 +00:00
AccumulatedFeeStartedAt time.Time `json:"accumulatedFeeStartedAt,omitempty"`
AccumulatedFees map[string]fixedpoint.Value `json:"accumulatedFees,omitempty"`
AccumulatedVolume fixedpoint.Value `json:"accumulatedVolume,omitempty"`
}
func (s *State) IsOver24Hours() bool {
return time.Now().Sub(s.AccumulatedFeeStartedAt) >= 24*time.Hour
}
func (s *State) Reset() {
t := time.Now()
dateTime := time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, t.Location())
log.Infof("resetting accumulated started time to: %s", dateTime)
s.AccumulatedFeeStartedAt = dateTime
s.AccumulatedFees = make(map[string]fixedpoint.Value)
s.AccumulatedVolume = 0
2021-03-22 07:50:47 +00:00
}
2021-03-22 01:54:38 +00:00
type Strategy struct {
*bbgo.Graceful
*bbgo.Notifiability
*bbgo.Persistence
Symbol string `json:"symbol"`
SourceExchange string `json:"sourceExchange"`
TradingExchange string `json:"tradingExchange"`
2022-01-14 04:10:40 +00:00
MinSpread fixedpoint.Value `json:"minSpread"`
Quantity fixedpoint.Value `json:"quantity"`
2021-03-22 01:54:38 +00:00
2021-03-22 10:48:18 +00:00
DailyFeeBudgets map[string]fixedpoint.Value `json:"dailyFeeBudgets,omitempty"`
DailyMaxVolume fixedpoint.Value `json:"dailyMaxVolume,omitempty"`
UpdateInterval types.Duration `json:"updateInterval"`
2021-12-27 17:47:07 +00:00
SimulateVolume bool `json:"simulateVolume"`
2021-03-22 01:54:38 +00:00
sourceSession, tradingSession *bbgo.ExchangeSession
sourceMarket, tradingMarket types.Market
state *State
2021-12-27 17:47:07 +00:00
mu sync.Mutex
lastSourceKLine, lastTradingKLine types.KLine
sourceBook, tradingBook *types.StreamOrderBook
groupID uint32
2021-03-22 01:54:38 +00:00
stopC chan struct{}
}
2021-03-22 10:48:18 +00:00
func (s *Strategy) isBudgetAllowed() bool {
if s.DailyFeeBudgets == nil {
return true
}
if s.state.AccumulatedFees == nil {
return true
}
for asset, budget := range s.DailyFeeBudgets {
if fee, ok := s.state.AccumulatedFees[asset]; ok {
if fee >= budget {
log.Warnf("accumulative fee %f exceeded the fee budget %f, skipping...", fee.Float64(), budget.Float64())
return false
}
}
}
return true
}
2021-03-22 01:54:38 +00:00
func (s *Strategy) handleTradeUpdate(trade types.Trade) {
log.Infof("received trade %+v", trade)
if trade.Symbol != s.Symbol {
return
}
2021-03-22 07:50:47 +00:00
2021-03-22 10:48:18 +00:00
if s.state.IsOver24Hours() {
s.state.Reset()
2021-03-22 07:50:47 +00:00
}
2021-03-22 10:48:18 +00:00
// safe check
if s.state.AccumulatedFees == nil {
s.state.AccumulatedFees = make(map[string]fixedpoint.Value)
}
2021-03-22 07:50:47 +00:00
2021-03-22 10:48:18 +00:00
s.state.AccumulatedFees[trade.FeeCurrency] += fixedpoint.NewFromFloat(trade.Fee)
s.state.AccumulatedVolume += fixedpoint.NewFromFloat(trade.Quantity)
log.Infof("accumulated fee: %f %s", s.state.AccumulatedFees[trade.FeeCurrency].Float64(), trade.FeeCurrency)
2021-03-22 01:54:38 +00:00
}
func (s *Strategy) CrossSubscribe(sessions map[string]*bbgo.ExchangeSession) {
sourceSession, ok := sessions[s.SourceExchange]
if !ok {
panic(fmt.Errorf("source session %s is not defined", s.SourceExchange))
}
sourceSession.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: "1m"})
2022-01-19 05:08:50 +00:00
sourceSession.Subscribe(types.BookChannel, s.Symbol, types.SubscribeOptions{})
2021-03-22 01:54:38 +00:00
tradingSession, ok := sessions[s.TradingExchange]
if !ok {
panic(fmt.Errorf("trading session %s is not defined", s.TradingExchange))
}
2021-12-27 18:14:49 +00:00
tradingSession.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: "1m"})
2022-01-19 05:08:50 +00:00
tradingSession.Subscribe(types.BookChannel, s.Symbol, types.SubscribeOptions{})
2021-03-22 01:54:38 +00:00
}
2021-03-22 10:48:18 +00:00
func (s *Strategy) CrossRun(ctx context.Context, _ bbgo.OrderExecutionRouter, sessions map[string]*bbgo.ExchangeSession) error {
2021-03-22 01:54:38 +00:00
if s.UpdateInterval == 0 {
s.UpdateInterval = types.Duration(time.Second)
}
sourceSession, ok := sessions[s.SourceExchange]
if !ok {
return fmt.Errorf("source session %s is not defined", s.SourceExchange)
}
s.sourceSession = sourceSession
tradingSession, ok := sessions[s.TradingExchange]
if !ok {
return fmt.Errorf("trading session %s is not defined", s.TradingExchange)
}
s.tradingSession = tradingSession
s.sourceMarket, ok = s.sourceSession.Market(s.Symbol)
if !ok {
return fmt.Errorf("source session market %s is not defined", s.Symbol)
}
s.tradingMarket, ok = s.tradingSession.Market(s.Symbol)
if !ok {
return fmt.Errorf("trading session market %s is not defined", s.Symbol)
}
2021-03-22 09:26:53 +00:00
s.stopC = make(chan struct{})
var state State
// load position
2021-03-22 10:48:18 +00:00
if err := s.Persistence.Load(&state, ID, stateKey); err != nil {
2021-03-22 09:26:53 +00:00
if err != service.ErrPersistenceNotExists {
return err
}
2021-03-22 10:48:18 +00:00
s.state = &State{}
s.state.Reset()
2021-03-22 09:26:53 +00:00
} else {
// loaded successfully
s.state = &state
log.Infof("state is restored: %+v", s.state)
2021-03-22 10:48:18 +00:00
if s.state.IsOver24Hours() {
log.Warn("state is over 24 hours, resetting to zero")
s.state.Reset()
}
2021-03-22 09:26:53 +00:00
}
s.Graceful.OnShutdown(func(ctx context.Context, wg *sync.WaitGroup) {
defer wg.Done()
close(s.stopC)
2021-03-22 10:48:18 +00:00
if err := s.Persistence.Save(&s.state, ID, stateKey); err != nil {
2021-03-22 09:26:53 +00:00
log.WithError(err).Errorf("can not save state: %+v", s.state)
} else {
log.Infof("state is saved => %+v", s.state)
}
})
2021-03-22 01:54:38 +00:00
// from here, set data binding
2021-05-27 19:13:50 +00:00
s.sourceSession.MarketDataStream.OnKLine(func(kline types.KLine) {
2021-12-27 17:47:07 +00:00
log.Infof("source exchange %s price: %f volume: %f", s.Symbol, kline.Close, kline.Volume)
2021-03-22 01:54:38 +00:00
s.mu.Lock()
2021-12-27 17:47:07 +00:00
s.lastSourceKLine = kline
s.mu.Unlock()
})
s.tradingSession.MarketDataStream.OnKLine(func(kline types.KLine) {
log.Infof("trading exchange %s price: %f volume: %f", s.Symbol, kline.Close, kline.Volume)
s.mu.Lock()
s.lastTradingKLine = kline
2021-03-22 01:54:38 +00:00
s.mu.Unlock()
})
s.sourceBook = types.NewStreamBook(s.Symbol)
2021-05-27 19:13:50 +00:00
s.sourceBook.BindStream(s.sourceSession.MarketDataStream)
2021-03-22 01:54:38 +00:00
s.tradingBook = types.NewStreamBook(s.Symbol)
2021-05-27 19:13:50 +00:00
s.tradingBook.BindStream(s.tradingSession.MarketDataStream)
2021-03-22 01:54:38 +00:00
s.tradingSession.UserDataStream.OnTradeUpdate(s.handleTradeUpdate)
2021-03-22 09:26:53 +00:00
2021-03-22 01:54:38 +00:00
instanceID := fmt.Sprintf("%s-%s", ID, s.Symbol)
2021-03-22 09:26:53 +00:00
s.groupID = max.GenerateGroupID(instanceID)
2021-03-22 01:54:38 +00:00
log.Infof("using group id %d from fnv32(%s)", s.groupID, instanceID)
go func() {
2022-01-14 04:03:29 +00:00
ticker := time.NewTicker(
util.MillisecondsJitter(s.UpdateInterval.Duration(), 1000),
)
2021-03-22 01:54:38 +00:00
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
2021-03-22 09:26:53 +00:00
case <-s.stopC:
return
2021-03-22 01:54:38 +00:00
case <-ticker.C:
2021-03-22 10:48:18 +00:00
if !s.isBudgetAllowed() {
continue
}
2022-01-14 04:03:29 +00:00
// < 10 seconds jitter sleep
delay := util.MillisecondsJitter(s.UpdateInterval.Duration(), 10*1000)
if delay < s.UpdateInterval.Duration() {
time.Sleep(delay)
}
bestBid, hasBid := s.tradingBook.BestBid()
bestAsk, hasAsk := s.tradingBook.BestAsk()
2021-03-22 01:54:38 +00:00
// try to use the bid/ask price from the trading book
if hasBid && hasAsk {
var spread = bestAsk.Price - bestBid.Price
var spreadPercentage = spread.Float64() / bestAsk.Price.Float64()
log.Infof("trading book spread=%f %f%%", spread.Float64(), spreadPercentage*100.0)
// use the source book price if the spread percentage greater than 10%
if spreadPercentage > 0.05 {
log.Warnf("spread too large (%f %f%%), using source book", spread.Float64(), spreadPercentage)
2022-01-14 04:03:29 +00:00
bestBid, hasBid = s.sourceBook.BestBid()
bestAsk, hasAsk = s.sourceBook.BestAsk()
2021-03-22 01:54:38 +00:00
}
2022-01-14 04:10:40 +00:00
if s.MinSpread > 0 {
if spread < s.MinSpread {
log.Warnf("spread < min spread, spread=%f minSpread=%f bid=%f ask=%f", spread.Float64(), s.MinSpread.Float64(), bestBid.Price.Float64(), bestAsk.Price.Float64())
continue
}
}
2021-03-22 10:48:18 +00:00
// if the spread is less than 100 ticks (100 pips), skip
if spread.Float64() < 100*s.tradingMarket.TickSize {
2021-03-22 01:54:38 +00:00
log.Warnf("spread too small, we can't place orders: spread=%f bid=%f ask=%f", spread.Float64(), bestBid.Price.Float64(), bestAsk.Price.Float64())
continue
}
} else {
2022-01-14 04:03:29 +00:00
bestBid, hasBid = s.sourceBook.BestBid()
bestAsk, hasAsk = s.sourceBook.BestAsk()
2021-03-22 01:54:38 +00:00
}
if !hasBid || !hasAsk {
log.Warn("no bids or asks on the source book or the trading book")
continue
}
var spread = bestAsk.Price - bestBid.Price
var spreadPercentage = spread.Float64() / bestAsk.Price.Float64()
log.Infof("spread=%f %f%% ask=%f bid=%f", spread.Float64(), spreadPercentage*100.0, bestAsk.Price.Float64(), bestBid.Price.Float64())
// var spreadPercentage = spread.Float64() / bestBid.Price.Float64()
var midPrice = (bestAsk.Price + bestBid.Price).Div(fixedpoint.NewFromFloat(2))
var price = midPrice.Float64()
log.Infof("mid price %f", midPrice.Float64())
var balances = s.tradingSession.Account.Balances()
2021-03-22 01:54:38 +00:00
var quantity = s.tradingMarket.MinQuantity
if s.Quantity > 0 {
quantity = s.Quantity.Float64()
quantity = math.Min(quantity, s.tradingMarket.MinQuantity)
2021-12-27 17:47:07 +00:00
} else if s.SimulateVolume {
s.mu.Lock()
if s.lastTradingKLine.Volume > 0 && s.lastSourceKLine.Volume > 0 {
volumeDiff := s.lastSourceKLine.Volume - s.lastTradingKLine.Volume
// change the current quantity only diff is positive
if volumeDiff > 0 {
quantity = volumeDiff
}
if baseBalance, ok := balances[s.tradingMarket.BaseCurrency]; ok {
quantity = math.Min(quantity, baseBalance.Available.Float64())
}
if quoteBalance, ok := balances[s.tradingMarket.QuoteCurrency]; ok {
maxQuantity := quoteBalance.Available.Float64() / price
quantity = math.Min(quantity, maxQuantity)
}
2021-12-27 17:47:07 +00:00
}
s.mu.Unlock()
2022-01-14 03:59:40 +00:00
} else {
// plus a 2% quantity jitter
quantity *= 1.0 + math.Max(0.02, rand.Float64())
}
2021-03-22 01:54:38 +00:00
var quoteAmount = price * quantity
if quoteAmount <= s.tradingMarket.MinNotional {
quantity = math.Max(
s.tradingMarket.MinQuantity,
s.tradingMarket.MinNotional*1.01/price)
}
createdOrders, err := tradingSession.Exchange.SubmitOrders(ctx, types.SubmitOrder{
2021-12-27 17:47:07 +00:00
Symbol: s.Symbol,
Side: types.SideTypeBuy,
Type: types.OrderTypeLimit,
Quantity: quantity,
Price: price,
Market: s.tradingMarket,
// TimeInForce: "GTC",
2021-12-27 17:47:07 +00:00
GroupID: s.groupID,
2021-03-22 01:54:38 +00:00
}, types.SubmitOrder{
2021-12-27 17:47:07 +00:00
Symbol: s.Symbol,
Side: types.SideTypeSell,
Type: types.OrderTypeLimit,
Quantity: quantity,
Price: price,
Market: s.tradingMarket,
// TimeInForce: "GTC",
2021-12-27 17:47:07 +00:00
GroupID: s.groupID,
2021-03-22 01:54:38 +00:00
})
if err != nil {
log.WithError(err).Error("order submit error")
}
time.Sleep(time.Second)
if err := tradingSession.Exchange.CancelOrders(ctx, createdOrders...); err != nil {
log.WithError(err).Error("cancel order error")
}
}
}
}()
return nil
}