Files
gocryptotrader/engine/engine.go
Andrew 92147cdc5f (Engine): Database system improvements (#358)
* Migrated to goose & sqlboiler

* create tests with sqlboiler

* code clean up

* Added gct -> sqlboiler config gen

* dropped pgx support

* dropped pgx support because who needs connection pools

* reenable sqlite audit tests

* first pass of migration changes

* stuff is broken :D

* sqlboiler :D

* end of date commit

* Added comments code clean up

* revert go module files back to upstream

* bug fix

* pushed go.mod update to use correc goose version

* renamed sqlite to sqlite3 for consistency across codebase and PR feedback changes

* makefile updates

* things are broken end of day commit

* added postgresql test

* use correct database name

* travis fixes for env vars

* travis fixes for env vars

* test fixes

* run migration on test setup

* test adding postgres support to appveyor

* Skip tests on appveyor due to issues with missing binaries

* oh yeah i have to support windows don't i

* bumped goose version up

* add postgres to osx

* fix travis config as osx does not support services move spin up to before_script

* added PGDATA path fix

* pass PG_DATA to pg_ctl

* added initdb to before install

* fixes to wording and bumps up goose version

* who needs ssl anyway

* moved ssl to correct section :D

* bumped goose version up

* unbreak travis

* unbreak travis

* fix if database is disabled in config

* move strings to consts

* converted more strings to const

* improvements to sqlboiler mmodel gen

* Added contrib\sqlboiler file

* sqlboiler windows contrib fixes

* bumped goose version up

* :D whoops

* further fixes to sql models

* further fixes to sql models

* database type fix for config gen

* README update

* go.mod clean up

* added config details for appveyor

* appveyor ordering fix

* force psql9.6

* appveyor config changes

* all the environmen vars

* model changes for psql

* model changes for psql

* sqlite model fixes

* attempt at osx fix

* added error check for migration

* typos and check against goose error instead of string :D

* updated sqlboiler commit id

* bump sqlboiler version again

* set decimal package to @0bb1631

* readme and makefile updates

* bump goose version update readme and add override flag to config gen

* README typo fix and lowered inserts in test down to 20 as we are only testing that inserts work running 200 was unnecessary

* added gctcli command for audit event

* Added debug output toggle to config added both postgres & sqlite support to gctcli command

* Wording changes on errors

* set sqlite to 1 connection to stop locke database issues

* Usage update for order

* README updates with config examples

* go.mod/sum tidy

* removed lines in import second

* removed lines in imports

* convert local time to utc for database and display output

* go mod clean up and error checking to time

* renamed all packages to sqlite3

* added windows command output for sql model gen

* time conversion fix

* time conversion on gctcli
2019-10-08 15:28:31 +11:00

477 lines
17 KiB
Go

package engine
import (
"errors"
"flag"
"fmt"
"runtime"
"sync"
"time"
"github.com/thrasher-corp/gocryptotrader/common"
"github.com/thrasher-corp/gocryptotrader/config"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/currency/coinmarketcap"
"github.com/thrasher-corp/gocryptotrader/dispatch"
exchange "github.com/thrasher-corp/gocryptotrader/exchanges"
"github.com/thrasher-corp/gocryptotrader/exchanges/request"
log "github.com/thrasher-corp/gocryptotrader/logger"
"github.com/thrasher-corp/gocryptotrader/portfolio"
"github.com/thrasher-corp/gocryptotrader/utils"
)
// Engine contains configuration, portfolio, exchange & ticker data and is the
// overarching type across this code base.
type Engine struct {
Config *config.Config
Portfolio *portfolio.Base
Exchanges []exchange.IBotExchange
ExchangeCurrencyPairManager *ExchangeCurrencyPairSyncer
NTPManager ntpManager
ConnectionManager connectionManager
DatabaseManager databaseManager
OrderManager orderManager
PortfolioManager portfolioManager
CommsManager commsManager
DepositAddressManager *DepositAddressManager
Settings Settings
Uptime time.Time
ServicesWG sync.WaitGroup
}
// Vars for engine
var (
Bot *Engine
)
func init() {
if Bot == nil {
return
}
}
// New starts a new engine
func New() (*Engine, error) {
var b Engine
b.Config = &config.Cfg
err := b.Config.LoadConfig("", false)
if err != nil {
return nil, fmt.Errorf("failed to load config. Err: %s", err)
}
return &b, nil
}
// NewFromSettings starts a new engine based on supplied settings
func NewFromSettings(settings *Settings) (*Engine, error) {
if settings == nil {
return nil, errors.New("engine: settings is nil")
}
var b Engine
b.Config = &config.Cfg
filePath, err := config.GetFilePath(settings.ConfigFile)
if err != nil {
return nil, err
}
log.Debugf(log.Global, "Loading config file %s..\n", filePath)
err = b.Config.LoadConfig(filePath, settings.EnableDryRun)
if err != nil {
return nil, fmt.Errorf("failed to load config. Err: %s", err)
}
err = common.CreateDir(settings.DataDir)
if err != nil {
return nil, fmt.Errorf("failed to open/create data directory: %s. Err: %s", settings.DataDir, err)
}
if *b.Config.Logging.Enabled {
log.SetupGlobalLogger()
log.SetupSubLoggers(b.Config.Logging.SubLoggers)
}
b.Settings.ConfigFile = filePath
b.Settings.DataDir = settings.DataDir
err = utils.AdjustGoMaxProcs(settings.GoMaxProcs)
if err != nil {
return nil, fmt.Errorf("unable to adjust runtime GOMAXPROCS value. Err: %s", err)
}
ValidateSettings(&b, settings)
return &b, nil
}
// ValidateSettings validates and sets all bot settings
func ValidateSettings(b *Engine, s *Settings) {
b.Settings.Verbose = s.Verbose
b.Settings.EnableDryRun = s.EnableDryRun
b.Settings.EnableAllExchanges = s.EnableAllExchanges
b.Settings.EnableAllPairs = s.EnableAllPairs
b.Settings.EnablePortfolioManager = s.EnablePortfolioManager
b.Settings.EnableCoinmarketcapAnalysis = s.EnableCoinmarketcapAnalysis
b.Settings.EnableDatabaseManager = s.EnableDatabaseManager
b.Settings.EnableDispatcher = s.EnableDispatcher
// TO-DO: FIXME
if flag.Lookup("grpc") != nil {
b.Settings.EnableGRPC = s.EnableGRPC
} else {
b.Settings.EnableGRPC = b.Config.RemoteControl.GRPC.Enabled
}
if flag.Lookup("grpcproxy") != nil {
b.Settings.EnableGRPCProxy = s.EnableGRPCProxy
} else {
b.Settings.EnableGRPCProxy = b.Config.RemoteControl.GRPC.GRPCProxyEnabled
}
if flag.Lookup("websocketrpc") != nil {
b.Settings.EnableWebsocketRPC = s.EnableWebsocketRPC
} else {
b.Settings.EnableWebsocketRPC = b.Config.RemoteControl.WebsocketRPC.Enabled
}
if flag.Lookup("deprecatedrpc") != nil {
b.Settings.EnableDeprecatedRPC = s.EnableDeprecatedRPC
} else {
b.Settings.EnableDeprecatedRPC = b.Config.RemoteControl.DeprecatedRPC.Enabled
}
b.Settings.EnableCommsRelayer = s.EnableCommsRelayer
b.Settings.EnableEventManager = s.EnableEventManager
if b.Settings.EnableEventManager {
if b.Settings.EventManagerDelay != time.Duration(0) && s.EventManagerDelay > 0 {
b.Settings.EventManagerDelay = s.EventManagerDelay
} else {
b.Settings.EventManagerDelay = EventSleepDelay
}
}
b.Settings.EnableConnectivityMonitor = s.EnableConnectivityMonitor
b.Settings.EnableNTPClient = s.EnableNTPClient
b.Settings.EnableOrderManager = s.EnableOrderManager
b.Settings.EnableExchangeSyncManager = s.EnableExchangeSyncManager
b.Settings.EnableDepositAddressManager = s.EnableDepositAddressManager
b.Settings.EnableTickerSyncing = s.EnableTickerSyncing
b.Settings.EnableOrderbookSyncing = s.EnableOrderbookSyncing
b.Settings.EnableExchangeAutoPairUpdates = s.EnableExchangeAutoPairUpdates
b.Settings.EnableExchangeWebsocketSupport = s.EnableExchangeWebsocketSupport
b.Settings.EnableExchangeRESTSupport = s.EnableExchangeRESTSupport
b.Settings.EnableExchangeVerbose = s.EnableExchangeVerbose
b.Settings.EnableExchangeHTTPRateLimiter = s.EnableExchangeHTTPDebugging
b.Settings.EnableExchangeHTTPDebugging = s.EnableExchangeHTTPDebugging
b.Settings.DisableExchangeAutoPairUpdates = s.DisableExchangeAutoPairUpdates
b.Settings.ExchangePurgeCredentials = s.ExchangePurgeCredentials
b.Settings.EnableWebsocketRoutine = s.EnableWebsocketRoutine
if !b.Settings.EnableExchangeHTTPRateLimiter {
request.DisableRateLimiter = true
}
// Checks if the flag values are different from the defaults
b.Settings.MaxHTTPRequestJobsLimit = s.MaxHTTPRequestJobsLimit
if b.Settings.MaxHTTPRequestJobsLimit != request.DefaultMaxRequestJobs && s.MaxHTTPRequestJobsLimit > 0 {
request.MaxRequestJobs = b.Settings.MaxHTTPRequestJobsLimit
}
b.Settings.RequestTimeoutRetryAttempts = s.RequestTimeoutRetryAttempts
if b.Settings.RequestTimeoutRetryAttempts != request.DefaultTimeoutRetryAttempts && s.RequestTimeoutRetryAttempts > 0 {
request.TimeoutRetryAttempts = b.Settings.RequestTimeoutRetryAttempts
}
b.Settings.ExchangeHTTPTimeout = s.ExchangeHTTPTimeout
if s.ExchangeHTTPTimeout != time.Duration(0) && s.ExchangeHTTPTimeout > 0 {
b.Settings.ExchangeHTTPTimeout = s.ExchangeHTTPTimeout
} else {
b.Settings.ExchangeHTTPTimeout = b.Config.GlobalHTTPTimeout
}
b.Settings.ExchangeHTTPUserAgent = s.ExchangeHTTPUserAgent
b.Settings.ExchangeHTTPProxy = s.ExchangeHTTPProxy
if s.GlobalHTTPTimeout != time.Duration(0) && s.GlobalHTTPTimeout > 0 {
b.Settings.GlobalHTTPTimeout = s.GlobalHTTPTimeout
} else {
b.Settings.GlobalHTTPTimeout = b.Config.GlobalHTTPTimeout
}
common.HTTPClient = common.NewHTTPClientWithTimeout(b.Settings.GlobalHTTPTimeout)
b.Settings.GlobalHTTPUserAgent = s.GlobalHTTPUserAgent
if b.Settings.GlobalHTTPUserAgent != "" {
common.HTTPUserAgent = b.Settings.GlobalHTTPUserAgent
}
b.Settings.GlobalHTTPProxy = s.GlobalHTTPProxy
b.Settings.DispatchMaxWorkerAmount = s.DispatchMaxWorkerAmount
}
// PrintSettings returns the engine settings
func PrintSettings(s *Settings) {
log.Debugln(log.Global)
log.Debugf(log.Global, "ENGINE SETTINGS")
log.Debugf(log.Global, "- CORE SETTINGS:")
log.Debugf(log.Global, "\t Verbose mode: %v", s.Verbose)
log.Debugf(log.Global, "\t Enable dry run mode: %v", s.EnableDryRun)
log.Debugf(log.Global, "\t Enable all exchanges: %v", s.EnableAllExchanges)
log.Debugf(log.Global, "\t Enable all pairs: %v", s.EnableAllPairs)
log.Debugf(log.Global, "\t Enable coinmarketcap analaysis: %v", s.EnableCoinmarketcapAnalysis)
log.Debugf(log.Global, "\t Enable portfolio manager: %v", s.EnablePortfolioManager)
log.Debugf(log.Global, "\t Enable gPRC: %v", s.EnableGRPC)
log.Debugf(log.Global, "\t Enable gRPC Proxy: %v", s.EnableGRPCProxy)
log.Debugf(log.Global, "\t Enable websocket RPC: %v", s.EnableWebsocketRPC)
log.Debugf(log.Global, "\t Enable deprecated RPC: %v", s.EnableDeprecatedRPC)
log.Debugf(log.Global, "\t Enable comms relayer: %v", s.EnableCommsRelayer)
log.Debugf(log.Global, "\t Enable event manager: %v", s.EnableEventManager)
log.Debugf(log.Global, "\t Event manager sleep delay: %v", s.EventManagerDelay)
log.Debugf(log.Global, "\t Enable order manager: %v", s.EnableOrderManager)
log.Debugf(log.Global, "\t Enable exchange sync manager: %v", s.EnableExchangeSyncManager)
log.Debugf(log.Global, "\t Enable deposit address manager: %v\n", s.EnableDepositAddressManager)
log.Debugf(log.Global, "\t Enable ticker syncing: %v", s.EnableTickerSyncing)
log.Debugf(log.Global, "\t Enable orderbook syncing: %v", s.EnableOrderbookSyncing)
log.Debugf(log.Global, "\t Enable websocket routine: %v\n", s.EnableWebsocketRoutine)
log.Debugf(log.Global, "\t Enable NTP client: %v", s.EnableNTPClient)
log.Debugf(log.Global, "\t Enable Database manager: %v", s.EnableDatabaseManager)
log.Debugf(log.Global, "\t Enable dispatcher: %v", s.EnableDispatcher)
log.Debugf(log.Global, "\t Dispatch package max worker amount: %d", s.DispatchMaxWorkerAmount)
log.Debugf(log.Global, "- FOREX SETTINGS:")
log.Debugf(log.Global, "\t Enable currency conveter: %v", s.EnableCurrencyConverter)
log.Debugf(log.Global, "\t Enable currency layer: %v", s.EnableCurrencyLayer)
log.Debugf(log.Global, "\t Enable fixer: %v", s.EnableFixer)
log.Debugf(log.Global, "\t Enable OpenExchangeRates: %v", s.EnableOpenExchangeRates)
log.Debugf(log.Global, "- EXCHANGE SETTINGS:")
log.Debugf(log.Global, "\t Enable exchange auto pair updates: %v", s.EnableExchangeAutoPairUpdates)
log.Debugf(log.Global, "\t Disable all exchange auto pair updates: %v", s.DisableExchangeAutoPairUpdates)
log.Debugf(log.Global, "\t Enable exchange websocket support: %v", s.EnableExchangeWebsocketSupport)
log.Debugf(log.Global, "\t Enable exchange verbose mode: %v", s.EnableExchangeVerbose)
log.Debugf(log.Global, "\t Enable exchange HTTP rate limiter: %v", s.EnableExchangeHTTPRateLimiter)
log.Debugf(log.Global, "\t Enable exchange HTTP debugging: %v", s.EnableExchangeHTTPDebugging)
log.Debugf(log.Global, "\t Exchange max HTTP request jobs: %v", s.MaxHTTPRequestJobsLimit)
log.Debugf(log.Global, "\t Exchange HTTP request timeout retry amount: %v", s.RequestTimeoutRetryAttempts)
log.Debugf(log.Global, "\t Exchange HTTP timeout: %v", s.ExchangeHTTPTimeout)
log.Debugf(log.Global, "\t Exchange HTTP user agent: %v", s.ExchangeHTTPUserAgent)
log.Debugf(log.Global, "\t Exchange HTTP proxy: %v\n", s.ExchangeHTTPProxy)
log.Debugf(log.Global, "- COMMON SETTINGS:")
log.Debugf(log.Global, "\t Global HTTP timeout: %v", s.GlobalHTTPTimeout)
log.Debugf(log.Global, "\t Global HTTP user agent: %v", s.GlobalHTTPUserAgent)
log.Debugf(log.Global, "\t Global HTTP proxy: %v", s.ExchangeHTTPProxy)
log.Debugln(log.Global)
}
// Start starts the engine
func (e *Engine) Start() error {
if e == nil {
return errors.New("engine instance is nil")
}
if e.Settings.EnableDatabaseManager {
if err := e.DatabaseManager.Start(); err != nil {
log.Errorf(log.Global, "Database manager unable to start: %v", err)
}
}
if e.Settings.EnableDispatcher {
if err := dispatch.Start(e.Settings.DispatchMaxWorkerAmount); err != nil {
log.Errorf(log.DispatchMgr, "Dispatcher unable to start: %v", err)
}
}
// Sets up internet connectivity monitor
if e.Settings.EnableConnectivityMonitor {
if err := e.ConnectionManager.Start(); err != nil {
log.Errorf(log.Global, "Connection manager unable to start: %v", err)
}
}
if e.Settings.EnableNTPClient {
if err := e.NTPManager.Start(); err != nil {
log.Errorf(log.Global, "NTP manager unable to start: %v", err)
}
}
e.Uptime = time.Now()
log.Debugf(log.Global, "Bot '%s' started.\n", e.Config.Name)
log.Debugf(log.Global, "Using data dir: %s\n", e.Settings.DataDir)
log.Debugf(log.Global,
"Using %d out of %d logical processors for runtime performance\n",
runtime.GOMAXPROCS(-1), runtime.NumCPU())
enabledExchanges := e.Config.CountEnabledExchanges()
if e.Settings.EnableAllExchanges {
enabledExchanges = len(e.Config.Exchanges)
}
log.Debugln(log.Global, "EXCHANGE COVERAGE")
log.Debugf(log.Global, "\t Available Exchanges: %d. Enabled Exchanges: %d.\n",
len(e.Config.Exchanges), enabledExchanges)
if e.Settings.ExchangePurgeCredentials {
log.Debugln(log.Global, "Purging exchange API credentials.")
e.Config.PurgeExchangeAPICredentials()
}
log.Debugln(log.Global, "Setting up exchanges..")
SetupExchanges()
if len(Bot.Exchanges) == 0 {
return errors.New("no exchanges are loaded")
}
if e.Settings.EnableCommsRelayer {
if err := e.CommsManager.Start(); err != nil {
log.Errorf(log.Global, "Communications manager unable to start: %v\n", err)
}
}
var newFxSettings []currency.FXSettings
for _, d := range e.Config.Currency.ForexProviders {
newFxSettings = append(newFxSettings, currency.FXSettings(d))
}
err := currency.RunStorageUpdater(currency.BotOverrides{
Coinmarketcap: e.Settings.EnableCoinmarketcapAnalysis,
FxCurrencyConverter: e.Settings.EnableCurrencyConverter,
FxCurrencyLayer: e.Settings.EnableCurrencyLayer,
FxFixer: e.Settings.EnableFixer,
FxOpenExchangeRates: e.Settings.EnableOpenExchangeRates,
},
&currency.MainConfiguration{
ForexProviders: newFxSettings,
CryptocurrencyProvider: coinmarketcap.Settings(e.Config.Currency.CryptocurrencyProvider),
Cryptocurrencies: e.Config.Currency.Cryptocurrencies,
FiatDisplayCurrency: e.Config.Currency.FiatDisplayCurrency,
CurrencyDelay: e.Config.Currency.CurrencyFileUpdateDuration,
FxRateDelay: e.Config.Currency.ForeignExchangeUpdateDuration,
},
e.Settings.DataDir,
e.Settings.Verbose)
if err != nil {
log.Errorf(log.Global, "currency updater system failed to start %v", err)
}
if e.Settings.EnableGRPC {
go StartRPCServer()
}
if e.Settings.EnableDeprecatedRPC {
go StartRESTServer()
}
if e.Settings.EnableWebsocketRPC {
go StartWebsocketServer()
StartWebsocketHandler()
}
if e.Settings.EnablePortfolioManager {
if err = e.PortfolioManager.Start(); err != nil {
log.Errorf(log.Global, "Fund manager unable to start: %v", err)
}
}
if e.Settings.EnableDepositAddressManager {
e.DepositAddressManager = new(DepositAddressManager)
e.DepositAddressManager.Sync()
}
if e.Settings.EnableOrderManager {
if err = e.OrderManager.Start(); err != nil {
log.Errorf(log.Global, "Order manager unable to start: %v", err)
}
}
if e.Settings.EnableExchangeSyncManager {
exchangeSyncCfg := CurrencyPairSyncerConfig{
SyncTicker: e.Settings.EnableTickerSyncing,
SyncOrderbook: e.Settings.EnableOrderbookSyncing,
SyncContinuously: true,
NumWorkers: 15,
}
e.ExchangeCurrencyPairManager, err = NewCurrencyPairSyncer(exchangeSyncCfg)
if err != nil {
log.Warnf(log.Global, "Unable to initialise exchange currency pair syncer. Err: %s", err)
} else {
go e.ExchangeCurrencyPairManager.Start()
}
}
if e.Settings.EnableEventManager {
go EventManger()
}
if e.Settings.EnableWebsocketRoutine {
go WebsocketRoutine()
}
return nil
}
// Stop correctly shuts down engine saving configuration files
func (e *Engine) Stop() {
log.Debugln(log.Global, "Engine shutting down..")
if len(portfolio.Portfolio.Addresses) != 0 {
e.Config.Portfolio = portfolio.Portfolio
}
if e.OrderManager.Started() {
if err := e.OrderManager.Stop(); err != nil {
log.Errorf(log.Global, "Order manager unable to stop. Error: %v", err)
}
}
if e.NTPManager.Started() {
if err := e.NTPManager.Stop(); err != nil {
log.Errorf(log.Global, "NTP manager unable to stop. Error: %v", err)
}
}
if e.CommsManager.Started() {
if err := e.CommsManager.Stop(); err != nil {
log.Errorf(log.Global, "Communication manager unable to stop. Error: %v", err)
}
}
if e.PortfolioManager.Started() {
if err := e.PortfolioManager.Stop(); err != nil {
log.Errorf(log.Global, "Fund manager unable to stop. Error: %v", err)
}
}
if e.ConnectionManager.Started() {
if err := e.ConnectionManager.Stop(); err != nil {
log.Errorf(log.Global, "Connection manager unable to stop. Error: %v", err)
}
}
if e.DatabaseManager.Started() {
if err := e.DatabaseManager.Stop(); err != nil {
log.Errorf(log.Global, "Database manager unable to stop. Error: %v", err)
}
}
if dispatch.IsRunning() {
if err := dispatch.Stop(); err != nil {
log.Errorf(log.DispatchMgr, "Dispatch system unable to stop. Error: %v", err)
}
}
if !e.Settings.EnableDryRun {
err := e.Config.SaveConfig(e.Settings.ConfigFile, false)
if err != nil {
log.Errorln(log.Global, "Unable to save config.")
} else {
log.Debugln(log.Global, "Config file saved successfully.")
}
}
// Wait for services to gracefully shutdown
e.ServicesWG.Wait()
err := log.CloseLogger()
if err != nil {
fmt.Printf("Failed to close logger %v", err)
}
}