bbgo_origin/pkg/bbgo/environment.go

1051 lines
28 KiB
Go
Raw Permalink Normal View History

2020-10-16 02:14:36 +00:00
package bbgo
import (
2021-02-21 08:52:47 +00:00
"bytes"
2020-10-16 02:14:36 +00:00
"context"
2020-10-20 04:11:44 +00:00
"fmt"
2021-02-21 08:52:47 +00:00
"image/png"
"io/ioutil"
2022-01-16 11:06:26 +00:00
stdlog "log"
2021-12-30 08:18:32 +00:00
"math/rand"
"os"
2021-02-21 08:52:47 +00:00
"strings"
2021-02-20 03:29:33 +00:00
"sync"
2020-10-16 02:14:36 +00:00
"time"
2021-02-21 08:52:47 +00:00
"github.com/pkg/errors"
"github.com/pquerna/otp"
2020-10-17 16:06:08 +00:00
log "github.com/sirupsen/logrus"
2022-01-16 11:06:26 +00:00
"github.com/slack-go/slack"
2020-12-29 08:00:03 +00:00
"github.com/spf13/viper"
2021-02-21 08:52:47 +00:00
"gopkg.in/tucnak/telebot.v2"
2020-10-16 02:14:36 +00:00
2022-09-09 09:40:17 +00:00
"github.com/c9s/bbgo/pkg/exchange"
2022-03-11 13:27:45 +00:00
"github.com/c9s/bbgo/pkg/fixedpoint"
2022-01-14 03:57:01 +00:00
"github.com/c9s/bbgo/pkg/interact"
2021-02-21 08:52:47 +00:00
"github.com/c9s/bbgo/pkg/notifier/slacknotifier"
"github.com/c9s/bbgo/pkg/notifier/telegramnotifier"
2020-10-16 02:14:36 +00:00
"github.com/c9s/bbgo/pkg/service"
googleservice "github.com/c9s/bbgo/pkg/service/google"
2021-02-21 08:52:47 +00:00
"github.com/c9s/bbgo/pkg/slack/slacklog"
2020-10-16 02:14:36 +00:00
"github.com/c9s/bbgo/pkg/types"
2020-10-30 21:21:17 +00:00
"github.com/c9s/bbgo/pkg/util"
2020-10-16 02:14:36 +00:00
)
2021-12-30 08:18:32 +00:00
func init() {
// randomize pulling
rand.Seed(time.Now().UnixNano())
}
2024-06-19 06:18:21 +00:00
var defaultSyncBufferPeriod = 30 * time.Minute
// IsBackTesting is a global variable that indicates the current environment is back-test or not.
var IsBackTesting = false
var BackTestService *service.BacktestService
func SetBackTesting(s *service.BacktestService) {
BackTestService = s
2022-12-05 18:30:04 +00:00
IsBackTesting = s != nil
}
var LoadedExchangeStrategies = make(map[string]SingleExchangeStrategy)
var LoadedCrossExchangeStrategies = make(map[string]CrossExchangeStrategy)
2020-10-28 23:54:59 +00:00
func RegisterStrategy(key string, s interface{}) {
loaded := 0
if d, ok := s.(SingleExchangeStrategy); ok {
2020-10-28 23:54:59 +00:00
LoadedExchangeStrategies[key] = d
loaded++
}
2020-10-28 23:54:59 +00:00
if d, ok := s.(CrossExchangeStrategy); ok {
2020-10-28 23:54:59 +00:00
LoadedCrossExchangeStrategies[key] = d
loaded++
}
2020-11-15 05:23:26 +00:00
if loaded == 0 {
panic(fmt.Errorf("%T does not implement SingleExchangeStrategy or CrossExchangeStrategy", s))
2020-10-28 23:54:59 +00:00
}
}
2020-10-20 05:52:25 +00:00
var emptyTime time.Time
2021-02-22 06:14:39 +00:00
type SyncStatus int
const (
SyncNotStarted SyncStatus = iota
Syncing
SyncDone
)
2020-10-16 02:14:36 +00:00
// Environment presents the real exchange data layer
type Environment struct {
// built-in service
DatabaseService *service.DatabaseService
OrderService *service.OrderService
TradeService *service.TradeService
ProfitService *service.ProfitService
PositionService *service.PositionService
BacktestService *service.BacktestService
RewardService *service.RewardService
MarginService *service.MarginService
SyncService *service.SyncService
AccountService *service.AccountService
WithdrawService *service.WithdrawService
DepositService *service.DepositService
PersistentService *service.PersistenceServiceFacade
2020-10-16 02:14:36 +00:00
// external services
GoogleSpreadSheetService *googleservice.SpreadSheetService
// startTime is the time of start point (which is used in the backtest)
2021-02-20 03:29:33 +00:00
startTime time.Time
// syncStartTime is the time point we want to start the sync (for trades and orders)
syncStartTime time.Time
syncMutex sync.Mutex
2021-02-21 11:36:03 +00:00
syncStatusMutex sync.Mutex
2021-02-22 06:14:39 +00:00
syncStatus SyncStatus
2022-03-11 08:58:45 +00:00
syncConfig *SyncConfig
2021-02-21 11:36:03 +00:00
loggingConfig *LoggingConfig
environmentConfig *EnvironmentConfig
2023-02-22 07:25:39 +00:00
2021-02-20 03:29:33 +00:00
sessions map[string]*ExchangeSession
2020-10-16 02:14:36 +00:00
}
func NewEnvironment() *Environment {
2022-06-08 06:37:03 +00:00
now := time.Now()
2020-10-16 02:14:36 +00:00
return &Environment{
2020-10-26 05:48:59 +00:00
// default trade scan time
2022-06-08 06:37:03 +00:00
syncStartTime: now.AddDate(-1, 0, 0), // defaults to sync from 1 year ago
2020-10-20 05:11:04 +00:00
sessions: make(map[string]*ExchangeSession),
2022-06-08 06:37:03 +00:00
startTime: now,
2021-02-20 16:58:34 +00:00
syncStatus: SyncNotStarted,
2020-10-16 02:14:36 +00:00
}
}
func (environ *Environment) Logger() log.FieldLogger {
if environ.loggingConfig != nil && len(environ.loggingConfig.Fields) > 0 {
return log.WithFields(environ.loggingConfig.Fields)
}
return log.StandardLogger()
}
func (environ *Environment) Session(name string) (*ExchangeSession, bool) {
s, ok := environ.sessions[name]
return s, ok
}
2020-11-07 12:34:34 +00:00
func (environ *Environment) Sessions() map[string]*ExchangeSession {
return environ.sessions
}
2023-02-22 07:25:39 +00:00
func (environ *Environment) SetLogging(config *LoggingConfig) {
environ.loggingConfig = config
}
2021-02-18 14:40:46 +00:00
func (environ *Environment) SelectSessions(names ...string) map[string]*ExchangeSession {
if len(names) == 0 {
return environ.sessions
}
sessions := make(map[string]*ExchangeSession)
for _, name := range names {
2021-02-20 03:29:33 +00:00
if s, ok := environ.Session(name); ok {
2021-02-18 14:40:46 +00:00
sessions[name] = s
}
}
return sessions
}
2024-01-24 07:18:31 +00:00
func (environ *Environment) ConfigureDatabase(ctx context.Context, config *Config) error {
2021-02-20 16:58:34 +00:00
// configureDB configures the database service based on the environment variable
2024-01-24 07:18:31 +00:00
var dbDriver string
var dbDSN string
var extraPkgNames []string
if config != nil && config.DatabaseConfig != nil {
dbDriver = config.DatabaseConfig.Driver
dbDSN = config.DatabaseConfig.DSN
extraPkgNames = config.DatabaseConfig.ExtraMigrationPackages
}
2021-02-20 16:58:34 +00:00
2024-01-24 07:18:31 +00:00
if val, ok := os.LookupEnv("DB_DRIVER"); ok {
dbDriver = val
}
2021-02-20 16:58:34 +00:00
2024-01-24 07:18:31 +00:00
if val, ok := os.LookupEnv("DB_DSN"); ok {
dbDSN = val
} else if val, ok := os.LookupEnv("SQLITE3_DSN"); ok && (dbDriver == "" || dbDriver == "sqlite3") {
dbDSN = val
dbDriver = "sqlite3"
} else if val, ok := os.LookupEnv("MYSQL_URL"); ok && (dbDriver == "" || dbDriver == "mysql") {
dbDSN = val
dbDriver = "mysql"
}
2021-02-20 16:58:34 +00:00
// database is optional
if dbDriver == "" || dbDSN == "" {
return nil
2021-02-20 16:58:34 +00:00
}
2024-01-24 07:18:31 +00:00
return environ.ConfigureDatabaseDriver(ctx, dbDriver, dbDSN, extraPkgNames...)
2021-02-20 16:58:34 +00:00
}
2024-01-24 07:18:31 +00:00
func (environ *Environment) ConfigureDatabaseDriver(ctx context.Context, driver string, dsn string, extraPkgNames ...string) error {
environ.DatabaseService = service.NewDatabaseService(driver, dsn)
2024-01-26 08:39:05 +00:00
environ.DatabaseService.AddMigrationPackages(extraPkgNames...)
err := environ.DatabaseService.Connect()
2021-02-02 09:26:35 +00:00
if err != nil {
return err
}
2021-02-16 08:30:01 +00:00
if err := environ.DatabaseService.Upgrade(ctx); err != nil {
2021-02-02 09:26:35 +00:00
return err
}
// get the db connection pool object to create other services
db := environ.DatabaseService.DB
environ.OrderService = &service.OrderService{DB: db}
2020-10-26 05:48:59 +00:00
environ.TradeService = &service.TradeService{DB: db}
environ.RewardService = &service.RewardService{DB: db}
environ.AccountService = &service.AccountService{DB: db}
environ.ProfitService = &service.ProfitService{DB: db}
2022-03-11 08:13:38 +00:00
environ.PositionService = &service.PositionService{DB: db}
2022-06-01 07:52:00 +00:00
environ.MarginService = &service.MarginService{DB: db}
environ.WithdrawService = &service.WithdrawService{DB: db}
environ.DepositService = &service.DepositService{DB: db}
environ.SyncService = &service.SyncService{
2021-03-14 03:04:56 +00:00
TradeService: environ.TradeService,
OrderService: environ.OrderService,
RewardService: environ.RewardService,
2022-06-01 07:52:00 +00:00
MarginService: environ.MarginService,
2021-03-14 03:04:56 +00:00
WithdrawService: &service.WithdrawService{DB: db},
DepositService: &service.DepositService{DB: db},
2020-10-26 05:48:59 +00:00
}
return nil
2020-10-26 05:48:59 +00:00
}
// AddExchangeSession adds the existing exchange session or pre-created exchange session
func (environ *Environment) AddExchangeSession(name string, session *ExchangeSession) *ExchangeSession {
2020-10-16 02:14:36 +00:00
environ.sessions[name] = session
return session
}
// AddExchange adds the given exchange with the session name, this is the default
func (environ *Environment) AddExchange(name string, exchange types.Exchange) (session *ExchangeSession) {
session = NewExchangeSession(name, exchange)
return environ.AddExchangeSession(name, session)
}
func (environ *Environment) ConfigureService(ctx context.Context, srvConfig *ServiceConfig) error {
if srvConfig.GoogleSpreadSheetService != nil {
environ.GoogleSpreadSheetService = googleservice.NewSpreadSheetService(ctx, srvConfig.GoogleSpreadSheetService.JsonTokenFile, srvConfig.GoogleSpreadSheetService.SpreadSheetID)
}
return nil
}
2021-02-21 08:52:47 +00:00
func (environ *Environment) ConfigureExchangeSessions(userConfig *Config) error {
// if sessions are not defined, we detect the sessions automatically
2020-12-29 08:00:03 +00:00
if len(userConfig.Sessions) == 0 {
return environ.AddExchangesByViperKeys()
}
return environ.AddExchangesFromSessionConfig(userConfig.Sessions)
}
func (environ *Environment) AddExchangesByViperKeys() error {
2021-05-26 16:35:51 +00:00
for _, n := range types.SupportedExchanges {
2020-12-29 08:00:03 +00:00
if viper.IsSet(string(n) + "-api-key") {
exMinimal, err := exchange.NewWithEnvVarPrefix(n, "")
2020-12-29 08:00:03 +00:00
if err != nil {
return err
}
if ex, ok := exMinimal.(types.Exchange); ok {
environ.AddExchange(n.String(), ex)
} else {
log.Errorf("exchange %T does not implement types.Exchange", exMinimal)
}
2020-12-29 08:00:03 +00:00
}
}
return nil
}
2021-02-02 09:26:35 +00:00
func (environ *Environment) AddExchangesFromSessionConfig(sessions map[string]*ExchangeSession) error {
for sessionName, session := range sessions {
if err := session.InitExchange(sessionName, nil); err != nil {
return err
}
environ.AddExchangeSession(sessionName, session)
2020-12-29 08:00:03 +00:00
}
return nil
}
func (environ *Environment) IsBackTesting() bool {
return environ.BacktestService != nil
}
// Init prepares the data that will be used by the strategies
2020-10-16 02:14:36 +00:00
func (environ *Environment) Init(ctx context.Context) (err error) {
for n := range environ.sessions {
var session = environ.sessions[n]
2021-05-07 16:45:24 +00:00
if err = session.Init(ctx, environ); err != nil {
2021-02-02 18:26:41 +00:00
// we can skip initialized sessions
if err != ErrSessionAlreadyInitialized {
return err
}
}
2021-05-07 16:45:24 +00:00
}
2021-05-07 16:45:24 +00:00
return
}
// Start initializes the symbols data streams
2021-05-07 16:45:24 +00:00
func (environ *Environment) Start(ctx context.Context) (err error) {
for n := range environ.sessions {
var session = environ.sessions[n]
if err = session.InitSymbols(ctx, environ); err != nil {
2021-05-02 15:58:34 +00:00
return err
}
}
2021-05-07 16:45:24 +00:00
return
}
2020-11-10 06:19:33 +00:00
func (environ *Environment) SetStartTime(t time.Time) *Environment {
environ.startTime = t
return environ
}
func (environ *Environment) StartTime() time.Time {
return environ.startTime
}
2021-02-20 03:29:33 +00:00
// SetSyncStartTime overrides the default trade scan time (-7 days)
func (environ *Environment) SetSyncStartTime(t time.Time) *Environment {
environ.syncStartTime = t
return environ
}
2022-03-11 08:58:45 +00:00
func (environ *Environment) BindSync(config *SyncConfig) {
// skip this if we are running back-test
if environ.BacktestService != nil {
return
}
// If trade service is configured, we have the db configured
if environ.TradeService == nil {
return
}
2022-03-11 08:58:45 +00:00
if config == nil || config.UserDataStream == nil {
return
}
2022-03-11 08:58:45 +00:00
environ.syncConfig = config
2022-04-27 05:25:42 +00:00
tradeWriterCreator := func(session *ExchangeSession) func(trade types.Trade) {
return func(trade types.Trade) {
trade.IsMargin = session.Margin
trade.IsFutures = session.Futures
if session.Margin {
trade.IsIsolated = session.IsolatedMargin
} else if session.Futures {
trade.IsIsolated = session.IsolatedFutures
}
// The StrategyID field and the PnL field needs to be updated by the strategy.
// trade.StrategyID, trade.PnL
2022-04-27 05:25:42 +00:00
if err := environ.TradeService.Insert(trade); err != nil {
log.WithError(err).Errorf("trade insert error: %+v", trade)
}
2022-01-27 10:13:15 +00:00
}
}
2022-04-27 05:25:42 +00:00
orderWriterCreator := func(session *ExchangeSession) func(order types.Order) {
return func(order types.Order) {
order.IsMargin = session.Margin
order.IsFutures = session.Futures
if session.Margin {
order.IsIsolated = session.IsolatedMargin
} else if session.Futures {
order.IsIsolated = session.IsolatedFutures
}
switch order.Status {
case types.OrderStatusFilled, types.OrderStatusCanceled:
if order.ExecutedQuantity.Sign() > 0 {
if err := environ.OrderService.Insert(order); err != nil {
log.WithError(err).Errorf("order insert error: %+v", order)
}
}
2022-01-27 10:13:15 +00:00
}
}
2022-01-27 10:13:15 +00:00
}
2022-01-27 10:13:15 +00:00
for _, session := range environ.sessions {
2022-04-27 09:13:58 +00:00
// avoid using the iterator variable.
s2 := session
2022-03-11 08:51:22 +00:00
// if trade sync is on, we will write all received trades
2022-03-11 08:58:45 +00:00
if config.UserDataStream.Trades {
2022-04-27 09:13:58 +00:00
tradeWriter := tradeWriterCreator(s2)
2022-01-27 10:13:15 +00:00
session.UserDataStream.OnTradeUpdate(tradeWriter)
}
2022-04-27 05:25:42 +00:00
2022-03-11 08:58:45 +00:00
if config.UserDataStream.FilledOrders {
2022-04-27 09:13:58 +00:00
orderWriter := orderWriterCreator(s2)
2022-01-27 10:13:15 +00:00
session.UserDataStream.OnOrderUpdate(orderWriter)
}
}
}
func (environ *Environment) Connect(ctx context.Context) error {
log.Debugf("starting interaction...")
if err := interact.Start(ctx); err != nil {
return err
}
for n := range environ.sessions {
// avoid using the placeholder variable for the session because we use that in the callbacks
var session = environ.sessions[n]
var logger = log.WithField("session", n)
2020-10-19 14:26:43 +00:00
if len(session.Subscriptions) == 0 {
logger.Warnf("exchange session %s has no subscriptions", session.Name)
} else {
// add the subscribe requests to the stream
for _, s := range session.Subscriptions {
logger.Infof("subscribing %s %s %v", s.Symbol, s.Channel, s.Options)
session.MarketDataStream.Subscribe(s.Channel, s.Symbol, s.Options)
}
}
logger.Infof("connecting %s market data stream...", session.Name)
if err := session.MarketDataStream.Connect(ctx); err != nil {
return err
}
if !session.PublicOnly {
logger.Infof("connecting %s user data stream...", session.Name)
if err := session.UserDataStream.Connect(ctx); err != nil {
return err
}
2020-10-16 02:14:36 +00:00
}
}
return nil
}
2021-02-22 06:14:39 +00:00
func (environ *Environment) IsSyncing() (status SyncStatus) {
2021-02-21 11:36:03 +00:00
environ.syncStatusMutex.Lock()
2021-02-22 06:14:39 +00:00
status = environ.syncStatus
environ.syncStatusMutex.Unlock()
return status
2021-02-21 11:36:03 +00:00
}
2021-02-22 06:14:39 +00:00
func (environ *Environment) setSyncing(status SyncStatus) {
2021-02-21 11:36:03 +00:00
environ.syncStatusMutex.Lock()
2021-02-22 06:14:39 +00:00
environ.syncStatus = status
2021-02-21 11:36:03 +00:00
environ.syncStatusMutex.Unlock()
2021-02-20 03:29:33 +00:00
}
2022-06-01 07:52:00 +00:00
func (environ *Environment) syncWithUserConfig(ctx context.Context, userConfig *Config) error {
sessions := environ.sessions
selectedSessions := userConfig.Sync.Sessions
if len(selectedSessions) > 0 {
sessions = environ.SelectSessions(selectedSessions...)
2021-02-28 07:05:49 +00:00
}
2023-12-07 08:07:23 +00:00
since := defaultSyncSinceTime()
2022-06-08 04:54:48 +00:00
if userConfig.Sync.Since != nil {
since = userConfig.Sync.Since.Time()
}
2023-12-07 08:07:23 +00:00
environ.SetSyncStartTime(since)
syncSymbolMap, restSymbols := categorizeSyncSymbol(userConfig.Sync.Symbols)
2022-06-01 07:52:00 +00:00
for _, session := range sessions {
syncSymbols := restSymbols
if ss, ok := syncSymbolMap[session.Name]; ok {
syncSymbols = append(syncSymbols, ss...)
}
2023-12-07 08:07:23 +00:00
if err := environ.syncSession(ctx, session, since, syncSymbols...); err != nil {
2022-06-01 07:52:00 +00:00
return err
}
2022-04-25 09:18:42 +00:00
2022-06-01 07:52:00 +00:00
if userConfig.Sync.DepositHistory {
2022-06-08 09:32:42 +00:00
if err := environ.SyncService.SyncDepositHistory(ctx, session.Exchange, since); err != nil {
return err
}
2022-06-01 07:52:00 +00:00
}
2022-04-25 09:18:42 +00:00
2022-06-01 07:52:00 +00:00
if userConfig.Sync.WithdrawHistory {
if err := environ.SyncService.SyncWithdrawHistory(ctx, session.Exchange, since); err != nil {
2022-06-01 07:52:00 +00:00
return err
2022-04-25 09:18:42 +00:00
}
2022-06-01 07:52:00 +00:00
}
2022-04-25 09:18:42 +00:00
2022-06-01 07:52:00 +00:00
if userConfig.Sync.RewardHistory {
2022-06-17 17:42:24 +00:00
if err := environ.SyncService.SyncRewardHistory(ctx, session.Exchange, since); err != nil {
2022-06-01 07:52:00 +00:00
return err
2022-04-25 09:18:42 +00:00
}
2022-06-01 07:52:00 +00:00
}
2022-04-25 09:18:42 +00:00
2022-06-01 07:52:00 +00:00
if userConfig.Sync.MarginHistory {
if err := environ.SyncService.SyncMarginHistory(ctx, session.Exchange,
since,
2022-06-01 07:52:00 +00:00
userConfig.Sync.MarginAssets...); err != nil {
return err
2022-04-25 09:18:42 +00:00
}
}
2022-06-01 07:52:00 +00:00
}
return nil
}
// Sync syncs all registered exchange sessions
func (environ *Environment) Sync(ctx context.Context, userConfig ...*Config) error {
if environ.SyncService == nil {
return nil
}
2022-04-25 09:18:42 +00:00
2022-06-01 07:52:00 +00:00
environ.syncMutex.Lock()
defer environ.syncMutex.Unlock()
environ.setSyncing(Syncing)
defer environ.setSyncing(SyncDone)
// sync by the defined user config
if len(userConfig) > 0 && userConfig[0] != nil && userConfig[0].Sync != nil {
return environ.syncWithUserConfig(ctx, userConfig[0])
}
// the default sync logics
2023-12-07 08:07:23 +00:00
since := defaultSyncSinceTime()
2021-02-20 03:29:33 +00:00
for _, session := range environ.sessions {
2023-12-07 08:07:23 +00:00
if err := environ.syncSession(ctx, session, since); err != nil {
2021-02-20 03:29:33 +00:00
return err
}
}
return nil
}
2022-05-04 06:40:35 +00:00
func (environ *Environment) RecordAsset(t time.Time, session *ExchangeSession, assets types.AssetMap) {
2022-05-03 11:26:52 +00:00
// skip for back-test
if environ.BacktestService != nil {
return
}
if environ.DatabaseService == nil || environ.AccountService == nil {
return
}
2022-05-04 06:40:35 +00:00
if err := environ.AccountService.InsertAsset(
t,
session.Name,
session.ExchangeName,
session.SubAccount,
session.Margin,
session.IsolatedMargin,
session.IsolatedMarginSymbol,
assets); err != nil {
log.WithError(err).Errorf("can not insert asset record")
}
2022-05-03 11:26:52 +00:00
}
2022-03-11 13:15:57 +00:00
func (environ *Environment) RecordPosition(position *types.Position, trade types.Trade, profit *types.Profit) {
2022-03-11 08:23:34 +00:00
// skip for back-test
if environ.BacktestService != nil {
return
}
2022-03-11 08:58:45 +00:00
if environ.DatabaseService == nil || environ.ProfitService == nil || environ.PositionService == nil {
2022-03-11 08:23:34 +00:00
return
}
2023-12-18 07:49:20 +00:00
// guard: set profit info to position if the strategy info is empty
2022-06-13 02:33:28 +00:00
if profit != nil {
if position.Strategy == "" && profit.Strategy != "" {
position.Strategy = profit.Strategy
}
2022-06-13 02:33:28 +00:00
if position.StrategyInstanceID == "" && profit.StrategyInstanceID != "" {
position.StrategyInstanceID = profit.StrategyInstanceID
}
}
2023-12-18 07:49:20 +00:00
log.Infof("recordPosition: position = %s, trade = %+v, profit = %+v", position.Base.String(), trade, profit)
2022-03-11 13:15:57 +00:00
if profit != nil {
2022-03-11 13:27:45 +00:00
if err := environ.PositionService.Insert(position, trade, profit.Profit); err != nil {
log.WithError(err).Errorf("can not insert position record")
}
2023-12-18 07:49:20 +00:00
2022-03-11 13:15:57 +00:00
if err := environ.ProfitService.Insert(*profit); err != nil {
log.WithError(err).Errorf("can not insert profit record: %+v", profit)
}
2022-03-11 13:27:45 +00:00
} else {
if err := environ.PositionService.Insert(position, trade, fixedpoint.Zero); err != nil {
log.WithError(err).Errorf("can not insert position record")
}
2022-03-11 08:23:34 +00:00
}
}
2022-03-06 07:37:41 +00:00
func (environ *Environment) RecordProfit(profit types.Profit) {
2022-03-11 08:23:34 +00:00
// skip for back-test
if environ.BacktestService != nil {
return
}
if environ.DatabaseService == nil {
return
}
if environ.ProfitService == nil {
return
}
2022-03-11 08:13:38 +00:00
if err := environ.ProfitService.Insert(profit); err != nil {
2022-03-06 07:37:41 +00:00
log.WithError(err).Errorf("can not insert profit record: %+v", profit)
}
}
2021-02-20 03:29:33 +00:00
func (environ *Environment) SyncSession(ctx context.Context, session *ExchangeSession, defaultSymbols ...string) error {
2021-04-09 04:44:30 +00:00
if environ.SyncService == nil {
return nil
}
2021-02-20 03:29:33 +00:00
environ.syncMutex.Lock()
defer environ.syncMutex.Unlock()
2021-02-22 06:14:39 +00:00
environ.setSyncing(Syncing)
defer environ.setSyncing(SyncDone)
2021-02-20 03:29:33 +00:00
2023-12-07 08:07:23 +00:00
since := defaultSyncSinceTime()
return environ.syncSession(ctx, session, since, defaultSymbols...)
2021-02-20 03:29:33 +00:00
}
2023-12-07 08:07:23 +00:00
func (environ *Environment) syncSession(
ctx context.Context, session *ExchangeSession, syncStartTime time.Time, defaultSymbols ...string,
) error {
symbols, err := session.getSessionSymbols(defaultSymbols...)
2021-02-19 02:42:24 +00:00
if err != nil {
return err
}
2021-02-22 07:01:05 +00:00
log.Infof("syncing symbols %v from session %s", symbols, session.Name)
2024-06-19 06:18:21 +00:00
syncBufferPeriod := -defaultSyncBufferPeriod
if environ.environmentConfig != nil && environ.environmentConfig.SyncBufferPeriod != nil {
2024-06-19 06:18:21 +00:00
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...)
}
2021-02-19 02:42:24 +00:00
func (environ *Environment) ConfigureNotificationSystem(ctx context.Context, userConfig *Config) error {
// setup default notification config
if userConfig.Notifications == nil {
2022-09-19 11:22:08 +00:00
userConfig.Notifications = &NotificationConfig{}
2021-02-21 08:52:47 +00:00
}
var isolation = GetIsolationFromContext(ctx)
var persistence = isolation.persistenceServiceFacade.Get()
err := environ.setupInteraction(persistence)
2022-01-14 16:32:21 +00:00
if err != nil {
return err
}
// setup slack
slackToken := viper.GetString("slack-token")
if len(slackToken) > 0 && userConfig.Notifications != nil {
2022-01-16 11:06:26 +00:00
environ.setupSlack(userConfig, slackToken, persistence)
2022-01-14 16:32:21 +00:00
}
// check if telegram bot token is defined
telegramBotToken := viper.GetString("telegram-bot-token")
if len(telegramBotToken) > 0 {
if err := environ.setupTelegram(userConfig, telegramBotToken, persistence); err != nil {
return err
}
}
if userConfig.Notifications != nil {
2022-09-19 11:28:29 +00:00
if err := environ.ConfigureNotification(userConfig.Notifications); err != nil {
2022-01-14 16:32:21 +00:00
return err
}
}
return nil
}
2022-09-19 11:28:29 +00:00
func (environ *Environment) ConfigureNotification(config *NotificationConfig) error {
if config.Switches != nil {
if config.Switches.Trade {
tradeHandler := func(trade types.Trade) {
Notify(trade)
}
for _, session := range environ.sessions {
session.UserDataStream.OnTradeUpdate(tradeHandler)
}
}
if config.Switches.OrderUpdate {
orderUpdateHandler := func(order types.Order) {
Notify(order)
}
for _, session := range environ.sessions {
session.UserDataStream.OnOrderUpdate(orderUpdateHandler)
}
}
}
return nil
}
// getAuthStoreID returns the authentication store id
// if telegram bot token is defined, the bot id will be used.
// if not, env var $USER will be used.
// if both are not defined, a default "default" will be used.
func getAuthStoreID() string {
telegramBotToken := viper.GetString("telegram-bot-token")
if len(telegramBotToken) > 0 {
tt := strings.Split(telegramBotToken, ":")
return tt[0]
}
userEnv := os.Getenv("USER")
if userEnv != "" {
return userEnv
}
return "default"
}
func (environ *Environment) setupInteraction(persistence service.PersistenceService) error {
2022-06-17 11:19:51 +00:00
var otpQRCodeImagePath = "otp.png"
var key *otp.Key
var keyURL string
var authStore = environ.getAuthStore(persistence)
if v, ok := util.GetEnvVarBool("FLUSH_OTP_KEY"); v && ok {
log.Warnf("flushing otp key...")
if err := authStore.Reset(); err != nil {
return err
}
}
if err := authStore.Load(&keyURL); err != nil {
log.Warnf("telegram session not found, generating new one-time password key for new telegram session...")
2021-02-21 08:52:47 +00:00
newKey, err := setupNewOTPKey(otpQRCodeImagePath)
2021-02-21 08:52:47 +00:00
if err != nil {
return errors.Wrapf(err, "failed to setup totp (time-based one time password) key")
2021-02-21 08:52:47 +00:00
}
key = newKey
keyURL = key.URL()
if err := authStore.Save(keyURL); err != nil {
return err
}
printOtpAuthGuide(otpQRCodeImagePath)
2021-02-21 08:52:47 +00:00
} else if keyURL != "" {
key, err = otp.NewKeyFromURL(keyURL)
if err != nil {
log.WithError(err).Errorf("can not load otp key from url: %s, generating new otp key", keyURL)
newKey, err := setupNewOTPKey(otpQRCodeImagePath)
if err != nil {
return errors.Wrapf(err, "failed to setup totp (time-based one time password) key")
}
key = newKey
keyURL = key.URL()
if err := authStore.Save(keyURL); err != nil {
return err
}
printOtpAuthGuide(otpQRCodeImagePath)
} else {
log.Infof("otp key loaded: %s", util.MaskKey(key.Secret()))
printOtpAuthGuide(otpQRCodeImagePath)
}
}
2021-02-21 08:52:47 +00:00
authStrict := false
authMode := interact.AuthModeToken
2022-01-14 16:29:35 +00:00
authToken := viper.GetString("telegram-bot-auth-token")
if authToken != "" && key != nil {
authStrict = true
} else if authToken != "" {
authMode = interact.AuthModeToken
} else if key != nil {
authMode = interact.AuthModeOTP
}
2022-01-14 16:29:35 +00:00
if authMode == interact.AuthModeToken {
log.Debugf("found interaction auth token, using token mode for authorization...")
printAuthTokenGuide(authToken)
}
interact.AddCustomInteraction(&interact.AuthInteract{
2022-06-17 07:04:23 +00:00
Strict: authStrict,
Mode: authMode,
Token: authToken, // can be empty string here
// pragma: allowlist nextline secret
OneTimePasswordKey: key, // can be nil here
})
2021-02-21 08:52:47 +00:00
return nil
}
func (environ *Environment) getAuthStore(persistence service.PersistenceService) service.Store {
id := getAuthStoreID()
return persistence.NewStore("bbgo", "auth", id)
}
2022-01-16 11:06:26 +00:00
func (environ *Environment) setupSlack(userConfig *Config, slackToken string, persistence service.PersistenceService) {
conf := userConfig.Notifications.Slack
if conf == nil {
return
}
2022-01-16 11:06:26 +00:00
if !strings.HasPrefix(slackToken, "xoxb-") {
log.Error("SLACK_BOT_TOKEN must have the prefix \"xoxb-\".")
return
}
2022-01-16 11:06:26 +00:00
// app-level token (for specific api)
slackAppToken := viper.GetString("slack-app-token")
2022-07-19 03:41:49 +00:00
if len(slackAppToken) > 0 && !strings.HasPrefix(slackAppToken, "xapp-") {
2022-01-16 11:06:26 +00:00
log.Errorf("SLACK_APP_TOKEN must have the prefix \"xapp-\".")
return
}
2022-01-16 11:06:26 +00:00
if conf.ErrorChannel != "" {
log.Debugf("found slack configured, setting up log hook...")
log.AddHook(slacklog.NewLogHook(slackToken, conf.ErrorChannel))
}
log.Debugf("adding slack notifier with default channel: %s", conf.DefaultChannel)
2022-04-25 10:56:19 +00:00
var slackOpts = []slack.Option{
2022-01-16 11:06:26 +00:00
slack.OptionLog(stdlog.New(os.Stdout, "api: ", stdlog.Lshortfile|stdlog.LstdFlags)),
2022-07-19 03:41:49 +00:00
}
if len(slackAppToken) > 0 {
slackOpts = append(slackOpts, slack.OptionAppLevelToken(slackAppToken))
2022-04-25 10:56:19 +00:00
}
if b, ok := util.GetEnvVarBool("DEBUG_SLACK"); ok {
slackOpts = append(slackOpts, slack.OptionDebug(b))
}
var client = slack.New(slackToken, slackOpts...)
2022-01-16 11:06:26 +00:00
var notifier = slacknotifier.New(client, conf.DefaultChannel)
Notification.AddNotifier(notifier)
2022-01-16 11:06:26 +00:00
// allocate a store, so that we can save the chatID for the owner
var messenger = interact.NewSlack(client)
var sessions = interact.SlackSessionMap{}
var sessionStore = persistence.NewStore("bbgo", "slack")
if err := sessionStore.Load(&sessions); err != nil {
2022-01-22 17:50:27 +00:00
2022-01-16 11:06:26 +00:00
} else {
2022-01-22 17:50:27 +00:00
// TODO: this is not necessary for slack, but we should find a way to restore the sessions
/*
for _, session := range sessions {
if session.IsAuthorized() {
// notifier.AddChat(session.Chat)
}
2022-01-16 11:06:26 +00:00
}
2022-01-22 17:50:27 +00:00
messenger.RestoreSessions(sessions)
messenger.OnAuthorized(func(userSession *interact.SlackSession) {
if userSession.IsAuthorized() {
// notifier.AddChat(userSession.Chat)
}
})
*/
2022-01-16 11:06:26 +00:00
}
interact.AddMessenger(messenger)
}
func (environ *Environment) setupTelegram(
userConfig *Config, telegramBotToken string, persistence service.PersistenceService,
) error {
tt := strings.Split(telegramBotToken, ":")
telegramID := tt[0]
bot, err := telebot.NewBot(telebot.Settings{
// You can also set custom API URL.
// If field is empty it equals to "https://api.telegram.org".
// URL: "http://195.129.111.17:8012",
Token: telegramBotToken,
Poller: &telebot.LongPoller{Timeout: 10 * time.Second},
})
if err != nil {
return err
}
var opts []telegramnotifier.Option
if userConfig.Notifications != nil && userConfig.Notifications.Telegram != nil {
log.Infof("telegram broadcast is enabled")
opts = append(opts, telegramnotifier.UseBroadcast())
}
var notifier = telegramnotifier.New(bot, opts...)
Notification.AddNotifier(notifier)
log.AddHook(telegramnotifier.NewLogHook(notifier))
// allocate a store, so that we can save the chatID for the owner
2022-01-16 11:06:26 +00:00
var messenger = interact.NewTelegram(bot)
var sessions = interact.TelegramSessionMap{}
var sessionStore = persistence.NewStore("bbgo", "telegram", telegramID)
2022-01-15 16:39:24 +00:00
if err := sessionStore.Load(&sessions); err != nil {
2022-01-19 10:29:24 +00:00
if err != service.ErrPersistenceNotExists {
log.WithError(err).Errorf("unexpected persistence error")
}
} else {
for _, session := range sessions {
if session.IsAuthorized() {
notifier.AddChat(session.Chat)
}
}
// you must restore the session after the notifier updates
messenger.RestoreSessions(sessions)
}
messenger.OnAuthorized(func(userSession *interact.TelegramSession) {
if userSession.IsAuthorized() {
notifier.AddChat(userSession.Chat)
}
2022-01-15 16:39:24 +00:00
log.Infof("user session %d got authorized, saving telegram sessions...", userSession.User.ID)
if err := sessionStore.Save(messenger.Sessions()); err != nil {
log.WithError(err).Errorf("telegram session save error")
}
})
2022-01-16 11:06:26 +00:00
interact.AddMessenger(messenger)
return nil
}
2021-02-21 08:52:47 +00:00
func writeOTPKeyAsQRCodePNG(key *otp.Key, imagePath string) error {
// Convert TOTP key into a PNG
var buf bytes.Buffer
img, err := key.Image(512, 512)
if err != nil {
return err
}
if err := png.Encode(&buf, img); err != nil {
return err
}
if err := ioutil.WriteFile(imagePath, buf.Bytes(), 0644); err != nil {
return err
}
return nil
}
// setupNewOTPKey generates a new otp key and save the secret as a qrcode image
func setupNewOTPKey(qrcodeImagePath string) (*otp.Key, error) {
key, err := service.NewDefaultTotpKey()
if err != nil {
return nil, errors.Wrapf(err, "failed to setup totp (time-based one time password) key")
}
printOtpKey(key)
if err := writeOTPKeyAsQRCodePNG(key, qrcodeImagePath); err != nil {
return nil, err
}
return key, nil
}
func printOtpKey(key *otp.Key) {
fmt.Println("")
2022-01-16 15:47:26 +00:00
fmt.Println("====================================================================")
fmt.Println(" PLEASE STORE YOUR OTP KEY SAFELY ")
fmt.Println("====================================================================")
fmt.Printf(" Issuer: %s\n", key.Issuer())
fmt.Printf(" AccountName: %s\n", key.AccountName())
fmt.Printf(" Secret: %s\n", key.Secret())
fmt.Printf(" Key URL: %s\n", key.URL())
2021-02-21 08:52:47 +00:00
fmt.Println("====================================================================")
fmt.Println("")
}
func printOtpAuthGuide(qrcodeImagePath string) {
2021-02-22 09:06:43 +00:00
fmt.Printf(`
To scan your OTP QR code, please run the following command:
open %s
2022-09-04 05:35:01 +00:00
For telegram, send the auth command with the generated one-time password to the bbgo bot you created to enable the notification:
2021-02-22 09:06:43 +00:00
/auth
2021-02-22 09:06:43 +00:00
`, qrcodeImagePath)
2021-02-21 08:52:47 +00:00
}
func printAuthTokenGuide(token string) {
2021-02-22 09:06:43 +00:00
fmt.Printf(`
For telegram, send the following command to the bbgo bot you created to enable the notification:
/auth
And then enter your token
2021-02-22 09:06:43 +00:00
%s
2021-02-22 09:06:43 +00:00
`, token)
2021-02-21 08:52:47 +00:00
}
func (session *ExchangeSession) getSessionSymbols(defaultSymbols ...string) ([]string, error) {
if session.IsolatedMargin {
return []string{session.IsolatedMarginSymbol}, nil
}
if len(defaultSymbols) > 0 {
return defaultSymbols, nil
}
return session.FindPossibleAssetSymbols()
}
2023-12-07 08:07:23 +00:00
func defaultSyncSinceTime() time.Time {
return time.Now().AddDate(0, -6, 0)
}