1

internal/trading/kline_store.go

@@ -12,10 +12,8 @@ type KlineStore struct {
 	store [3]*collect.ConcurrentMap[string, *collect.ConcurrentMap[types.Interval, *KlineSeries]] // K线列表: []exchange<instId, interval, klines>
 }
 
-func NewKlineSeriesStore(vmdb vmts.VictoriaMetricsTSDB) (kss *KlineStore) {
-	kss = &KlineStore{
-		vmdb: vmdb,
-	}
+func NewKlineSeriesStore() (kss *KlineStore) {
+	kss = &KlineStore{}
 	kss.store[pb.ExchangeType_OKX] = collect.NewConcurrentMap[string, *collect.ConcurrentMap[types.Interval, *KlineSeries]](64, func(s string) string { return s })
 	// kss.klines[pb.ExchangeType_BINANCE] =
 	return

internal/trading/trading_service.go

@@ -36,6 +36,8 @@ func NewTradingService(
 
 // 初始化历史k线, 订阅实时k线
 func (svr *TradingService) Init() (err error) {
+	svr.klineStore = NewKlineSeriesStore()
+
 	// get instance
 	exchangeTradeInsts, err := svr.marketClientAside.ListExchangeTradeInstance(context.Background(), pb.ExchangeType_OKX)
 	if err != nil {
@@ -109,14 +111,13 @@ func (svr *TradingService) subscribeKlines(reconnect bool) {
 			kline := new(types.Kline)
 			kline.ParsePBKline(msg.Kline.Exchange, k)
 			zlog.Debugf("recv: streamId=%d, %v, %s, %#v", msg.Kline.StreamId, msg.Kline.Exchange, msg.Kline.InstId, kline)
-
 			svr.klineStore.Update(msg.Kline.Exchange, msg.Kline.InstId, kline)
 		}
 	}
 }
 
 var SubscribeKlineIntervals = []string{
-	string(types.Interval1s),
+	// string(types.Interval1s),
 	string(types.Interval1m),
 	string(types.Interval3m),
 	string(types.Interval5m),

pkg/utils/lang/tuple.go

@@ -0,0 +1,6 @@
+package lang
+
+type Tuple2[V0, V1 any] struct {
+	V0 V0
+	V1 V1
+}