bbgo_origin/pkg/bbgo/session.go

604 lines
17 KiB
Go
Raw Normal View History

2020-10-16 02:14:36 +00:00
package bbgo
2020-10-18 04:23:00 +00:00
import (
2021-01-26 09:21:18 +00:00
"context"
"fmt"
"strings"
2021-01-26 09:21:18 +00:00
"time"
2021-05-16 07:03:36 +00:00
"github.com/c9s/bbgo/pkg/fixedpoint"
log "github.com/sirupsen/logrus"
2021-03-22 07:46:55 +00:00
"github.com/spf13/viper"
2020-10-28 01:13:57 +00:00
"github.com/c9s/bbgo/pkg/indicator"
"github.com/c9s/bbgo/pkg/service"
2020-10-18 04:23:00 +00:00
"github.com/c9s/bbgo/pkg/types"
2021-02-19 02:42:24 +00:00
"github.com/c9s/bbgo/pkg/util"
2020-10-18 04:23:00 +00:00
)
2020-10-16 02:14:36 +00:00
2021-05-06 17:30:09 +00:00
var fiatCurrencies = []string{"USDC", "USDT", "USD", "TWD", "EUR", "GBP"}
2020-10-28 01:13:57 +00:00
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
store *MarketDataStore
2020-10-28 01:13:57 +00:00
}
func NewStandardIndicatorSet(symbol string, store *MarketDataStore) *StandardIndicatorSet {
2020-10-28 01:13:57 +00:00
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),
store: store,
2020-10-28 01:13:57 +00:00
}
// let us pre-defined commonly used intervals
2020-10-28 01:43:19 +00:00
for interval := range types.SupportedIntervals {
2020-10-28 01:13:57 +00:00
for _, window := range []int{7, 25, 99} {
2020-10-29 05:08:33 +00:00
iw := types.IntervalWindow{Interval: interval, Window: window}
set.sma[iw] = &indicator.SMA{IntervalWindow: iw}
set.sma[iw].Bind(store)
set.ewma[iw] = &indicator.EWMA{IntervalWindow: iw}
set.ewma[iw].Bind(store)
2020-10-28 01:13:57 +00:00
}
2020-10-29 09:51:20 +00:00
// setup boll indicator, we may refactor boll indicator by subscribing SMA indicator,
2020-10-29 09:51:20 +00:00
// 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
2020-10-29 09:51:20 +00:00
iw := types.IntervalWindow{Interval: interval, Window: 21}
set.boll[iw] = &indicator.BOLL{IntervalWindow: iw, K: 2.0}
set.boll[iw].Bind(store)
2020-10-28 01:13:57 +00:00
}
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 {
2020-11-11 15:18:53 +00:00
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
2020-10-28 01:13:57 +00:00
}
return inc
2020-10-28 01:13:57 +00:00
}
// 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
}
// ExchangeSession presents the exchange connection Session
2020-10-16 02:14:36 +00:00
// 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
2021-02-02 18:26:41 +00:00
Notifiability `json:"-" yaml:"-"`
2021-02-02 09:26:35 +00:00
// ---------------------------
// Session config fields
// ---------------------------
2020-10-16 02:14:36 +00:00
2021-02-02 09:26:35 +00:00
// Exchange Session name
Name string `json:"name,omitempty" yaml:"name,omitempty"`
ExchangeName string `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"`
SubAccount string `json:"subAccount,omitempty" yaml:"subAccount,omitempty"`
2020-10-16 02:14:36 +00:00
// Withdrawal is used for enabling withdrawal functions
2021-05-16 09:58:51 +00:00
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"`
2021-02-02 09:26:35 +00:00
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"`
2021-02-02 09:26:35 +00:00
// ---------------------------
// Runtime fields
// ---------------------------
2021-02-02 09:26:35 +00:00
// The exchange account states
2021-02-03 01:34:53 +00:00
Account *types.Account `json:"-" yaml:"-"`
2021-02-03 01:34:53 +00:00
IsInitialized bool `json:"-" yaml:"-"`
OrderExecutor *ExchangeOrderExecutor `json:"orderExecutor,omitempty" yaml:"orderExecutor,omitempty"`
2020-10-16 02:14:36 +00:00
// Stream is the connection stream of the exchange
2021-02-02 18:26:41 +00:00
Stream types.Stream `json:"-" yaml:"-"`
2020-10-16 02:14:36 +00:00
2021-02-02 18:26:41 +00:00
Subscriptions map[types.Subscription]types.Subscription `json:"-" yaml:"-"`
2020-10-16 02:14:36 +00:00
2021-02-02 18:26:41 +00:00
Exchange types.Exchange `json:"-" yaml:"-"`
2020-10-16 02:14:36 +00:00
// Trades collects the executed trades from the exchange
// map: symbol -> []trade
Trades map[string]*types.TradeSlice `json:"-" yaml:"-"`
2020-10-18 04:30:13 +00:00
// markets defines market configuration of a symbol
markets map[string]types.Market
2020-10-16 02:14:36 +00:00
2020-11-10 11:06:20 +00:00
// startPrices is used for backtest
startPrices map[string]float64
2020-11-10 11:06:20 +00:00
lastPrices map[string]float64
lastPriceUpdatedAt time.Time
2020-10-16 02:14:36 +00:00
2020-10-28 01:13:57 +00:00
// marketDataStores contains the market data store of each market
marketDataStores map[string]*MarketDataStore
2020-10-22 02:54:03 +00:00
positions map[string]*Position
2021-01-20 08:28:27 +00:00
2020-10-28 01:13:57 +00:00
// standard indicators of each market
standardIndicatorSets map[string]*StandardIndicatorSet
2020-10-28 01:13:57 +00:00
orderStores map[string]*OrderStore
2021-01-21 06:51:37 +00:00
usedSymbols map[string]struct{}
initializedSymbols map[string]struct{}
2021-02-01 09:13:54 +00:00
logger *log.Entry
2020-10-17 15:51:44 +00:00
}
func NewExchangeSession(name string, exchange types.Exchange) *ExchangeSession {
session := &ExchangeSession{
Notifiability: Notifiability{
SymbolChannelRouter: NewPatternChannelRouter(nil),
SessionChannelRouter: NewPatternChannelRouter(nil),
ObjectChannelRouter: NewObjectChannelRouter(),
},
Name: name,
Exchange: exchange,
Stream: exchange.NewStream(),
Subscriptions: make(map[types.Subscription]types.Subscription),
Account: &types.Account{},
2021-01-21 07:10:40 +00:00
Trades: make(map[string]*types.TradeSlice),
markets: make(map[string]types.Market),
2020-11-10 11:06:20 +00:00
startPrices: make(map[string]float64),
lastPrices: make(map[string]float64),
2021-01-20 08:28:27 +00:00
positions: make(map[string]*Position),
marketDataStores: make(map[string]*MarketDataStore),
standardIndicatorSets: make(map[string]*StandardIndicatorSet),
2021-01-21 06:51:37 +00:00
orderStores: make(map[string]*OrderStore),
usedSymbols: make(map[string]struct{}),
initializedSymbols: make(map[string]struct{}),
2021-02-01 09:13:54 +00:00
logger: log.WithField("session", name),
}
session.OrderExecutor = &ExchangeOrderExecutor{
// copy the notification system so that we can route
Notifiability: session.Notifiability,
Session: session,
}
return session
}
2021-05-02 15:58:34 +00:00
// 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 {
2021-02-02 18:26:41 +00:00
return ErrSessionAlreadyInitialized
}
var log = log.WithField("session", session.Name)
2021-03-22 07:46:55 +00:00
if !viper.GetBool("bbgo-markets-cache") {
markets, err := session.Exchange.QueryMarkets(ctx)
if err != nil {
return err
}
session.markets = markets
} else {
// load markets first
var markets, err = LoadExchangeMarketsWithCache(ctx, session.Exchange)
if err != nil {
return err
}
2021-03-22 07:46:55 +00:00
if len(markets) == 0 {
return fmt.Errorf("market config should not be empty")
}
2021-03-22 07:46:55 +00:00
session.markets = markets
}
2021-02-02 18:26:41 +00:00
// query and initialize the balances
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)
2021-02-02 18:26:41 +00:00
// forward trade updates and order updates to the order executor
session.Stream.OnTradeUpdate(session.OrderExecutor.EmitTradeUpdate)
session.Stream.OnOrderUpdate(session.OrderExecutor.EmitOrderUpdate)
session.Account.BindStream(session.Stream)
// insert trade into db right before everything
2021-05-07 17:09:06 +00:00
// TODO: we should insert the backtest trades into the database,
// however we should clean up the trades before we start the next backtesting
if environ.TradeService != nil && environ.BacktestService == nil {
session.Stream.OnTradeUpdate(func(trade types.Trade) {
if err := environ.TradeService.Insert(trade); err != nil {
log.WithError(err).Errorf("trade insert error: %+v", trade)
}
})
}
session.Stream.OnKLineClosed(func(kline types.KLine) {
2021-02-10 14:40:36 +00:00
log.WithField("marketData", "kline").Infof("kline closed: %+v", kline)
})
// update last prices
session.Stream.OnKLineClosed(func(kline types.KLine) {
if _, ok := session.startPrices[kline.Symbol]; !ok {
session.startPrices[kline.Symbol] = kline.Open
}
session.lastPrices[kline.Symbol] = kline.Close
})
2021-05-02 15:58:34 +00:00
session.IsInitialized = true
return nil
}
func (session *ExchangeSession) InitSymbols(ctx context.Context, environ *Environment) error {
2021-04-09 04:43:13 +00:00
if err := session.initUsedSymbols(ctx, environ); err != nil {
return err
}
return nil
}
2021-04-09 04:43:13 +00:00
// initUsedSymbols uses usedSymbols to initialize the related data structure
func (session *ExchangeSession) initUsedSymbols(ctx context.Context, environ *Environment) error {
for symbol := range session.usedSymbols {
2021-05-02 15:58:34 +00:00
if err := session.initSymbol(ctx, environ, symbol); err != nil {
return err
}
}
return nil
}
2021-05-02 15:58:34 +00:00
// 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 {
2021-04-09 04:43:13 +00:00
// 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 {
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.Stream.OnTradeUpdate(func(trade types.Trade) {
session.Trades[symbol].Append(trade)
})
position := &Position{
Symbol: symbol,
BaseCurrency: market.BaseCurrency,
QuoteCurrency: market.QuoteCurrency,
}
position.AddTrades(trades)
position.BindStream(session.Stream)
session.positions[symbol] = position
orderStore := NewOrderStore(symbol)
orderStore.AddOrderUpdate = true
orderStore.BindStream(session.Stream)
session.orderStores[symbol] = orderStore
marketDataStore := NewMarketDataStore(symbol)
marketDataStore.BindStream(session.Stream)
session.marketDataStores[symbol] = marketDataStore
standardIndicatorSet := NewStandardIndicatorSet(symbol, marketDataStore)
session.standardIndicatorSets[symbol] = standardIndicatorSet
// used kline intervals by the given symbol
var usedKLineIntervals = map[types.Interval]struct{}{}
// always subscribe the 1m kline so we can make sure the connection persists.
usedKLineIntervals[types.Interval1m] = struct{}{}
for _, sub := range session.Subscriptions {
if sub.Channel != types.KLineChannel {
continue
}
if sub.Options.Interval == "" {
continue
}
if sub.Symbol == symbol {
usedKLineIntervals[types.Interval(sub.Options.Interval)] = struct{}{}
}
2020-10-17 15:51:44 +00:00
}
var lastPriceTime time.Time
for interval := range usedKLineIntervals {
// avoid querying the last unclosed kline
endTime := environ.startTime.Add(- interval.Duration())
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 interval %s (end time <= %s)", interval, environ.startTime)
continue
}
// update last prices by the given kline
lastKLine := kLines[len(kLines)-1]
if lastPriceTime == emptyTime {
session.lastPrices[symbol] = lastKLine.Close
lastPriceTime = lastKLine.EndTime
} else if lastKLine.EndTime.After(lastPriceTime) {
session.lastPrices[symbol] = lastKLine.Close
lastPriceTime = lastKLine.EndTime
}
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
2020-10-16 02:14:36 +00:00
}
2020-10-28 01:13:57 +00:00
func (session *ExchangeSession) StandardIndicatorSet(symbol string) (*StandardIndicatorSet, bool) {
set, ok := session.standardIndicatorSets[symbol]
return set, ok
}
2021-01-20 08:29:15 +00:00
func (session *ExchangeSession) Position(symbol string) (pos *Position, ok bool) {
pos, ok = session.positions[symbol]
2021-04-28 11:32:49 +00:00
if ok {
return pos, ok
}
market, ok := session.markets[symbol]
if !ok {
return nil, false
}
pos = &Position{
Symbol: symbol,
BaseCurrency: market.BaseCurrency,
QuoteCurrency: market.QuoteCurrency,
}
ok = true
session.positions[symbol] = pos
2021-01-20 08:29:15 +00:00
return pos, ok
}
2021-02-02 18:26:41 +00:00
func (session *ExchangeSession) Positions() map[string]*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]
2020-10-18 04:27:11 +00:00
return s, ok
}
2020-11-10 11:06:20 +00:00
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
}
2021-02-02 18:26:41 +00:00
func (session *ExchangeSession) LastPrices() map[string]float64 {
return session.lastPrices
}
func (session *ExchangeSession) Market(symbol string) (market types.Market, ok bool) {
2020-10-18 04:30:13 +00:00
market, ok = session.markets[symbol]
return market, ok
}
2021-02-02 18:26:41 +00:00
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
}
2021-02-02 18:26:41 +00:00
func (session *ExchangeSession) OrderStores() map[string]*OrderStore {
return session.orderStores
}
2020-10-16 02:14:36 +00:00
// 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")
}
2020-10-16 02:14:36 +00:00
sub := types.Subscription{
Channel: channel,
Symbol: symbol,
Options: options,
}
// add to the loaded symbol table
session.usedSymbols[symbol] = struct{}{}
2020-10-16 02:14:36 +00:00
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
switch order.Type {
case types.OrderTypeStopMarket, types.OrderTypeStopLimit:
order.StopPriceString = market.FormatPrice(order.StopPrice)
}
switch order.Type {
case types.OrderTypeMarket, types.OrderTypeStopMarket:
order.Price = 0.0
order.PriceString = ""
default:
order.PriceString = market.FormatPrice(order.Price)
}
order.QuantityString = market.FormatQuantity(order.Quantity)
return order, nil
}
2021-01-26 09:21:18 +00:00
func (session *ExchangeSession) UpdatePrices(ctx context.Context) (err error) {
if session.lastPriceUpdatedAt.After(time.Now().Add(- time.Hour)) {
return nil
}
2021-01-26 09:21:18 +00:00
balances := session.Account.Balances()
2021-02-06 21:11:58 +00:00
symbols := make([]string, len(balances))
2021-01-26 09:21:18 +00:00
for _, b := range balances {
symbols = append(symbols, b.Currency+"USDT")
2021-02-06 21:11:58 +00:00
}
2021-01-26 09:21:18 +00:00
2021-02-06 21:11:58 +00:00
tickers, err := session.Exchange.QueryTickers(ctx, symbols...)
if err != nil || len(tickers) == 0 {
return err
}
2021-01-26 09:21:18 +00:00
2021-02-06 21:11:58 +00:00
for k, v := range tickers {
session.lastPrices[k] = v.Last
2021-01-26 09:21:18 +00:00
}
session.lastPriceUpdatedAt = time.Now()
2021-01-26 09:21:18 +00:00
return err
}
2021-02-19 02:42:24 +00:00
func (session *ExchangeSession) FindPossibleSymbols() (symbols []string, err error) {
// If the session is an isolated margin session, there will be only the isolated margin symbol
2021-02-22 07:01:05 +00:00
if session.Margin && session.IsolatedMargin {
2021-02-19 02:42:24 +00:00
return []string{
session.IsolatedMarginSymbol,
}, nil
}
var balances = session.Account.Balances()
var fiatAssets []string
for _, currency := range 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
}