environment: check syncBufferPeriod

This commit is contained in:
c9s 2024-06-19 14:18:21 +08:00
parent 6b6bf2f722
commit b2722d9e44
No known key found for this signature in database
GPG Key ID: 7385E7E464CB0A54
6 changed files with 37 additions and 11 deletions

View File

@ -346,6 +346,8 @@ type EnvironmentConfig struct {
DisableMarketDataStore bool `json:"disableMarketDataStore"` DisableMarketDataStore bool `json:"disableMarketDataStore"`
MaxSessionTradeBufferSize int `json:"maxSessionTradeBufferSize"` MaxSessionTradeBufferSize int `json:"maxSessionTradeBufferSize"`
SyncBufferPeriod *types.Duration `json:"syncBufferPeriod"`
} }
type Config struct { type Config struct {

View File

@ -37,6 +37,8 @@ func init() {
rand.Seed(time.Now().UnixNano()) rand.Seed(time.Now().UnixNano())
} }
var defaultSyncBufferPeriod = 30 * time.Minute
// IsBackTesting is a global variable that indicates the current environment is back-test or not. // IsBackTesting is a global variable that indicates the current environment is back-test or not.
var IsBackTesting = false var IsBackTesting = false
@ -645,7 +647,17 @@ func (environ *Environment) syncSession(
log.Infof("syncing symbols %v from session %s", symbols, session.Name) log.Infof("syncing symbols %v from session %s", symbols, session.Name)
return environ.SyncService.SyncSessionSymbols(ctx, session.Exchange, syncStartTime, symbols...) syncBufferPeriod := -defaultSyncBufferPeriod
if environ.environmentConfig.SyncBufferPeriod != nil {
syncBufferPeriod = -environ.environmentConfig.SyncBufferPeriod.Duration()
}
if syncBufferPeriod > 0 {
log.Warnf("syncBufferPeriod should be a negative number, given: %d", syncBufferPeriod)
}
syncEndTime := time.Now().Add(syncBufferPeriod)
return environ.SyncService.SyncSessionSymbols(ctx, session.Exchange, syncStartTime, syncEndTime, symbols...)
} }
func (environ *Environment) ConfigureNotificationSystem(ctx context.Context, userConfig *Config) error { func (environ *Environment) ConfigureNotificationSystem(ctx context.Context, userConfig *Config) error {

View File

@ -19,7 +19,10 @@ type OrderService struct {
DB *sqlx.DB DB *sqlx.DB
} }
func (s *OrderService) Sync(ctx context.Context, exchange types.Exchange, symbol string, startTime time.Time) error { func (s *OrderService) Sync(
ctx context.Context, exchange types.Exchange, symbol string,
startTime, endTime time.Time,
) error {
isMargin, isFutures, isIsolated, isolatedSymbol := exchange2.GetSessionAttributes(exchange) isMargin, isFutures, isIsolated, isolatedSymbol := exchange2.GetSessionAttributes(exchange)
// override symbol if isolatedSymbol is not empty // override symbol if isolatedSymbol is not empty
if isIsolated && len(isolatedSymbol) > 0 { if isIsolated && len(isolatedSymbol) > 0 {
@ -77,7 +80,7 @@ func (s *OrderService) Sync(ctx context.Context, exchange types.Exchange, symbol
} }
for _, sel := range tasks { for _, sel := range tasks {
if err := sel.execute(ctx, s.DB, startTime); err != nil { if err := sel.execute(ctx, s.DB, startTime, endTime); err != nil {
return err return err
} }
} }

View File

@ -27,7 +27,9 @@ type SyncService struct {
// SyncSessionSymbols syncs the trades from the given exchange session // SyncSessionSymbols syncs the trades from the given exchange session
func (s *SyncService) SyncSessionSymbols( func (s *SyncService) SyncSessionSymbols(
ctx context.Context, exchange types.Exchange, startTime time.Time, symbols ...string, ctx context.Context, exchange types.Exchange,
startTime, endTime time.Time,
symbols ...string,
) error { ) error {
markets, err := cache.LoadExchangeMarketsWithCache(ctx, exchange) markets, err := cache.LoadExchangeMarketsWithCache(ctx, exchange)
if err != nil { if err != nil {
@ -41,12 +43,12 @@ func (s *SyncService) SyncSessionSymbols(
} }
log.Infof("syncing %s %s trades from %s...", exchange.Name(), symbol, startTime) log.Infof("syncing %s %s trades from %s...", exchange.Name(), symbol, startTime)
if err := s.TradeService.Sync(ctx, exchange, symbol, startTime); err != nil { if err := s.TradeService.Sync(ctx, exchange, symbol, startTime, endTime); err != nil {
return err return err
} }
log.Infof("syncing %s %s orders from %s...", exchange.Name(), symbol, startTime) log.Infof("syncing %s %s orders from %s...", exchange.Name(), symbol, startTime)
if err := s.OrderService.Sync(ctx, exchange, symbol, startTime); err != nil { if err := s.OrderService.Sync(ctx, exchange, symbol, startTime, endTime); err != nil {
return err return err
} }
} }

View File

@ -54,7 +54,10 @@ type SyncTask struct {
LogInsert bool LogInsert bool
} }
func (sel SyncTask) execute(ctx context.Context, db *sqlx.DB, startTime time.Time, args ...time.Time) error { func (sel SyncTask) execute(
ctx context.Context,
db *sqlx.DB, startTime time.Time, endTimeArgs ...time.Time,
) error {
batchBufferRefVal := reflect.MakeSlice(reflect.SliceOf(reflect.TypeOf(sel.Type)), 0, sel.BatchInsertBuffer) batchBufferRefVal := reflect.MakeSlice(reflect.SliceOf(reflect.TypeOf(sel.Type)), 0, sel.BatchInsertBuffer)
// query from db // query from db
@ -84,8 +87,8 @@ func (sel SyncTask) execute(ctx context.Context, db *sqlx.DB, startTime time.Tim
startTime = lastRecordTime(sel, recordSliceRef, startTime) startTime = lastRecordTime(sel, recordSliceRef, startTime)
endTime := time.Now() endTime := time.Now()
if len(args) > 0 { if len(endTimeArgs) > 0 {
endTime = args[0] endTime = endTimeArgs[0]
} }
// asset "" means all assets // asset "" means all assets

View File

@ -58,7 +58,11 @@ func NewTradeService(db *sqlx.DB) *TradeService {
return &TradeService{db} return &TradeService{db}
} }
func (s *TradeService) Sync(ctx context.Context, exchange types.Exchange, symbol string, startTime time.Time) error { func (s *TradeService) Sync(
ctx context.Context,
exchange types.Exchange, symbol string,
startTime, endTime time.Time,
) error {
isMargin, isFutures, isIsolated, isolatedSymbol := exchange2.GetSessionAttributes(exchange) isMargin, isFutures, isIsolated, isolatedSymbol := exchange2.GetSessionAttributes(exchange)
// override symbol if isolatedSymbol is not empty // override symbol if isolatedSymbol is not empty
if isIsolated && len(isolatedSymbol) > 0 { if isIsolated && len(isolatedSymbol) > 0 {
@ -106,7 +110,7 @@ func (s *TradeService) Sync(ctx context.Context, exchange types.Exchange, symbol
} }
for _, sel := range tasks { for _, sel := range tasks {
if err := sel.execute(ctx, s.DB, startTime); err != nil { if err := sel.execute(ctx, s.DB, startTime, endTime); err != nil {
return err return err
} }
} }