package trading import ( "fmt" "sig-pub/api/pb" "sig-pub/pkg/data" "sig-pub/pkg/types" "sig-pub/pkg/types/series" "sig-pub/pkg/zlog" "sync" "sync/atomic" ) const ( MaxSeriesKlines = 1000 // 有效k线数量 MaxCapSeriesKlines = 1280 // 最大k线数量, slice扩容12次后cap=1280 ) // TradeInstanceKlineSeries 单个交易产品所有周期k线 type TradeInstanceKlineSeries struct { IntervalKlines *types.IntervalState[*KlineSeries] Status atomic.Int32 // 交易产品状态 } func NewTradeInstanceKlineSeries(exchange pb.ExchangeType, instId string) *TradeInstanceKlineSeries { si := new(TradeInstanceKlineSeries) si.Status.Store(int32(data.StatusNone)) si.IntervalKlines = types.NewIntervalState[*KlineSeries]() for interval := range types.SupportedIntervals { si.IntervalKlines.Set(interval, NewKlineSeries(exchange, instId, interval)) } return si } type KlineSeries struct { sync.RWMutex Exchange pb.ExchangeType InstId string Interval types.Interval IntervalAdder types.IntervalAdder lastTs int64 klines []*types.Kline } func NewKlineSeries(exchange pb.ExchangeType, instId string, interval types.Interval) *KlineSeries { intervalAdder, ok := types.SupportedIntervals[interval] if !ok { panic(fmt.Errorf("unsupport interval: %s", interval)) } return &KlineSeries{ Exchange: exchange, InstId: instId, Interval: interval, IntervalAdder: intervalAdder, klines: make([]*types.Kline, 0, MaxSeriesKlines/10), } } // Get [0]当前k线 func (s *KlineSeries) Get(offset int16) (k types.Kline, ok bool) { if ok = offset >= 0 && offset < MaxSeriesKlines; !ok { return } s.RLock() defer s.RUnlock() length := len(s.klines) index := (length - 1) - int(offset) if ok = index > 0 && index < length; !ok { return } return *(s.klines[index]), true } // Series 时间降序序列[count...offset] // offset: 从序列尾部开始偏移量 // count: 从offset位置开始向序列头部k线条数 func (s *KlineSeries) Series(offset, count int16) (klines series.Klines, ok bool) { if ok = offset >= 0 && offset < MaxSeriesKlines; !ok { return } if ok = count > 0 && offset+count < MaxSeriesKlines; !ok { return } s.RLock() defer s.RUnlock() length := len(s.klines) indexEnd := (length - 1) - int(offset) indexStart := (length - 1) - int(offset) - int(count) + 1 if ok = indexEnd >= 0 && indexEnd < length && indexStart >= 0 && indexStart < length; !ok { return } total := indexEnd - indexStart + 1 klines = make(series.Klines, total) for i := indexStart; i <= indexEnd; i++ { offset := total - 1 - (i - indexStart) klines[offset] = *(s.klines[i]) } return klines, true } func (s *KlineSeries) LastTs() int64 { return s.lastTs } // 检查k线序列完整 func (s *KlineSeries) Update(kline *types.Kline) (lastTs int64, serial bool) { s.Lock() defer s.Unlock() serial = true lastTs = s.lastTs if kline.Ts <= s.lastTs { return } // 检查k线是否连续 if len(s.klines) > 0 { expectTs := s.IntervalAdder(s.lastTs, 1) if kline.Ts != expectTs { serial = false miss := (kline.Ts-s.lastTs)/s.IntervalAdder(0, 1) - 1 zlog.Warningf("k线不连续: instId=%s(%s), interval=%s, miss=%d, lastTs=%d, got=%d, expected=%d", s.InstId, s.Exchange, s.Interval, miss, s.lastTs, kline.Ts, expectTs) return } } if len(s.klines) < MaxCapSeriesKlines { s.klines = append(s.klines, kline) // 容量超过0.65, 直接扩容到最大 if capacity := cap(s.klines); capacity < MaxCapSeriesKlines && capacity > MaxCapSeriesKlines*0.65 { klines := make([]*types.Kline, len(s.klines), MaxCapSeriesKlines) copy(klines, s.klines) s.klines = klines } } else { // 循环复用切片空间,避免扩容 length := len(s.klines) copy(s.klines, s.klines[length-MaxSeriesKlines+1:]) s.klines[MaxSeriesKlines-1] = kline s.klines = s.klines[:MaxSeriesKlines] } s.lastTs = kline.Ts return }