diff --git a/README.md b/README.md index 0a82fa2..610d146 100644 --- a/README.md +++ b/README.md @@ -46,5 +46,5 @@ k线推送 - closed -> cache -> async tsdb -> kline signal - query(监控成功率/缓存命中率) -> cache -> tsdb - - +价格/交易量精度->tsdb读写存储 +kline时间窗口 diff --git a/config/config.toml b/config/config.toml index 114ce4e..26e2268 100644 --- a/config/config.toml +++ b/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 diff --git a/internal/exchange/exchange_grpc_server.go b/internal/exchange/exchange_grpc_server.go index bb29ab9..8b3a064 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/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()