mirror of
https://github.com/c9s/bbgo.git
synced 2024-09-21 08:41:10 +00:00
323 lines
8.8 KiB
Go
323 lines
8.8 KiB
Go
package gap
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"math"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/sirupsen/logrus"
|
|
|
|
"github.com/c9s/bbgo/pkg/bbgo"
|
|
"github.com/c9s/bbgo/pkg/exchange/max"
|
|
"github.com/c9s/bbgo/pkg/fixedpoint"
|
|
"github.com/c9s/bbgo/pkg/service"
|
|
"github.com/c9s/bbgo/pkg/types"
|
|
)
|
|
|
|
const ID = "gap"
|
|
|
|
const stateKey = "state-v1"
|
|
|
|
var log = logrus.WithField("strategy", ID)
|
|
|
|
func init() {
|
|
bbgo.RegisterStrategy(ID, &Strategy{})
|
|
}
|
|
|
|
func (s *Strategy) ID() string {
|
|
return ID
|
|
}
|
|
|
|
type State struct {
|
|
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
|
|
}
|
|
|
|
type Strategy struct {
|
|
*bbgo.Graceful
|
|
*bbgo.Notifiability
|
|
*bbgo.Persistence
|
|
|
|
Symbol string `json:"symbol"`
|
|
SourceExchange string `json:"sourceExchange"`
|
|
TradingExchange string `json:"tradingExchange"`
|
|
|
|
DailyFeeBudgets map[string]fixedpoint.Value `json:"dailyFeeBudgets,omitempty"`
|
|
DailyMaxVolume fixedpoint.Value `json:"dailyMaxVolume,omitempty"`
|
|
UpdateInterval types.Duration `json:"updateInterval"`
|
|
|
|
sourceSession, tradingSession *bbgo.ExchangeSession
|
|
sourceMarket, tradingMarket types.Market
|
|
|
|
state *State
|
|
|
|
mu sync.Mutex
|
|
lastKLine types.KLine
|
|
sourceBook, tradingBook *types.StreamOrderBook
|
|
groupID uint32
|
|
|
|
stopC chan struct{}
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
func (s *Strategy) handleTradeUpdate(trade types.Trade) {
|
|
log.Infof("received trade %+v", trade)
|
|
|
|
if trade.Symbol != s.Symbol {
|
|
return
|
|
}
|
|
|
|
if s.state.IsOver24Hours() {
|
|
s.state.Reset()
|
|
}
|
|
|
|
// safe check
|
|
if s.state.AccumulatedFees == nil {
|
|
s.state.AccumulatedFees = make(map[string]fixedpoint.Value)
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
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"})
|
|
sourceSession.Subscribe(types.BookChannel, s.Symbol, types.SubscribeOptions{Depth: "5"})
|
|
|
|
tradingSession, ok := sessions[s.TradingExchange]
|
|
if !ok {
|
|
panic(fmt.Errorf("trading session %s is not defined", s.TradingExchange))
|
|
}
|
|
|
|
tradingSession.Subscribe(types.BookChannel, s.Symbol, types.SubscribeOptions{Interval: "1m"})
|
|
}
|
|
|
|
func (s *Strategy) CrossRun(ctx context.Context, _ bbgo.OrderExecutionRouter, sessions map[string]*bbgo.ExchangeSession) error {
|
|
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)
|
|
}
|
|
|
|
s.stopC = make(chan struct{})
|
|
|
|
var state State
|
|
// load position
|
|
if err := s.Persistence.Load(&state, ID, stateKey); err != nil {
|
|
if err != service.ErrPersistenceNotExists {
|
|
return err
|
|
}
|
|
|
|
s.state = &State{}
|
|
s.state.Reset()
|
|
} else {
|
|
// loaded successfully
|
|
s.state = &state
|
|
log.Infof("state is restored: %+v", s.state)
|
|
|
|
if s.state.IsOver24Hours() {
|
|
log.Warn("state is over 24 hours, resetting to zero")
|
|
s.state.Reset()
|
|
}
|
|
}
|
|
|
|
s.Graceful.OnShutdown(func(ctx context.Context, wg *sync.WaitGroup) {
|
|
defer wg.Done()
|
|
|
|
close(s.stopC)
|
|
|
|
if err := s.Persistence.Save(&s.state, ID, stateKey); err != nil {
|
|
log.WithError(err).Errorf("can not save state: %+v", s.state)
|
|
} else {
|
|
log.Infof("state is saved => %+v", s.state)
|
|
}
|
|
})
|
|
|
|
// from here, set data binding
|
|
s.sourceSession.MarketDataStream.OnKLine(func(kline types.KLine) {
|
|
log.Infof("source exchange %s price: %f", s.Symbol, kline.Close)
|
|
s.mu.Lock()
|
|
s.lastKLine = kline
|
|
s.mu.Unlock()
|
|
})
|
|
|
|
s.sourceBook = types.NewStreamBook(s.Symbol)
|
|
s.sourceBook.BindStream(s.sourceSession.MarketDataStream)
|
|
|
|
s.tradingBook = types.NewStreamBook(s.Symbol)
|
|
s.tradingBook.BindStream(s.tradingSession.MarketDataStream)
|
|
|
|
s.tradingSession.UserDataStream.OnTradeUpdate(s.handleTradeUpdate)
|
|
|
|
instanceID := fmt.Sprintf("%s-%s", ID, s.Symbol)
|
|
s.groupID = max.GenerateGroupID(instanceID)
|
|
log.Infof("using group id %d from fnv32(%s)", s.groupID, instanceID)
|
|
|
|
go func() {
|
|
ticker := time.NewTicker(s.UpdateInterval.Duration())
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
|
|
case <-s.stopC:
|
|
return
|
|
|
|
case <-ticker.C:
|
|
if !s.isBudgetAllowed() {
|
|
continue
|
|
}
|
|
|
|
sourceBook := s.sourceBook.Copy()
|
|
book := s.tradingBook.Copy()
|
|
bestBid, hasBid := book.BestBid()
|
|
bestAsk, hasAsk := book.BestAsk()
|
|
|
|
// 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)
|
|
bestBid, hasBid = sourceBook.BestBid()
|
|
bestAsk, hasAsk = sourceBook.BestAsk()
|
|
}
|
|
|
|
// if the spread is less than 100 ticks (100 pips), skip
|
|
if spread.Float64() < 100*s.tradingMarket.TickSize {
|
|
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 {
|
|
bestBid, hasBid = sourceBook.BestBid()
|
|
bestAsk, hasAsk = sourceBook.BestAsk()
|
|
}
|
|
|
|
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 quantity = s.tradingMarket.MinQuantity
|
|
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{
|
|
Symbol: s.Symbol,
|
|
Side: types.SideTypeBuy,
|
|
Type: types.OrderTypeLimit,
|
|
Quantity: quantity,
|
|
Price: price,
|
|
Market: s.tradingMarket,
|
|
TimeInForce: "GTC",
|
|
GroupID: s.groupID,
|
|
}, types.SubmitOrder{
|
|
Symbol: s.Symbol,
|
|
Side: types.SideTypeSell,
|
|
Type: types.OrderTypeLimit,
|
|
Quantity: quantity,
|
|
Price: price,
|
|
Market: s.tradingMarket,
|
|
TimeInForce: "GTC",
|
|
GroupID: s.groupID,
|
|
})
|
|
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
|
|
}
|