Daily engine improvements

Link up various subsystems to be managed atomically with the ability to start/stop them
New subsystem APIs
Comms changes

engine/comms_relayer.go

@@ -0,0 +1,87 @@
+package engine
+
+import (
+	"errors"
+	"sync/atomic"
+
+	"github.com/thrasher-/gocryptotrader/communications"
+	"github.com/thrasher-/gocryptotrader/communications/base"
+	log "github.com/thrasher-/gocryptotrader/logger"
+)
+
+// commsManager starts the NTP manager
+type commsManager struct {
+	started  int32
+	stopped  int32
+	shutdown chan struct{}
+	relayMsg chan base.Event
+	comms    *communications.Communications
+}
+
+func (c *commsManager) Started() bool {
+	return atomic.LoadInt32(&c.started) == 1
+}
+
+func (c *commsManager) Start() (err error) {
+	if atomic.AddInt32(&c.started, 1) != 1 {
+		return errors.New("communications manager already started")
+	}
+
+	defer func() {
+		if err != nil {
+			atomic.CompareAndSwapInt32(&c.started, 1, 0)
+		}
+	}()
+
+	log.Debugln("Communications manager starting...")
+	commsCfg := Bot.Config.GetCommunicationsConfig()
+	c.comms, err = communications.NewComm(&commsCfg)
+	if err != nil {
+		return err
+	}
+
+	c.shutdown = make(chan struct{})
+	c.relayMsg = make(chan base.Event)
+	go c.run()
+	log.Debugln("Communications manager started.")
+	return nil
+}
+
+func (c *commsManager) Stop() error {
+	if atomic.LoadInt32(&c.started) == 0 {
+		return errors.New("communications manager not started")
+	}
+
+	if atomic.AddInt32(&c.stopped, 1) != 1 {
+		return errors.New("communications manager is already stopped")
+	}
+
+	close(c.shutdown)
+	log.Debugln("Communications manager shutting down...")
+	return nil
+}
+
+func (c *commsManager) PushEvent(evt base.Event) {
+	if !c.Started() {
+		return
+	}
+	c.relayMsg <- evt
+}
+
+func (c *commsManager) run() {
+	defer func() {
+		// TO-DO shutdown comms connections for connected services (Slack etc)
+		atomic.CompareAndSwapInt32(&c.stopped, 1, 0)
+		atomic.CompareAndSwapInt32(&c.started, 1, 0)
+		log.Debugln("Communications manager shutdown.")
+	}()
+
+	for {
+		select {
+		case msg := <-c.relayMsg:
+			c.comms.PushEvent(msg)
+		case <-c.shutdown:
+			return
+		}
+	}
+}

engine/connection.go

@@ -49,6 +49,8 @@ func (c *connectionManager) Stop() error {
 
 	log.Debugln("Connection manager shutting down...")
 	c.conn.Shutdown()
+	atomic.CompareAndSwapInt32(&c.stopped, 1, 0)
+	atomic.CompareAndSwapInt32(&c.started, 1, 0)
 	log.Debugln("Connection manager stopped.")
 	return nil
 }

engine/engine.go

@@ -12,15 +12,12 @@ import (
 	"time"
 
 	"github.com/thrasher-/gocryptotrader/common"
-	"github.com/thrasher-/gocryptotrader/communications"
 	"github.com/thrasher-/gocryptotrader/config"
 	"github.com/thrasher-/gocryptotrader/currency"
 	"github.com/thrasher-/gocryptotrader/currency/coinmarketcap"
-	"github.com/thrasher-/gocryptotrader/engine/events"
 	exchange "github.com/thrasher-/gocryptotrader/exchanges"
 	"github.com/thrasher-/gocryptotrader/exchanges/request"
 	log "github.com/thrasher-/gocryptotrader/logger"
-	"github.com/thrasher-/gocryptotrader/ntpclient"
 	"github.com/thrasher-/gocryptotrader/portfolio"
 	"github.com/thrasher-/gocryptotrader/utils"
 )
@@ -32,10 +29,11 @@ type Engine struct {
 	Portfolio                      *portfolio.Base
 	Exchanges                      []exchange.IBotExchange
 	ExchangeCurrencyPairManager    *ExchangeCurrencyPairSyncer
+	NTPManager                     ntpManager
 	ConnectionManager              connectionManager
 	OrderManager                   orderManager
 	PortfolioManager               portfolioManager
-	CommsRelayer                   *communications.Communications
+	CommsManager                   commsManager
 	Shutdown                       chan struct{}
 	Settings                       Settings
 	CryptocurrencyDepositAddresses map[string]map[string]string
@@ -148,11 +146,10 @@ func ValidateSettings(b *Engine, s *Settings) {
 	b.Settings.EnableEventManager = s.EnableEventManager
 
 	if b.Settings.EnableEventManager {
-		events.Verbose = b.Settings.Verbose
 		if b.Settings.EventManagerDelay != time.Duration(0) && s.EventManagerDelay > 0 {
 			b.Settings.EventManagerDelay = s.EventManagerDelay
 		} else {
-			b.Settings.EventManagerDelay = events.SleepDelay
+			b.Settings.EventManagerDelay = EventSleepDelay
 		}
 	}
 
@@ -273,27 +270,8 @@ func (e *Engine) Start() {
 	}
 
 	if e.Settings.EnableNTPClient {
-		if e.Config.NTPClient.Level != -1 {
-			NTPTime, errNTP := ntpclient.NTPClient(e.Config.NTPClient.Pool)
-			currentTime := time.Now()
-			if errNTP != nil {
-				log.Warnf("NTPClient failed to create: %v", errNTP)
-			} else {
-				NTPcurrentTimeDifference := NTPTime.Sub(currentTime)
-				configNTPTime := *e.Config.NTPClient.AllowedDifference
-				configNTPNegativeTime := (*e.Config.NTPClient.AllowedNegativeDifference - (*e.Config.NTPClient.AllowedNegativeDifference * 2))
-				if NTPcurrentTimeDifference > configNTPTime || NTPcurrentTimeDifference < configNTPNegativeTime {
-					log.Warnf("Time out of sync (NTP): %v | (time.Now()): %v | (Difference): %v | (Allowed): +%v / %v", NTPTime, currentTime, NTPcurrentTimeDifference, configNTPTime, configNTPNegativeTime)
-					if e.Config.NTPClient.Level == 0 {
-						disable, errNTP := e.Config.DisableNTPCheck(os.Stdin)
-						if errNTP != nil {
-							log.Errorf("failed to disable ntp time check reason: %v", errNTP)
-						} else {
-							log.Info(disable)
-						}
-					}
-				}
-			}
+		if err := e.NTPManager.Start(); err != nil {
+			log.Errorf("NTP manager unable to start: %v", err)
 		}
 	}
 
@@ -322,10 +300,9 @@ func (e *Engine) Start() {
 	}
 
 	if e.Settings.EnableCommsRelayer {
-		log.Debugln("Starting communication mediums..")
-		commsCfg := e.Config.GetCommunicationsConfig()
-		e.CommsRelayer = communications.NewComm(&commsCfg)
-		e.CommsRelayer.GetEnabledCommunicationMediums()
+		if err := e.CommsManager.Start(); err != nil {
+			log.Errorf("Communications manager unable to start: %v", err)
+		}
 	}
 
 	var newFxSettings []currency.FXSettings
@@ -398,7 +375,7 @@ func (e *Engine) Start() {
 	}
 
 	if e.Settings.EnableEventManager {
-		go events.EventManger()
+		go EventManger()
 	}
 
 	<-e.Shutdown
@@ -419,6 +396,18 @@ func (e *Engine) Stop() {
 		}
 	}
 
+	if e.NTPManager.Started() {
+		if err := e.NTPManager.Stop(); err != nil {
+			log.Errorf("NTP manager unable to stop. Error: %v", err)
+		}
+	}
+
+	if e.CommsManager.Started() {
+		if err := e.CommsManager.Stop(); err != nil {
+			log.Errorf("Communication manager unable to stop. Error: %v", err)
+		}
+	}
+
 	if e.PortfolioManager.Started() {
 		if err := e.PortfolioManager.Stop(); err != nil {
 			log.Errorf("Fund manager unable to stop. Error: %v", err)

engine/events.go

@@ -1,4 +1,4 @@
-package events
+package engine
 
 import (
 	"errors"
@@ -6,7 +6,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/thrasher-/gocryptotrader/communications"
 	"github.com/thrasher-/gocryptotrader/communications/base"
 	"github.com/thrasher-/gocryptotrader/config"
 	"github.com/thrasher-/gocryptotrader/currency"
@@ -16,6 +15,8 @@ import (
 	log "github.com/thrasher-/gocryptotrader/logger"
 )
 
+// TO-DO MAKE THIS A SERVICE SUBSYSTEM
+
 // Event const vars
 const (
 	ItemPrice     = "PRICE"
@@ -32,7 +33,6 @@ const (
 	ActionTest         = "ACTION_TEST"
 
 	defaultSleepDelay = time.Millisecond * 500
-	defaultVerbose    = true
 )
 
 // vars related to events package
@@ -41,16 +41,11 @@ var (
 	errInvalidCondition = errors.New("invalid conditional option")
 	errInvalidAction    = errors.New("invalid action")
 	errExchangeDisabled = errors.New("desired exchange is disabled")
-
-	SleepDelay = defaultSleepDelay
-	Verbose    = defaultVerbose
-
-	// NOTE comms is an interim implementation
-	comms *communications.Communications
+	EventSleepDelay     = defaultSleepDelay
 )
 
-// ConditionParams holds the event condition variables
-type ConditionParams struct {
+// EventConditionParams holds the event condition variables
+type EventConditionParams struct {
 	Condition string
 	Price     float64
 
@@ -64,7 +59,7 @@ type Event struct {
 	ID        int64
 	Exchange  string
 	Item      string
-	Condition ConditionParams
+	Condition EventConditionParams
 	Pair      currency.Pair
 	Asset     assets.AssetType
 	Action    string
@@ -75,15 +70,9 @@ type Event struct {
 // appended
 var Events []*Event
 
-// SetComms is an interim function that will support a median integration. This
-// sets the current comms package.
-func SetComms(commsP *communications.Communications) {
-	comms = commsP
-}
-
 // Add adds an event to the Events chain and returns an index/eventID
 // and an error
-func Add(exchange, item string, condition ConditionParams, currencyPair currency.Pair, asset assets.AssetType, action string) (int64, error) {
+func Add(exchange, item string, condition EventConditionParams, currencyPair currency.Pair, asset assets.AssetType, action string) (int64, error) {
 	err := IsValidEvent(exchange, item, condition, action)
 	if err != nil {
 		return 0, err
@@ -139,7 +128,7 @@ func (e *Event) ExecuteAction() bool {
 		if action[0] == ActionSMSNotify {
 			message := fmt.Sprintf("Event triggered: %s", e.String())
 			if action[1] == "ALL" {
-				comms.PushEvent(base.Event{
+				Bot.CommsManager.PushEvent(base.Event{
 					Type:    "event",
 					Message: message,
 				})
@@ -164,7 +153,7 @@ func (e *Event) processTicker() bool {
 
 	t, err := ticker.GetTicker(e.Exchange, e.Pair, e.Asset)
 	if err != nil {
-		if Verbose {
+		if Bot.Settings.Verbose {
 			log.Debugf("Events: failed to get ticker. Err: %s", err)
 		}
 		return false
@@ -173,7 +162,7 @@ func (e *Event) processTicker() bool {
 	lastPrice := t.Last
 
 	if lastPrice == 0 {
-		if Verbose {
+		if Bot.Settings.Verbose {
 			log.Debugln("Events: ticker last price is 0")
 		}
 		return false
@@ -211,7 +200,7 @@ func (e *Event) processCondition(actual, threshold float64) bool {
 func (e *Event) processOrderbook() bool {
 	ob, err := orderbook.Get(e.Exchange, e.Pair, e.Asset)
 	if err != nil {
-		if Verbose {
+		if Bot.Settings.Verbose {
 			log.Debugf("Events: Failed to get orderbook. Err: %s", err)
 		}
 		return false
@@ -242,18 +231,17 @@ func (e *Event) processOrderbook() bool {
 	return success
 }
 
-// CheckCondition will check the event structure to see if there is a condition
+// CheckEventCondition will check the event structure to see if there is a condition
 // met
-func (e *Event) CheckCondition() bool {
+func (e *Event) CheckEventCondition() bool {
 	if e.Item == ItemPrice {
 		return e.processTicker()
 	}
-
 	return e.processOrderbook()
 }
 
 // IsValidEvent checks the actions to be taken and returns an error if incorrect
-func IsValidEvent(exchange, item string, condition ConditionParams, action string) error {
+func IsValidEvent(exchange, item string, condition EventConditionParams, action string) error {
 	exchange = strings.ToUpper(exchange)
 	item = strings.ToUpper(item)
 	action = strings.ToUpper(action)
@@ -290,7 +278,7 @@ func IsValidEvent(exchange, item string, condition ConditionParams, action strin
 		}
 
 		if a[1] != "ALL" {
-			comms.PushEvent(base.Event{Type: a[1]})
+			Bot.CommsManager.PushEvent(base.Event{Type: a[1]})
 		}
 	} else if action != ActionConsolePrint && action != ActionTest {
 		return errInvalidAction
@@ -302,17 +290,17 @@ func IsValidEvent(exchange, item string, condition ConditionParams, action strin
 // EventManger is the overarching routine that will iterate through the Events
 // chain
 func EventManger() {
-	log.Debugf("EventManager started. SleepDelay: %v", SleepDelay.String())
+	log.Debugf("EventManager started. SleepDelay: %v", EventSleepDelay.String())
 
 	for {
 		total, executed := GetEventCounter()
 		if total > 0 && executed != total {
 			for _, event := range Events {
 				if !event.Executed {
-					if Verbose {
+					if Bot.Settings.Verbose {
 						log.Debugf("Events: Processing event %s.", event.String())
 					}
-					success := event.CheckCondition()
+					success := event.CheckEventCondition()
 					if success {
 						log.Debugf(
 							"Events: ID: %d triggered on %s successfully.\n", event.ID,
@@ -323,7 +311,7 @@ func EventManger() {
 				}
 			}
 		}
-		time.Sleep(SleepDelay)
+		time.Sleep(EventSleepDelay)
 	}
 }
 

engine/events_test.go

@@ -1,4 +1,4 @@
-package events
+package engine
 
 //
 // import (

engine/helpers.go

@@ -29,6 +29,87 @@ import (
 	"github.com/thrasher-/gocryptotrader/utils"
 )
 
+// GetSubsystemsStatus returns the status of various subsystems
+func GetSubsystemsStatus() map[string]bool {
+	systems := make(map[string]bool)
+	systems["communications"] = Bot.CommsManager.Started()
+	systems["internet_monitor"] = Bot.ConnectionManager.Started()
+	systems["orders"] = Bot.OrderManager.Started()
+	systems["portfolio"] = Bot.PortfolioManager.Started()
+	systems["ntp_timekeeper"] = Bot.NTPManager.Started()
+	systems["exchange_syncer"] = Bot.Settings.EnableExchangeSyncManager
+	systems["grpc"] = Bot.Settings.EnableGRPC
+	systems["grpc_proxy"] = Bot.Settings.EnableGRPCProxy
+	systems["deprecated_rpc"] = Bot.Settings.EnableDeprecatedRPC
+	systems["websocket_rpc"] = Bot.Settings.EnableWebsocketRPC
+	return systems
+}
+
+// RPCEndpoint stores an RPC endpoint status and addr
+type RPCEndpoint struct {
+	Started    bool
+	ListenAddr string
+}
+
+// GetRPCEndpoints returns a list of RPC endpoints and their listen addrs
+func GetRPCEndpoints() map[string]RPCEndpoint {
+	endpoints := make(map[string]RPCEndpoint)
+	endpoints["grpc"] = RPCEndpoint{
+		Started:    Bot.Settings.EnableGRPC,
+		ListenAddr: "grpc://" + Bot.Config.RemoteControl.GRPC.ListenAddress,
+	}
+	endpoints["grpc_proxy"] = RPCEndpoint{
+		Started:    Bot.Settings.EnableGRPCProxy,
+		ListenAddr: "http://" + Bot.Config.RemoteControl.GRPC.GRPCProxyListenAddress,
+	}
+	endpoints["deprecated_rpc"] = RPCEndpoint{
+		Started:    Bot.Settings.EnableDeprecatedRPC,
+		ListenAddr: "http://" + Bot.Config.RemoteControl.DeprecatedRPC.ListenAddress,
+	}
+	endpoints["websocket_rpc"] = RPCEndpoint{
+		Started:    Bot.Settings.EnableWebsocketRPC,
+		ListenAddr: "ws://" + Bot.Config.RemoteControl.WebsocketRPC.ListenAddress,
+	}
+	return endpoints
+}
+
+// SetSubsystem enables or disables an engine subsystem
+func SetSubsystem(subsys string, enable bool) error {
+	switch strings.ToLower(subsys) {
+	case "communications":
+		if enable {
+			return Bot.CommsManager.Start()
+		}
+		return Bot.CommsManager.Stop()
+	case "internet_monitor":
+		if enable {
+			return Bot.ConnectionManager.Start()
+		}
+		return Bot.CommsManager.Stop()
+	case "orders":
+		if enable {
+			return Bot.OrderManager.Start()
+		}
+		return Bot.OrderManager.Stop()
+	case "portfolio":
+		if enable {
+			return Bot.PortfolioManager.Start()
+		}
+		return Bot.OrderManager.Stop()
+	case "ntp_timekeeper":
+		if enable {
+			return Bot.NTPManager.Start()
+		}
+		return Bot.NTPManager.Stop()
+	case "exchange_syncer":
+		if enable {
+			Bot.ExchangeCurrencyPairManager.Start()
+		}
+		Bot.ExchangeCurrencyPairManager.Stop()
+	}
+	return errors.New("subsystem not found")
+}
+
 // GetExchangeOTPs returns OTP codes for all exchanges which have a otpsecret
 // stored
 func GetExchangeOTPs() (map[string]string, error) {

engine/orders.go

@@ -72,9 +72,17 @@ func (o *orderManager) Start() error {
 	return nil
 }
 func (o *orderManager) Stop() error {
+	if atomic.LoadInt32(&o.started) == 0 {
+		return errors.New("order manager not started")
+	}
+
 	if atomic.AddInt32(&o.stopped, 1) != 1 {
 		return errors.New("order manager is already stopped")
 	}
+	defer func() {
+		atomic.CompareAndSwapInt32(&o.stopped, 1, 0)
+		atomic.CompareAndSwapInt32(&o.started, 1, 0)
+	}()
 
 	log.Debugln("Order manager shutting down...")
 	close(o.shutdown)
@@ -101,7 +109,7 @@ func (o *orderManager) gracefulShutdown() {
 					msg := fmt.Sprintf("Order manager: Exchange %s unable to cancel order ID=%v. Err: %s",
 						k, v[y].ID, err)
 					log.Debugln(msg)
-					Bot.CommsRelayer.PushEvent(base.Event{
+					Bot.CommsManager.PushEvent(base.Event{
 						Type:    "order",
 						Message: msg,
 					})
@@ -111,7 +119,7 @@ func (o *orderManager) gracefulShutdown() {
 				msg := fmt.Sprintf("Order manager: Exchange %s order ID=%v cancelled.",
 					k, v[y].ID)
 				log.Debugln(msg)
-				Bot.CommsRelayer.PushEvent(base.Event{
+				Bot.CommsManager.PushEvent(base.Event{
 					Type:    "order",
 					Message: msg,
 				})
@@ -222,7 +230,7 @@ func (o *orderManager) Submit(exchName string, order *exchange.OrderSubmission)
 	msg := fmt.Sprintf("Order manager: Exchange %s submitted order ID=%v [Ours: %v] pair=%v price=%v amount=%v side=%v type=%v.",
 		exchName, result.OrderID, id.String(), order.Pair, order.Price, order.Amount, order.OrderSide, order.OrderType)
 	log.Debugln(msg)
-	Bot.CommsRelayer.PushEvent(base.Event{
+	Bot.CommsManager.PushEvent(base.Event{
 		Type:    "order",
 		Message: msg,
 	})
@@ -257,7 +265,7 @@ func (o *orderManager) processOrders() {
 				msg := fmt.Sprintf("Order manager: Exchange %s added order ID=%v pair=%v price=%v amount=%v side=%v type=%v.",
 					order.Exchange, order.ID, order.CurrencyPair, order.Price, order.Amount, order.OrderSide, order.OrderType)
 				log.Debug(msg)
-				Bot.CommsRelayer.PushEvent(base.Event{
+				Bot.CommsManager.PushEvent(base.Event{
 					Type:    "order",
 					Message: msg,
 				})

engine/portfolio.go

@@ -51,9 +51,11 @@ func (p *portfolioManager) run() {
 	Bot.ServicesWG.Add(1)
 	tick := time.NewTicker(PortfolioSleepDelay)
 	defer func() {
-		log.Debugf("Portfolio manager shutdown.")
+		atomic.CompareAndSwapInt32(&p.stopped, 1, 0)
+		atomic.CompareAndSwapInt32(&p.started, 1, 0)
 		tick.Stop()
 		Bot.ServicesWG.Done()
+		log.Debugf("Portfolio manager shutdown.")
 	}()
 
 	for {

engine/rpcserver.go

@@ -15,7 +15,6 @@ import (
 	"github.com/thrasher-/gocryptotrader/common"
 	"github.com/thrasher-/gocryptotrader/common/crypto"
 	"github.com/thrasher-/gocryptotrader/currency"
-	"github.com/thrasher-/gocryptotrader/engine/events"
 	exchange "github.com/thrasher-/gocryptotrader/exchanges"
 	"github.com/thrasher-/gocryptotrader/exchanges/assets"
 	"github.com/thrasher-/gocryptotrader/gctrpc"
@@ -152,8 +151,47 @@ func (s *RPCServer) GetInfo(ctx context.Context, r *gctrpc.GetInfoRequest) (*gct
 		AvailableExchanges:   int64(len(Bot.Config.Exchanges)),
 		DefaultFiatCurrency:  Bot.Config.Currency.FiatDisplayCurrency.String(),
 		DefaultForexProvider: Bot.Config.GetPrimaryForexProvider(),
+		SubsystemStatus:      GetSubsystemsStatus(),
 	}
+	endpoints := GetRPCEndpoints()
+	resp.RpcEndpoints = make(map[string]*gctrpc.RPCEndpoint)
+	for k, v := range endpoints {
+		resp.RpcEndpoints[k] = &gctrpc.RPCEndpoint{
+			Started:       v.Started,
+			ListenAddress: v.ListenAddr,
+		}
+	}
+	return &resp, nil
+}
 
+// GetSubsystems returns a list of subsystems and their status
+func (s *RPCServer) GetSubsystems(ctx context.Context, r *gctrpc.GetSubsystemsRequest) (*gctrpc.GetSusbsytemsResponse, error) {
+	return &gctrpc.GetSusbsytemsResponse{SubsystemsStatus: GetSubsystemsStatus()}, nil
+}
+
+// EnableSubsystem enables a engine subsytem
+func (s *RPCServer) EnableSubsystem(ctx context.Context, r *gctrpc.GenericSubsystemRequest) (*gctrpc.GenericSubsystemResponse, error) {
+	err := SetSubsystem(r.Subsystem, true)
+	return &gctrpc.GenericSubsystemResponse{}, err
+}
+
+// DisableSubsystem disables a engine subsytem
+func (s *RPCServer) DisableSubsystem(ctx context.Context, r *gctrpc.GenericSubsystemRequest) (*gctrpc.GenericSubsystemResponse, error) {
+	err := SetSubsystem(r.Subsystem, false)
+	return &gctrpc.GenericSubsystemResponse{}, err
+}
+
+// GetRPCEndpoints returns a list of API endpoints
+func (s *RPCServer) GetRPCEndpoints(ctx context.Context, r *gctrpc.GetRPCEndpointsRequest) (*gctrpc.GetRPCEndpointsResponse, error) {
+	endpoints := GetRPCEndpoints()
+	var resp gctrpc.GetRPCEndpointsResponse
+	resp.Endpoints = make(map[string]*gctrpc.RPCEndpoint)
+	for k, v := range endpoints {
+		resp.Endpoints[k] = &gctrpc.RPCEndpoint{
+			Started:       v.Started,
+			ListenAddress: v.ListenAddr,
+		}
+	}
 	return &resp, nil
 }
 
@@ -638,7 +676,7 @@ func (s *RPCServer) GetEvents(ctx context.Context, r *gctrpc.GetEventsRequest) (
 
 // AddEvent adds an event
 func (s *RPCServer) AddEvent(ctx context.Context, r *gctrpc.AddEventRequest) (*gctrpc.AddEventResponse, error) {
-	evtCondition := events.ConditionParams{
+	evtCondition := EventConditionParams{
 		CheckBids:        r.ConditionParams.CheckBids,
 		CheckBidsAndAsks: r.ConditionParams.CheckBidsAndAsks,
 		Condition:        r.ConditionParams.Condition,
@@ -649,7 +687,7 @@ func (s *RPCServer) AddEvent(ctx context.Context, r *gctrpc.AddEventRequest) (*g
 	p := currency.NewPairWithDelimiter(r.Pair.Base,
 		r.Pair.Quote, r.Pair.Delimiter)
 
-	id, err := events.Add(r.Exchange, r.Item, evtCondition, p, assets.AssetType(r.AssetType), r.Action)
+	id, err := Add(r.Exchange, r.Item, evtCondition, p, assets.AssetType(r.AssetType), r.Action)
 	if err != nil {
 		return nil, err
 	}
@@ -659,7 +697,7 @@ func (s *RPCServer) AddEvent(ctx context.Context, r *gctrpc.AddEventRequest) (*g
 
 // RemoveEvent removes an event, specified by an event ID
 func (s *RPCServer) RemoveEvent(ctx context.Context, r *gctrpc.RemoveEventRequest) (*gctrpc.RemoveEventResponse, error) {
-	events.Remove(r.Id)
+	Remove(r.Id)
 	return &gctrpc.RemoveEventResponse{}, nil
 }
 

engine/timekeeper.go

@@ -0,0 +1,140 @@
+package engine
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"sync/atomic"
+	"time"
+
+	log "github.com/thrasher-/gocryptotrader/logger"
+	ntpclient "github.com/thrasher-/gocryptotrader/ntpclient"
+)
+
+// vars related to the NTP manager
+var (
+	NTPCheckInterval = time.Second * 30
+	NTPRetryLimit    = 3
+	errNTPDisabled   = errors.New("ntp client disabled")
+)
+
+// ntpManager starts the NTP manager
+type ntpManager struct {
+	started       int32
+	stopped       int32
+	inititalCheck bool
+	shutdown      chan struct{}
+}
+
+func (n *ntpManager) Started() bool {
+	return atomic.LoadInt32(&n.started) == 1
+}
+
+func (n *ntpManager) Start() (err error) {
+	if atomic.AddInt32(&n.started, 1) != 1 {
+		return errors.New("NTP manager already started")
+	}
+
+	var disable bool
+	defer func() {
+		if err != nil || disable {
+			atomic.CompareAndSwapInt32(&n.started, 1, 0)
+		}
+	}()
+
+	log.Debugln("NTP manager starting...")
+	if Bot.Config.NTPClient.Level == 0 {
+		// Initial NTP check (prompts user on how we should proceed)
+		n.inititalCheck = true
+
+		// Sometimes the NTP client can have transient issues due to UDP, try
+		// the default retry limits before giving up
+		for i := 0; i < NTPRetryLimit; i++ {
+			err = n.processTime()
+			switch err {
+			case nil:
+				break
+			case errNTPDisabled:
+				log.Debugf("NTP manager: User disabled NTP prompts. Exiting.")
+				disable = true
+				err = nil
+				return
+			default:
+				if i == NTPRetryLimit-1 {
+					return err
+				}
+			}
+		}
+	}
+	n.shutdown = make(chan struct{})
+	go n.run()
+	log.Debugln("NTP manager started.")
+	return nil
+}
+
+func (n *ntpManager) Stop() error {
+	if atomic.LoadInt32(&n.started) == 0 {
+		return errors.New("NTP manager not started")
+	}
+
+	if atomic.AddInt32(&n.stopped, 1) != 1 {
+		return errors.New("NTP manager is already stopped")
+	}
+
+	close(n.shutdown)
+	log.Debugln("NTP manager shutting down...")
+	return nil
+}
+
+func (n *ntpManager) run() {
+	t := time.NewTicker(NTPCheckInterval)
+	defer func() {
+		t.Stop()
+		atomic.CompareAndSwapInt32(&n.stopped, 1, 0)
+		atomic.CompareAndSwapInt32(&n.started, 1, 0)
+		log.Debugln("NTP manager shutdown.")
+	}()
+
+	for {
+		select {
+		case <-n.shutdown:
+			return
+		case <-t.C:
+			n.processTime()
+			if Bot.Config.NTPClient.Level == 0 {
+				close(n.shutdown)
+			}
+		}
+	}
+}
+
+func (n *ntpManager) FetchNTPTime() (time.Time, error) {
+	return ntpclient.NTPClient(Bot.Config.NTPClient.Pool)
+}
+
+func (n *ntpManager) processTime() error {
+	NTPTime, err := n.FetchNTPTime()
+	if err != nil {
+		return err
+	}
+
+	currentTime := time.Now()
+	NTPcurrentTimeDifference := NTPTime.Sub(currentTime)
+	configNTPTime := *Bot.Config.NTPClient.AllowedDifference
+	configNTPNegativeTime := (*Bot.Config.NTPClient.AllowedNegativeDifference - (*Bot.Config.NTPClient.AllowedNegativeDifference * 2))
+	if NTPcurrentTimeDifference > configNTPTime || NTPcurrentTimeDifference < configNTPNegativeTime {
+		log.Warnf("NTP manager: Time out of sync (NTP): %v | (time.Now()): %v | (Difference): %v | (Allowed): +%v / %v", NTPTime, currentTime, NTPcurrentTimeDifference, configNTPTime, configNTPNegativeTime)
+		if n.inititalCheck {
+			n.inititalCheck = false
+			disable, err := Bot.Config.DisableNTPCheck(os.Stdin)
+			if err != nil {
+				return fmt.Errorf("unable to disable NTP check: %s", err)
+			}
+			log.Info(disable)
+			if Bot.Config.NTPClient.Level == -1 {
+				return errNTPDisabled
+			}
+		}
+	}
+	return nil
+}