You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
122 lines
3.3 KiB
122 lines
3.3 KiB
package trading |
|
|
|
import ( |
|
"context" |
|
"fmt" |
|
"io" |
|
"sig-pub/api/pb" |
|
"sig-pub/pkg/indicator" |
|
"sig-pub/pkg/types" |
|
"sig-pub/pkg/types/series" |
|
"sig-pub/pkg/zlog" |
|
|
|
"google.golang.org/grpc" |
|
) |
|
|
|
type IOffsetIndicatorContext interface { |
|
indicator.IIndicatorContext |
|
SetOffset(offset int16) |
|
} |
|
|
|
// IndicatorContext 指标上下文, 提供k线序列给指标计算使用 |
|
type IndicatorContext struct { |
|
IOffsetIndicatorContext |
|
kSeries *KlineSeries |
|
offset int16 |
|
} |
|
|
|
func NewIndicatorContext(kSeries *KlineSeries) *IndicatorContext { |
|
return &IndicatorContext{ |
|
kSeries: kSeries, |
|
} |
|
} |
|
|
|
func (c *IndicatorContext) SetOffset(offset int16) { |
|
c.offset = offset |
|
} |
|
|
|
func (c *IndicatorContext) Get(offset int16) (kline types.Kline) { |
|
offset += c.offset |
|
k, ok := c.kSeries.Get(offset) |
|
if !ok { |
|
lastTs := c.kSeries.LastTs() |
|
zlog.Warningf("get kline series offset out of range: offset=%d, lastTs=%d", offset, lastTs) |
|
panic(fmt.Errorf("get kline series offset out of range: offset=%d", offset)) |
|
} |
|
return k |
|
} |
|
|
|
func (c *IndicatorContext) Series(offset, count int16) (klines series.Klines) { |
|
offset += c.offset |
|
ks, ok := c.kSeries.Series(offset, count) |
|
if !ok { |
|
zlog.Warningf("get kline series offset out of range: offset=%d, count=%d, lastTs=%d", offset, count, c.kSeries.LastTs()) |
|
panic(fmt.Errorf("get kline series offset out of range: offset=%d, count=%d", offset, count)) |
|
} |
|
return ks |
|
} |
|
|
|
type HistoryIndicatorContext struct { |
|
IOffsetIndicatorContext |
|
exchangeClient pb.ExchangeServiceClient |
|
context *IndicatorContext |
|
} |
|
|
|
func NewHistoryIndicatorContext(exchangeClient pb.ExchangeServiceClient) *HistoryIndicatorContext { |
|
return &HistoryIndicatorContext{ |
|
exchangeClient: exchangeClient, |
|
} |
|
} |
|
func (c *HistoryIndicatorContext) Init(exchange pb.ExchangeType, instId string, interval types.Interval, before, after int64) (totalK int, err error) { |
|
// fetch history series |
|
req := &pb.ReqHistoryKlineStream{ |
|
Exchange: exchange, |
|
InstId: instId, |
|
Interval: string(interval), |
|
Count: 0, |
|
Before: before, |
|
After: after, |
|
} |
|
stream, err := c.exchangeClient.HistoryKlineStream(context.Background(), req, grpc.UseCompressor("snappy")) |
|
if err != nil { |
|
zlog.Errorf("fetch history kline stream error: instId=%s(%s), interval=%s, %#v, err=%v", instId, exchange, interval, req, err) |
|
return |
|
} |
|
|
|
klineSeries := NewKlineSeries(exchange, instId, interval) |
|
for { |
|
msg, err0 := stream.Recv() |
|
if err0 == io.EOF { |
|
break |
|
} |
|
if err0 != nil { |
|
err = err0 |
|
zlog.Error("fetch kline stream recv error: ", err0) |
|
return |
|
} |
|
// zlog.Debugf("recv: %s(%s), %s, branch=%d, ts=%d~%d", instId, exchange, interval, len(msg.Klines), msg.Klines[0].Ts, msg.Klines[len(msg.Klines)-1].Ts) |
|
totalK += len(msg.Klines) |
|
for _, k := range msg.Klines { |
|
kline := new(types.Kline) |
|
kline.ParsePBKline(exchange, k) |
|
if lastTs, ok := klineSeries.Update(kline); !ok { |
|
err = fmt.Errorf("history stream kline not series: last=%d", lastTs) |
|
return |
|
} |
|
} |
|
} |
|
c.context = NewIndicatorContext(klineSeries) |
|
return |
|
} |
|
|
|
func (c *HistoryIndicatorContext) SetOffset(offset int16) { |
|
c.context.SetOffset(offset) |
|
} |
|
|
|
func (c *HistoryIndicatorContext) Get(offset int16) (kline types.Kline) { |
|
return c.context.Get(offset) |
|
} |
|
|
|
func (c *HistoryIndicatorContext) Series(offset, count int16) (klines series.Klines) { |
|
return c.context.Series(offset, count) |
|
}
|
|
|