fix buildAggSeries for custom series

Author: anyongjin

data/common.go

@@ -262,7 +262,7 @@ func (j *PairTFCache) fillLacks(pair string, subTfSecs int, startMS, endMS int64
 	j.WaitBar = nil
 	if len(preRows) > 0 {
 		fromTFMS := int64(subTfSecs * 1000)
-		oldBars, _, err := buildOHLCVSeries(exs, j.TimeFrame, preRows, tfMSecs, 0, nil, fromTFMS, j.AlignOffMS, false)
+		oldBars, _, err := buildAggSeries(exs, j.TimeFrame, preRows, tfMSecs, 0, nil, fromTFMS, j.AlignOffMS, false)
 		if err != nil {
 			return nil, err
 		}

data/feeder.go

@@ -311,25 +311,16 @@ func applyAdjSeriesList(exs *orm.ExSymbol, adjs []*orm.AdjInfo, rows []*orm.Data
 	return orm.BarsToSeries(exs, rows[0].TimeFrame, orm.ApplyAdj(adjs, bars, adjMode, cutEnd, limit), nil, rows[0].IsWarmUp, true)
 }
 
-func buildOHLCVSeries(exs *orm.ExSymbol, tf string, rows []*orm.DataSeries, toTFMSecs int64, preFire float64, prev []*orm.DataSeries, fromTFMS, offMS int64, opts ...bool) ([]*orm.DataSeries, bool, *errs.Error) {
+func buildAggSeries(exs *orm.ExSymbol, tf string, rows []*orm.DataSeries, toTFMSecs int64, preFire float64, prev []*orm.DataSeries, fromTFMS, offMS int64, opts ...bool) ([]*orm.DataSeries, bool, *errs.Error) {
 	isWarmUp := false
 	if len(opts) > 0 {
 		isWarmUp = opts[len(opts)-1]
 	}
-	bars, err := orm.SeriesToBars(exs, rows)
-	if err != nil {
-		return nil, false, err
-	}
-	prevBars, err := orm.SeriesToBars(exs, prev)
+	aggRows, lastOk, err := orm.ResampleDataSeries(exs, tf, rows, prev, toTFMSecs, preFire, fromTFMS, offMS, isWarmUp)
 	if err != nil {
-		return nil, false, err
-	}
-	aggBars, lastOk := utils.BuildOHLCV(bars, toTFMSecs, preFire, prevBars, fromTFMS, offMS)
-	targetTF := tf
-	if targetTF == "" {
-		targetTF = utils2.SecsToTF(int(toTFMSecs / 1000))
+		return nil, false, errs.New(core.ErrInvalidBars, err)
 	}
-	return orm.BarsToSeries(exs, targetTF, aggBars, nil, isWarmUp, true), lastOk, nil
+	return aggRows, lastOk, nil
 }
 
 type IDataFeeder interface {
@@ -551,7 +542,7 @@ func (f *SeriesFeeder) onNewData(barTfMSecs int64, rows []*orm.DataSeries) (bool
 		if state.WaitBar != nil {
 			olds = append(olds, state.WaitBar)
 		}
-		ohlcvs, lastOk, err = buildOHLCVSeries(f.ExSymbol, state.TimeFrame, rows, staMSecs, f.PreFire, olds, barTfMSecs, state.AlignOffMS, false, f.isWarmUp)
+		ohlcvs, lastOk, err = buildAggSeries(f.ExSymbol, state.TimeFrame, rows, staMSecs, f.PreFire, olds, barTfMSecs, state.AlignOffMS, false, f.isWarmUp)
 		if err != nil {
 			return false, err
 		}
@@ -584,7 +575,7 @@ func (f *SeriesFeeder) onNewData(barTfMSecs int64, rows []*orm.DataSeries) (bool
 		if barTfMSecs < staMSecs {
 			// The last unfinished data should be kept here
 			// 这里应该保留最后未完成的数据
-			ohlcvs, _, err = buildOHLCVSeries(f.ExSymbol, state.TimeFrame, rows, staMSecs, f.PreFire, nil, barTfMSecs, state.AlignOffMS, false, f.isWarmUp)
+			ohlcvs, _, err = buildAggSeries(f.ExSymbol, state.TimeFrame, rows, staMSecs, f.PreFire, nil, barTfMSecs, state.AlignOffMS, false, f.isWarmUp)
 			if err != nil {
 				return false, err
 			}
@@ -612,7 +603,7 @@ func (f *SeriesFeeder) onNewData(barTfMSecs int64, rows []*orm.DataSeries) (bool
 				olds = append(olds, state.WaitBar)
 			}
 			bigTfMSecs := int64(state.TFSecs * 1000)
-			curOhlcvs, lastDone, err := buildOHLCVSeries(f.ExSymbol, state.TimeFrame, curRows, bigTfMSecs, f.PreFire, olds, srcMSecs, srcAlignOff, false, f.isWarmUp)
+			curOhlcvs, lastDone, err := buildAggSeries(f.ExSymbol, state.TimeFrame, curRows, bigTfMSecs, f.PreFire, olds, srcMSecs, srcAlignOff, false, f.isWarmUp)
 			if err != nil {
 				return false, err
 			}

data/watcher.go

@@ -273,7 +273,7 @@ func (w *SeriesWatcher) onSpiderBar(raw *utils.IOMsgRaw) {
 			olds = append(olds, job.WaitBar)
 		}
 		jobMSecs := int64(job.TFSecs * 1000)
-		aggRows, lastDone, err := buildOHLCVSeries(orm.GetExSymbol2(exgName, market, pair), utils2.SecsToTF(job.TFSecs), curRows, jobMSecs, 0, olds, tfMSecs, job.AlignOffMS, false)
+		aggRows, lastDone, err := buildAggSeries(orm.GetExSymbol2(exgName, market, pair), utils2.SecsToTF(job.TFSecs), curRows, jobMSecs, 0, olds, tfMSecs, job.AlignOffMS, false)
 		if err != nil {
 			log.Error("build series fail", zap.String("pair", pair), zap.Error(err))
 			return

orm/series.go

@@ -2,6 +2,7 @@ package orm
 
 import (
 	"fmt"
+	"sort"
 	"strconv"
 
 	"github.com/banbox/banexg"
@@ -342,10 +343,7 @@ func ResampleSeriesRecords(info *SeriesInfo, exs *ExSymbol, rows []*DataRecord,
 		}
 		values := make(map[string]any, len(fields))
 		for _, field := range fields {
-			fn, ok := GetAggRuleFunc("last")
-			if exs != nil {
-				fn, ok = GetAggRuleFunc(exs.AggRule(field.Name))
-			}
+			fn, ok := GetAggRuleFunc(seriesAggRule(info, exs, field.Name))
 			if !ok {
 				fn, _ = GetAggRuleFunc("last")
 			}
@@ -386,6 +384,109 @@ func ResampleSeriesRecords(info *SeriesInfo, exs *ExSymbol, rows []*DataRecord,
 	return out, nil
 }
 
+func ResampleDataSeries(exs *ExSymbol, tf string, rows, prev []*DataSeries, toTFMS int64, preFire float64, fromTFMS, offMS int64, isWarmUp bool) ([]*DataSeries, bool, error) {
+	if len(rows) == 0 {
+		return nil, false, nil
+	}
+	if tf == "" {
+		tf = utils2.SecsToTF(int(toTFMS / 1000))
+	}
+	source := NormalizeSeriesSource(rows[0].Source)
+	info := NewSeriesInfo(source, tf, inferSeriesFields(prev, rows))
+	records := make([]*DataRecord, 0, len(prev)+len(rows))
+	for _, row := range prev {
+		if rec := SeriesToRecord(row); rec != nil {
+			records = append(records, rec)
+		}
+	}
+	for _, row := range rows {
+		if rec := SeriesToRecord(row); rec != nil {
+			records = append(records, rec)
+		}
+	}
+	offsetMS := int64(float64(toTFMS)*preFire) + offMS
+	aggRecords, err := ResampleSeriesRecords(info, exs, records, toTFMS, offsetMS)
+	if err != nil {
+		return nil, false, err
+	}
+	out := RecordsToSeries(info, exs, aggRecords)
+	for _, row := range out {
+		row.IsWarmUp = isWarmUp
+	}
+	lastFinished := false
+	if fromTFMS > 0 && len(out) > 0 {
+		_, offset := utils2.GetTfAlignOrigin(int(toTFMS / 1000))
+		alignOffMS := int64(offset * 1000)
+		finishMS := utils2.AlignTfMSecsOffset(rows[len(rows)-1].TimeMS+fromTFMS+offsetMS, toTFMS, alignOffMS)
+		lastFinished = finishMS > out[len(out)-1].TimeMS
+	}
+	return out, lastFinished, nil
+}
+
+func inferSeriesFields(groups ...[]*DataSeries) []SeriesField {
+	seen := make(map[string]any)
+	for _, rows := range groups {
+		for _, row := range rows {
+			if row == nil {
+				continue
+			}
+			for key, val := range row.Values {
+				if _, ok := seen[key]; !ok {
+					seen[key] = val
+				}
+			}
+		}
+	}
+	keys := make([]string, 0, len(seen))
+	for key := range seen {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	fields := make([]SeriesField, 0, len(keys))
+	for _, key := range keys {
+		fields = append(fields, SeriesField{Name: key, Type: inferSeriesFieldType(seen[key])})
+	}
+	return fields
+}
+
+func inferSeriesFieldType(val any) string {
+	switch val.(type) {
+	case bool:
+		return "bool"
+	case string:
+		return "string"
+	case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64:
+		return "int"
+	case float32, float64:
+		return "float"
+	default:
+		return "json"
+	}
+}
+
+func seriesAggRule(info *SeriesInfo, exs *ExSymbol, field string) string {
+	if exs != nil {
+		if rule, ok := exs.AggRuleMap()[field]; ok {
+			return rule
+		}
+	}
+	if info != nil && NormalizeSeriesSource(info.Name) == SeriesSourceKline {
+		switch field {
+		case "open":
+			return "first"
+		case "high":
+			return "max"
+		case "low":
+			return "min"
+		case "close":
+			return "last"
+		case "volume", "quote", "buy_volume", "trade_num":
+			return "sum"
+		}
+	}
+	return "last"
+}
+
 func seriesFloatValue(values map[string]any, key string) (float64, error) {
 	if values == nil {
 		return 0, fmt.Errorf("series event missing field %q", key)

orm/series_test.go

@@ -111,6 +111,28 @@ func TestResampleSeriesRecordsUsesAggRules(t *testing.T) {
 	}
 }
 
+func TestResampleDataSeriesUsesAggRulesForGenericSeries(t *testing.T) {
+	exs := &ExSymbol{ID: 3, AggRules: `{"rate":"avg","volume":"sum"}`}
+	rows := []*DataSeries{
+		{Source: "funding", Sid: 3, TimeMS: 1_700_000_040_000, EndMS: 1_700_000_100_000, Closed: true, Values: map[string]any{"rate": 1.0, "volume": 2.0}},
+		{Source: "funding", Sid: 3, TimeMS: 1_700_000_100_000, EndMS: 1_700_000_160_000, Closed: true, Values: map[string]any{"rate": 3.0, "volume": 5.0}},
+	}
+
+	got, done, err := ResampleDataSeries(exs, "2m", rows, nil, 120_000, 0, 60_000, 0, true)
+	if err != nil {
+		t.Fatalf("ResampleDataSeries returned error: %v", err)
+	}
+	if !done || len(got) != 1 {
+		t.Fatalf("expected one finished row, done=%v len=%d", done, len(got))
+	}
+	if got[0].Source != "funding" || got[0].TimeFrame != "2m" || !got[0].IsWarmUp {
+		t.Fatalf("unexpected series header: %+v", got[0])
+	}
+	if got[0].Values["rate"] != 2.0 || got[0].Values["volume"] != 7.0 {
+		t.Fatalf("unexpected values: %+v", got[0].Values)
+	}
+}
+
 func testPairTFKline(symbol, tf string, ts int64) *banexg.PairTFKline {
 	return &banexg.PairTFKline{
 		Kline:     banexg.Kline{Time: ts},