mirror of
https://github.com/c9s/bbgo.git
synced 2024-11-22 23:05:15 +00:00
888 lines
27 KiB
Go
888 lines
27 KiB
Go
package bbgo
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
log "github.com/sirupsen/logrus"
|
|
"github.com/spf13/viper"
|
|
|
|
"github.com/c9s/bbgo/pkg/cmd/cmdutil"
|
|
"github.com/c9s/bbgo/pkg/fixedpoint"
|
|
"github.com/c9s/bbgo/pkg/indicator"
|
|
"github.com/c9s/bbgo/pkg/service"
|
|
"github.com/c9s/bbgo/pkg/types"
|
|
"github.com/c9s/bbgo/pkg/util"
|
|
)
|
|
|
|
var (
|
|
debugEWMA = false
|
|
debugSMA = false
|
|
)
|
|
|
|
func init() {
|
|
// when using --dotenv option, the dotenv is loaded from command.PersistentPreRunE, not init.
|
|
// hence here the env var won't enable the debug flag
|
|
util.SetEnvVarBool("DEBUG_EWMA", &debugEWMA)
|
|
util.SetEnvVarBool("DEBUG_SMA", &debugSMA)
|
|
}
|
|
|
|
type StandardIndicatorSet struct {
|
|
Symbol string
|
|
// Standard indicators
|
|
// interval -> window
|
|
sma map[types.IntervalWindow]*indicator.SMA
|
|
ewma map[types.IntervalWindow]*indicator.EWMA
|
|
boll map[types.IntervalWindow]*indicator.BOLL
|
|
stoch map[types.IntervalWindow]*indicator.STOCH
|
|
volatility map[types.IntervalWindow]*indicator.VOLATILITY
|
|
|
|
store *MarketDataStore
|
|
}
|
|
|
|
func NewStandardIndicatorSet(symbol string, store *MarketDataStore) *StandardIndicatorSet {
|
|
set := &StandardIndicatorSet{
|
|
Symbol: symbol,
|
|
sma: make(map[types.IntervalWindow]*indicator.SMA),
|
|
ewma: make(map[types.IntervalWindow]*indicator.EWMA),
|
|
boll: make(map[types.IntervalWindow]*indicator.BOLL),
|
|
stoch: make(map[types.IntervalWindow]*indicator.STOCH),
|
|
volatility: make(map[types.IntervalWindow]*indicator.VOLATILITY),
|
|
store: store,
|
|
}
|
|
|
|
// let us pre-defined commonly used intervals
|
|
for interval := range types.SupportedIntervals {
|
|
for _, window := range []int{7, 25, 99} {
|
|
iw := types.IntervalWindow{Interval: interval, Window: window}
|
|
set.sma[iw] = &indicator.SMA{IntervalWindow: iw}
|
|
set.sma[iw].Bind(store)
|
|
if debugSMA {
|
|
set.sma[iw].OnUpdate(func(value float64) {
|
|
log.Infof("%s SMA %s: %f", symbol, iw.String(), value)
|
|
})
|
|
}
|
|
|
|
set.ewma[iw] = &indicator.EWMA{IntervalWindow: iw}
|
|
set.ewma[iw].Bind(store)
|
|
|
|
// if debug EWMA is enabled, we add the debug handler
|
|
if debugEWMA {
|
|
set.ewma[iw].OnUpdate(func(value float64) {
|
|
log.Infof("%s EWMA %s: %f", symbol, iw.String(), value)
|
|
})
|
|
}
|
|
|
|
}
|
|
|
|
// setup boll indicator, we may refactor boll indicator by subscribing SMA indicator,
|
|
// however, since general used BOLLINGER band use window 21, which is not in the existing SMA indicator sets.
|
|
// Pull out the bandwidth configuration as the boll Key
|
|
iw := types.IntervalWindow{Interval: interval, Window: 21}
|
|
set.boll[iw] = &indicator.BOLL{IntervalWindow: iw, K: 2.0}
|
|
set.boll[iw].Bind(store)
|
|
}
|
|
|
|
return set
|
|
}
|
|
|
|
// BOLL returns the bollinger band indicator of the given interval and the window,
|
|
// Please note that the K for std dev is fixed and defaults to 2.0
|
|
func (set *StandardIndicatorSet) BOLL(iw types.IntervalWindow, bandWidth float64) *indicator.BOLL {
|
|
inc, ok := set.boll[iw]
|
|
if !ok {
|
|
inc = &indicator.BOLL{IntervalWindow: iw, K: bandWidth}
|
|
inc.Bind(set.store)
|
|
set.boll[iw] = inc
|
|
}
|
|
|
|
return inc
|
|
}
|
|
|
|
// SMA returns the simple moving average indicator of the given interval and the window size.
|
|
func (set *StandardIndicatorSet) SMA(iw types.IntervalWindow) *indicator.SMA {
|
|
inc, ok := set.sma[iw]
|
|
if !ok {
|
|
inc = &indicator.SMA{IntervalWindow: iw}
|
|
inc.Bind(set.store)
|
|
set.sma[iw] = inc
|
|
}
|
|
|
|
return inc
|
|
}
|
|
|
|
// EWMA returns the exponential weighed moving average indicator of the given interval and the window size.
|
|
func (set *StandardIndicatorSet) EWMA(iw types.IntervalWindow) *indicator.EWMA {
|
|
inc, ok := set.ewma[iw]
|
|
if !ok {
|
|
inc = &indicator.EWMA{IntervalWindow: iw}
|
|
inc.Bind(set.store)
|
|
set.ewma[iw] = inc
|
|
}
|
|
|
|
return inc
|
|
}
|
|
|
|
func (set *StandardIndicatorSet) STOCH(iw types.IntervalWindow) *indicator.STOCH {
|
|
inc, ok := set.stoch[iw]
|
|
if !ok {
|
|
inc = &indicator.STOCH{IntervalWindow: iw}
|
|
inc.Bind(set.store)
|
|
set.stoch[iw] = inc
|
|
}
|
|
|
|
return inc
|
|
}
|
|
|
|
// VOLATILITY returns the volatility(stddev) indicator of the given interval and the window size.
|
|
func (set *StandardIndicatorSet) VOLATILITY(iw types.IntervalWindow) *indicator.VOLATILITY {
|
|
inc, ok := set.volatility[iw]
|
|
if !ok {
|
|
inc = &indicator.VOLATILITY{IntervalWindow: iw}
|
|
inc.Bind(set.store)
|
|
set.volatility[iw] = inc
|
|
}
|
|
|
|
return inc
|
|
}
|
|
|
|
// ExchangeSession presents the exchange connection Session
|
|
// It also maintains and collects the data returned from the stream.
|
|
type ExchangeSession struct {
|
|
// exchange Session based notification system
|
|
// we make it as a value field so that we can configure it separately
|
|
Notifiability `json:"-" yaml:"-"`
|
|
|
|
// ---------------------------
|
|
// Session config fields
|
|
// ---------------------------
|
|
|
|
// Exchange Session name
|
|
Name string `json:"name,omitempty" yaml:"name,omitempty"`
|
|
ExchangeName types.ExchangeName `json:"exchange" yaml:"exchange"`
|
|
EnvVarPrefix string `json:"envVarPrefix" yaml:"envVarPrefix"`
|
|
Key string `json:"key,omitempty" yaml:"key,omitempty"`
|
|
Secret string `json:"secret,omitempty" yaml:"secret,omitempty"`
|
|
Passphrase string `json:"passphrase,omitempty" yaml:"passphrase,omitempty"`
|
|
SubAccount string `json:"subAccount,omitempty" yaml:"subAccount,omitempty"`
|
|
|
|
// Withdrawal is used for enabling withdrawal functions
|
|
Withdrawal bool `json:"withdrawal,omitempty" yaml:"withdrawal,omitempty"`
|
|
MakerFeeRate fixedpoint.Value `json:"makerFeeRate,omitempty" yaml:"makerFeeRate,omitempty"`
|
|
TakerFeeRate fixedpoint.Value `json:"takerFeeRate,omitempty" yaml:"takerFeeRate,omitempty"`
|
|
|
|
PublicOnly bool `json:"publicOnly,omitempty" yaml:"publicOnly"`
|
|
Margin bool `json:"margin,omitempty" yaml:"margin"`
|
|
IsolatedMargin bool `json:"isolatedMargin,omitempty" yaml:"isolatedMargin,omitempty"`
|
|
IsolatedMarginSymbol string `json:"isolatedMarginSymbol,omitempty" yaml:"isolatedMarginSymbol,omitempty"`
|
|
|
|
Futures bool `json:"futures,omitempty" yaml:"futures"`
|
|
IsolatedFutures bool `json:"isolatedFutures,omitempty" yaml:"isolatedFutures,omitempty"`
|
|
IsolatedFuturesSymbol string `json:"isolatedFuturesSymbol,omitempty" yaml:"isolatedFuturesSymbol,omitempty"`
|
|
|
|
// ---------------------------
|
|
// Runtime fields
|
|
// ---------------------------
|
|
|
|
// The exchange account states
|
|
Account *types.Account `json:"-" yaml:"-"`
|
|
|
|
IsInitialized bool `json:"-" yaml:"-"`
|
|
|
|
OrderExecutor *ExchangeOrderExecutor `json:"orderExecutor,omitempty" yaml:"orderExecutor,omitempty"`
|
|
|
|
// UserDataStream is the connection stream of the exchange
|
|
UserDataStream types.Stream `json:"-" yaml:"-"`
|
|
MarketDataStream types.Stream `json:"-" yaml:"-"`
|
|
|
|
// Subscriptions
|
|
// this is a read-only field when running strategy
|
|
Subscriptions map[types.Subscription]types.Subscription `json:"-" yaml:"-"`
|
|
|
|
Exchange types.Exchange `json:"-" yaml:"-"`
|
|
|
|
// Trades collects the executed trades from the exchange
|
|
// map: symbol -> []trade
|
|
Trades map[string]*types.TradeSlice `json:"-" yaml:"-"`
|
|
|
|
// markets defines market configuration of a symbol
|
|
markets map[string]types.Market
|
|
|
|
// orderBooks stores the streaming order book
|
|
orderBooks map[string]*types.StreamOrderBook
|
|
|
|
// startPrices is used for backtest
|
|
startPrices map[string]float64
|
|
|
|
lastPrices map[string]float64
|
|
lastPriceUpdatedAt time.Time
|
|
|
|
// marketDataStores contains the market data store of each market
|
|
marketDataStores map[string]*MarketDataStore
|
|
|
|
positions map[string]*types.Position
|
|
|
|
// standard indicators of each market
|
|
standardIndicatorSets map[string]*StandardIndicatorSet
|
|
|
|
orderStores map[string]*OrderStore
|
|
|
|
usedSymbols map[string]struct{}
|
|
initializedSymbols map[string]struct{}
|
|
|
|
logger *log.Entry
|
|
}
|
|
|
|
func NewExchangeSession(name string, exchange types.Exchange) *ExchangeSession {
|
|
userDataStream := exchange.NewStream()
|
|
marketDataStream := exchange.NewStream()
|
|
marketDataStream.SetPublicOnly()
|
|
|
|
session := &ExchangeSession{
|
|
Notifiability: Notifiability{
|
|
SymbolChannelRouter: NewPatternChannelRouter(nil),
|
|
SessionChannelRouter: NewPatternChannelRouter(nil),
|
|
ObjectChannelRouter: NewObjectChannelRouter(),
|
|
},
|
|
|
|
Name: name,
|
|
Exchange: exchange,
|
|
UserDataStream: userDataStream,
|
|
MarketDataStream: marketDataStream,
|
|
Subscriptions: make(map[types.Subscription]types.Subscription),
|
|
Account: &types.Account{},
|
|
Trades: make(map[string]*types.TradeSlice),
|
|
|
|
orderBooks: make(map[string]*types.StreamOrderBook),
|
|
markets: make(map[string]types.Market),
|
|
startPrices: make(map[string]float64),
|
|
lastPrices: make(map[string]float64),
|
|
positions: make(map[string]*types.Position),
|
|
marketDataStores: make(map[string]*MarketDataStore),
|
|
standardIndicatorSets: make(map[string]*StandardIndicatorSet),
|
|
orderStores: make(map[string]*OrderStore),
|
|
usedSymbols: make(map[string]struct{}),
|
|
initializedSymbols: make(map[string]struct{}),
|
|
logger: log.WithField("session", name),
|
|
}
|
|
|
|
session.OrderExecutor = &ExchangeOrderExecutor{
|
|
// copy the notification system so that we can route
|
|
Notifiability: session.Notifiability,
|
|
Session: session,
|
|
}
|
|
|
|
return session
|
|
}
|
|
|
|
// Init initializes the basic data structure and market information by its exchange.
|
|
// Note that the subscribed symbols are not loaded in this stage.
|
|
func (session *ExchangeSession) Init(ctx context.Context, environ *Environment) error {
|
|
if session.IsInitialized {
|
|
return ErrSessionAlreadyInitialized
|
|
}
|
|
|
|
var log = log.WithField("session", session.Name)
|
|
|
|
// load markets first
|
|
|
|
var disableMarketsCache = false
|
|
var markets types.MarketMap
|
|
var err error
|
|
if util.SetEnvVarBool("DISABLE_MARKETS_CACHE", &disableMarketsCache); disableMarketsCache {
|
|
markets, err = session.Exchange.QueryMarkets(ctx)
|
|
} else {
|
|
markets, err = LoadExchangeMarketsWithCache(ctx, session.Exchange)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
if len(markets) == 0 {
|
|
return fmt.Errorf("market config should not be empty")
|
|
}
|
|
|
|
session.markets = markets
|
|
|
|
// query and initialize the balances
|
|
if !session.PublicOnly {
|
|
log.Infof("querying balances from session %s...", session.Name)
|
|
balances, err := session.Exchange.QueryAccountBalances(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
log.Infof("%s account", session.Name)
|
|
balances.Print()
|
|
session.Account.UpdateBalances(balances)
|
|
|
|
// forward trade updates and order updates to the order executor
|
|
session.UserDataStream.OnTradeUpdate(session.OrderExecutor.EmitTradeUpdate)
|
|
session.UserDataStream.OnOrderUpdate(session.OrderExecutor.EmitOrderUpdate)
|
|
session.Account.BindStream(session.UserDataStream)
|
|
|
|
session.bindConnectionStatusNotification(session.UserDataStream, "user data")
|
|
|
|
// if metrics mode is enabled, we bind the callbacks to update metrics
|
|
if viper.GetBool("metrics") {
|
|
session.metricsBalancesUpdater(balances)
|
|
session.bindUserDataStreamMetrics(session.UserDataStream)
|
|
}
|
|
}
|
|
|
|
// TODO: move this logic to Environment struct
|
|
// if back-test service is not set, meaning we are not back-testing
|
|
// we should insert trade into db right before everything
|
|
if environ.BacktestService == nil {
|
|
// if trade service is configured, we have the db configured
|
|
if environ.TradeService != nil {
|
|
session.UserDataStream.OnTradeUpdate(func(trade types.Trade) {
|
|
if err := environ.TradeService.Insert(trade); err != nil {
|
|
log.WithError(err).Errorf("trade insert error: %+v", trade)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
if viper.GetBool("debug-kline") {
|
|
session.MarketDataStream.OnKLine(func(kline types.KLine) {
|
|
log.WithField("marketData", "kline").Infof("kline: %+v", kline)
|
|
})
|
|
session.MarketDataStream.OnKLineClosed(func(kline types.KLine) {
|
|
log.WithField("marketData", "kline").Infof("kline closed: %+v", kline)
|
|
})
|
|
}
|
|
|
|
// update last prices
|
|
session.MarketDataStream.OnKLineClosed(func(kline types.KLine) {
|
|
if _, ok := session.startPrices[kline.Symbol]; !ok {
|
|
session.startPrices[kline.Symbol] = kline.Open
|
|
}
|
|
|
|
session.lastPrices[kline.Symbol] = kline.Close
|
|
})
|
|
|
|
session.IsInitialized = true
|
|
return nil
|
|
}
|
|
|
|
func (session *ExchangeSession) InitSymbols(ctx context.Context, environ *Environment) error {
|
|
if err := session.initUsedSymbols(ctx, environ); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// initUsedSymbols uses usedSymbols to initialize the related data structure
|
|
func (session *ExchangeSession) initUsedSymbols(ctx context.Context, environ *Environment) error {
|
|
for symbol := range session.usedSymbols {
|
|
if err := session.initSymbol(ctx, environ, symbol); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// initSymbol loads trades for the symbol, bind stream callbacks, init positions, market data store.
|
|
// please note, initSymbol can not be called for the same symbol for twice
|
|
func (session *ExchangeSession) initSymbol(ctx context.Context, environ *Environment, symbol string) error {
|
|
if _, ok := session.initializedSymbols[symbol]; ok {
|
|
// return fmt.Errorf("symbol %s is already initialized", symbol)
|
|
return nil
|
|
}
|
|
|
|
market, ok := session.markets[symbol]
|
|
if !ok {
|
|
return fmt.Errorf("market %s is not defined", symbol)
|
|
}
|
|
|
|
var err error
|
|
var trades []types.Trade
|
|
if environ.SyncService != nil && environ.BacktestService == nil {
|
|
tradingFeeCurrency := session.Exchange.PlatformFeeCurrency()
|
|
if strings.HasPrefix(symbol, tradingFeeCurrency) {
|
|
trades, err = environ.TradeService.QueryForTradingFeeCurrency(session.Exchange.Name(), symbol, tradingFeeCurrency)
|
|
} else {
|
|
trades, err = environ.TradeService.Query(service.QueryTradesOptions{
|
|
Exchange: session.Exchange.Name(),
|
|
Symbol: symbol,
|
|
})
|
|
}
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
log.Infof("symbol %s: %d trades loaded", symbol, len(trades))
|
|
}
|
|
|
|
session.Trades[symbol] = &types.TradeSlice{Trades: trades}
|
|
session.UserDataStream.OnTradeUpdate(func(trade types.Trade) {
|
|
if trade.Symbol == symbol {
|
|
session.Trades[symbol].Append(trade)
|
|
}
|
|
})
|
|
|
|
position := &types.Position{
|
|
Symbol: symbol,
|
|
BaseCurrency: market.BaseCurrency,
|
|
QuoteCurrency: market.QuoteCurrency,
|
|
}
|
|
position.AddTrades(trades)
|
|
position.BindStream(session.UserDataStream)
|
|
session.positions[symbol] = position
|
|
|
|
orderStore := NewOrderStore(symbol)
|
|
orderStore.AddOrderUpdate = true
|
|
|
|
orderStore.BindStream(session.UserDataStream)
|
|
session.orderStores[symbol] = orderStore
|
|
|
|
marketDataStore := NewMarketDataStore(symbol)
|
|
marketDataStore.BindStream(session.MarketDataStream)
|
|
session.marketDataStores[symbol] = marketDataStore
|
|
|
|
standardIndicatorSet := NewStandardIndicatorSet(symbol, marketDataStore)
|
|
session.standardIndicatorSets[symbol] = standardIndicatorSet
|
|
|
|
// used kline intervals by the given symbol
|
|
var klineSubscriptions = map[types.Interval]struct{}{}
|
|
|
|
// always subscribe the 1m kline so we can make sure the connection persists.
|
|
klineSubscriptions[types.Interval1m] = struct{}{}
|
|
|
|
// Aggregate the intervals that we are using in the subscriptions.
|
|
for _, sub := range session.Subscriptions {
|
|
switch sub.Channel {
|
|
case types.BookChannel:
|
|
book := types.NewStreamBook(sub.Symbol)
|
|
book.BindStream(session.MarketDataStream)
|
|
session.orderBooks[sub.Symbol] = book
|
|
|
|
case types.KLineChannel:
|
|
if sub.Options.Interval == "" {
|
|
continue
|
|
}
|
|
|
|
if sub.Symbol == symbol {
|
|
klineSubscriptions[types.Interval(sub.Options.Interval)] = struct{}{}
|
|
}
|
|
}
|
|
}
|
|
|
|
for interval := range klineSubscriptions {
|
|
// avoid querying the last unclosed kline
|
|
endTime := environ.startTime
|
|
kLines, err := session.Exchange.QueryKLines(ctx, symbol, interval, types.KLineQueryOptions{
|
|
EndTime: &endTime,
|
|
Limit: 1000, // indicators need at least 100
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if len(kLines) == 0 {
|
|
log.Warnf("no kline data for %s %s (end time <= %s)", symbol, interval, environ.startTime)
|
|
continue
|
|
}
|
|
|
|
// update last prices by the given kline
|
|
lastKLine := kLines[len(kLines)-1]
|
|
if interval == types.Interval1m {
|
|
log.Infof("last kline %+v", lastKLine)
|
|
session.lastPrices[symbol] = lastKLine.Close
|
|
}
|
|
|
|
for _, k := range kLines {
|
|
// let market data store trigger the update, so that the indicator could be updated too.
|
|
marketDataStore.AddKLine(k)
|
|
}
|
|
}
|
|
|
|
log.Infof("%s last price: %f", symbol, session.lastPrices[symbol])
|
|
|
|
session.initializedSymbols[symbol] = struct{}{}
|
|
return nil
|
|
}
|
|
|
|
func (session *ExchangeSession) StandardIndicatorSet(symbol string) (*StandardIndicatorSet, bool) {
|
|
set, ok := session.standardIndicatorSets[symbol]
|
|
return set, ok
|
|
}
|
|
|
|
func (session *ExchangeSession) Position(symbol string) (pos *types.Position, ok bool) {
|
|
pos, ok = session.positions[symbol]
|
|
if ok {
|
|
return pos, ok
|
|
}
|
|
|
|
market, ok := session.markets[symbol]
|
|
if !ok {
|
|
return nil, false
|
|
}
|
|
|
|
pos = &types.Position{
|
|
Symbol: symbol,
|
|
BaseCurrency: market.BaseCurrency,
|
|
QuoteCurrency: market.QuoteCurrency,
|
|
}
|
|
ok = true
|
|
session.positions[symbol] = pos
|
|
return pos, ok
|
|
}
|
|
|
|
func (session *ExchangeSession) Positions() map[string]*types.Position {
|
|
return session.positions
|
|
}
|
|
|
|
// MarketDataStore returns the market data store of a symbol
|
|
func (session *ExchangeSession) MarketDataStore(symbol string) (s *MarketDataStore, ok bool) {
|
|
s, ok = session.marketDataStores[symbol]
|
|
return s, ok
|
|
}
|
|
|
|
// MarketDataStore returns the market data store of a symbol
|
|
func (session *ExchangeSession) OrderBook(symbol string) (s *types.StreamOrderBook, ok bool) {
|
|
s, ok = session.orderBooks[symbol]
|
|
return s, ok
|
|
}
|
|
|
|
func (session *ExchangeSession) StartPrice(symbol string) (price float64, ok bool) {
|
|
price, ok = session.startPrices[symbol]
|
|
return price, ok
|
|
}
|
|
|
|
func (session *ExchangeSession) LastPrice(symbol string) (price float64, ok bool) {
|
|
price, ok = session.lastPrices[symbol]
|
|
return price, ok
|
|
}
|
|
|
|
func (session *ExchangeSession) LastPrices() map[string]float64 {
|
|
return session.lastPrices
|
|
}
|
|
|
|
func (session *ExchangeSession) Market(symbol string) (market types.Market, ok bool) {
|
|
market, ok = session.markets[symbol]
|
|
return market, ok
|
|
}
|
|
|
|
func (session *ExchangeSession) Markets() map[string]types.Market {
|
|
return session.markets
|
|
}
|
|
|
|
func (session *ExchangeSession) OrderStore(symbol string) (store *OrderStore, ok bool) {
|
|
store, ok = session.orderStores[symbol]
|
|
return store, ok
|
|
}
|
|
|
|
func (session *ExchangeSession) OrderStores() map[string]*OrderStore {
|
|
return session.orderStores
|
|
}
|
|
|
|
// 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 {
|
|
if channel == types.KLineChannel && len(options.Interval) == 0 {
|
|
panic("subscription interval for kline can not be empty")
|
|
}
|
|
|
|
sub := types.Subscription{
|
|
Channel: channel,
|
|
Symbol: symbol,
|
|
Options: options,
|
|
}
|
|
|
|
// add to the loaded symbol table
|
|
session.usedSymbols[symbol] = struct{}{}
|
|
session.Subscriptions[sub] = sub
|
|
return session
|
|
}
|
|
|
|
func (session *ExchangeSession) FormatOrder(order types.SubmitOrder) (types.SubmitOrder, error) {
|
|
market, ok := session.Market(order.Symbol)
|
|
if !ok {
|
|
return order, fmt.Errorf("market is not defined: %s", order.Symbol)
|
|
}
|
|
|
|
order.Market = market
|
|
return order, nil
|
|
}
|
|
|
|
func (session *ExchangeSession) UpdatePrices(ctx context.Context) (err error) {
|
|
if session.lastPriceUpdatedAt.After(time.Now().Add(-time.Hour)) {
|
|
return nil
|
|
}
|
|
|
|
balances := session.Account.Balances()
|
|
|
|
var symbols []string
|
|
for _, b := range balances {
|
|
symbols = append(symbols, b.Currency+"USDT")
|
|
symbols = append(symbols, "USDT"+b.Currency)
|
|
}
|
|
|
|
tickers, err := session.Exchange.QueryTickers(ctx, symbols...)
|
|
|
|
if err != nil || len(tickers) == 0 {
|
|
return err
|
|
}
|
|
|
|
for k, v := range tickers {
|
|
session.lastPrices[k] = v.Last
|
|
}
|
|
|
|
session.lastPriceUpdatedAt = time.Now()
|
|
return err
|
|
}
|
|
|
|
func (session *ExchangeSession) FindPossibleSymbols() (symbols []string, err error) {
|
|
// If the session is an isolated margin session, there will be only the isolated margin symbol
|
|
if session.Margin && session.IsolatedMargin {
|
|
return []string{
|
|
session.IsolatedMarginSymbol,
|
|
}, nil
|
|
}
|
|
|
|
var balances = session.Account.Balances()
|
|
var fiatAssets []string
|
|
|
|
for _, currency := range types.FiatCurrencies {
|
|
if balance, ok := balances[currency]; ok && balance.Total() > 0 {
|
|
fiatAssets = append(fiatAssets, currency)
|
|
}
|
|
}
|
|
|
|
var symbolMap = map[string]struct{}{}
|
|
|
|
for _, market := range session.Markets() {
|
|
// ignore the markets that are not fiat currency markets
|
|
if !util.StringSliceContains(fiatAssets, market.QuoteCurrency) {
|
|
continue
|
|
}
|
|
|
|
// ignore the asset that we don't have in the balance sheet
|
|
balance, hasAsset := balances[market.BaseCurrency]
|
|
if !hasAsset || balance.Total() == 0 {
|
|
continue
|
|
}
|
|
|
|
symbolMap[market.Symbol] = struct{}{}
|
|
}
|
|
|
|
for s := range symbolMap {
|
|
symbols = append(symbols, s)
|
|
}
|
|
|
|
return symbols, nil
|
|
}
|
|
|
|
// InitExchange initialize the exchange instance and allocate memory for fields
|
|
// In this stage, the session var could be loaded from the JSON config, so the pointer fields are still nil
|
|
// The Init method will be called after this stage, environment.Init will call the session.Init method later.
|
|
func (session *ExchangeSession) InitExchange(name string, exchange types.Exchange) error {
|
|
var err error
|
|
var exchangeName = session.ExchangeName
|
|
if exchange == nil {
|
|
if session.PublicOnly {
|
|
exchange, err = cmdutil.NewExchangePublic(exchangeName)
|
|
} else {
|
|
if session.Key != "" && session.Secret != "" {
|
|
exchange, err = cmdutil.NewExchangeStandard(exchangeName, session.Key, session.Secret, session.Passphrase, session.SubAccount)
|
|
} else {
|
|
exchange, err = cmdutil.NewExchangeWithEnvVarPrefix(exchangeName, session.EnvVarPrefix)
|
|
}
|
|
}
|
|
}
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// configure exchange
|
|
if session.Margin {
|
|
marginExchange, ok := exchange.(types.MarginExchange)
|
|
if !ok {
|
|
return fmt.Errorf("exchange %s does not support margin", exchangeName)
|
|
}
|
|
|
|
if session.IsolatedMargin {
|
|
marginExchange.UseIsolatedMargin(session.IsolatedMarginSymbol)
|
|
} else {
|
|
marginExchange.UseMargin()
|
|
}
|
|
}
|
|
|
|
if session.Futures {
|
|
futuresExchange, ok := exchange.(types.FuturesExchange)
|
|
if !ok {
|
|
return fmt.Errorf("exchange %s does not support futures", exchangeName)
|
|
}
|
|
|
|
if session.IsolatedFutures {
|
|
futuresExchange.UseIsolatedFutures(session.IsolatedFuturesSymbol)
|
|
} else {
|
|
futuresExchange.UseFutures()
|
|
}
|
|
}
|
|
|
|
session.Name = name
|
|
session.Notifiability = Notifiability{
|
|
SymbolChannelRouter: NewPatternChannelRouter(nil),
|
|
SessionChannelRouter: NewPatternChannelRouter(nil),
|
|
ObjectChannelRouter: NewObjectChannelRouter(),
|
|
}
|
|
session.Exchange = exchange
|
|
session.UserDataStream = exchange.NewStream()
|
|
session.MarketDataStream = exchange.NewStream()
|
|
session.MarketDataStream.SetPublicOnly()
|
|
|
|
// pointer fields
|
|
session.Subscriptions = make(map[types.Subscription]types.Subscription)
|
|
session.Account = &types.Account{}
|
|
session.Trades = make(map[string]*types.TradeSlice)
|
|
|
|
session.orderBooks = make(map[string]*types.StreamOrderBook)
|
|
session.markets = make(map[string]types.Market)
|
|
session.lastPrices = make(map[string]float64)
|
|
session.startPrices = make(map[string]float64)
|
|
session.marketDataStores = make(map[string]*MarketDataStore)
|
|
session.positions = make(map[string]*types.Position)
|
|
session.standardIndicatorSets = make(map[string]*StandardIndicatorSet)
|
|
session.orderStores = make(map[string]*OrderStore)
|
|
session.OrderExecutor = &ExchangeOrderExecutor{
|
|
// copy the notification system so that we can route
|
|
Notifiability: session.Notifiability,
|
|
Session: session,
|
|
}
|
|
|
|
session.usedSymbols = make(map[string]struct{})
|
|
session.initializedSymbols = make(map[string]struct{})
|
|
session.logger = log.WithField("session", name)
|
|
return nil
|
|
}
|
|
|
|
func (session *ExchangeSession) MarginType() string {
|
|
margin := "none"
|
|
if session.Margin {
|
|
margin = "margin"
|
|
if session.IsolatedMargin {
|
|
margin = "isolated"
|
|
}
|
|
}
|
|
return margin
|
|
}
|
|
|
|
func (session *ExchangeSession) metricsBalancesUpdater(balances types.BalanceMap) {
|
|
for currency, balance := range balances {
|
|
labels := prometheus.Labels{
|
|
"exchange": session.ExchangeName.String(),
|
|
"margin": session.MarginType(),
|
|
"symbol": session.IsolatedMarginSymbol,
|
|
"currency": currency,
|
|
}
|
|
|
|
metricsTotalBalances.With(labels).Set(balance.Total().Float64())
|
|
metricsLockedBalances.With(labels).Set(balance.Locked.Float64())
|
|
metricsAvailableBalances.With(labels).Set(balance.Available.Float64())
|
|
metricsLastUpdateTimeBalance.With(prometheus.Labels{
|
|
"exchange": session.ExchangeName.String(),
|
|
"margin": session.MarginType(),
|
|
"channel": "user",
|
|
"data_type": "balance",
|
|
"symbol": "",
|
|
"currency": currency,
|
|
}).SetToCurrentTime()
|
|
}
|
|
|
|
}
|
|
|
|
func (session *ExchangeSession) metricsOrderUpdater(order types.Order) {
|
|
metricsLastUpdateTimeBalance.With(prometheus.Labels{
|
|
"exchange": session.ExchangeName.String(),
|
|
"margin": session.MarginType(),
|
|
"channel": "user",
|
|
"data_type": "order",
|
|
"symbol": order.Symbol,
|
|
"currency": "",
|
|
}).SetToCurrentTime()
|
|
}
|
|
|
|
func (session *ExchangeSession) metricsTradeUpdater(trade types.Trade) {
|
|
labels := prometheus.Labels{
|
|
"exchange": session.ExchangeName.String(),
|
|
"margin": session.MarginType(),
|
|
"side": trade.Side.String(),
|
|
"symbol": trade.Symbol,
|
|
"liquidity": trade.Liquidity(),
|
|
}
|
|
metricsTradingVolume.With(labels).Add(trade.Quantity * trade.Price)
|
|
metricsTradesTotal.With(labels).Inc()
|
|
metricsLastUpdateTimeBalance.With(prometheus.Labels{
|
|
"exchange": session.ExchangeName.String(),
|
|
"margin": session.MarginType(),
|
|
"channel": "user",
|
|
"data_type": "trade",
|
|
"symbol": trade.Symbol,
|
|
"currency": "",
|
|
}).SetToCurrentTime()
|
|
}
|
|
|
|
func (session *ExchangeSession) bindMarketDataStreamMetrics(stream types.Stream) {
|
|
stream.OnBookUpdate(func(book types.SliceOrderBook) {
|
|
metricsLastUpdateTimeBalance.With(prometheus.Labels{
|
|
"exchange": session.ExchangeName.String(),
|
|
"margin": session.MarginType(),
|
|
"channel": "market",
|
|
"data_type": "book",
|
|
"symbol": book.Symbol,
|
|
"currency": "",
|
|
}).SetToCurrentTime()
|
|
})
|
|
stream.OnKLineClosed(func(kline types.KLine) {
|
|
metricsLastUpdateTimeBalance.With(prometheus.Labels{
|
|
"exchange": session.ExchangeName.String(),
|
|
"margin": session.MarginType(),
|
|
"channel": "market",
|
|
"data_type": "kline",
|
|
"symbol": kline.Symbol,
|
|
"currency": "",
|
|
}).SetToCurrentTime()
|
|
})
|
|
}
|
|
|
|
func (session *ExchangeSession) bindUserDataStreamMetrics(stream types.Stream) {
|
|
stream.OnBalanceUpdate(session.metricsBalancesUpdater)
|
|
stream.OnBalanceSnapshot(session.metricsBalancesUpdater)
|
|
stream.OnTradeUpdate(session.metricsTradeUpdater)
|
|
stream.OnOrderUpdate(session.metricsOrderUpdater)
|
|
stream.OnDisconnect(func() {
|
|
metricsConnectionStatus.With(prometheus.Labels{
|
|
"channel": "user",
|
|
"exchange": session.ExchangeName.String(),
|
|
"margin": session.MarginType(),
|
|
"symbol": session.IsolatedMarginSymbol,
|
|
}).Set(0.0)
|
|
})
|
|
stream.OnConnect(func() {
|
|
metricsConnectionStatus.With(prometheus.Labels{
|
|
"channel": "user",
|
|
"exchange": session.ExchangeName.String(),
|
|
"margin": session.MarginType(),
|
|
"symbol": session.IsolatedMarginSymbol,
|
|
}).Set(1.0)
|
|
})
|
|
}
|
|
|
|
func (session *ExchangeSession) bindConnectionStatusNotification(stream types.Stream, streamName string) {
|
|
stream.OnDisconnect(func() {
|
|
session.Notifiability.Notify("session %s %s stream disconnected", session.Name, streamName)
|
|
})
|
|
stream.OnConnect(func() {
|
|
session.Notifiability.Notify("session %s %s stream connected", session.Name, streamName)
|
|
})
|
|
}
|