diff --git a/internal/trading/kline_store.go b/internal/trading/kline_store.go index 1716ad0..beff5b4 100644 --- a/internal/trading/kline_store.go +++ b/internal/trading/kline_store.go @@ -12,10 +12,8 @@ type KlineStore struct { store [3]*collect.ConcurrentMap[string, *collect.ConcurrentMap[types.Interval, *KlineSeries]] // K线列表: []exchange } -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 diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 50f3a1b..d0689bd 100644 --- a/internal/trading/trading_service.go +++ b/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), diff --git a/pkg/utils/lang/tuple.go b/pkg/utils/lang/tuple.go new file mode 100644 index 0000000..a06e4a8 --- /dev/null +++ b/pkg/utils/lang/tuple.go @@ -0,0 +1,6 @@ +package lang + +type Tuple2[V0, V1 any] struct { + V0 V0 + V1 V1 +}