bbgo_origin/pkg/bbgo/tradecollector.go

238 lines
5.8 KiB
Go
Raw Normal View History

2021-06-24 07:38:44 +00:00
package bbgo
import (
"context"
"sync"
2021-10-13 23:56:24 +00:00
"time"
2021-06-24 07:38:44 +00:00
log "github.com/sirupsen/logrus"
"github.com/c9s/bbgo/pkg/fixedpoint"
2021-06-24 07:38:44 +00:00
"github.com/c9s/bbgo/pkg/sigchan"
"github.com/c9s/bbgo/pkg/types"
)
//go:generate callbackgen -type TradeCollector
type TradeCollector struct {
Symbol string
orderSig sigchan.Chan
tradeStore *TradeStore
tradeC chan types.Trade
position *types.Position
2021-06-24 07:38:44 +00:00
orderStore *OrderStore
doneTrades map[types.TradeKey]struct{}
2021-06-24 07:38:44 +00:00
mu sync.Mutex
recoverCallbacks []func(trade types.Trade)
tradeCallbacks []func(trade types.Trade, profit, netProfit fixedpoint.Value)
positionUpdateCallbacks []func(position *types.Position)
profitCallbacks []func(trade types.Trade, profit *types.Profit)
2021-06-24 07:38:44 +00:00
}
func NewTradeCollector(symbol string, position *types.Position, orderStore *OrderStore) *TradeCollector {
2021-06-24 07:38:44 +00:00
return &TradeCollector{
Symbol: symbol,
orderSig: sigchan.New(1),
tradeC: make(chan types.Trade, 100),
tradeStore: NewTradeStore(),
doneTrades: make(map[types.TradeKey]struct{}),
2021-06-24 07:38:44 +00:00
position: position,
orderStore: orderStore,
}
}
2022-03-14 06:30:41 +00:00
// OrderStore returns the order store used by the trade collector
func (c *TradeCollector) OrderStore() *OrderStore {
return c.orderStore
}
2022-03-14 06:30:41 +00:00
// Position returns the position used by the trade collector
func (c *TradeCollector) Position() *types.Position {
return c.position
}
func (c *TradeCollector) TradeStore() *TradeStore {
return c.tradeStore
}
func (c *TradeCollector) SetPosition(position *types.Position) {
c.position = position
}
// QueueTrade sends the trade object to the trade channel,
// so that the goroutine can receive the trade and process in the background.
func (c *TradeCollector) QueueTrade(trade types.Trade) {
2021-06-24 07:38:44 +00:00
c.tradeC <- trade
}
// BindStreamForBackground bind the stream callback for background processing
func (c *TradeCollector) BindStreamForBackground(stream types.Stream) {
stream.OnTradeUpdate(c.QueueTrade)
}
2021-06-24 07:38:44 +00:00
func (c *TradeCollector) BindStream(stream types.Stream) {
2022-01-06 17:17:07 +00:00
stream.OnTradeUpdate(func(trade types.Trade) {
c.ProcessTrade(trade)
})
2021-06-24 07:38:44 +00:00
}
2021-10-05 13:39:10 +00:00
// Emit triggers the trade processing (position update)
// If you sent order, and the order store is updated, you can call this method
// so that trades will be processed in the next round of the goroutine loop
2021-06-24 07:38:44 +00:00
func (c *TradeCollector) Emit() {
c.orderSig.Emit()
}
func (c *TradeCollector) Recover(ctx context.Context, ex types.ExchangeTradeHistoryService, symbol string, from time.Time) error {
trades, err := ex.QueryTrades(ctx, symbol, &types.TradeQueryOptions{
StartTime: &from,
})
if err != nil {
return err
}
for _, td := range trades {
log.Debugf("processing trade: %s", td.String())
if c.ProcessTrade(td) {
log.Infof("recovered trade: %s", td.String())
c.EmitRecover(td)
}
}
return nil
}
func (c *TradeCollector) setDone(key types.TradeKey) {
c.mu.Lock()
c.doneTrades[key] = struct{}{}
c.mu.Unlock()
}
2021-10-05 13:39:10 +00:00
// Process filters the received trades and see if there are orders matching the trades
// if we have the order in the order store, then the trade will be considered for the position.
// profit will also be calculated.
func (c *TradeCollector) Process() bool {
2021-10-05 13:29:45 +00:00
positionChanged := false
2021-10-05 13:29:45 +00:00
c.tradeStore.Filter(func(trade types.Trade) bool {
key := trade.Key()
2022-08-09 06:06:52 +00:00
c.mu.Lock()
defer c.mu.Unlock()
// if it's already done, remove the trade from the trade store
2022-01-06 17:03:12 +00:00
if _, done := c.doneTrades[key]; done {
return true
}
2021-10-05 13:29:45 +00:00
if c.orderStore.Exists(trade.OrderID) {
if c.position != nil {
profit, netProfit, madeProfit := c.position.AddTrade(trade)
if madeProfit {
p := c.position.NewProfit(trade, profit, netProfit)
c.EmitTrade(trade, profit, netProfit)
c.EmitProfit(trade, &p)
} else {
c.EmitTrade(trade, fixedpoint.Zero, fixedpoint.Zero)
c.EmitProfit(trade, nil)
}
positionChanged = true
2022-03-11 13:15:57 +00:00
} else {
2022-03-11 13:27:45 +00:00
c.EmitTrade(trade, fixedpoint.Zero, fixedpoint.Zero)
2021-10-05 13:29:45 +00:00
}
2022-08-09 06:06:52 +00:00
c.doneTrades[key] = struct{}{}
2021-10-05 13:29:45 +00:00
return true
}
return false
})
if positionChanged && c.position != nil {
2021-10-05 13:29:45 +00:00
c.EmitPositionUpdate(c.position)
}
return positionChanged
2021-10-05 13:29:45 +00:00
}
2022-01-07 08:52:45 +00:00
// processTrade takes a trade and see if there is a matched order
// if the order is found, then we add the trade to the position
// return true when the given trade is added
// return false when the given trade is not added
2022-01-06 17:03:12 +00:00
func (c *TradeCollector) processTrade(trade types.Trade) bool {
2022-08-09 06:06:52 +00:00
c.mu.Lock()
defer c.mu.Unlock()
2022-08-09 06:06:52 +00:00
key := trade.Key()
// if it's already done, remove the trade from the trade store
if _, done := c.doneTrades[key]; done {
return false
}
2022-08-09 06:06:52 +00:00
if c.orderStore.Exists(trade.OrderID) {
if c.position != nil {
profit, netProfit, madeProfit := c.position.AddTrade(trade)
if madeProfit {
p := c.position.NewProfit(trade, profit, netProfit)
c.EmitTrade(trade, profit, netProfit)
c.EmitProfit(trade, &p)
} else {
c.EmitTrade(trade, fixedpoint.Zero, fixedpoint.Zero)
c.EmitProfit(trade, nil)
}
c.EmitPositionUpdate(c.position)
2022-03-11 13:15:57 +00:00
} else {
2022-03-11 13:27:45 +00:00
c.EmitTrade(trade, fixedpoint.Zero, fixedpoint.Zero)
}
2022-08-09 06:06:52 +00:00
c.doneTrades[key] = struct{}{}
2022-01-06 17:03:12 +00:00
return true
}
return false
}
2022-01-07 08:52:45 +00:00
// return true when the given trade is added
// return false when the given trade is not added
2022-01-06 17:17:07 +00:00
func (c *TradeCollector) ProcessTrade(trade types.Trade) bool {
2022-01-07 08:52:45 +00:00
key := trade.Key()
// if it's already done, remove the trade from the trade store
c.mu.Lock()
2022-01-07 08:52:45 +00:00
if _, done := c.doneTrades[key]; done {
2022-01-06 17:17:07 +00:00
return false
}
c.mu.Unlock()
2022-01-07 08:52:45 +00:00
if c.processTrade(trade) {
return true
}
c.tradeStore.Add(trade)
return false
}
// Run is a goroutine executed in the background
2021-10-13 23:56:24 +00:00
// Do not use this function if you need back-testing
2021-06-24 07:38:44 +00:00
func (c *TradeCollector) Run(ctx context.Context) {
2021-10-13 23:56:24 +00:00
var ticker = time.NewTicker(3 * time.Second)
2021-06-24 07:38:44 +00:00
for {
select {
case <-ctx.Done():
return
2021-10-13 23:56:24 +00:00
case <-ticker.C:
c.Process()
2021-06-24 07:38:44 +00:00
case <-c.orderSig:
2021-10-05 13:29:45 +00:00
c.Process()
2021-06-24 07:38:44 +00:00
case trade := <-c.tradeC:
2022-01-06 17:03:12 +00:00
c.ProcessTrade(trade)
2021-06-24 07:38:44 +00:00
}
}
}