lychiyu/bbgo_origin
1
0
Fork 0
mirror of https://github.com/c9s/bbgo.git synced 2024-11-26 00:35:15 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

imporve CrossExchange subscription handling

Browse Source
This commit is contained in:
c9s 2020-11-15 13:27:33 +08:00
parent 94aaaf21b0
commit ded970f5a4
2 changed files with 17 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
pkg/bbgo/environment.go
View File

@ -365,14 +365,13 @@ func (environ *Environment) Connect(ctx context.Context) error {
var logger = log.WithField("session", n) var logger = log.WithField("session", n)
if len(session.Subscriptions) == 0 { if len(session.Subscriptions) == 0 {
logger.Warnf("no subscriptions, exchange session %s will not be connected", session.Name) logger.Warnf("exchange session %s has no subscriptions", session.Name)
continue } else {
} // add the subscribe requests to the stream
for _, s := range session.Subscriptions {
// add the subscribe requests to the stream logger.Infof("subscribing %s %s %v", s.Symbol, s.Channel, s.Options)
for _, s := range session.Subscriptions { session.Stream.Subscribe(s.Channel, s.Symbol, s.Options)
logger.Infof("subscribing %s %s %v", s.Symbol, s.Channel, s.Options) }
session.Stream.Subscribe(s.Channel, s.Symbol, s.Options)
} }
logger.Infof("connecting session %s...", session.Name) logger.Infof("connecting session %s...", session.Name)

10
pkg/bbgo/trader.go
View File

@ -23,6 +23,10 @@ type ExchangeSessionSubscriber interface {
Subscribe(session *ExchangeSession) Subscribe(session *ExchangeSession)
} }
type CrossExchangeSessionSubscriber interface {
Subscribe(sessions map[string]*ExchangeSession)
}
type CrossExchangeStrategy interface { type CrossExchangeStrategy interface {
Run(ctx context.Context, orderExecutionRouter OrderExecutionRouter, sessions map[string]*ExchangeSession) error Run(ctx context.Context, orderExecutionRouter OrderExecutionRouter, sessions map[string]*ExchangeSession) error
} }
@ -220,6 +224,12 @@ func (trader *Trader) Run(ctx context.Context) error {
sessions: trader.environment.sessions, sessions: trader.environment.sessions,
} }
for _, strategy := range trader.crossExchangeStrategies {
if subscriber, ok := strategy.(CrossExchangeSessionSubscriber); ok {
subscriber.Subscribe(trader.environment.sessions)
}
}
for _, strategy := range trader.crossExchangeStrategies { for _, strategy := range trader.crossExchangeStrategies {
if err := strategy.Run(ctx, router, trader.environment.sessions); err != nil { if err := strategy.Run(ctx, router, trader.environment.sessions); err != nil {
return err return err