39 changed files with 901 additions and 671 deletions
@ -0,0 +1,50 @@ |
|||||||
|
package exchange |
||||||
|
|
||||||
|
import ( |
||||||
|
"context" |
||||||
|
"fmt" |
||||||
|
"sig-pub/api/pb" |
||||||
|
"sig-pub/pkg/storage/kvrocks" |
||||||
|
vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" |
||||||
|
"sig-pub/pkg/types" |
||||||
|
) |
||||||
|
|
||||||
|
type ExchangeDataPersist struct { |
||||||
|
vmtsdb *vmts.VictoriaMetricsTSDB |
||||||
|
kvdb *kvrocks.KVRocksDB |
||||||
|
// exchangeFetch ExchangeFetcher
|
||||||
|
} |
||||||
|
|
||||||
|
func NewExchangeDataService( |
||||||
|
vmdb *vmts.VictoriaMetricsTSDB, |
||||||
|
kvdb *kvrocks.KVRocksDB, |
||||||
|
) *ExchangeDataPersist { |
||||||
|
return &ExchangeDataPersist{ |
||||||
|
vmtsdb: vmdb, |
||||||
|
kvdb: kvdb, |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
func (p *ExchangeDataPersist) Init() (err error) { |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
func (p *ExchangeDataPersist) SaveKlines(inst types.TradeInstance, klines []*types.Kline) (err error) { |
||||||
|
err = p.vmtsdb.SaveKlines(inst, klines) |
||||||
|
if err != nil { |
||||||
|
return |
||||||
|
} |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
func (p *ExchangeDataPersist) GetHistoryKlineMarkTs(exchange pb.ExchangeType, instId string, interval types.Interval) (ts int64, err error) { |
||||||
|
tsKey := fmt.Sprintf(HistoryKlineTsKey, exchange, instId, interval) |
||||||
|
ts, err = p.kvdb.GetI64(context.Background(), tsKey) |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
func (p *ExchangeDataPersist) SaveHistoryKlineMarkTs(exchange pb.ExchangeType, instId string, interval types.Interval, ts int64) (tsKey string, err error) { |
||||||
|
tsKey = fmt.Sprintf(HistoryKlineTsKey, exchange, instId, interval) |
||||||
|
err = p.kvdb.SetI64(context.Background(), tsKey, ts) |
||||||
|
return |
||||||
|
} |
||||||
@ -1,34 +0,0 @@ |
|||||||
package exchange |
|
||||||
|
|
||||||
import ( |
|
||||||
vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" |
|
||||||
"sig-pub/pkg/types" |
|
||||||
) |
|
||||||
|
|
||||||
type ExchangeDataService struct { |
|
||||||
vmdb *vmts.VictoriaMetricsTSDB |
|
||||||
exchangeFetch ExchangeFetcher |
|
||||||
} |
|
||||||
|
|
||||||
func NewExchangeDataService( |
|
||||||
vmdb *vmts.VictoriaMetricsTSDB, |
|
||||||
exchangeFetch ExchangeFetcher, |
|
||||||
) *ExchangeDataService { |
|
||||||
return &ExchangeDataService{ |
|
||||||
vmdb: vmdb, |
|
||||||
exchangeFetch: exchangeFetch, |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func (svc *ExchangeDataService) Init() (err error) { |
|
||||||
return |
|
||||||
} |
|
||||||
|
|
||||||
func (svc *ExchangeDataService) SaveKlines(inst types.TradeInstance, klines []*types.Kline) (err error) { |
|
||||||
err = svc.vmdb.SaveKlines(inst, klines) |
|
||||||
if err != nil { |
|
||||||
return |
|
||||||
} |
|
||||||
// todo klines 时间点回溯检查
|
|
||||||
return |
|
||||||
} |
|
||||||
@ -0,0 +1,478 @@ |
|||||||
|
package exchange |
||||||
|
|
||||||
|
import ( |
||||||
|
"context" |
||||||
|
"errors" |
||||||
|
"fmt" |
||||||
|
"runtime" |
||||||
|
"sig-pub/api/pb" |
||||||
|
"sig-pub/pkg/aside" |
||||||
|
"sig-pub/pkg/data" |
||||||
|
"sig-pub/pkg/types" |
||||||
|
"sig-pub/pkg/utils/times" |
||||||
|
"sig-pub/pkg/zlog" |
||||||
|
"sort" |
||||||
|
"sync" |
||||||
|
"sync/atomic" |
||||||
|
"time" |
||||||
|
|
||||||
|
"google.golang.org/grpc" |
||||||
|
) |
||||||
|
|
||||||
|
// ExchangeService 交易所服务
|
||||||
|
type ExchangeService struct { |
||||||
|
exchangeMap map[pb.ExchangeType]*Exchange |
||||||
|
tradeInstanceAside *aside.TradeInstanceAside |
||||||
|
exchangeDataService *ExchangeDataPersist |
||||||
|
|
||||||
|
klinePublisher *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]] |
||||||
|
} |
||||||
|
|
||||||
|
// exchanges: 支持的数据源交易所
|
||||||
|
func NewExchangeService( |
||||||
|
tradeInstanceAside *aside.TradeInstanceAside, |
||||||
|
exchangeDataService *ExchangeDataPersist, |
||||||
|
exchanges ...*Exchange, |
||||||
|
) *ExchangeService { |
||||||
|
exchangeMap := make(map[pb.ExchangeType]*Exchange) |
||||||
|
for _, exchange := range exchanges { |
||||||
|
exchangeMap[exchange.ExchangeType] = exchange |
||||||
|
} |
||||||
|
|
||||||
|
return &ExchangeService{ |
||||||
|
exchangeMap: exchangeMap, |
||||||
|
tradeInstanceAside: tradeInstanceAside, |
||||||
|
exchangeDataService: exchangeDataService, |
||||||
|
klinePublisher: NewPublisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]](16), |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
func (svc *ExchangeService) Init() (err error) { |
||||||
|
svc.subscribeExchanges() |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
// GetKlineSubscriber 订阅k线订阅器
|
||||||
|
func (svc *ExchangeService) GetKlineSubscriber() (subscriber *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]]) { |
||||||
|
subscriber = svc.klinePublisher |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
// 订阅交易所推送行情
|
||||||
|
func (svc *ExchangeService) subscribeExchanges() { |
||||||
|
// consumerKline
|
||||||
|
// 交易所订阅交易产品
|
||||||
|
for _, exchange := range svc.exchangeMap { |
||||||
|
go func(exchange *Exchange) { |
||||||
|
// get exchange trade instances
|
||||||
|
insts, err := svc.tradeInstanceAside.ListExchangeTradeInstance(context.Background(), exchange.ExchangeType) |
||||||
|
if err != nil { |
||||||
|
zlog.Error(err) |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
var exchangeInstIds []string |
||||||
|
var processingInsts []types.TradeInstance |
||||||
|
for _, inst := range insts { |
||||||
|
exchangeInstIds = append(exchangeInstIds, inst.ExchangeInstId) |
||||||
|
tradeInst := &types.TradeInstance{ |
||||||
|
InstId: inst.InstId, |
||||||
|
Status: inst.Status, |
||||||
|
PriceSz: 0, |
||||||
|
QuantitySz: 0, |
||||||
|
ExchangeInstId: inst.ExchangeInstId, |
||||||
|
Exchange: exchange.ExchangeType, |
||||||
|
} |
||||||
|
exchange.TradeInstIds.Store(inst.InstId, inst.ExchangeInstId) |
||||||
|
exchange.ExchangeInsts.Store(inst.ExchangeInstId, &ExchangeTradeInstance{ |
||||||
|
Inst: tradeInst, |
||||||
|
LiveKline: types.NewIntervalState[types.Kline](), |
||||||
|
LiveKStartTs: types.NewIntervalState[int64](), |
||||||
|
HistoryMarkTs: types.NewIntervalState[int64](), |
||||||
|
}) |
||||||
|
// 待初始化币种数据
|
||||||
|
if inst.Status == int32(data.StatusProcessing) { |
||||||
|
processingInsts = append(processingInsts, *tradeInst) |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// instIds := []string{"BTC-USDT", "DOGE-USDT-SWAP"}
|
||||||
|
err = exchange.Subscriber.SubscribeKline(exchangeInstIds...) |
||||||
|
if err != nil { |
||||||
|
zlog.Error(err) |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
go func() { |
||||||
|
c := exchange.Subscriber.ConsumerKline() |
||||||
|
svc.consumerKline(exchange, c) |
||||||
|
// todo subscribe books 订单簿
|
||||||
|
zlog.Infof("unsubscribe exchange: %s", exchange.ExchangeType) |
||||||
|
}() |
||||||
|
|
||||||
|
// 初始化k线数据
|
||||||
|
go svc.initialKlines(exchange, processingInsts) |
||||||
|
}(exchange) |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// consumerKline 消费交易所k线数据
|
||||||
|
func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.ChannelKline) { |
||||||
|
exchangeType := exchange.ExchangeType |
||||||
|
|
||||||
|
// publish to subscribers
|
||||||
|
pubStreamKlineMap := make(map[string]*pb.StreamKline) |
||||||
|
|
||||||
|
for { |
||||||
|
clear(pubStreamKlineMap) |
||||||
|
|
||||||
|
channelK, ok := <-c |
||||||
|
if !ok { |
||||||
|
return |
||||||
|
} |
||||||
|
if len(channelK.Klines) == 0 { |
||||||
|
continue |
||||||
|
} |
||||||
|
if len(channelK.Klines) > 1 { |
||||||
|
sort.Slice(channelK.Klines, func(i, j int) bool { |
||||||
|
return channelK.Klines[i].Ts < channelK.Klines[j].Ts |
||||||
|
}) |
||||||
|
} |
||||||
|
firstKline, lastKline := channelK.Klines[0], channelK.Klines[len(channelK.Klines)-1] |
||||||
|
|
||||||
|
// 交易所 instid 转 sig-instid
|
||||||
|
var tradeInst *types.TradeInstance |
||||||
|
exchangeInst, ok := exchange.ExchangeInsts.Load(channelK.ExgInstId) |
||||||
|
if !ok || exchangeInst == nil || exchangeInst.Inst == nil { |
||||||
|
zlog.Errorf("unknown exchange instId: %v, %s", channelK.Exchange, channelK.ExgInstId) |
||||||
|
continue |
||||||
|
} |
||||||
|
tradeInst = exchangeInst.Inst |
||||||
|
|
||||||
|
// 标记交易产品开始订阅k线时间
|
||||||
|
exchangeInst.LiveKStartTs.SetIf(firstKline.Interval, firstKline.Ts, func(old int64) bool { return old == 0 }) |
||||||
|
|
||||||
|
// 标记实时k线
|
||||||
|
exchangeInst.LiveKline.Set(lastKline.Interval, *lastKline) |
||||||
|
// 记录实时价格
|
||||||
|
exchangeInst.Last = lastKline.Close |
||||||
|
|
||||||
|
var confirmKlines []*types.Kline |
||||||
|
for _, kline := range channelK.Klines { |
||||||
|
// zlog.Infof("recv kline: %#v", kline)
|
||||||
|
confirm := 0 |
||||||
|
if kline.Confirm { |
||||||
|
confirm = 1 |
||||||
|
confirmKlines = append(confirmKlines, kline) |
||||||
|
} |
||||||
|
pubKey := fmt.Sprintf("/kline/%s/%s/%s/%d", exchangeType, tradeInst.InstId, kline.Interval, confirm) |
||||||
|
msg, ok := pubStreamKlineMap[pubKey] |
||||||
|
if !ok { |
||||||
|
msg = new(pb.StreamKline) |
||||||
|
msg.InstId = tradeInst.InstId |
||||||
|
msg.Exchange = channelK.Exchange |
||||||
|
pubStreamKlineMap[pubKey] = msg |
||||||
|
} |
||||||
|
pbk := kline.ToPBKline() |
||||||
|
msg.Klines = append(msg.Klines, pbk) |
||||||
|
} |
||||||
|
|
||||||
|
if len(confirmKlines) > 0 { |
||||||
|
// tsdb storage todo 异步处理
|
||||||
|
err := svc.exchangeDataService.SaveKlines(*tradeInst, confirmKlines) |
||||||
|
if err != nil { |
||||||
|
zlog.Errorf("kline save to tsdb error: ", err) |
||||||
|
} |
||||||
|
|
||||||
|
// 初始化状态完成, 检查k线时间戳标记
|
||||||
|
if exchangeInst.Status.Load() == int32(data.StatusOk) { |
||||||
|
lastConfirmKline := confirmKlines[len(confirmKlines)-1] |
||||||
|
historyMark := exchangeInst.HistoryMarkTs.Get(lastConfirmKline.Interval) |
||||||
|
_ = historyMark |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// publish grpc stream klines
|
||||||
|
for pubKey, kline := range pubStreamKlineMap { |
||||||
|
if len(kline.Klines) == 0 { |
||||||
|
continue |
||||||
|
} |
||||||
|
subs := svc.klinePublisher.Publisher(pubKey) |
||||||
|
for _, sub := range subs { |
||||||
|
if err := sub.Send(&pb.RspStreamSubscribeKline{Kline: kline}); err != nil { |
||||||
|
zlog.Error(err) |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
const ( |
||||||
|
KlineBefore0 int64 = 1672502400000 // k线开始数据 2023-01-01 00:00:00 GMT+8
|
||||||
|
HistoryKlineTsKey string = "history-kline-ts:%s:%s:%s" // exchange:sig-instid:interval
|
||||||
|
SingleKlineFetchTaskMaxFailTimes int32 = 100 // 单个k线拉取任务最大失败次数
|
||||||
|
) |
||||||
|
|
||||||
|
type fetchKlineTask struct { |
||||||
|
inst types.TradeInstance |
||||||
|
interval types.Interval |
||||||
|
afterTs int64 |
||||||
|
beforeTs int64 |
||||||
|
times int32 // 重试次数
|
||||||
|
} |
||||||
|
|
||||||
|
func (t fetchKlineTask) logKey() string { |
||||||
|
return fmt.Sprintf("%s:%s:%s:%d:%d", t.inst.Exchange, t.inst.InstId, t.interval, t.beforeTs, t.afterTs) |
||||||
|
} |
||||||
|
|
||||||
|
// initialKline 初始化交易产品历史k线数据
|
||||||
|
func (svc *ExchangeService) initialKlines(exchange *Exchange, insts []types.TradeInstance) { |
||||||
|
// 记录成功和失败的交易产品
|
||||||
|
var success, failed []types.TradeInstance |
||||||
|
|
||||||
|
for _, inst := range insts { |
||||||
|
err := svc.initialTradeInstanceKlines(exchange, inst) |
||||||
|
if err != nil { |
||||||
|
zlog.Errorf("initial fetch trade instance error: %s(%s), err=%v", inst.InstId, inst.Exchange, err) |
||||||
|
failed = append(failed, inst) |
||||||
|
} else { |
||||||
|
success = append(success, inst) |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
zlog.Infof("%d insts initial finished, success %d, failed %d", len(insts), len(success), len(failed)) |
||||||
|
} |
||||||
|
|
||||||
|
// initTradeInstanceKlines 初始化交易产品历史k线数据
|
||||||
|
func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, tradeInst types.TradeInstance) (err error) { |
||||||
|
exchangeInst, ok := exchange.ExchangeInsts.Load(tradeInst.ExchangeInstId) |
||||||
|
if !ok { |
||||||
|
err = fmt.Errorf("not load exchange trade instance: %s", tradeInst.ExchangeInstId) |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
// 并发数
|
||||||
|
concurrent := max(8, runtime.NumCPU()*2) |
||||||
|
// 任务 channel
|
||||||
|
taskCh := make(chan fetchKlineTask, concurrent) |
||||||
|
retryTaskCh := make(chan fetchKlineTask, concurrent) |
||||||
|
|
||||||
|
// 发布任务数, 成功任务数, 失败任务次数
|
||||||
|
var pubTasks, subTasks, failTasks atomic.Int32 |
||||||
|
var pubTaskDone atomic.Bool // 所有任务已发布
|
||||||
|
ctx, cancel := context.WithCancel(context.Background()) |
||||||
|
|
||||||
|
defer func() { |
||||||
|
if err != nil { |
||||||
|
// 交易所k线初始化失败
|
||||||
|
exchangeInst.Status.Store(int32(data.StatusFailed)) |
||||||
|
return |
||||||
|
} |
||||||
|
exchangeInst.HistoryMarkTs.Range(func(_ int, interval types.Interval, ts int64) { |
||||||
|
tsKey, ex := svc.exchangeDataService.SaveHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval, ts) |
||||||
|
if ex != nil { |
||||||
|
zlog.Errorf("history mark inititaled ts error: key=%s, ts=%d, %v", tsKey, ts, ex) |
||||||
|
} |
||||||
|
}) |
||||||
|
exchangeInst.Status.Store(int32(data.StatusOk)) |
||||||
|
}() |
||||||
|
|
||||||
|
go func() { |
||||||
|
defer func() { |
||||||
|
pubTaskDone.Store(true) |
||||||
|
// 无任务处理
|
||||||
|
if subTasks.Load() == 0 { |
||||||
|
cancel() |
||||||
|
} |
||||||
|
zlog.Infof("trade instance initial kline %s(%s), pub %d fetch tasks", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load()) |
||||||
|
}() |
||||||
|
|
||||||
|
for interval, intervalAdder := range types.SupportedIntervals { |
||||||
|
// interval := types.Interval1d
|
||||||
|
// intervalAdder := types.SupportedIntervals[interval]
|
||||||
|
// history 未补全前, history写 kvdb ts mark, 补全后 ws live 写 ts mark
|
||||||
|
beforeTs, ex := svc.exchangeDataService.GetHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval) |
||||||
|
if ex != nil { |
||||||
|
err = ex |
||||||
|
zlog.Error(err) |
||||||
|
cancel() |
||||||
|
return |
||||||
|
} |
||||||
|
if beforeTs == 0 { |
||||||
|
beforeTs = intervalAdder(KlineBefore0, -1) |
||||||
|
} |
||||||
|
|
||||||
|
exchangeInst.HistoryMarkTs.Set(interval, beforeTs) |
||||||
|
for { |
||||||
|
// 判定订阅任务发布完成
|
||||||
|
liveStartTs := exchangeInst.LiveKStartTs.Get(interval) |
||||||
|
if liveStartTs != 0 && beforeTs >= liveStartTs { |
||||||
|
break |
||||||
|
} |
||||||
|
if beforeTs > time.Now().UnixMilli() { |
||||||
|
break |
||||||
|
} |
||||||
|
|
||||||
|
afterTs := intervalAdder(beforeTs, 101) |
||||||
|
task := fetchKlineTask{ |
||||||
|
inst: tradeInst, |
||||||
|
interval: interval, |
||||||
|
afterTs: afterTs, |
||||||
|
beforeTs: beforeTs, |
||||||
|
times: 0, |
||||||
|
} |
||||||
|
|
||||||
|
// 发布任务
|
||||||
|
select { |
||||||
|
case taskCh <- task: |
||||||
|
pubTasks.Add(1) |
||||||
|
case <-ctx.Done(): |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
beforeTs = intervalAdder(afterTs, -1) |
||||||
|
} |
||||||
|
} |
||||||
|
}() |
||||||
|
|
||||||
|
// 任务消费器 多协程并行
|
||||||
|
wg := new(sync.WaitGroup) |
||||||
|
for range concurrent { |
||||||
|
wg.Add(1) |
||||||
|
go func() { |
||||||
|
defer wg.Done() |
||||||
|
|
||||||
|
var task fetchKlineTask |
||||||
|
for { |
||||||
|
select { |
||||||
|
case <-ctx.Done(): |
||||||
|
return |
||||||
|
case task = <-taskCh: |
||||||
|
case task = <-retryTaskCh: |
||||||
|
} |
||||||
|
|
||||||
|
if task.times > 0 { |
||||||
|
zlog.Infof("retry fetch history kline task %d times: task -> %s", task.times, task.logKey()) |
||||||
|
} |
||||||
|
|
||||||
|
if lastKlineTs, ex := svc.fetchTaskKlines(exchange, task); ex != nil { |
||||||
|
failTasks.Add(1) |
||||||
|
if task.times >= SingleKlineFetchTaskMaxFailTimes { |
||||||
|
err = fmt.Errorf("task failed to many times %d, key: %s, err: %v", task.times, task.logKey(), ex) |
||||||
|
cancel() |
||||||
|
return |
||||||
|
} |
||||||
|
// retry task
|
||||||
|
task.times++ |
||||||
|
select { |
||||||
|
case retryTaskCh <- task: |
||||||
|
case <-ctx.Done(): |
||||||
|
return |
||||||
|
} |
||||||
|
} else { |
||||||
|
// 周期任务最后kline时间
|
||||||
|
if lastKlineTs != 0 { |
||||||
|
exchangeInst.HistoryMarkTs.SetIf(task.interval, lastKlineTs, func(old int64) bool { |
||||||
|
return lastKlineTs > old |
||||||
|
}) |
||||||
|
// historyMarkTsMu.Lock()
|
||||||
|
// historyMarkTs[task.interval] = max(historyMarkTs[task.interval], lastKlineTs)
|
||||||
|
// historyMarkTsMu.Unlock()
|
||||||
|
} |
||||||
|
|
||||||
|
zlog.Debugf("trade instance initial kline tasks processing: %s(%s), pub %d, sub %d, fail %d", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load()) |
||||||
|
// 任务都已执行成功结束
|
||||||
|
subs := subTasks.Add(1) |
||||||
|
if pubTaskDone.Load() && subs >= pubTasks.Load() { |
||||||
|
zlog.Infof("trade instance initial kline tasks success finished, %s(%s), pub %d, sub %d, fail %d", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load()) |
||||||
|
cancel() |
||||||
|
return |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
}() |
||||||
|
} |
||||||
|
wg.Wait() |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
func (svc *ExchangeService) fetchTaskKlines(exchange *Exchange, task fetchKlineTask) (lastKlineTs int64, err error) { |
||||||
|
interval, afterTs, beforeTs := task.interval, task.afterTs, task.beforeTs |
||||||
|
klines, err := exchange.Fetcher.FetchHistoryKlines(context.Background(), task.inst.ExchangeInstId, interval, afterTs, beforeTs) |
||||||
|
if err != nil { |
||||||
|
zlog.Errorf("fetch history kline task error: task -> %s, err -> %v", task.logKey(), err) |
||||||
|
return |
||||||
|
} |
||||||
|
if len(klines) == 0 { |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
loc, _ := time.LoadLocation("Asia/Shanghai") |
||||||
|
|
||||||
|
sts, ets := klines[0].Ts, klines[len(klines)-1].Ts |
||||||
|
lastKlineTs = max(sts, ets) |
||||||
|
ss := time.UnixMilli(sts).In(loc).Format(times.FORMAT_DATE) |
||||||
|
ee := time.UnixMilli(ets).In(loc).Format(times.FORMAT_DATE) |
||||||
|
zlog.Infof("fetch interval %s %d~%d klines: ret=%d~%d, %d klines, %s~%s", interval, beforeTs, afterTs, ets, sts, len(klines), ee, ss) |
||||||
|
|
||||||
|
// store to tsdb
|
||||||
|
err = svc.exchangeDataService.SaveKlines(task.inst, klines) |
||||||
|
if err != nil { |
||||||
|
zlog.Errorf("save history klines to tsdb error: task -> %s, err -> %v", task.logKey(), err) |
||||||
|
return |
||||||
|
} |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
// Exchanges 支持的交易所列表
|
||||||
|
func (svc *ExchangeService) Exchanges() (exchanges []pb.ExchangeType, err error) { |
||||||
|
for exchange := range svc.exchangeMap { |
||||||
|
exchanges = append(exchanges, exchange) |
||||||
|
} |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
// ExchangeInstanceState 交易所交易产品状态
|
||||||
|
func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeTypes []pb.ExchangeType, instIds []string) (states []*pb.TradeInstanceState, err error) { |
||||||
|
var exchanges []*Exchange |
||||||
|
if allExchange { |
||||||
|
for _, exg := range svc.exchangeMap { |
||||||
|
exchanges = append(exchanges, exg) |
||||||
|
} |
||||||
|
} else { |
||||||
|
for _, exchangeType := range exchangeTypes { |
||||||
|
exg, ok := svc.exchangeMap[exchangeType] |
||||||
|
if !ok { |
||||||
|
err = fmt.Errorf("not support exchange: %v", exchangeType) |
||||||
|
return |
||||||
|
} |
||||||
|
exchanges = append(exchanges, exg) |
||||||
|
} |
||||||
|
} |
||||||
|
if len(exchanges) == 0 { |
||||||
|
err = errors.New("no support exchanges") |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
for _, exchange := range exchanges { |
||||||
|
for _, instId := range instIds { |
||||||
|
// trade instId to exchangeInstId
|
||||||
|
exchangeInstId, ok := exchange.TradeInstIds.Load(instId) |
||||||
|
if !ok { |
||||||
|
continue |
||||||
|
} |
||||||
|
inst, ok := exchange.ExchangeInsts.Load(exchangeInstId) |
||||||
|
if !ok { |
||||||
|
continue |
||||||
|
} |
||||||
|
|
||||||
|
state := &pb.TradeInstanceState{ |
||||||
|
Exchange: exchange.ExchangeType, |
||||||
|
InstId: instId, |
||||||
|
Last: inst.Last.String(), |
||||||
|
} |
||||||
|
states = append(states, state) |
||||||
|
} |
||||||
|
} |
||||||
|
return |
||||||
|
} |
||||||
@ -1,3 +1,20 @@ |
|||||||
package indicator |
package indicator |
||||||
|
|
||||||
// load indicator plugin
|
// load indicator plugin
|
||||||
|
|
||||||
|
// 热指标 自动加载/实时更新/内存缓存
|
||||||
|
// 历史指标 实时计算
|
||||||
|
// 自定义插件化指标
|
||||||
|
|
||||||
|
type IndicatorService struct { |
||||||
|
} |
||||||
|
|
||||||
|
func NewIndicatorService() *IndicatorService { |
||||||
|
return &IndicatorService{} |
||||||
|
} |
||||||
|
|
||||||
|
// 加载热指标
|
||||||
|
// 订阅k线数据 更新指标
|
||||||
|
func (svc *IndicatorService) Init() { |
||||||
|
|
||||||
|
} |
||||||
|
|||||||
@ -0,0 +1,44 @@ |
|||||||
|
package indicator |
||||||
|
|
||||||
|
import ( |
||||||
|
"fmt" |
||||||
|
"sig-pub/pkg/types" |
||||||
|
"sig-pub/pkg/types/series" |
||||||
|
|
||||||
|
"github.com/spf13/cast" |
||||||
|
) |
||||||
|
|
||||||
|
// RSI: 相对强弱指数 (RSI)
|
||||||
|
// rsi define: https://www.investopedia.com/terms/r/rsi.asp
|
||||||
|
type RSI struct { |
||||||
|
series.Series |
||||||
|
values series.Floats |
||||||
|
prices series.Floats |
||||||
|
|
||||||
|
baseline int32 |
||||||
|
} |
||||||
|
|
||||||
|
func NewRSI(baseline int32) *RSI { |
||||||
|
return &RSI{ |
||||||
|
baseline: baseline, |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
func (ind *RSI) Init(prams map[string]any) { |
||||||
|
cast.ToIntE("1") |
||||||
|
} |
||||||
|
|
||||||
|
func (ind *RSI) Update(klines []types.Kline) (err error) { |
||||||
|
for _, kline := range klines { |
||||||
|
c, ok := kline.Close.Float64() |
||||||
|
if !ok { |
||||||
|
err = fmt.Errorf("kline close to float64 error: %s", kline.Close.String()) |
||||||
|
return |
||||||
|
} |
||||||
|
ind.prices.Push(c) |
||||||
|
} |
||||||
|
|
||||||
|
diff := ind.prices.Diff() |
||||||
|
_ = diff |
||||||
|
return |
||||||
|
} |
||||||
@ -1,36 +0,0 @@ |
|||||||
package types |
|
||||||
|
|
||||||
import ( |
|
||||||
"fmt" |
|
||||||
"sig-pub/api/pb" |
|
||||||
) |
|
||||||
|
|
||||||
type Exchange string |
|
||||||
|
|
||||||
var ( |
|
||||||
ExchangeSIG = Exchange(pb.Exchange_SIG.String()) |
|
||||||
ExchangeOKX = Exchange(pb.Exchange_OKX.String()) // okx
|
|
||||||
ExchangeBINANCE = Exchange(pb.Exchange_BINANCE.String()) // 币安
|
|
||||||
) |
|
||||||
|
|
||||||
func (ex Exchange) Exchange2PB() (pb.Exchange, error) { |
|
||||||
switch ex { |
|
||||||
case ExchangeOKX: |
|
||||||
return pb.Exchange_OKX, nil |
|
||||||
case ExchangeBINANCE: |
|
||||||
return pb.Exchange_BINANCE, nil |
|
||||||
default: |
|
||||||
return pb.Exchange_SIG, fmt.Errorf("unknown exchange: %v", ex) |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
func ExchangePBParse(pbex pb.Exchange) (Exchange, bool) { |
|
||||||
switch pbex { |
|
||||||
case pb.Exchange_OKX: |
|
||||||
return ExchangeOKX, true |
|
||||||
case pb.Exchange_BINANCE: |
|
||||||
return ExchangeBINANCE, true |
|
||||||
default: |
|
||||||
return ExchangeSIG, false |
|
||||||
} |
|
||||||
} |
|
||||||
@ -0,0 +1,33 @@ |
|||||||
|
package series |
||||||
|
|
||||||
|
import "github.com/govalues/decimal" |
||||||
|
|
||||||
|
// 值列表
|
||||||
|
type Decimals []decimal.Decimal |
||||||
|
|
||||||
|
func NewDecimals(a ...decimal.Decimal) Decimals { |
||||||
|
return Decimals(a) |
||||||
|
} |
||||||
|
|
||||||
|
func (s *Decimals) Push(v decimal.Decimal) { |
||||||
|
*s = append(*s, v) |
||||||
|
} |
||||||
|
|
||||||
|
func (s *Decimals) Append(vs ...decimal.Decimal) { |
||||||
|
*s = append(*s, vs...) |
||||||
|
} |
||||||
|
|
||||||
|
func (s Decimals) Diff() (values Decimals, err error) { |
||||||
|
var r decimal.Decimal |
||||||
|
for i, v := range s { |
||||||
|
if i == 0 { |
||||||
|
values.Push(decimal.Zero) |
||||||
|
continue |
||||||
|
} |
||||||
|
if r, err = v.Sub(s[i-1]); err != nil { |
||||||
|
return |
||||||
|
} |
||||||
|
values.Push(r) |
||||||
|
} |
||||||
|
return |
||||||
|
} |
||||||
@ -0,0 +1,26 @@ |
|||||||
|
package series |
||||||
|
|
||||||
|
type Floats []float64 |
||||||
|
|
||||||
|
func NewFloats(a ...float64) Floats { |
||||||
|
return Floats(a) |
||||||
|
} |
||||||
|
|
||||||
|
func (s *Floats) Push(v float64) { |
||||||
|
*s = append(*s, v) |
||||||
|
} |
||||||
|
|
||||||
|
func (s *Floats) Append(vs ...float64) { |
||||||
|
*s = append(*s, vs...) |
||||||
|
} |
||||||
|
|
||||||
|
func (s Floats) Diff() (values Floats) { |
||||||
|
for i, v := range s { |
||||||
|
if i == 0 { |
||||||
|
values.Push(0) |
||||||
|
continue |
||||||
|
} |
||||||
|
values.Push(v - s[i-1]) |
||||||
|
} |
||||||
|
return values |
||||||
|
} |
||||||
Loading…
Reference in new issue