Trades: Abstract and simplify trade processor

gbjk · Dec 20, 2024 · c37c010 · c37c010
1 parent 50448ec
commit c37c010
Show file tree

Hide file tree

Showing 67 changed files with 247 additions and 295 deletions.
diff --git a/backtester/data/kline/api/api.go b/backtester/data/kline/api/api.go
@@ -30,7 +30,7 @@ func LoadData(ctx context.Context, dataType int64, startDate, endDate time.Time,
 			return nil, fmt.Errorf("could not retrieve candle data for %v %v %v, %v", exch.GetName(), a, fPair, err)
 		}
 	case common.DataTrade:
-		var trades []trade.Data
+		var trades []trade.Trade
 		trades, err = exch.GetHistoricTrades(ctx,
 			fPair,
 			a,

diff --git a/backtester/data/kline/csv/csv.go b/backtester/data/kline/csv/csv.go
@@ -105,7 +105,7 @@ func LoadData(dataType int64, filepath, exchangeName string, interval time.Durat
 		}
 		resp.Item = &candles
 	case common.DataTrade:
-		var trades []trade.Data
+		var trades []trade.Trade
 		for {
 			row, errCSV := csvData.Read()
 			if errCSV != nil {
@@ -115,7 +115,7 @@ func LoadData(dataType int64, filepath, exchangeName string, interval time.Durat
 				return nil, errCSV
 			}
 
-			t := trade.Data{}
+			t := trade.Trade{}
 			v, errParse := strconv.ParseInt(row[0], 10, 32)
 			if errParse != nil {
 				return nil, errParse

diff --git a/backtester/data/kline/live/live.go b/backtester/data/kline/live/live.go
@@ -48,7 +48,7 @@ func LoadData(ctx context.Context, timeToRetrieve time.Time, exch exchange.IBotE
 			return nil, fmt.Errorf("could not retrieve live candle data for %v %v %v, %v", exch.GetName(), a, currencyPair, err)
 		}
 	case common.DataTrade:
-		var trades []trade.Data
+		var trades []trade.Trade
 		trades, err = exch.GetHistoricTrades(ctx,
 			pFmt,
 			a,

diff --git a/cmd/exchange_wrapper_issues/main.go b/cmd/exchange_wrapper_issues/main.go
@@ -93,7 +93,7 @@ func main() {
 		wrapperConfig.OrderSubmission.OrderType = orderTypeOverride
 	}
 	if orderSideOverride != "BUY" {
-		wrapperConfig.OrderSubmission.OrderSide = orderSideOverride
+		wrapperConfig.OrderSubmission.OrderSide = orderSideOverrideDefaultSaveInterval
 	}
 	if orderPriceOverride > 0 {
 		wrapperConfig.OrderSubmission.Price = orderPriceOverride
@@ -432,7 +432,7 @@ func testWrappers(e exchange.IBotExchange, base *exchange.Base, config *Config)
 				Response:   jsonifyInterface([]interface{}{nil}),
 			})
 
-			var getHistoricTradesResponse []trade.Data
+			var getHistoricTradesResponse []trade.Trade
 			getHistoricTradesResponse, err = e.GetHistoricTrades(context.TODO(), p, assetTypes[i], time.Now().Add(-time.Hour), time.Now())
 			msg = ""
 			if err != nil {
@@ -446,7 +446,7 @@ func testWrappers(e exchange.IBotExchange, base *exchange.Base, config *Config)
 				Response:   jsonifyInterface([]interface{}{getHistoricTradesResponse}),
 			})
 
-			var getRecentTradesResponse []trade.Data
+			var getRecentTradesResponse []trade.Trade
 			getRecentTradesResponse, err = e.GetRecentTrades(context.TODO(), p, assetTypes[i])
 			msg = ""
 			if err != nil {

diff --git a/engine/datahistory_manager_test.go b/engine/datahistory_manager_test.go
@@ -1499,7 +1499,7 @@ func (d dataHistoryJobResultService) GetJobResultsBetween(_ string, _, _ time.Ti
 	return nil, nil
 }
 
-func dataHistoryTraderLoader(exch, a, base, quote string, start, _ time.Time) ([]trade.Data, error) {
+func dataHistoryTraderLoader(exch, a, base, quote string, start, _ time.Time) ([]trade.Trade, error) {
 	cp, err := currency.NewPairFromStrings(base, quote)
 	if err != nil {
 		return nil, err
@@ -1508,7 +1508,7 @@ func dataHistoryTraderLoader(exch, a, base, quote string, start, _ time.Time) ([
 	if err != nil {
 		return nil, err
 	}
-	return []trade.Data{
+	return []trade.Trade{
 		{
 			Exchange:     exch,
 			CurrencyPair: cp,
@@ -1546,7 +1546,7 @@ func dataHistoryCandleLoader(exch string, cp currency.Pair, a asset.Item, interv
 	}, nil
 }
 
-func dataHistoryTradeSaver(...trade.Data) error {
+func dataHistoryTradeSaver(...trade.Trade) error {
 	return nil
 }
 
@@ -1595,8 +1595,8 @@ func (f dhmExchange) GetHistoricCandlesExtended(_ context.Context, p currency.Pa
 	}, nil
 }
 
-func (f dhmExchange) GetHistoricTrades(_ context.Context, p currency.Pair, a asset.Item, startTime, _ time.Time) ([]trade.Data, error) {
-	return []trade.Data{
+func (f dhmExchange) GetHistoricTrades(_ context.Context, p currency.Pair, a asset.Item, startTime, _ time.Time) ([]trade.Trade, error) {
+	return []trade.Trade{
 		{
 			Exchange:     testExchange,
 			CurrencyPair: p,

diff --git a/engine/datahistory_manager_types.go b/engine/datahistory_manager_types.go
@@ -133,8 +133,8 @@ type DataHistoryManager struct {
 	maxResultInsertions        int64
 	verbose                    bool
 	candleLoader               func(string, currency.Pair, asset.Item, kline.Interval, time.Time, time.Time) (*kline.Item, error)
-	tradeLoader                func(string, string, string, string, time.Time, time.Time) ([]trade.Data, error)
-	tradeSaver                 func(...trade.Data) error
+	tradeLoader                func(string, string, string, string, time.Time, time.Time) ([]trade.Trade, error)
+	tradeSaver                 func(...trade.Trade) error
 	candleSaver                func(*kline.Item, bool) (uint64, error)
 }
 

diff --git a/engine/engine.go b/engine/engine.go
@@ -214,13 +214,13 @@ func validateSettings(b *Engine, s *Settings, flagSet FlagSet) {
 		b.Settings.EventManagerDelay = EventSleepDelay
 	}
 
-	if b.Settings.TradeBufferProcessingInterval != trade.DefaultProcessorIntervalTime {
+	if b.Settings.TradeBufferProcessingInterval != trade.DefaultSaveInterval {
 		if b.Settings.TradeBufferProcessingInterval >= time.Second {
 			trade.BufferProcessorIntervalTime = b.Settings.TradeBufferProcessingInterval
 		} else {
-			b.Settings.TradeBufferProcessingInterval = trade.DefaultProcessorIntervalTime
+			b.Settings.TradeBufferProcessingInterval = trade.DefaultSaveInterval
 			gctlog.Warnf(gctlog.Global, "-tradeprocessinginterval must be >= to 1 second, using default value of %v",
-				trade.DefaultProcessorIntervalTime)
+				trade.DefaultSaveInterval)
 		}
 	}
 

diff --git a/engine/rpcserver.go b/engine/rpcserver.go
@@ -3219,7 +3219,7 @@ func (s *RPCServer) GetSavedTrades(_ context.Context, r *gctrpc.GetSavedTradesRe
 	if err != nil {
 		return nil, err
 	}
-	var trades []trade.Data
+	var trades []trade.Trade
 	trades, err = trade.GetTradesInRange(r.Exchange, r.AssetType, r.Pair.Base, r.Pair.Quote, start, end)
 	if err != nil {
 		return nil, err
@@ -3468,7 +3468,7 @@ func (s *RPCServer) FindMissingSavedTradeIntervals(_ context.Context, r *gctrpc.
 		iterationTime = iterationTime.Add(time.Hour)
 	}
 
-	var trades []trade.Data
+	var trades []trade.Trade
 	trades, err = trade.GetTradesInRange(
 		r.ExchangeName,
 		r.AssetType,
@@ -3563,7 +3563,7 @@ func (s *RPCServer) GetHistoricTrades(r *gctrpc.GetSavedTradesRequest, stream gc
 	if err != nil {
 		return err
 	}
-	var trades []trade.Data
+	var trades []trade.Trade
 	start, err := time.Parse(common.SimpleTimeFormatWithTimezone, r.Start)
 	if err != nil {
 		return fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
@@ -3644,7 +3644,7 @@ func (s *RPCServer) GetRecentTrades(ctx context.Context, r *gctrpc.GetSavedTrade
 		return nil, err
 	}
 
-	var trades []trade.Data
+	var trades []trade.Trade
 	trades, err = exch.GetRecentTrades(ctx, cp, a)
 	if err != nil {
 		return nil, err

diff --git a/engine/rpcserver_test.go b/engine/rpcserver_test.go
@@ -884,7 +884,7 @@ func TestGetHistoricCandles(t *testing.T) {
 	if len(results.Candle) == 0 {
 		t.Error("expected results")
 	}
-	err = trade.SaveTradesToDatabase(trade.Data{
+	err = trade.SaveTradesToDatabase(trade.Trade{
 		TID:          "test123",
 		Exchange:     testExchange,
 		CurrencyPair: cp,
@@ -956,7 +956,7 @@ func TestFindMissingSavedTradeIntervals(t *testing.T) {
 		t.Errorf("expected a status message")
 	}
 	// one trade response
-	err = trade.SaveTradesToDatabase(trade.Data{
+	err = trade.SaveTradesToDatabase(trade.Trade{
 		TID:          "test1234",
 		Exchange:     testExchange,
 		CurrencyPair: cp,
@@ -989,7 +989,7 @@ func TestFindMissingSavedTradeIntervals(t *testing.T) {
 	}
 
 	// two trades response
-	err = trade.SaveTradesToDatabase(trade.Data{
+	err = trade.SaveTradesToDatabase(trade.Trade{
 		TID:          "test123",
 		Exchange:     testExchange,
 		CurrencyPair: cp,

diff --git a/engine/websocketroutine_manager.go b/engine/websocketroutine_manager.go
@@ -347,7 +347,7 @@ func (m *WebsocketRoutineManager) websocketDataHandler(exchName string, data int
 				m.printAccountHoldingsChangeSummary(d[x])
 			}
 		}
-	case []trade.Data:
+	case []trade.Trade:
 		if m.verbose {
 			log.Infof(log.Trade, "%+v", d)
 		}

diff --git a/exchanges/alphapoint/alphapoint_wrapper.go b/exchanges/alphapoint/alphapoint_wrapper.go
@@ -262,12 +262,12 @@ func (a *Alphapoint) GetWithdrawalsHistory(_ context.Context, _ currency.Code, _
 }
 
 // GetRecentTrades returns the most recent trades for a currency and asset
-func (a *Alphapoint) GetRecentTrades(_ context.Context, _ currency.Pair, _ asset.Item) ([]trade.Data, error) {
+func (a *Alphapoint) GetRecentTrades(_ context.Context, _ currency.Pair, _ asset.Item) ([]trade.Trade, error) {
 	return nil, common.ErrNotYetImplemented
 }
 
 // GetHistoricTrades returns historic trade data within the timeframe provided
-func (a *Alphapoint) GetHistoricTrades(_ context.Context, _ currency.Pair, _ asset.Item, _, _ time.Time) ([]trade.Data, error) {
+func (a *Alphapoint) GetHistoricTrades(_ context.Context, _ currency.Pair, _ asset.Item, _, _ time.Time) ([]trade.Trade, error) {
 	return nil, common.ErrNotYetImplemented
 }
 

diff --git a/exchanges/binance/binance_websocket.go b/exchanges/binance/binance_websocket.go
@@ -338,7 +338,7 @@ func (b *Binance) wsHandleData(respRaw []byte) error {
 				err)
 		}
 		return b.Websocket.Trade.Update(saveTradeData,
-			trade.Data{
+			trade.Trade{
 				CurrencyPair: pair,
 				Timestamp:    t.TimeStamp,
 				Price:        t.Price.Float64(),

diff --git a/exchanges/binance/binance_wrapper.go b/exchanges/binance/binance_wrapper.go
@@ -766,14 +766,14 @@ func (b *Binance) GetWithdrawalsHistory(ctx context.Context, c currency.Code, _
 }
 
 // GetRecentTrades returns the most recent trades for a currency and asset
-func (b *Binance) GetRecentTrades(ctx context.Context, p currency.Pair, a asset.Item) ([]trade.Data, error) {
+func (b *Binance) GetRecentTrades(ctx context.Context, p currency.Pair, a asset.Item) ([]trade.Trade, error) {
 	const limit = 1000
 	rFmt, err := b.GetPairFormat(a, true)
 	if err != nil {
 		return nil, err
 	}
 	pFmt := p.Format(rFmt)
-	resp := make([]trade.Data, 0, limit)
+	resp := make([]trade.Trade, 0, limit)
 	switch a {
 	case asset.Spot:
 		tradeData, err := b.GetMostRecentTrades(ctx,
@@ -783,7 +783,7 @@ func (b *Binance) GetRecentTrades(ctx context.Context, p currency.Pair, a asset.
 		}
 
 		for i := range tradeData {
-			resp = append(resp, trade.Data{
+			resp = append(resp, trade.Trade{
 				TID:          strconv.FormatInt(tradeData[i].ID, 10),
 				Exchange:     b.Name,
 				CurrencyPair: p,
@@ -800,7 +800,7 @@ func (b *Binance) GetRecentTrades(ctx context.Context, p currency.Pair, a asset.
 		}
 
 		for i := range tradeData {
-			resp = append(resp, trade.Data{
+			resp = append(resp, trade.Trade{
 				TID:          strconv.FormatInt(tradeData[i].ID, 10),
 				Exchange:     b.Name,
 				CurrencyPair: p,
@@ -817,7 +817,7 @@ func (b *Binance) GetRecentTrades(ctx context.Context, p currency.Pair, a asset.
 		}
 
 		for i := range tradeData {
-			resp = append(resp, trade.Data{
+			resp = append(resp, trade.Trade{
 				TID:          strconv.FormatInt(tradeData[i].ID, 10),
 				Exchange:     b.Name,
 				CurrencyPair: p,
@@ -830,7 +830,7 @@ func (b *Binance) GetRecentTrades(ctx context.Context, p currency.Pair, a asset.
 	}
 
 	if b.IsSaveTradeDataEnabled() {
-		err := trade.AddTradesToBuffer(b.Name, resp...)
+		err := trade.Add(b.Name, resp...)
 		if err != nil {
 			return nil, err
 		}
@@ -841,7 +841,7 @@ func (b *Binance) GetRecentTrades(ctx context.Context, p currency.Pair, a asset.
 }
 
 // GetHistoricTrades returns historic trade data within the timeframe provided
-func (b *Binance) GetHistoricTrades(ctx context.Context, p currency.Pair, a asset.Item, from, to time.Time) ([]trade.Data, error) {
+func (b *Binance) GetHistoricTrades(ctx context.Context, p currency.Pair, a asset.Item, from, to time.Time) ([]trade.Trade, error) {
 	if err := b.CurrencyPairs.IsAssetEnabled(a); err != nil {
 		return nil, err
 	}
@@ -862,9 +862,9 @@ func (b *Binance) GetHistoricTrades(ctx context.Context, p currency.Pair, a asse
 	if err != nil {
 		return nil, fmt.Errorf("%w %v", err, pFmt)
 	}
-	result := make([]trade.Data, len(trades))
+	result := make([]trade.Trade, len(trades))
 	for i := range trades {
-		result[i] = trade.Data{
+		result[i] = trade.Trade{
 			CurrencyPair: p,
 			TID:          strconv.FormatInt(trades[i].ATradeID, 10),
 			Amount:       trades[i].Quantity,

diff --git a/exchanges/binanceus/binanceus_types.go b/exchanges/binanceus/binanceus_types.go
@@ -146,8 +146,8 @@ type AggregatedTrade struct {
 }
 
 // toTradeData this method converts the AggregatedTrade data into an instance of trade.Data
-func (a *AggregatedTrade) toTradeData(p currency.Pair, exchange string, aType asset.Item) *trade.Data {
-	return &trade.Data{
+func (a *AggregatedTrade) toTradeData(p currency.Pair, exchange string, aType asset.Item) *trade.Trade {
+	return &trade.Trade{
 		CurrencyPair: p,
 		TID:          strconv.FormatInt(a.ATradeID, 10),
 		Amount:       a.Quantity,

diff --git a/exchanges/binanceus/binanceus_websocket.go b/exchanges/binanceus/binanceus_websocket.go
@@ -346,7 +346,7 @@ func (bi *Binanceus) wsHandleData(respRaw []byte) error {
 					}
 
 					return bi.Websocket.Trade.Update(saveTradeData,
-						trade.Data{
+						trade.Trade{
 							CurrencyPair: pair,
 							Timestamp:    t.TimeStamp,
 							Price:        price,

diff --git a/exchanges/binanceus/binanceus_wrapper.go b/exchanges/binanceus/binanceus_wrapper.go
@@ -464,7 +464,7 @@ func (bi *Binanceus) GetWithdrawalsHistory(ctx context.Context, c currency.Code,
 }
 
 // GetRecentTrades returns the most recent trades for a currency and asset
-func (bi *Binanceus) GetRecentTrades(ctx context.Context, p currency.Pair, assetType asset.Item) ([]trade.Data, error) {
+func (bi *Binanceus) GetRecentTrades(ctx context.Context, p currency.Pair, assetType asset.Item) ([]trade.Trade, error) {
 	if p.IsEmpty() {
 		return nil, currency.ErrCurrencyPairEmpty
 	}
@@ -476,9 +476,9 @@ func (bi *Binanceus) GetRecentTrades(ctx context.Context, p currency.Pair, asset
 	if err != nil {
 		return nil, err
 	}
-	resp := make([]trade.Data, len(tradeData))
+	resp := make([]trade.Trade, len(tradeData))
 	for i := range tradeData {
-		resp[i] = trade.Data{
+		resp[i] = trade.Trade{
 			TID:          strconv.FormatInt(tradeData[i].ID, 10),
 			Exchange:     bi.Name,
 			AssetType:    assetType,
@@ -490,7 +490,7 @@ func (bi *Binanceus) GetRecentTrades(ctx context.Context, p currency.Pair, asset
 	}
 
 	if bi.IsSaveTradeDataEnabled() {
-		err := trade.AddTradesToBuffer(bi.Name, resp...)
+		err := trade.Add(bi.Name, resp...)
 		if err != nil {
 			return nil, err
 		}
@@ -500,7 +500,7 @@ func (bi *Binanceus) GetRecentTrades(ctx context.Context, p currency.Pair, asset
 }
 
 // GetHistoricTrades returns historic trade data within the timeframe provided
-func (bi *Binanceus) GetHistoricTrades(ctx context.Context, p currency.Pair, assetType asset.Item, timestampStart, timestampEnd time.Time) ([]trade.Data, error) {
+func (bi *Binanceus) GetHistoricTrades(ctx context.Context, p currency.Pair, assetType asset.Item, timestampStart, timestampEnd time.Time) ([]trade.Trade, error) {
 	if p.IsEmpty() {
 		return nil, currency.ErrCurrencyPairEmpty
 	}
@@ -516,7 +516,7 @@ func (bi *Binanceus) GetHistoricTrades(ctx context.Context, p currency.Pair, ass
 	if err != nil {
 		return nil, err
 	}
-	result := make([]trade.Data, len(trades))
+	result := make([]trade.Trade, len(trades))
 	exName := bi.Name
 	for i := range trades {
 		t := trades[i].toTradeData(p, exName, assetType)

diff --git a/exchanges/bitfinex/bitfinex_websocket.go b/exchanges/bitfinex/bitfinex_websocket.go
@@ -977,7 +977,7 @@ func (b *Bitfinex) handleWSTradesUpdate(c *subscription.Subscription, eventType
 			}
 			wsTrade.Price = price
 		}
-		tradeHolder = append(tradeHolder, wsTrade)
+		tradeHolder = append(tTradeHolder, wsTrade)
 	}
 	trades := make([]trade.Data, len(tradeHolder))
 	for i := range tradeHolder {
@@ -989,7 +989,7 @@ func (b *Bitfinex) handleWSTradesUpdate(c *subscription.Subscription, eventType
 		}
 		price := tradeHolder[i].Price
 		if price == 0 && tradeHolder[i].Rate > 0 {
-			price = tradeHolder[i].Rate
+			price = tradeHoldTrade].Rate
 		}
 		trades[i] = trade.Data{
 			TID:          strconv.FormatInt(tradeHolder[i].ID, 10),