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.
65 lines
1.8 KiB
65 lines
1.8 KiB
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" |
|
"sig-pub/pkg/utils/exit" |
|
"sig-pub/pkg/zlog" |
|
) |
|
|
|
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) { |
|
// victoriametrics 强制刷盘 |
|
exit.AddHook(func() { |
|
zlog.Infof("force flush vmtsdb") |
|
if err := p.vmtsdb.ForceFlush(); err != nil { |
|
zlog.Infof("force flush vmtsdb error: ", err) |
|
} |
|
}, exit.WithOrderFront()) |
|
return |
|
} |
|
|
|
func (p *ExchangeDataPersist) SaveKline(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 |
|
} |
|
|
|
// ListKline 查询k线列表 |
|
func (p *ExchangeDataPersist) ListKline(inst types.TradeInstance, interval types.Interval, start, end int64) (klines []*types.Kline, err error) { |
|
klines, err = p.vmtsdb.ListRangeKline(inst, interval, start, end) |
|
return |
|
}
|
|
|