mirror of
https://github.com/c9s/bbgo.git
synced 2024-11-10 09:11:55 +00:00
split files
This commit is contained in:
c9s
parent
98192ae91f
commit
ee86a71ebb
158
pkg/bbgo/environment.go
Normal file
158
pkg/bbgo/environment.go
Normal file
|
|
@ -0,0 +1,158 @@
|
|||
package bbgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jmoiron/sqlx"
|
||||
"github.com/sirupsen/logrus"
|
||||
"github.com/spf13/viper"
|
||||
|
||||
"github.com/c9s/bbgo/cmd/cmdutil"
|
||||
"github.com/c9s/bbgo/pkg/service"
|
||||
"github.com/c9s/bbgo/pkg/types"
|
||||
)
|
||||
|
||||
// Environment presents the real exchange data layer
|
||||
type Environment struct {
|
||||
TradeService *service.TradeService
|
||||
TradeSync *service.TradeSync
|
||||
|
||||
sessions map[string]*ExchangeSession
|
||||
}
|
||||
|
||||
func NewDefaultEnvironment(db *sqlx.DB) *Environment {
|
||||
environment := NewEnvironment(db)
|
||||
|
||||
for _, n := range SupportedExchanges {
|
||||
if viper.IsSet(string(n) + "-api-key") {
|
||||
exchange, err := cmdutil.NewExchange(n)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
environment.AddExchange(string(n), exchange)
|
||||
}
|
||||
}
|
||||
|
||||
return environment
|
||||
}
|
||||
|
||||
func NewEnvironment(db *sqlx.DB) *Environment {
|
||||
tradeService := &service.TradeService{DB: db}
|
||||
return &Environment{
|
||||
TradeService: tradeService,
|
||||
TradeSync: &service.TradeSync{
|
||||
Service: tradeService,
|
||||
},
|
||||
sessions: make(map[string]*ExchangeSession),
|
||||
}
|
||||
}
|
||||
|
||||
func (environ *Environment) AddExchange(name string, exchange types.Exchange) (session *ExchangeSession) {
|
||||
session = &ExchangeSession{
|
||||
Name: name,
|
||||
Exchange: exchange,
|
||||
Subscriptions: make(map[types.Subscription]types.Subscription),
|
||||
Markets: make(map[string]types.Market),
|
||||
Trades: make(map[string][]types.Trade),
|
||||
LastPrices: make(map[string]float64),
|
||||
}
|
||||
|
||||
environ.sessions[name] = session
|
||||
return session
|
||||
}
|
||||
|
||||
func (environ *Environment) Init(ctx context.Context) (err error) {
|
||||
startTime := time.Now().AddDate(0, 0, -7) // sync from 7 days ago
|
||||
|
||||
for _, session := range environ.sessions {
|
||||
loadedSymbols := make(map[string]struct{})
|
||||
for _, sub := range session.Subscriptions {
|
||||
loadedSymbols[sub.Symbol] = struct{}{}
|
||||
}
|
||||
|
||||
markets, err := session.Exchange.QueryMarkets(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
session.Markets = markets
|
||||
|
||||
for symbol := range loadedSymbols {
|
||||
if err := environ.TradeSync.Sync(ctx, session.Exchange, symbol, startTime); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var trades []types.Trade
|
||||
|
||||
tradingFeeCurrency := session.Exchange.PlatformFeeCurrency()
|
||||
if strings.HasPrefix(symbol, tradingFeeCurrency) {
|
||||
trades, err = environ.TradeService.QueryForTradingFeeCurrency(symbol, tradingFeeCurrency)
|
||||
} else {
|
||||
trades, err = environ.TradeService.Query(symbol)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
logrus.Infof("symbol %s: %d trades loaded", symbol, len(trades))
|
||||
session.Trades[symbol] = trades
|
||||
|
||||
currentPrice, err := session.Exchange.QueryAveragePrice(ctx, symbol)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
session.LastPrices[symbol] = currentPrice
|
||||
}
|
||||
|
||||
balances, err := session.Exchange.QueryAccountBalances(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
stream := session.Exchange.NewStream()
|
||||
|
||||
session.Stream = stream
|
||||
|
||||
session.Account = &Account{balances: balances}
|
||||
session.Account.BindStream(session.Stream)
|
||||
|
||||
marketDataStore := NewMarketDataStore()
|
||||
marketDataStore.BindStream(session.Stream)
|
||||
|
||||
// update last prices
|
||||
session.Stream.OnKLineClosed(func(kline types.KLine) {
|
||||
session.LastPrices[kline.Symbol] = kline.Close
|
||||
})
|
||||
|
||||
session.Stream.OnTrade(func(trade *types.Trade) {
|
||||
// append trades
|
||||
session.Trades[trade.Symbol] = append(session.Trades[trade.Symbol], *trade)
|
||||
|
||||
if err := environ.TradeService.Insert(*trade); err != nil {
|
||||
logrus.WithError(err).Errorf("trade insert error: %+v", *trade)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (environ *Environment) Connect(ctx context.Context) error {
|
||||
for _, session := range environ.sessions {
|
||||
for _, s := range session.Subscriptions {
|
||||
logrus.Infof("subscribing %s %s %v", s.Symbol, s.Channel, s.Options)
|
||||
session.Stream.Subscribe(s.Channel, s.Symbol, s.Options)
|
||||
}
|
||||
|
||||
if err := session.Stream.Connect(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
43
pkg/bbgo/session.go
Normal file
43
pkg/bbgo/session.go
Normal file
|
|
@ -0,0 +1,43 @@
|
|||
package bbgo
|
||||
|
||||
import "github.com/c9s/bbgo/pkg/types"
|
||||
|
||||
// ExchangeSession presents the exchange connection session
|
||||
// It also maintains and collects the data returned from the stream.
|
||||
type ExchangeSession struct {
|
||||
// Exchange session name
|
||||
Name string
|
||||
|
||||
// The exchange account states
|
||||
Account *Account
|
||||
|
||||
// Stream is the connection stream of the exchange
|
||||
Stream types.Stream
|
||||
|
||||
Subscriptions map[types.Subscription]types.Subscription
|
||||
|
||||
Exchange types.Exchange
|
||||
|
||||
// Markets defines market configuration of a symbol
|
||||
Markets map[string]types.Market
|
||||
|
||||
LastPrices map[string]float64
|
||||
|
||||
// Trades collects the executed trades from the exchange
|
||||
// map: symbol -> []trade
|
||||
Trades map[string][]types.Trade
|
||||
|
||||
MarketDataStore *MarketDataStore
|
||||
}
|
||||
|
||||
// Subscribe save the subscription info, later it will be assigned to the stream
|
||||
func (session *ExchangeSession) Subscribe(channel types.Channel, symbol string, options types.SubscribeOptions) *ExchangeSession {
|
||||
sub := types.Subscription{
|
||||
Channel: channel,
|
||||
Symbol: symbol,
|
||||
Options: options,
|
||||
}
|
||||
|
||||
session.Subscriptions[sub] = sub
|
||||
return session
|
||||
}
|
||||
|
|
@ -2,16 +2,10 @@ package bbgo
|
|||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jmoiron/sqlx"
|
||||
"github.com/pkg/errors"
|
||||
log "github.com/sirupsen/logrus"
|
||||
"github.com/spf13/viper"
|
||||
|
||||
"github.com/c9s/bbgo/cmd/cmdutil"
|
||||
"github.com/c9s/bbgo/pkg/service"
|
||||
"github.com/c9s/bbgo/pkg/types"
|
||||
|
||||
_ "github.com/go-sql-driver/mysql"
|
||||
|
|
@ -37,187 +31,6 @@ type CrossExchangeStrategy interface {
|
|||
Run(ctx context.Context, orderExecutionRouter types.OrderExecutionRouter, sessions map[string]*ExchangeSession) error
|
||||
}
|
||||
|
||||
// ExchangeSession presents the exchange connection session
|
||||
// It also maintains and collects the data returned from the stream.
|
||||
type ExchangeSession struct {
|
||||
// Exchange session name
|
||||
Name string
|
||||
|
||||
// The exchange account states
|
||||
Account *Account
|
||||
|
||||
// Stream is the connection stream of the exchange
|
||||
Stream types.Stream
|
||||
|
||||
Subscriptions map[types.Subscription]types.Subscription
|
||||
|
||||
Exchange types.Exchange
|
||||
|
||||
// Markets defines market configuration of a symbol
|
||||
Markets map[string]types.Market
|
||||
|
||||
LastPrices map[string]float64
|
||||
|
||||
// Trades collects the executed trades from the exchange
|
||||
// map: symbol -> []trade
|
||||
Trades map[string][]types.Trade
|
||||
|
||||
MarketDataStore *MarketDataStore
|
||||
}
|
||||
|
||||
func (session *ExchangeSession) Subscribe(channel types.Channel, symbol string, options types.SubscribeOptions) *ExchangeSession {
|
||||
sub := types.Subscription{
|
||||
Channel: channel,
|
||||
Symbol: symbol,
|
||||
Options: options,
|
||||
}
|
||||
session.Subscriptions[sub] = sub
|
||||
return session
|
||||
}
|
||||
|
||||
// Environment presents the real exchange data layer
|
||||
type Environment struct {
|
||||
TradeService *service.TradeService
|
||||
TradeSync *service.TradeSync
|
||||
|
||||
sessions map[string]*ExchangeSession
|
||||
}
|
||||
|
||||
func NewDefaultEnvironment(db *sqlx.DB) *Environment {
|
||||
environment := NewEnvironment(db)
|
||||
|
||||
for _, n := range SupportedExchanges {
|
||||
if viper.IsSet(string(n) + "-api-key") {
|
||||
exchange, err := cmdutil.NewExchange(n)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
environment.AddExchange(string(n), exchange)
|
||||
}
|
||||
}
|
||||
|
||||
return environment
|
||||
}
|
||||
|
||||
func NewEnvironment(db *sqlx.DB) *Environment {
|
||||
tradeService := &service.TradeService{DB: db}
|
||||
return &Environment{
|
||||
TradeService: tradeService,
|
||||
TradeSync: &service.TradeSync{
|
||||
Service: tradeService,
|
||||
},
|
||||
sessions: make(map[string]*ExchangeSession),
|
||||
}
|
||||
}
|
||||
|
||||
func (environ *Environment) AddExchange(name string, exchange types.Exchange) (session *ExchangeSession) {
|
||||
session = &ExchangeSession{
|
||||
Name: name,
|
||||
Exchange: exchange,
|
||||
Subscriptions: make(map[types.Subscription]types.Subscription),
|
||||
Markets: make(map[string]types.Market),
|
||||
Trades: make(map[string][]types.Trade),
|
||||
LastPrices: make(map[string]float64),
|
||||
}
|
||||
|
||||
environ.sessions[name] = session
|
||||
return session
|
||||
}
|
||||
|
||||
func (environ *Environment) Init(ctx context.Context) (err error) {
|
||||
startTime := time.Now().AddDate(0, 0, -7) // sync from 7 days ago
|
||||
|
||||
for _, session := range environ.sessions {
|
||||
loadedSymbols := make(map[string]struct{})
|
||||
for _, sub := range session.Subscriptions {
|
||||
loadedSymbols[sub.Symbol] = struct{}{}
|
||||
}
|
||||
|
||||
markets, err := session.Exchange.QueryMarkets(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
session.Markets = markets
|
||||
|
||||
for symbol := range loadedSymbols {
|
||||
if err := environ.TradeSync.Sync(ctx, session.Exchange, symbol, startTime); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var trades []types.Trade
|
||||
|
||||
tradingFeeCurrency := session.Exchange.PlatformFeeCurrency()
|
||||
if strings.HasPrefix(symbol, tradingFeeCurrency) {
|
||||
trades, err = environ.TradeService.QueryForTradingFeeCurrency(symbol, tradingFeeCurrency)
|
||||
} else {
|
||||
trades, err = environ.TradeService.Query(symbol)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.Infof("symbol %s: %d trades loaded", symbol, len(trades))
|
||||
session.Trades[symbol] = trades
|
||||
|
||||
currentPrice, err := session.Exchange.QueryAveragePrice(ctx, symbol)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
session.LastPrices[symbol] = currentPrice
|
||||
}
|
||||
|
||||
balances, err := session.Exchange.QueryAccountBalances(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
stream := session.Exchange.NewStream()
|
||||
|
||||
session.Stream = stream
|
||||
|
||||
session.Account = &Account{balances: balances}
|
||||
session.Account.BindStream(session.Stream)
|
||||
|
||||
marketDataStore := NewMarketDataStore()
|
||||
marketDataStore.BindStream(session.Stream)
|
||||
|
||||
// update last prices
|
||||
session.Stream.OnKLineClosed(func(kline types.KLine) {
|
||||
session.LastPrices[kline.Symbol] = kline.Close
|
||||
})
|
||||
|
||||
session.Stream.OnTrade(func(trade *types.Trade) {
|
||||
// append trades
|
||||
session.Trades[trade.Symbol] = append(session.Trades[trade.Symbol], *trade)
|
||||
|
||||
if err := environ.TradeService.Insert(*trade); err != nil {
|
||||
log.WithError(err).Errorf("trade insert error: %+v", *trade)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (environ *Environment) Connect(ctx context.Context) error {
|
||||
for _, session := range environ.sessions {
|
||||
for _, s := range session.Subscriptions {
|
||||
log.Infof("subscribing %s %s %v", s.Symbol, s.Channel, s.Options)
|
||||
session.Stream.Subscribe(s.Channel, s.Symbol, s.Options)
|
||||
}
|
||||
|
||||
if err := session.Stream.Connect(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type Notifiability struct {
|
||||
notifiers []Notifier
|
||||
}
|
||||
|
|
|
|||
|
|
@ -38,6 +38,8 @@ func (e *Exchange) Name() types.ExchangeName {
|
|||
}
|
||||
|
||||
func (e *Exchange) QueryMarkets(ctx context.Context) (types.MarketMap, error) {
|
||||
log.Info("querying market info...")
|
||||
|
||||
exchangeInfo, err := e.Client.NewExchangeInfoService().Do(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
|
|
|||
|
|
@ -35,6 +35,8 @@ func (e *Exchange) Name() types.ExchangeName {
|
|||
}
|
||||
|
||||
func (e *Exchange) QueryMarkets(ctx context.Context) (types.MarketMap, error) {
|
||||
log.Info("querying market info...")
|
||||
|
||||
remoteMarkets, err := e.client.PublicService.Markets()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user