kline/exchanges: automatic creation of unsupported candle intervals (#1091)

* kline: Add builder and testing * Ideas * kline: deploy builder functionality across GCT * exchanges: implement across gct * exchanges: Add tests and fix implementations before kline package testing and veri. * kline: Add tests and start to fix ConvertToNewInterval * kline: fix ConvertToNewInterval add tests * kline: complete overarching tests now on to exchanges * kline: finish exchange tests and implement limits * exchanges: more fixes * linter: fix * engine: fix tests * kraken: fix recent trades and other fixes * zb: fix tests * bithumb: fix empty insertion * kline: refactor/optimize CreateKline function * kline: remove the mooos! * kline: prealloc CalculateCandleDateRanges * linter: fix * exchanges: prealloc extended * fix whoopsie * reverse fix because this is a whoopsie * okx: fix risidual issues * linter: fix * kline: initial nits from @gloriouscode * kline: rename builder -> request and cascade change * linter: fix + test * kline: update forced alignment on start and end times when CreateKlineRequest is called. * nits: more more more * NITS: Addressed * tests: fix race issue * Update exchanges/kline/request.go Co-authored-by: Scott <gloriousCode@users.noreply.github.com> * kline: add method AddPadding() to automatically fill in holes in kline.Request functionality and reject if missing data when converting * kline: Add params start and end to addPadding() to insert blanks in between block * kline: remove test comment code as it's not needed anymore * kline: fix lint and test * kline: sort slice without extra bool check every iteration * okx: fix issues with timeing and candles and such from niterinos & address typo * Update exchanges/kline/kline.go Co-authored-by: Scott <gloriousCode@users.noreply.github.com> * glorious: niterinos * Update exchanges/poloniex/poloniex_wrapper.go Co-authored-by: Scott <gloriousCode@users.noreply.github.com> * glorious: nits now onto conflicts YAYA!!! * Update exchanges/exchange_test.go Co-authored-by: Scott <gloriousCode@users.noreply.github.com> * glorious: nits again * thrasher: nitters * thrasher: niterinos - adds partial flag for incomplete recent candles and fetching. * kline: rm fmtizzle packageizzle * glorious: nitters * glorious: more niterinos * fix last niterinos Co-authored-by: Ryan O'Hara-Reid <ryan.oharareid@thrasher.io> Co-authored-by: Scott <gloriousCode@users.noreply.github.com>
2026-05-24 15:10:19 +00:00 · 2023-01-17 16:22:33 +11:00
parent 72f36d70d1
commit 83cfefa45c
110 changed files with 11312 additions and 5768 deletions
--- a/engine/datahistory_manager.go
+++ b/engine/datahistory_manager.go
@@ -182,7 +182,7 @@ func (m *DataHistoryManager) compareJobsToData(jobs ...*DataHistoryJob) error {
 		if err != nil {
 			return err
 		}
-		var candles kline.Item
+		var candles *kline.Item
 		switch jobs[i].DataType {
 		case dataHistoryCandleDataType,
 			dataHistoryCandleValidationDataType,
@@ -707,9 +707,9 @@ func (m *DataHistoryManager) processCandleData(job *DataHistoryJob, exch exchang
 	candles, err := exch.GetHistoricCandlesExtended(context.TODO(),
 		job.Pair,
 		job.Asset,
+		job.Interval,
 		startRange,
-		endRange,
-		job.Interval)
+		endRange)
 	if err != nil {
 		r.Result += "could not get candles: " + err.Error() + ". "
 		r.Status = dataHistoryStatusFailed
@@ -725,7 +725,7 @@ func (m *DataHistoryManager) processCandleData(job *DataHistoryJob, exch exchang
 		}
 	}
 	candles.SourceJobID = job.ID
-	err = m.saveCandlesInBatches(job, &candles, r)
+	err = m.saveCandlesInBatches(job, candles, r)
 	return r, err
 }

@@ -838,7 +838,7 @@ func (m *DataHistoryManager) convertTradesToCandles(job *DataHistoryJob, startRa
 		return r, nil //nolint:nilerr // error is returned in the job result
 	}
 	candles.SourceJobID = job.ID
-	err = m.saveCandlesInBatches(job, &candles, r)
+	err = m.saveCandlesInBatches(job, candles, r)
 	return r, err
 }

@@ -870,14 +870,14 @@ func (m *DataHistoryManager) convertCandleData(job *DataHistoryJob, startRange,
 		r.Status = dataHistoryStatusFailed
 		return r, nil //nolint:nilerr // error is returned in the job result
 	}
-	newCandles, err := kline.ConvertToNewInterval(&candles, job.ConversionInterval)
+	newCandles, err := candles.ConvertToNewInterval(job.ConversionInterval)
 	if err != nil {
 		r.Result = "could not convert candles in range: " + err.Error()
 		r.Status = dataHistoryStatusFailed
 		return r, nil //nolint:nilerr // error is returned in the job result
 	}
 	newCandles.SourceJobID = job.ID
-	err = m.saveCandlesInBatches(job, &candles, r)
+	err = m.saveCandlesInBatches(job, candles, r)
 	return r, err
 }

@@ -910,9 +910,9 @@ func (m *DataHistoryManager) validateCandles(job *DataHistoryJob, exch exchange.
 	apiCandles, err := exch.GetHistoricCandlesExtended(context.TODO(),
 		job.Pair,
 		job.Asset,
+		job.Interval,
 		startRange,
-		endRange,
-		job.Interval)
+		endRange)
 	if err != nil {
 		r.Result = "could not get API candles: " + err.Error()
 		r.Status = dataHistoryStatusFailed
@@ -1012,7 +1012,7 @@ func (m *DataHistoryManager) validateCandles(job *DataHistoryJob, exch exchange.
 	if len(validationIssues) > 0 {
 		r.Result = strings.Join(validationIssues, " -- ")
 	}
-	err = m.saveCandlesInBatches(job, &apiCandles, r)
+	err = m.saveCandlesInBatches(job, apiCandles, r)
 	return r, err
 }

@@ -1230,7 +1230,8 @@ func (m *DataHistoryManager) validateJob(job *DataHistoryJob) error {
 	}

 	b := exch.GetBase()
-	if !b.Features.Enabled.Kline.Intervals[job.Interval.Word()] &&
+	// TODO: In future allow custom candles.
+	if !b.Features.Enabled.Kline.Intervals.ExchangeSupported(job.Interval) &&
 		(job.DataType == dataHistoryCandleDataType || job.DataType == dataHistoryCandleValidationDataType) {
 		return fmt.Errorf("job interval %s %s %w %s", job.Nickname, job.Interval.Word(), kline.ErrUnsupportedInterval, job.Exchange)
 	}
--- a/engine/datahistory_manager_test.go
+++ b/engine/datahistory_manager_test.go
@@ -1163,7 +1163,7 @@ func TestUpscaleJobCandleData(t *testing.T) {
 		Exchange:           testExchange,
 		Asset:              asset.Spot,
 		Pair:               currency.NewPair(currency.BTC, currency.USDT),
-		StartDate:          time.Now().Add(-kline.OneHour.Duration() * 2),
+		StartDate:          time.Now().Add(-kline.OneHour.Duration() * 24),
 		EndDate:            time.Now(),
 		Interval:           kline.OneHour,
 		ConversionInterval: kline.OneDay,
@@ -1530,22 +1530,26 @@ func dataHistoryTraderLoader(exch, a, base, quote string, start, _ time.Time) ([
 	}, nil
 }

-func dataHistoryCandleLoader(exch string, cp currency.Pair, a asset.Item, i kline.Interval, start, _ time.Time) (kline.Item, error) {
-	return kline.Item{
+func dataHistoryCandleLoader(exch string, cp currency.Pair, a asset.Item, i kline.Interval, start, _ time.Time) (*kline.Item, error) {
+	start = start.Truncate(i.Duration())
+	var candles []kline.Candle
+	for x := 0; x < 24; x++ {
+		candles = append(candles, kline.Candle{
+			Time:   start,
+			Open:   1,
+			High:   10,
+			Low:    1,
+			Close:  4,
+			Volume: 8,
+		})
+		start = start.Add(i.Duration())
+	}
+	return &kline.Item{
 		Exchange: exch,
 		Pair:     cp,
 		Asset:    a,
 		Interval: i,
-		Candles: []kline.Candle{
-			{
-				Time:   start,
-				Open:   1,
-				High:   10,
-				Low:    1,
-				Close:  4,
-				Volume: 8,
-			},
-		},
+		Candles:  candles,
 	}, nil
 }

@@ -1563,8 +1567,8 @@ type dhmExchange struct {
 	exchange.IBotExchange
 }

-func (f dhmExchange) GetHistoricCandlesExtended(ctx context.Context, p currency.Pair, a asset.Item, timeStart, _ time.Time, interval kline.Interval) (kline.Item, error) {
-	return kline.Item{
+func (f dhmExchange) GetHistoricCandlesExtended(ctx context.Context, p currency.Pair, a asset.Item, interval kline.Interval, timeStart, _ time.Time) (*kline.Item, error) {
+	return &kline.Item{
 		Exchange: testExchange,
 		Pair:     p,
 		Asset:    a,
--- a/engine/datahistory_manager_types.go
+++ b/engine/datahistory_manager_types.go
@@ -132,7 +132,7 @@ type DataHistoryManager struct {
 	maxJobsPerCycle            int64
 	maxResultInsertions        int64
 	verbose                    bool
-	candleLoader               func(string, currency.Pair, asset.Item, kline.Interval, time.Time, time.Time) (kline.Item, error)
+	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
 	candleSaver                func(*kline.Item, bool) (uint64, error)
--- a/engine/rpcserver.go
+++ b/engine/rpcserver.go
@@ -2447,11 +2447,11 @@ func (s *RPCServer) GetHistoricCandles(ctx context.Context, r *gctrpc.GetHistori
 	resp := gctrpc.GetHistoricCandlesResponse{
 		Interval: interval.Short(),
 		Pair:     r.Pair,
-		Start:    r.Start,
-		End:      r.End,
+		Start:    start.UTC().Format(common.SimpleTimeFormatWithTimezone),
+		End:      end.UTC().Format(common.SimpleTimeFormatWithTimezone),
 	}

-	var klineItem kline.Item
+	var klineItem *kline.Item
 	if r.UseDb {
 		klineItem, err = kline.LoadFromDatabase(r.Exchange,
 			pair,
@@ -2459,34 +2459,20 @@ func (s *RPCServer) GetHistoricCandles(ctx context.Context, r *gctrpc.GetHistori
 			interval,
 			start,
 			end)
-		if err != nil {
-			return nil, err
-		}
 	} else {
 		if r.ExRequest {
-			klineItem, err = exch.GetHistoricCandlesExtended(ctx,
-				pair,
-				a,
-				start,
-				end,
-				interval)
+			klineItem, err = exch.GetHistoricCandlesExtended(ctx, pair, a, interval, start, end)
 		} else {
-			klineItem, err = exch.GetHistoricCandles(ctx,
-				pair,
-				a,
-				start,
-				end,
-				interval)
+			klineItem, err = exch.GetHistoricCandles(ctx, pair, a, interval, start, end)
 		}
 	}
-
 	if err != nil {
 		return nil, err
 	}

 	if r.FillMissingWithTrades {
 		var tradeDataKline *kline.Item
-		tradeDataKline, err = fillMissingCandlesWithStoredTrades(start, end, &klineItem)
+		tradeDataKline, err = fillMissingCandlesWithStoredTrades(start, end, klineItem)
 		if err != nil {
 			return nil, err
 		}
@@ -2496,17 +2482,18 @@ func (s *RPCServer) GetHistoricCandles(ctx context.Context, r *gctrpc.GetHistori
 	resp.Exchange = klineItem.Exchange
 	for i := range klineItem.Candles {
 		resp.Candle = append(resp.Candle, &gctrpc.Candle{
-			Time:   klineItem.Candles[i].Time.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone),
-			Low:    klineItem.Candles[i].Low,
-			High:   klineItem.Candles[i].High,
-			Open:   klineItem.Candles[i].Open,
-			Close:  klineItem.Candles[i].Close,
-			Volume: klineItem.Candles[i].Volume,
+			Time:      klineItem.Candles[i].Time.UTC().Format(common.SimpleTimeFormatWithTimezone),
+			Low:       klineItem.Candles[i].Low,
+			High:      klineItem.Candles[i].High,
+			Open:      klineItem.Candles[i].Open,
+			Close:     klineItem.Candles[i].Close,
+			Volume:    klineItem.Candles[i].Volume,
+			IsPartial: klineItem.Candles[i].ValidationIssues == kline.PartialCandle,
 		})
 	}

 	if r.Sync && !r.UseDb {
-		_, err = kline.StoreInDatabase(&klineItem, r.Force)
+		_, err = kline.StoreInDatabase(klineItem, r.Force)
 		if err != nil {
 			if errors.Is(err, exchangeDB.ErrNoExchangeFound) {
 				return nil, errors.New("exchange was not found in database, you can seed existing data or insert a new exchange via the dbseed")
@@ -2533,7 +2520,7 @@ func fillMissingCandlesWithStoredTrades(startTime, endTime time.Time, klineItem
 		if ranges[i].HasDataInRange {
 			continue
 		}
-		var tradeCandles kline.Item
+		var tradeCandles *kline.Item
 		trades, err := trade.GetTradesInRange(
 			klineItem.Exchange,
 			klineItem.Asset.String(),
@@ -3260,17 +3247,16 @@ func (s *RPCServer) ConvertTradesToCandles(_ context.Context, r *gctrpc.ConvertT
 		return nil, err
 	}

-	var trades []trade.Data
-	trades, err = trade.GetTradesInRange(r.Exchange, r.AssetType, r.Pair.Base, r.Pair.Quote, start, end)
+	trades, err := trade.GetTradesInRange(r.Exchange, r.AssetType, r.Pair.Base, r.Pair.Quote, start, end)
 	if err != nil {
 		return nil, err
 	}
 	if len(trades) == 0 {
 		return nil, errNoTrades
 	}
+
 	interval := kline.Interval(r.TimeInterval)
-	var klineItem kline.Item
-	klineItem, err = trade.ConvertTradesToCandles(interval, trades...)
+	klineItem, err := trade.ConvertTradesToCandles(interval, trades...)
 	if err != nil {
 		return nil, err
 	}
@@ -3287,17 +3273,18 @@ func (s *RPCServer) ConvertTradesToCandles(_ context.Context, r *gctrpc.ConvertT
 	}
 	for i := range klineItem.Candles {
 		resp.Candle = append(resp.Candle, &gctrpc.Candle{
-			Time:   klineItem.Candles[i].Time.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone),
-			Low:    klineItem.Candles[i].Low,
-			High:   klineItem.Candles[i].High,
-			Open:   klineItem.Candles[i].Open,
-			Close:  klineItem.Candles[i].Close,
-			Volume: klineItem.Candles[i].Volume,
+			Time:      klineItem.Candles[i].Time.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone),
+			Low:       klineItem.Candles[i].Low,
+			High:      klineItem.Candles[i].High,
+			Open:      klineItem.Candles[i].Open,
+			Close:     klineItem.Candles[i].Close,
+			Volume:    klineItem.Candles[i].Volume,
+			IsPartial: klineItem.Candles[i].ValidationIssues == kline.PartialCandle,
 		})
 	}

 	if r.Sync {
-		_, err = kline.StoreInDatabase(&klineItem, r.Force)
+		_, err = kline.StoreInDatabase(klineItem, r.Force)
 		if err != nil {
 			return nil, err
 		}
@@ -5023,19 +5010,11 @@ func (s *RPCServer) GetTechnicalAnalysis(ctx context.Context, r *gctrpc.GetTechn
 		return nil, err
 	}

-	klineInterval := kline.Interval(r.Interval)
-
-	err = exch.GetBase().ValidateKline(pair, as, klineInterval)
-	if err != nil {
-		return nil, err
-	}
-
-	klines, err := exch.GetHistoricCandlesExtended(ctx,
-		pair,
+	klines, err := exch.GetHistoricCandlesExtended(ctx, pair,
 		as,
+		kline.Interval(r.Interval),
 		r.Start.AsTime(),
-		r.End.AsTime(),
-		klineInterval)
+		r.End.AsTime())
 	if err != nil {
 		return nil, err
 	}
@@ -5102,15 +5081,19 @@ func (s *RPCServer) GetTechnicalAnalysis(ctx context.Context, r *gctrpc.GetTechn
 			return nil, err
 		}

-		var otherKlines kline.Item
+		var otherKlines *kline.Item
 		otherKlines, err = otherExch.GetHistoricCandlesExtended(ctx,
-			otherPair, otherAs, r.Start.AsTime(), r.End.AsTime(), klineInterval)
+			otherPair,
+			otherAs,
+			kline.Interval(r.Interval),
+			r.Start.AsTime(),
+			r.End.AsTime())
 		if err != nil {
 			return nil, err
 		}

 		var correlation []float64
-		correlation, err = klines.GetCorrelationCoefficient(&otherKlines, r.Period)
+		correlation, err = klines.GetCorrelationCoefficient(otherKlines, r.Period)
 		if err != nil {
 			return nil, err
 		}
--- a/engine/rpcserver_test.go
+++ b/engine/rpcserver_test.go
@@ -49,6 +49,8 @@ const (
 	fakeExchangeName      = "fake"
 )

+var errExpectedTestError = errors.New("expected test error")
+
 // fExchange is a fake exchange with function overrides
 // we're not testing an actual exchange's implemented functions
 type fExchange struct {
@@ -139,8 +141,8 @@ func (f fExchange) GetFundingRates(ctx context.Context, request *order.FundingRa
 	}, nil
 }

-func (f fExchange) GetHistoricCandles(ctx context.Context, p currency.Pair, a asset.Item, timeStart, _ time.Time, interval kline.Interval) (kline.Item, error) {
-	return kline.Item{
+func (f fExchange) GetHistoricCandles(ctx context.Context, p currency.Pair, a asset.Item, interval kline.Interval, timeStart, _ time.Time) (*kline.Item, error) {
+	return &kline.Item{
 		Exchange: fakeExchangeName,
 		Pair:     p,
 		Asset:    a,
@@ -174,8 +176,11 @@ func generateCandles(amount int, timeStart time.Time, interval kline.Interval) [
 	return candy
 }

-func (f fExchange) GetHistoricCandlesExtended(ctx context.Context, p currency.Pair, a asset.Item, timeStart, _ time.Time, interval kline.Interval) (kline.Item, error) {
-	return kline.Item{
+func (f fExchange) GetHistoricCandlesExtended(ctx context.Context, p currency.Pair, a asset.Item, interval kline.Interval, timeStart, _ time.Time) (*kline.Item, error) {
+	if interval == 0 {
+		return nil, errExpectedTestError
+	}
+	return &kline.Item{
 		Exchange: fakeExchangeName,
 		Pair:     p,
 		Asset:    a,
@@ -2489,9 +2494,7 @@ func TestGetTechnicalAnalysis(t *testing.T) {
 		Enabled:      currency.Pairs{cp},
 	}

-	b.Features.Enabled.Kline.Intervals = map[string]bool{
-		kline.OneDay.Word(): true,
-	}
+	b.Features.Enabled.Kline.Intervals = kline.DeployExchangeIntervals(kline.OneDay)
 	em.Add(fExchange{IBotExchange: exch})
 	s := RPCServer{
 		Engine: &Engine{
@@ -2520,8 +2523,8 @@ func TestGetTechnicalAnalysis(t *testing.T) {
 		AssetType: "upsideprofitcontract",
 		Pair:      &gctrpc.CurrencyPair{},
 	})
-	if !errors.Is(err, kline.ErrValidatingParams) {
-		t.Fatalf("received: '%v' but expected: '%v'", err, kline.ErrValidatingParams)
+	if !errors.Is(err, errExpectedTestError) {
+		t.Fatalf("received: '%v' but expected: '%v'", err, errExpectedTestError)
 	}

 	_, err = s.GetTechnicalAnalysis(context.Background(), &gctrpc.GetTechnicalAnalysisRequest{