1

README.md

@@ -46,5 +46,5 @@ k线推送
 - closed -> cache -> async tsdb -> kline signal
 - query(监控成功率/缓存命中率) -> cache -> tsdb
 
-
-
+价格/交易量精度->tsdb读写存储
+kline时间窗口

config/config.toml

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

internal/exchange/exchange_grpc_server.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"runtime"
 	"sig-pub/api/pb"
 	"sig-pub/pkg/aside"
 	"sig-pub/pkg/data"
@@ -307,8 +308,10 @@ func (t initialKlineTask) logKey() string {
 
 // initialKline 初始化交易产品历史k线数据
 func (svc *ExchangeGrpcServer) initialKlines(exchange *Exchange, insts []types.TradeInstance) {
+	concurrent := max(8, runtime.NumCPU()*2)
+
 	// 任务 channel
-	taskCh := make(chan initialKlineTask, 8)
+	taskCh := make(chan initialKlineTask, concurrent)
 
 	// 任务生成
 	go func() {
@@ -344,6 +347,15 @@ func (svc *ExchangeGrpcServer) initialKlines(exchange *Exchange, insts []types.T
 				}
 				// 订阅完成
 				if inst.LiveMarkTs != 0 && beforeTs > inst.LiveMarkTs {
+					// history status -> ok
+					break
+				}
+				if beforeTs > time.Now().UnixMilli() {
+					if inst.LiveMarkTs != 0 {
+						// history status -> ok
+					} else {
+						// live status -> not ok
+					}
 					break
 				}
 			}
@@ -356,7 +368,7 @@ func (svc *ExchangeGrpcServer) initialKlines(exchange *Exchange, insts []types.T
 
 	// 任务消费器 8协程并行
 	wg := new(sync.WaitGroup)
-	for range 11 {
+	for range concurrent {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()