init kline

cmd/exchange/exchange_test.go

@@ -47,7 +47,7 @@ func TestExchange(t *testing.T) {
 			klines = klines[:0]
 		}
 
-		zlog.Infof("channel: %s, instId: %s, datas: %#v", candle.InstId, candle.Exchange, candle.Klines[0])
+		zlog.Infof("channel: %s, instId: %s, datas: %#v", candle.ExgInstId, candle.Exchange, candle.Klines[0])
 	}
 }
 

config/config.toml

@@ -59,7 +59,7 @@ receiveBuffer = 4096
 marketSubscribeLimit = 16
 consumeBatch = 1024
 consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume
-httpProxy = "http://192.168.1.5:7890"
+httpProxy = "http://192.168.1.6:7890"
 
 # 模拟盘API交易地址如下：
 # REST：https://www.okx.com

docker-compose.yml

@@ -27,14 +27,12 @@ services:
     container_name: sig-kvrocks
     hostname: sig-kvrocks
     user: 'root'
+    network_mode: host
     # restart: always
-    sysctls:
-      net.core.somaxconn: 1024
     volumes:
       - "/etc/localtime:/etc/localtime:ro"
+      - "./config/kvrocks/kvrocks.conf:/var/lib/kvrocks/kvrocks.conf:ro"
       - "./fs/kvrocks_data:/var/lib/kvrocks"
-    ports:
-      - '7666:6666'
     command: --bind 0.0.0.0 --dir /var/lib/kvrocks
   sig-mysql:
     container_name: sig-mysql

internal/exchange/exchange.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sig-pub/pkg/types"
-	"sync"
+	"sig-pub/pkg/utils/collect"
 )
 
 // 交易所行情数据订阅
@@ -35,8 +35,16 @@ type Exchange struct {
 	ExType     types.Exchange
 	Fetcher    ExchangeFetcher
 	Subscriber ExchangeSubscriber
-	Insts      map[string]*types.TradeInstance // <ExchangeInstId, Inst>
+	Insts      *collect.SyncMap[string, *ExchangeTradeInstance] // map[string]*ExchangeTradeInstance // <ExchangeInstId, Inst>
-	sync.RWMutex
+	// sync.RWMutex
+}
+
+// 交易所交易产品
+type ExchangeTradeInstance struct {
+	Inst          *types.TradeInstance
+	Status        int32 // 交易产品状态, 0.初始化中 1.正常
+	LiveMarkTs    int64 // websocket订阅k线标记时间戳
+	HistoryMarkTs int64 // 拉取历史k线标记时间戳
 }
 
 func NewExchange(fetcher ExchangeFetcher, subscriber ExchangeSubscriber) *Exchange {
@@ -49,6 +57,6 @@ func NewExchange(fetcher ExchangeFetcher, subscriber ExchangeSubscriber) *Exchan
 		ExType:     exType,
 		Fetcher:    fetcher,
 		Subscriber: subscriber,
-		Insts:      make(map[string]*types.TradeInstance),
+		Insts:      collect.NewSyncMap[string, *ExchangeTradeInstance](),
 	}
 }

internal/exchange/exchange_grpc_server.go

@@ -9,9 +9,11 @@ import (
 	"sig-pub/pkg/data"
 	"sig-pub/pkg/storage/kvrocks"
 	"sig-pub/pkg/types"
+	"sig-pub/pkg/utils/times"
 	"sig-pub/pkg/zlog"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"google.golang.org/grpc"
 )
@@ -68,7 +70,6 @@ func (svc *ExchangeGrpcServer) subscribeExchanges() {
 
 			var exchangeInstIds []string
 			var processingInsts []types.TradeInstance
-			exchange.Lock()
 			for _, inst := range insts {
 				exchangeInstIds = append(exchangeInstIds, inst.ExchangeInstId)
 				tradeInst := &types.TradeInstance{
@@ -79,13 +80,17 @@ func (svc *ExchangeGrpcServer) subscribeExchanges() {
 					ExchangeInstId: inst.ExchangeInstId,
 					Exchange:       exchange.ExType,
 				}
-				exchange.Insts[inst.ExchangeInstId] = tradeInst
+				exchange.Insts.Store(inst.ExchangeInstId, &ExchangeTradeInstance{
+					Inst:          tradeInst,
+					Status:        0,
+					LiveMarkTs:    0,
+					HistoryMarkTs: 0,
+				})
 				// 待初始化币种数据
 				if inst.Status == data.StatusProcessing {
 					processingInsts = append(processingInsts, *tradeInst)
 				}
 			}
-			exchange.Unlock()
 
 			// instIds := []string{"BTC-USDT", "DOGE-USDT-SWAP"}
 			err = exchange.Subscriber.SubscribeKline(exchangeInstIds...)
@@ -119,13 +124,14 @@ func (svc *ExchangeGrpcServer) consumerKline(exchange *Exchange, c <-chan *types
 
 		// 交易所 instid 转 sig-instid
 		var exInst *types.TradeInstance
-		exchange.RLock()
+		if inst, ok := exchange.Insts.Load(channelK.ExgInstId); ok && inst != nil {
-		if inst, ok := exchange.Insts[channelK.InstId]; ok && inst != nil {
+			exInst = inst.Inst
-			exchange.RUnlock()
+			// 标记交易产品开始订阅k线时间
-			exInst = inst
+			if inst.LiveMarkTs == 0 && len(channelK.Klines) > 0 {
+				inst.LiveMarkTs = channelK.Klines[0].Ts
+			}
 		} else {
-			exchange.RUnlock()
+			zlog.Errorf("unknown exchange instId: %v, %s", channelK.Exchange, channelK.ExgInstId)
-			zlog.Errorf("unknown exchange instId: %v, %s", channelK.Exchange, channelK.InstId)
 			continue
 		}
 
@@ -284,59 +290,70 @@ func (svc *ExchangeGrpcServer) SubscribeKline(stream grpc.BidiStreamingServer[pb
 
 const (
 	KlineBefore0      int64  = 1672502400000               // k线开始数据 2023-01-01 00:00:00 GMT+8
-	HistoryKlineTsKey string = "history-kline-ts:%s/%s/%s" // exchange:sig-instid:interval
+	HistoryKlineTsKey string = "history-kline-ts:%s:%s:%s" // exchange:sig-instid:interval
 )
 
-// initialKline 初始化交易产品历史k线数据
+type initialKlineTask struct {
-func (svc *ExchangeGrpcServer) initialKlines(exchange *Exchange, insts []types.TradeInstance) {
-	var err error
-	defer func() {
-		if err != nil {
-			zlog.Error(err)
-		}
-	}()
-
-	type task struct {
 	inst     types.TradeInstance
 	interval types.Interval
 	afterTs  int64
 	beforeTs int64
-	}
+	times    int32 // 重试次数
+}
 
+func (t initialKlineTask) logKey() string {
+	return fmt.Sprintf("%s:%s:%s:%d:%d", t.inst.Exchange, t.inst.InstId, t.interval, t.beforeTs, t.afterTs)
+}
+
+// initialKline 初始化交易产品历史k线数据
+func (svc *ExchangeGrpcServer) initialKlines(exchange *Exchange, insts []types.TradeInstance) {
 	// 任务 channel
-	ch := make(chan task)
+	taskCh := make(chan initialKlineTask, 8)
 
-	// 任务生成器
+	// 任务生成
 	go func() {
 		for _, inst := range insts {
 			// for interval, intervalAdder := range types.SupportedIntervals {
-			interval := types.Interval1h
+			interval := types.Interval1d
 			intervalAdder := types.SupportedIntervals[interval]
+			// history 未补全前, history写 kvdb ts mark, 补全后 ws live 写 ts mark
 			tsKey := fmt.Sprintf(HistoryKlineTsKey, inst.Exchange, inst.InstId, interval)
-			beforeTs, e := svc.kvdb.GetI64(context.Background(), tsKey)
+			beforeTs, err := svc.kvdb.GetI64(context.Background(), tsKey)
-			if e != nil {
+			if err != nil {
-				err = e
+				zlog.Error(err)
-				return
+				panic(err)
 			}
 			if beforeTs == 0 {
 				beforeTs = intervalAdder(KlineBefore0, -1)
 			}
 			for {
 				afterTs := intervalAdder(beforeTs, 101)
-				ch <- task{
+				taskCh <- initialKlineTask{
 					inst:     inst,
 					interval: interval,
 					afterTs:  afterTs,
 					beforeTs: beforeTs,
+					times:    0,
+				}
+				beforeTs = intervalAdder(afterTs, -1)
+
+				// 对比 ws 获取的实时k线
+				inst, ok := exchange.Insts.Load(inst.ExchangeInstId)
+				if !ok {
+					break
+				}
+				// 订阅完成
+				if inst.LiveMarkTs != 0 && beforeTs > inst.LiveMarkTs {
+					break
 				}
-				beforeTs = afterTs
-				// todo compare ws live ts
 			}
 			// }
 		}
-		close(ch)
+		close(taskCh)
 	}()
 
+	loc, _ := time.LoadLocation("Asia/Shanghai")
+
 	// 任务消费器 8协程并行
 	wg := new(sync.WaitGroup)
 	for range 11 {
@@ -344,27 +361,40 @@ func (svc *ExchangeGrpcServer) initialKlines(exchange *Exchange, insts []types.T
 		go func() {
 			defer wg.Done()
 			for {
-				task, ok := <-ch
+				task, ok := <-taskCh
 				if !ok {
 					break
 				}
 
+				if task.times > 0 {
+					zlog.Infof("retry fetch history kline task %d times: task -> %s", task.times, task.logKey())
+				}
+
 				interval, afterTs, beforeTs := task.interval, task.afterTs, task.beforeTs
-				klines, e := exchange.Fetcher.FetchHistoryKlines(context.Background(), task.inst.ExchangeInstId, interval, afterTs, beforeTs)
+				klines, err := exchange.Fetcher.FetchHistoryKlines(context.Background(), task.inst.ExchangeInstId, interval, afterTs, beforeTs)
-				if e != nil {
+				if err != nil {
-					// todo retry
+					zlog.Errorf("fetch history kline task error: task -> %s, err -> %v", task.logKey(), err)
-					err = e
+					// retry task
+					task.times++
+					taskCh <- task
 					return
 				}
 				if len(klines) == 0 {
-					continue // todo ....
+					continue
 				}
 
-				zlog.Infof("fetch interval %s %d~%d klines: ret=%d~%d, %d klines", interval, afterTs, beforeTs, klines[0].Ts, klines[len(klines)-1].Ts, len(klines))
+				sts, ets := klines[0].Ts, klines[len(klines)-1].Ts
+				ss := time.UnixMilli(sts).In(loc).Format(times.FORMAT_DATE)
+				ee := time.UnixMilli(ets).In(loc).Format(times.FORMAT_DATE)
+				zlog.Infof("fetch interval %s %d~%d klines: ret=%d~%d, %d klines, %s~%s", interval, afterTs, beforeTs, sts, ets, len(klines), ss, ee)
 
 				// store to tsdb
 				err = svc.exchangeDataService.SaveKlines(task.inst, klines)
 				if err != nil {
+					zlog.Errorf("save history klines to tsdb error: task -> %s, err -> %v", task.logKey(), err)
+					// retry task
+					task.times++
+					taskCh <- task
 					return
 				}
 			}

internal/exchange/okx/channel_kline.go

@@ -86,7 +86,7 @@ func (c *ChannelCandle) GetSubscribes() (instIds []string) {
 // func candleData2Klines() (klines []*types.Kline) {
 func candleData2Klines(channelData *ChannelData[*CandleData]) (r *types.ChannelKline, err error) {
 	r = &types.ChannelKline{
-		InstId:   channelData.InstId,
+		ExgInstId: channelData.InstId,
 		Exchange:  types.ExchangeOKX,
 	}
 	for _, data := range *channelData.Data {

internal/exchange/okx/okx_fetch.go

@@ -84,6 +84,7 @@ func (f *OkxFetcher) FetchHistoryKlines(ctx context.Context, okxInstId string, i
 		return
 	}
 
+	// http status: 429 Too Many Requests
 	status := resp.StatusCode()
 	if status != 200 {
 		err = fmt.Errorf("request history klines status error: %s, %s", url, resp.Status())

pkg/types/kline.go

@@ -55,8 +55,7 @@ func (k *Kline) ToPBKline() (kline *pb.Kline) {
 
 // ChannelKline k线订阅消息
 type ChannelKline struct {
-	InstId         string   `json:"instId"`         // 交易产品id，如 BTC-USDT-SWAP
+	ExgInstId string   `json:"instId"`   // 交易所交易产品id，如 BTC_USDT_SWAP
-	ExchangeInstId string   `json:"exchangeInstId"` // 交易所交易产品id
 	Exchange  Exchange `json:"exchange"` // 交易所
 	Klines    []*Kline `json:"klines"`
 }

pkg/utils/collect/sync_map.go

@@ -0,0 +1,34 @@
+package collect
+
+import "sync"
+
+// SyncMap sync.Map 泛型包装
+type SyncMap[K comparable, V any] struct {
+	m *sync.Map
+}
+
+func NewSyncMap[K comparable, V any]() *SyncMap[K, V] {
+	return &SyncMap[K, V]{
+		m: new(sync.Map),
+	}
+}
+
+// Store 放置新值
+func (m *SyncMap[K, V]) Store(k K, v V) {
+	m.m.Store(k, v)
+}
+
+func (m *SyncMap[K, V]) Load(k K) (v V, ok bool) {
+	value, ok := m.m.Load(k)
+	if !ok {
+		return
+	}
+	v = value.(V)
+	return
+}
+
+func (m *SyncMap[K, V]) Range(f func(k K, v V) bool) {
+	m.m.Range(func(key, value any) bool {
+		return f(key.(K), value.(V))
+	})
+}

pkg/utils/times/times.go

@@ -0,0 +1,9 @@
+package times
+
+const FORMAT string = "2006-01-02 15:04:05"
+const FORMAT2 string = "2006/01/02 15:04:05"
+const FORMAT_DATE string = "2006-01-02"
+const FORMAT_DATE2 string = "2006/01/02"
+const FORMAT_MONTH string = "2006-01"
+const FORMAT_TIME string = "15:04:05"
+const FORMAT_TIME_Minute string = "15:04"