lychiyu/bbgo_origin
1
0
Fork 0
mirror of https://github.com/c9s/bbgo.git synced 2024-11-26 08:45:16 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

grid: refactor trade handler with trade collector

Browse Source
This commit is contained in:
c9s 2021-11-05 00:30:04 +08:00
parent 7787edffa0
commit 7db7596abe
1 changed files with 19 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
pkg/strategy/grid/strategy.go
View File

@ -38,7 +38,7 @@ type State struct {
// [source Order ID] -> arbitrage order // [source Order ID] -> arbitrage order
ArbitrageOrders map[uint64]types.Order `json:"arbitrageOrders"` ArbitrageOrders map[uint64]types.Order `json:"arbitrageOrders"`
ProfitStats bbgo.ProfitStats `json:"profitStats,omitempty"` ProfitStats bbgo.ProfitStats `json:"profitStats,omitempty"`
} }
type Strategy struct { type Strategy struct {
@ -99,6 +99,8 @@ type Strategy struct {
// activeOrders is the locally maintained active order book of the maker orders. // activeOrders is the locally maintained active order book of the maker orders.
activeOrders *bbgo.LocalActiveOrderBook activeOrders *bbgo.LocalActiveOrderBook
tradeCollector *bbgo.TradeCollector
// groupID is the group ID used for the strategy instance for canceling orders // groupID is the group ID used for the strategy instance for canceling orders
groupID uint32 groupID uint32
} }
@ -375,29 +377,9 @@ func (s *Strategy) placeGridOrders(orderExecutor bbgo.OrderExecutor, session *bb
} }
func (s *Strategy) tradeUpdateHandler(trade types.Trade) { func (s *Strategy) tradeUpdateHandler(trade types.Trade) {
if trade.Symbol != s.Symbol { if s.TradeService != nil {
return if err := s.TradeService.Mark(context.Background(), trade.ID, ID); err != nil {
} log.WithError(err).Error("trade mark error")
if s.orderStore.Exists(trade.OrderID) {
log.Infof("received trade update of order %d: %+v", trade.OrderID, trade)
if s.TradeService != nil {
if err := s.TradeService.Mark(context.Background(), trade.ID, ID); err != nil {
log.WithError(err).Error("trade mark error")
}
}
if trade.Side == types.SideTypeSelf {
return
}
profit, netProfit, madeProfit := s.state.Position.AddTrade(trade)
if madeProfit {
s.Notify("%s average cost profit: %f %s, net profit =~ %f %s",
s.Symbol,
profit.Float64(), s.Market.QuoteCurrency,
netProfit.Float64(), s.Market.QuoteCurrency)
} }
} }
} }
@ -430,7 +412,7 @@ func (s *Strategy) handleFilledOrder(filledOrder types.Order) {
} }
if amount <= s.Market.MinNotional { if amount <= s.Market.MinNotional {
quantity = bbgo.AdjustFloatQuantityByMinAmount(quantity, price, s.Market.MinNotional * 1.001) quantity = bbgo.AdjustFloatQuantityByMinAmount(quantity, price, s.Market.MinNotional*1.001)
// update amount // update amount
amount = quantity * price amount = quantity * price
@ -589,7 +571,7 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
log.Infof("using group id %d from fnv(%s)", s.groupID, instanceID) log.Infof("using group id %d from fnv(%s)", s.groupID, instanceID)
if err := s.LoadState(); err != nil { if err := s.LoadState(); err != nil {
return err return err
} }
s.Notify("grid %s position", s.Symbol, s.state.Position) s.Notify("grid %s position", s.Symbol, s.state.Position)
@ -602,6 +584,17 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
s.activeOrders.OnFilled(s.handleFilledOrder) s.activeOrders.OnFilled(s.handleFilledOrder)
s.activeOrders.BindStream(session.UserDataStream) s.activeOrders.BindStream(session.UserDataStream)
s.tradeCollector = bbgo.NewTradeCollector(s.Symbol, s.state.Position, s.orderStore)
s.tradeCollector.OnTrade(func(trade types.Trade) {
s.Notifiability.Notify(trade)
s.state.ProfitStats.AddTrade(trade)
s.tradeUpdateHandler(trade)
})
s.tradeCollector.OnPositionUpdate(func(position *bbgo.Position) {
s.Notifiability.Notify(position)
})
s.tradeCollector.BindStream(session.UserDataStream)
s.Graceful.OnShutdown(func(ctx context.Context, wg *sync.WaitGroup) { s.Graceful.OnShutdown(func(ctx context.Context, wg *sync.WaitGroup) {
defer wg.Done() defer wg.Done()
@ -618,8 +611,6 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
} }
}) })
session.UserDataStream.OnTradeUpdate(s.tradeUpdateHandler)
session.UserDataStream.OnStart(func() { session.UserDataStream.OnStart(func() {
// if we have orders in the state data, we can restore them // if we have orders in the state data, we can restore them
if len(s.state.Orders) > 0 { if len(s.state.Orders) > 0 {