diff --git a/internal/exchange/exchange_data_persist.go b/internal/exchange/exchange_data_persist.go index 4d717cd..7edaafd 100644 --- a/internal/exchange/exchange_data_persist.go +++ b/internal/exchange/exchange_data_persist.go @@ -7,6 +7,7 @@ import ( "sig-pub/pkg/storage/kvrocks" vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" "sig-pub/pkg/types" + "sig-pub/pkg/utils/collect" ) type ExchangeDataPersist struct { @@ -52,5 +53,6 @@ func (p *ExchangeDataPersist) SaveHistoryKlineMarkTs(exchange pb.ExchangeType, i // ListKline 查询k线列表 func (p *ExchangeDataPersist) ListKline(inst types.TradeInstance, interval types.Interval, start, end int64) (klines []*types.Kline, err error) { klines, err = p.vmtsdb.ListRangeKline(inst, interval, start, end) + collect.Reverse(klines) return } diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index 4ee8329..1806bca 100644 --- a/internal/exchange/exchange_service.go +++ b/internal/exchange/exchange_service.go @@ -12,7 +12,6 @@ import ( "sig-pub/pkg/utils/collect" "sig-pub/pkg/utils/times" "sig-pub/pkg/zlog" - "sort" "sync" "sync/atomic" "time" @@ -136,11 +135,8 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch if len(channelK.Klines) == 0 { continue } - if len(channelK.Klines) > 1 { - sort.Slice(channelK.Klines, func(i, j int) bool { - return channelK.Klines[i].Ts < channelK.Klines[j].Ts - }) - } + // 排序后取出头尾k线 + collect.SortAsc(channelK.Klines, func(k *types.Kline) int64 { return k.Ts }) firstKline, lastKline := channelK.Klines[0], channelK.Klines[len(channelK.Klines)-1] // 交易所 instid 转 sig-instid diff --git a/internal/indicator/indicator.go b/internal/indicator/indicator.go index d1463fe..a03d0cb 100644 --- a/internal/indicator/indicator.go +++ b/internal/indicator/indicator.go @@ -1,20 +1,48 @@ package indicator +import ( + vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" + "sync" +) + // load indicator plugin // 热指标 自动加载/实时更新/内存缓存 // 历史指标 实时计算 // 自定义插件化指标 - type IndicatorService struct { + vmdb vmts.VictoriaMetricsTSDB } -func NewIndicatorService() *IndicatorService { - return &IndicatorService{} +func NewIndicatorService(vmdb vmts.VictoriaMetricsTSDB) *IndicatorService { + return &IndicatorService{ + vmdb: vmdb, + } } // 加载热指标 // 订阅k线数据 更新指标 func (svc *IndicatorService) Init() { + // Regist(RSI) + // if Indicators[RSI] +} + +// 注册 +func (svc *IndicatorService) Register(indicatorName string, indicatorInterface any) { + +} + +// Indicator 获取指标 +// key: 指标名称-指标参数hash +// args: 策略参数(柯里化), 如 window, 6, round, 14... +func (svc *IndicatorService) Indicator(indicatorName string, args ...any) { + // 调用方: strategy, webview, other indicator + // strategy: IsCross(RSI6[10:1], RSI14[10:1])... + // webview: RSI6(stime, etime, interval) + // other indicator: RSI6[10] * MACD[10] +} +// kline -> indicator root -> /internal/force_flush -> all indicators (wg concurrent) -> all strategy (concurrent) +type Ind struct { + LiveMu sync.RWMutex // 实时k线锁, k线更新后指标更新时写锁 } diff --git a/internal/indicator/rsi.go b/internal/indicator/rsi.go new file mode 100644 index 0000000..73b09f2 --- /dev/null +++ b/internal/indicator/rsi.go @@ -0,0 +1,35 @@ +package indicator + +import ( + "fmt" + "sig-pub/api/pb" + "sig-pub/pkg/types" +) + +type RSI struct { + // types.IntervalWindow + window int // 窗口大小 +} + +func (ind *RSI) IntervalWindow() { + // 指标参数注入 +} + +func (ind *RSI) QueryRange(exchange pb.ExchangeType, instId string, interval types.Interval, rsi int) (query string, err error) { + var r types.MeticMatrix + _ = r + intervalAdder, ok := types.SupportedIntervals[interval] + if !ok { + err = fmt.Errorf("unsupport interval %s", interval) + return + } + minutes := intervalAdder(0, int64(rsi)) / 1000 / 60 + + query = fmt.Sprintf(` + 100 - 100 / (1 + ( + avg_over_time(clamp_min(delta(%s{kind="close", interval="%s", exchange="%s"}), 0)[%dm]) / + avg_over_time(abs(clamp_max(delta(%s{kind="close", interval="%s", exchange="%s"}), 0))[%dm]) + )) + `, instId, interval, exchange, minutes, instId, interval, exchange, minutes) + return +} diff --git a/pkg/aside/trade_instance_client.go b/pkg/aside/trade_instance_client.go index aaf24d9..342284e 100644 --- a/pkg/aside/trade_instance_client.go +++ b/pkg/aside/trade_instance_client.go @@ -6,6 +6,8 @@ import ( "sig-pub/pkg/data/entity" "sig-pub/pkg/mapping" "sig-pub/pkg/utils/kvcache" + "sig-pub/pkg/utils/retry" + "sig-pub/pkg/zlog" "time" cache "github.com/fanjindong/go-cache" @@ -62,10 +64,12 @@ func (c *TradeInstanceAside) getTradeInstance0(ctx context.Context, instId strin // ListExchangeTradeInstance 获取交易所支持的交易实例 func (c *TradeInstanceAside) ListExchangeTradeInstance(ctx context.Context, exchange pb.ExchangeType) (marketInsts []*pb.MarketTradeInstance, err error) { - rsp, err := c.marketClient.ListMarketTradeInstance(ctx, &pb.ReqListMarketTradeInstance{Exchange: exchange}) - if err != nil { - return - } - marketInsts = rsp.ExchangeInsts - return + return retry.DoWithStepDelay(10, time.Second, func(retryTimes uint32) ([]*pb.MarketTradeInstance, error) { + rsp, err := c.marketClient.ListMarketTradeInstance(ctx, &pb.ReqListMarketTradeInstance{Exchange: exchange}) + if err != nil { + zlog.Errorf("list market trade instances error: retry %d times, %v", retryTimes, err) + return nil, err + } + return rsp.ExchangeInsts, nil + }) } diff --git a/pkg/storage/tsdb/victoria_metrics/data.go b/pkg/storage/tsdb/victoria_metrics/data.go new file mode 100644 index 0000000..91af393 --- /dev/null +++ b/pkg/storage/tsdb/victoria_metrics/data.go @@ -0,0 +1,36 @@ +package vmts + +import "github.com/govalues/decimal" + +type VMMetricKline struct { + Metric struct { + Name string `json:"__name__"` + Interval string `json:"interval"` + Kind string `json:"kind"` + } `json:"metric"` + Values []decimal.Decimal `json:"values"` + Timestamps []int64 `json:"timestamps"` +} + +const ( + QueryStatusSuccess = "success" + QueryStatusError = "error" +) + +type ResponseQuery struct { + Status string `json:"status"` // success, error + ErrorType string `json:"errorType"` // 422 + Error string `json:"error"` + Stats struct { + SeriesFetched string `json:"seriesFetched"` + ExecutionTimeMsec int `json:"executionTimeMsec"` + } `json:"stats"` + Data struct { + ResultType string `json:"resultType"` // matrix, vector + Result []struct { + Metric map[string]string `json:"metric"` // {__name__: 'BTC_USDT', exchange: 'OKX', interval: '5m', kind: 'close'} + Values [][]any `json:"values"` // data.type = matrix [[1759975200, '122407.82'], [1759975500, '122385.92']] + Value []any `json:"value"` // data.type = vector [1759975200, '122407.82'] + } `json:"result"` + } `json:"data"` +} diff --git a/pkg/storage/tsdb/victoria_metrics/vm.go b/pkg/storage/tsdb/victoria_metrics/vm.go index b893799..6553fd6 100644 --- a/pkg/storage/tsdb/victoria_metrics/vm.go +++ b/pkg/storage/tsdb/victoria_metrics/vm.go @@ -12,7 +12,9 @@ import ( "sig-pub/pkg/config" "sig-pub/pkg/types" "sig-pub/pkg/utils/collect" + "sig-pub/pkg/utils/conver" "sig-pub/pkg/zlog" + "strconv" "github.com/bytedance/sonic" "github.com/govalues/decimal" @@ -86,7 +88,7 @@ func compressData(data []byte) ([]byte, error) { return b.Bytes(), nil } -// 获取原始k线列表 [end, ..., start] +// 获取原始k线列表 [start, ..., end] func (vm *VictoriaMetricsTSDB) ListRangeKline(inst types.TradeInstance, interval types.Interval, start, end int64) (klines []*types.Kline, err error) { defer func() { if r := recover(); r != nil { @@ -183,19 +185,90 @@ func (vm *VictoriaMetricsTSDB) ListRangeKline(inst types.TradeInstance, interval } // zlog.Infof("response body line: %#v", vmMetric) } - collect.Reverse(klines) // for _, kline := range klines { // zlog.Infof("kline: %#v", kline) // } return } -type VMMetricKline struct { - Metric struct { - Name string `json:"__name__"` - Interval string `json:"interval"` - Kind string `json:"kind"` - } `json:"metric"` - Values []decimal.Decimal `json:"values"` - Timestamps []int64 `json:"timestamps"` +// 数据延时 强制刷盘 +// https://www.victoriametrics.com.cn/docs/query/#latency +func (vm *VictoriaMetricsTSDB) ForceFlush() (err error) { + resp, err := http.Get(fmt.Sprintf("%s/internal/force_flush", vm.addr)) + if err != nil { + return + } + resp.Body.Close() + return +} + +// https://www.victoriametrics.com.cn/docs/query/#instant-query +func (vm *VictoriaMetricsTSDB) QueryInstant(time int64, step types.Interval) { + +} + +// QueryRange 查询结果 query_range 接口调用 [start, ..., end] +// https://www.victoriametrics.com.cn/docs/query/#range-query +func (vm *VictoriaMetricsTSDB) QueryRange(start, end int64, step types.Interval, query string) (meticMatrixs []*types.MeticMatrix, err error) { + params := fmt.Sprintf("latency_offset=0&start=%d&end=%d&step=%s&query=%s", start, end, step, url.QueryEscape(query)) + resp, err := http.Get(fmt.Sprintf("%s/api/v1/query_range?%s", vm.addr, params)) + if err != nil { + return + } + defer func() { + if err := resp.Body.Close(); err != nil { + zlog.Error("close vmtsdb query_range response error:", err) + } + }() + + bytes, err := io.ReadAll(resp.Body) + if err != nil { + return + } + var r = new(ResponseQuery) + if err = sonic.Unmarshal(bytes, r); err != nil { + return + } + + if r.Status != QueryStatusSuccess { + err = fmt.Errorf("query_range response error: errorType=%s, error=%s", r.ErrorType, r.Error) + return + } + + if collect.NotIn(r.Data.ResultType, "vector", "matrix") { + err = fmt.Errorf("query_range ResultType error: %s", r.Data.ResultType) + return + } + + for _, result := range r.Data.Result { + m := &types.MeticMatrix{ + Name: result.Metric["__name__"], + Exchange: result.Metric["exchange"], + Interval: result.Metric["interval"], + Kind: result.Metric["kind"], + } + + var values [][]any + if r.Data.ResultType == "vector" { + values = [][]any{result.Value} + } + if r.Data.ResultType == "matrix" { + values = result.Values + } + + m.Timestamps = make([]int64, 0, len(values)) + m.Values = make([]float64, 0, len(values)) + for _, value := range values { + ts := int64(conver.ToFloat64(value[0]) * 1000) + v, e := strconv.ParseFloat(value[1].(string), 64) + if e != nil { + err = e + return + } + m.Timestamps = append(m.Timestamps, ts) + m.Values = append(m.Values, v) + } + meticMatrixs = append(meticMatrixs, m) + } + return } diff --git a/pkg/storage/tsdb/victoria_metrics/vm_test.go b/pkg/storage/tsdb/victoria_metrics/vm_test.go index 5f4f00e..3c19e4f 100644 --- a/pkg/storage/tsdb/victoria_metrics/vm_test.go +++ b/pkg/storage/tsdb/victoria_metrics/vm_test.go @@ -4,14 +4,18 @@ import ( "sig-pub/api/pb" "sig-pub/pkg/config" "sig-pub/pkg/types" + "sig-pub/pkg/utils/collect" "sig-pub/pkg/zlog" "testing" + "time" ) -var test_addr = "http://127.0.0.1:8428" +func getTestVMDB() *VictoriaMetricsTSDB { + return NewVictoriaMetricsTSDB(config.VictoriaMetricsConfig{Addr: "http://127.0.0.1:8428"}) +} func TestGetRangeKline(t *testing.T) { - vmdb := NewVictoriaMetricsTSDB(config.VictoriaMetricsConfig{Addr: test_addr}) + vmdb := getTestVMDB() inst := types.TradeInstance{ InstId: "BTC_USDT", Exchange: pb.ExchangeType_OKX, @@ -22,7 +26,37 @@ func TestGetRangeKline(t *testing.T) { t.Error(err) return } + collect.Reverse(klines) for _, k := range klines { zlog.Infof("%#v", k) } } + +func TestQueryRange(t *testing.T) { + vmdb := getTestVMDB() + ts1 := time.Now().UnixMilli() + // 计算MA5 + ma5, err := vmdb.QueryRange(1759975200000, 1760061600000, types.Interval5m, ` + avg_over_time(BTC_USDT{kind="close", interval="5m", exchange="OKX"}[25m]) + `) + if err != nil { + t.Error(err) + return + } + ts2 := time.Now().UnixMilli() + t.Logf("MA5 use %dms: %#v\n", ts2-ts1, ma5) + + // 计算RSI6 + rsi6, err := vmdb.QueryRange(1759975200000, 1760061600000, types.Interval5m, ` + 100 - 100 / (1 + ( + avg_over_time(clamp_min(delta(BTC_USDT{kind="close", interval="5m", exchange="OKX"}), 0)[30m]) / + avg_over_time(abs(clamp_max(delta(BTC_USDT{kind="close", interval="5m", exchange="OKX"}), 0))[30m]) + )) + `) + if err != nil { + t.Error(err) + return + } + ts3 := time.Now().UnixMilli() + t.Logf("RSI6 use %dms: %#v\n", ts3-ts2, rsi6) +} diff --git a/pkg/types/indicator.go b/pkg/types/indicator.go new file mode 100644 index 0000000..c6b6db3 --- /dev/null +++ b/pkg/types/indicator.go @@ -0,0 +1,11 @@ +package types + +// MeticMatrix 指标向量矩阵 +type MeticMatrix struct { + Name string `json:"name"` // 指标名称 + Exchange string `json:"exchange"` + Interval string `json:"interval"` + Kind string `json:"kind"` + Timestamps []int64 `json:"timestamps"` // len(timestamps) == len(values) + Values []float64 `json:"values"` +} diff --git a/pkg/types/interval.go b/pkg/types/interval.go index 70e8c56..fa6c2f5 100644 --- a/pkg/types/interval.go +++ b/pkg/types/interval.go @@ -57,6 +57,9 @@ type IntervalMap map[Interval]IntervalAdder type IntervalAdder func(ts, mul int64) (ret int64) +// todo 支持其他周期方案: +// 1.订阅1m实时k线和拉取历史k线时进行聚合 +// 2.PromQL: open/close -> first_over_time/last_over_time, vol->sum_over_time var SupportedIntervals = IntervalMap{ // Interval1s: func(ts, mul int64) (ret int64) { return ts + (1000 * mul) }, Interval1m: func(ts, mul int64) (ret int64) { return ts + (1 * 60 * 1000 * mul) }, diff --git a/pkg/utils/collect/collect.go b/pkg/utils/collect/collect.go index f5646ac..95db500 100644 --- a/pkg/utils/collect/collect.go +++ b/pkg/utils/collect/collect.go @@ -40,6 +40,26 @@ func Filter[T any](slice []T, predicate func(int, T) bool) []T { return res } +// SortAsc 切片升序排序 +func SortAsc[T any, C cmp.Ordered](slice []T, compare func(T) C) { + if len(slice) < 2 { + return + } + sort.Slice(slice, func(i, j int) bool { + return compare(slice[i]) < compare(slice[j]) + }) +} + +// SortDesc 切片降序排序 +func SortDesc[T any, C cmp.Ordered](slice []T, compare func(T) C) { + if len(slice) < 2 { + return + } + sort.Slice(slice, func(i, j int) bool { + return compare(slice[i]) > compare(slice[j]) + }) +} + // Tail 获取切片尾部元素 // dv: 空切片默认值 func Tail[T any](slice []T, dv T) T { diff --git a/pkg/utils/conver/cast.go b/pkg/utils/conver/cast.go new file mode 100644 index 0000000..ad59b90 --- /dev/null +++ b/pkg/utils/conver/cast.go @@ -0,0 +1,164 @@ +package conver + +import ( + "math" + "strconv" +) + +func ToInt64(inter any, defaultVal ...int64) int64 { + var def int64 = 0 + if len(defaultVal) > 0 { + def = defaultVal[0] + } + + switch v := inter.(type) { + default: + return def + case int: + return int64(v) + case int8: + return int64(v) + case int16: + return int64(v) + case int32: + return int64(v) + case int64: + return v + case uint8: + return int64(v) + case uint16: + return int64(v) + case uint32: + return int64(v) + case uint64: + return int64(v) + case float32: + return int64(v) + case float64: + if math.IsNaN(v) { + return def + } + return int64(v) + case bool: + if v { + return 1 + } + return 0 + case string: + a, e := strconv.Atoi(v) + if e != nil { + return def + } + return int64(a) + case *uint64: + if v == nil { + return def + } + return int64(*v) + case *int64: + if v == nil { + return def + } + return *v + } +} + +func ToFloat64(inter any, defaultVal ...float64) (r float64) { + var def float64 = 0 + if len(defaultVal) > 0 { + def = defaultVal[0] + } + defer func() { + if math.IsNaN(r) { + r = def + } + }() + + switch v := inter.(type) { + default: + return def + case int: + return float64(v) + case int8: + return float64(v) + case int16: + return float64(v) + case int32: + return float64(v) + case int64: + return float64(v) + case uint8: + return float64(v) + case uint16: + return float64(v) + case uint32: + return float64(v) + case uint64: + return float64(v) + case float32: + return float64(v) + case float64: + return float64(v) + case *float64: + if v == nil { + return def + } + return *v + case *int64: + if v == nil { + return def + } + return float64(*v) + case string: + a, e := strconv.ParseFloat(v, 64) + if e != nil { + return def + } + return a + } +} +func ToBool(inter any, defaultVal ...bool) bool { + var def bool + if len(defaultVal) > 0 { + def = defaultVal[0] + } + + switch v := inter.(type) { + default: + return def + case bool: + return v + case int: + return v > 0 + case int8: + return v > 0 + case int16: + return v > 0 + case int32: + return v > 0 + case int64: + return v > 0 + case uint8: + return v > 0 + case uint16: + return v > 0 + case uint32: + return v > 0 + case uint64: + return v > 0 + case float32: + return v > 0 + case float64: + return v > 0 + case string: + a, e := strconv.ParseBool(v) + if e == nil { + return a + } + b, e := strconv.Atoi(v) + if e != nil { + return def + } + return b > 0 + } +} diff --git a/pkg/utils/retry/retry.go b/pkg/utils/retry/retry.go new file mode 100644 index 0000000..0900e2e --- /dev/null +++ b/pkg/utils/retry/retry.go @@ -0,0 +1,37 @@ +package retry + +import ( + "time" +) + +// DoWithFixDelay 根据固定延迟重试函数 +func DoWithFixDelay[T any](maxRetryTimes uint32, delay time.Duration, handler func(retryTimes uint32) (T, error)) (r T, lastErr error) { + for retry := range maxRetryTimes { + if retry > 0 { + time.Sleep(delay) + } + r, err := handler(retry) + if err != nil { + lastErr = err + continue + } + return r, nil + } + return +} + +// DoWithStepDelay 根据步进延迟重试函数 1s, 2s, 4s... +func DoWithStepDelay[T any](maxRetryTimes uint32, initialDelay time.Duration, handler func(retryTimes uint32) (T, error)) (r T, lastErr error) { + for retry := range maxRetryTimes { + if retry > 0 { + time.Sleep(time.Duration(1<<(retry-1)) * initialDelay) + } + r, err := handler(retry) + if err != nil { + lastErr = err + continue + } + return r, nil + } + return +} diff --git a/pkg/utils/retry/retry_test.go b/pkg/utils/retry/retry_test.go new file mode 100644 index 0000000..46f389b --- /dev/null +++ b/pkg/utils/retry/retry_test.go @@ -0,0 +1,25 @@ +package retry + +import ( + "fmt" + "testing" + "time" +) + +func TestRetryStep(t *testing.T) { + t.Logf("start retry: unix=%d", time.Now().Unix()) + r, err := DoWithStepDelay(5, time.Second, func(retryTimes uint32) (uint32, error) { + t.Logf("retry %d times, unix=%d", retryTimes, time.Now().Unix()) + if retryTimes < 4 { + return retryTimes, fmt.Errorf("test error: %d", retryTimes) + } + return retryTimes, nil + }) + if err != nil { + t.Error(err) + return + } + if r != 4 { + t.Errorf("error result: %d", r) + } +}