Feature: Data history manager engine subsystem (#693)

* Adds lovely initial concept for historical data doer

* Adds ability to save tasks. Adds config. Adds startStop to engine

* Has a database microservice without use of globals! Further infrastructure design. Adds readme

* Commentary to help design

* Adds migrations for database

* readme and adds database models

* Some modelling that doesn't work end of day

* Completes datahistoryjob sql.Begins datahistoryjobresult

* Adds datahistoryjob functions to retreive job results. Adapts subsystem

* Adds process for upserting jobs and job results to the database

* Broken end of day weird sqlboiler crap

* Fixes issue with SQL generation.

* RPC generation and addition of basic upsert command

* Renames types

* Adds rpc functions

* quick commit before context swithc. Exchanges aren't being populated

* Begin the tests!

* complete sql tests. stop failed jobs. CLI command creation

* Defines rpc commands

* Fleshes out RPC implementation

* Expands testing

* Expands testing, removes double remove

* Adds coverage of data history subsystem, expands errors and nil checks

* Minor logic improvement

* streamlines datahistory test setup

* End of day minor linting

* Lint, convert simplify, rpc expansion, type expansion, readme expansion

* Documentation update

* Renames for consistency

* Completes RPC server commands

* Fixes tests

* Speeds up testing by reducing unnecessary actions. Adds maxjobspercycle config

* Comments for everything

* Adds missing result string. checks interval supported. default start end cli

* Fixes ID problem. Improves binance trade fetch. job ranges are processed

* adds dbservice coverage. adds rpcserver coverage

* docs regen, uses dbcon interface, reverts binance, fixes races, toggle manager

* Speed up tests, remove bad global usage, fix uuid check

* Adds verbose. Updates docs. Fixes postgres

* Minor changes to logging and start stop

* Fixes postgres db tests, fixes postgres column typo

* Fixes old string typo,removes constraint,error parsing for nonreaders

* prevents dhm running when table doesn't exist. Adds prereq documentation

* Adds parallel, rmlines, err fix, comment fix, minor param fixes

* doc regen, common time range check and test updating

* Fixes job validation issues. Updates candle range checker.

* Ensures test cannot fail due to time.Now() shenanigans

* Fixes oopsie, adds documentation and a warn

* Fixes another time test, adjusts copy

* Drastically speeds up data history manager tests via function overrides

* Fixes summary bug and better logs

* Fixes local time test, fixes websocket tests

* removes defaults and comment,updates error messages,sets cli command args

* Fixes FTX trade processing

* Fixes issue where jobs got stuck if data wasn't returned but retrieval was successful

* Improves test speed. Simplifies trade verification SQL. Adds command help

* Fixes the oopsies

* Fixes use of query within transaction. Fixes trade err

* oopsie, not needed

* Adds missing data status. Properly ends job even when data is missing

* errors are more verbose and so have more words to describe them

* Doc regen for new status

* tiny test tinkering

* str := string("Removes .String()").String()

* Merge fixups

* Fixes a data race discovered during github actions

* Allows websocket test to pass consistently

* Fixes merge issue preventing datahistorymanager from starting via config

* Niterinos cmd defaults and explanations

* fixes default oopsie

* Fixes lack of nil protection

* Additional oopsie

* More detailed error for validating job exchange
This commit is contained in:
Scott
2021-07-01 16:21:48 +10:00
committed by GitHub
parent c109cfb6b4
commit 197ef2df21
133 changed files with 17770 additions and 1367 deletions

View File

@@ -1,7 +1,6 @@
package engine
import (
"errors"
"fmt"
"sync/atomic"
@@ -21,8 +20,6 @@ type CommunicationManager struct {
comms *communications.Communications
}
var errNilConfig = errors.New("received nil communications config")
// SetupCommunicationManager creates a communications manager
func SetupCommunicationManager(cfg *base.CommunicationsConfig) (*CommunicationManager, error) {
if cfg == nil {

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Communication_manager
# GoCryptoTrader package Communication manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Communication_manager
## Current Features for Communication manager
+ The communication manager subsystem is used to push events raised in GoCryptoTrader to any enabled communication system such as a Slack server
+ In order to modify the behaviour of the communication manager subsystem, you can edit the following inside your config file under `communications`:

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Connection_manager
# GoCryptoTrader package Connection manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Connection_manager
## Current Features for Connection manager
+ The connection manager subsystem is used to periodically check whether the application is connected to the internet and will provide alerts of any changes
+ In order to modify the behaviour of the connection manager subsystem, you can edit the following inside your config file under `connectionMonitor`:

View File

@@ -16,9 +16,7 @@ import (
// DatabaseConnectionManagerName is an exported subsystem name
const DatabaseConnectionManagerName = "database"
var (
errDatabaseDisabled = errors.New("database support disabled")
)
var errDatabaseDisabled = errors.New("database support disabled")
// DatabaseConnectionManager holds the database connection and its status
type DatabaseConnectionManager struct {
@@ -43,6 +41,15 @@ func (m *DatabaseConnectionManager) IsRunning() bool {
return atomic.LoadInt32(&m.started) == 1
}
// GetInstance returns a limited scoped database instance
func (m *DatabaseConnectionManager) GetInstance() database.IDatabase {
if m == nil || atomic.LoadInt32(&m.started) == 0 {
return nil
}
return m.dbConn
}
// SetupDatabaseConnectionManager creates a new database manager
func SetupDatabaseConnectionManager(cfg *database.Config) (*DatabaseConnectionManager, error) {
if cfg == nil {
@@ -67,6 +74,14 @@ func SetupDatabaseConnectionManager(cfg *database.Config) (*DatabaseConnectionMa
return m, nil
}
// IsConnected is an exported check to verify if the database is connected
func (m *DatabaseConnectionManager) IsConnected() bool {
if m == nil || atomic.LoadInt32(&m.started) == 0 {
return false
}
return m.dbConn.IsConnected()
}
// Start sets up the database connection manager to maintain a SQL connection
func (m *DatabaseConnectionManager) Start(wg *sync.WaitGroup) (err error) {
if m == nil {
@@ -92,14 +107,14 @@ func (m *DatabaseConnectionManager) Start(wg *sync.WaitGroup) (err error) {
m.host,
m.database,
m.driver)
m.dbConn, err = dbpsql.Connect()
m.dbConn, err = dbpsql.Connect(m.dbConn.GetConfig())
case database.DBSQLite,
database.DBSQLite3:
log.Debugf(log.DatabaseMgr,
"Attempting to establish database connection to %s utilising %s driver\n",
m.database,
m.driver)
m.dbConn, err = dbsqlite3.Connect()
m.dbConn, err = dbsqlite3.Connect(m.database)
default:
return database.ErrNoDatabaseProvided
}

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Database_connection
# GoCryptoTrader package Database connection
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Database_connection
## Current Features for Database connection
+ The database connection manager subsystem is used to periodically check whether the application is connected to the database and will provide alerts of any changes
+ In order to modify the behaviour of the database connection manager subsystem, you can edit the following inside your config file under `database`:

View File

@@ -23,7 +23,7 @@ func CreateDatabase(t *testing.T) string {
return tmpDir
}
func Cleanup(t *testing.T, tmpDir string) {
func Cleanup(tmpDir string) {
if database.DB.IsConnected() {
err := database.DB.CloseConnection()
if err != nil {
@@ -53,7 +53,7 @@ func TestSetupDatabaseConnectionManager(t *testing.T) {
func TestStartSQLite(t *testing.T) {
tmpDir := CreateDatabase(t)
defer Cleanup(t, tmpDir)
defer Cleanup(tmpDir)
m, err := SetupDatabaseConnectionManager(&database.Config{})
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
@@ -92,7 +92,7 @@ func TestStartSQLite(t *testing.T) {
// This test does not care for a successful connection
func TestStartPostgres(t *testing.T) {
tmpDir := CreateDatabase(t)
defer Cleanup(t, tmpDir)
defer Cleanup(tmpDir)
m, err := SetupDatabaseConnectionManager(&database.Config{})
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
@@ -116,7 +116,7 @@ func TestStartPostgres(t *testing.T) {
func TestDatabaseConnectionManagerIsRunning(t *testing.T) {
tmpDir := CreateDatabase(t)
defer Cleanup(t, tmpDir)
defer Cleanup(tmpDir)
m, err := SetupDatabaseConnectionManager(&database.Config{
Enabled: true,
Driver: database.DBSQLite,
@@ -147,7 +147,7 @@ func TestDatabaseConnectionManagerIsRunning(t *testing.T) {
func TestDatabaseConnectionManagerStop(t *testing.T) {
tmpDir := CreateDatabase(t)
defer Cleanup(t, tmpDir)
defer Cleanup(tmpDir)
m, err := SetupDatabaseConnectionManager(&database.Config{
Enabled: true,
Driver: database.DBSQLite,
@@ -184,7 +184,7 @@ func TestDatabaseConnectionManagerStop(t *testing.T) {
func TestCheckConnection(t *testing.T) {
tmpDir := CreateDatabase(t)
defer Cleanup(t, tmpDir)
defer Cleanup(tmpDir)
var m *DatabaseConnectionManager
err := m.checkConnection()
if !errors.Is(err, ErrNilSubsystem) {
@@ -235,7 +235,42 @@ func TestCheckConnection(t *testing.T) {
m.dbConn.SetConnected(false)
err = m.checkConnection()
if !errors.Is(err, database.ErrDatabaseNotConnected) {
t.Errorf("error '%v', expected '%v'", err, database.ErrDatabaseNotConnected)
}
}
func TestGetInstance(t *testing.T) {
tmpDir := CreateDatabase(t)
defer Cleanup(tmpDir)
m, err := SetupDatabaseConnectionManager(&database.Config{
Enabled: true,
Driver: database.DBSQLite,
ConnectionDetails: drivers.ConnectionDetails{
Host: "localhost",
Database: "test.db",
},
})
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
db := m.GetInstance()
if db != nil {
t.Error("expected nil")
}
var wg sync.WaitGroup
err = m.Start(&wg)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
db = m.GetInstance()
if db == nil {
t.Error("expected not nil")
}
m = nil
db = m.GetInstance()
if db != nil {
t.Error("expected nil")
}
}

View File

@@ -0,0 +1,930 @@
package engine
import (
"database/sql"
"errors"
"fmt"
"strings"
"sync/atomic"
"time"
"github.com/gofrs/uuid"
"github.com/thrasher-corp/gocryptotrader/common"
"github.com/thrasher-corp/gocryptotrader/config"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/database/repository/candle"
"github.com/thrasher-corp/gocryptotrader/database/repository/datahistoryjob"
"github.com/thrasher-corp/gocryptotrader/database/repository/datahistoryjobresult"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
"github.com/thrasher-corp/gocryptotrader/exchanges/trade"
"github.com/thrasher-corp/gocryptotrader/log"
)
// SetupDataHistoryManager creates a data history manager subsystem
func SetupDataHistoryManager(em iExchangeManager, dcm iDatabaseConnectionManager, cfg *config.DataHistoryManager) (*DataHistoryManager, error) {
if em == nil {
return nil, errNilExchangeManager
}
if dcm == nil {
return nil, errNilDatabaseConnectionManager
}
if cfg == nil {
return nil, errNilConfig
}
if cfg.CheckInterval <= 0 {
cfg.CheckInterval = defaultDataHistoryTicker
}
if cfg.MaxJobsPerCycle == 0 {
cfg.MaxJobsPerCycle = defaultDataHistoryMaxJobsPerCycle
}
db := dcm.GetInstance()
dhj, err := datahistoryjob.Setup(db)
if err != nil {
return nil, err
}
dhjr, err := datahistoryjobresult.Setup(db)
if err != nil {
return nil, err
}
return &DataHistoryManager{
exchangeManager: em,
databaseConnectionInstance: db,
shutdown: make(chan struct{}),
interval: time.NewTicker(cfg.CheckInterval),
jobDB: dhj,
jobResultDB: dhjr,
maxJobsPerCycle: cfg.MaxJobsPerCycle,
verbose: cfg.Verbose,
tradeLoader: trade.HasTradesInRanges,
candleLoader: kline.LoadFromDatabase,
}, nil
}
// Start runs the subsystem
func (m *DataHistoryManager) Start() error {
if m == nil {
return ErrNilSubsystem
}
if !atomic.CompareAndSwapInt32(&m.started, 0, 1) {
return ErrSubSystemAlreadyStarted
}
m.shutdown = make(chan struct{})
m.run()
log.Debugf(log.DataHistory, "Data history manager %v", MsgSubSystemStarted)
return nil
}
// IsRunning checks whether the subsystem is running
func (m *DataHistoryManager) IsRunning() bool {
if m == nil {
return false
}
return atomic.LoadInt32(&m.started) == 1
}
// Stop stops the subsystem
func (m *DataHistoryManager) Stop() error {
if m == nil {
return ErrNilSubsystem
}
if !atomic.CompareAndSwapInt32(&m.started, 1, 0) {
return ErrSubSystemNotStarted
}
close(m.shutdown)
log.Debugf(log.DataHistory, "Data history manager %v", MsgSubSystemShutdown)
return nil
}
// retrieveJobs will connect to the database and look for existing jobs
func (m *DataHistoryManager) retrieveJobs() ([]*DataHistoryJob, error) {
if m == nil {
return nil, ErrNilSubsystem
}
if atomic.LoadInt32(&m.started) == 0 {
return nil, ErrSubSystemNotStarted
}
dbJobs, err := m.jobDB.GetAllIncompleteJobsAndResults()
if err != nil {
return nil, err
}
var response []*DataHistoryJob
for i := range dbJobs {
dbJob, err := m.convertDBModelToJob(&dbJobs[i])
if err != nil {
return nil, err
}
err = m.validateJob(dbJob)
if err != nil {
log.Error(log.DataHistory, err)
continue
}
response = append(response, dbJob)
}
return response, nil
}
// PrepareJobs will validate the config jobs, verify their status with the database
// and return all valid jobs to be processed
// m.jobs will be overridden by this function
func (m *DataHistoryManager) PrepareJobs() ([]*DataHistoryJob, error) {
if m == nil {
return nil, ErrNilSubsystem
}
if atomic.LoadInt32(&m.started) == 0 {
return nil, ErrSubSystemNotStarted
}
m.m.Lock()
defer m.m.Unlock()
jobs, err := m.retrieveJobs()
if err != nil {
defer func() {
err = m.Stop()
if err != nil {
log.Error(log.DataHistory, err)
}
}()
return nil, fmt.Errorf("error retrieving jobs, has everything been setup? Data history manager will shut down. %w", err)
}
err = m.compareJobsToData(jobs...)
if err != nil {
return nil, err
}
return jobs, nil
}
func (m *DataHistoryManager) compareJobsToData(jobs ...*DataHistoryJob) error {
if m == nil {
return ErrNilSubsystem
}
if atomic.LoadInt32(&m.started) == 0 {
return ErrSubSystemNotStarted
}
var err error
for i := range jobs {
jobs[i].rangeHolder, err = kline.CalculateCandleDateRanges(jobs[i].StartDate, jobs[i].EndDate, jobs[i].Interval, uint32(jobs[i].RequestSizeLimit))
if err != nil {
return err
}
var candles kline.Item
switch jobs[i].DataType {
case dataHistoryCandleDataType:
candles, err = m.candleLoader(jobs[i].Exchange, jobs[i].Pair, jobs[i].Asset, jobs[i].Interval, jobs[i].StartDate, jobs[i].EndDate)
if err != nil && !errors.Is(err, candle.ErrNoCandleDataFound) {
return fmt.Errorf("%s could not load candle data: %w", jobs[i].Nickname, err)
}
jobs[i].rangeHolder.SetHasDataFromCandles(candles.Candles)
case dataHistoryTradeDataType:
err := m.tradeLoader(jobs[i].Exchange, jobs[i].Asset.String(), jobs[i].Pair.Base.String(), jobs[i].Pair.Quote.String(), jobs[i].rangeHolder)
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("%s could not load trade data: %w", jobs[i].Nickname, err)
}
default:
return fmt.Errorf("%s %w %s", jobs[i].Nickname, errUnknownDataType, jobs[i].DataType)
}
}
return nil
}
func (m *DataHistoryManager) run() {
go func() {
validJobs, err := m.PrepareJobs()
if err != nil {
log.Error(log.DataHistory, err)
}
m.m.Lock()
m.jobs = validJobs
m.m.Unlock()
for {
select {
case <-m.shutdown:
return
case <-m.interval.C:
if m.databaseConnectionInstance.IsConnected() {
go func() {
if err := m.runJobs(); err != nil {
log.Error(log.DataHistory, err)
}
}()
}
}
}
}()
}
func (m *DataHistoryManager) runJobs() error {
if m == nil {
return ErrNilSubsystem
}
if atomic.LoadInt32(&m.started) == 0 {
return ErrSubSystemNotStarted
}
if !atomic.CompareAndSwapInt32(&m.processing, 0, 1) {
return fmt.Errorf("runJobs %w", errAlreadyRunning)
}
defer atomic.StoreInt32(&m.processing, 0)
validJobs, err := m.PrepareJobs()
if err != nil {
return err
}
m.m.Lock()
defer func() {
m.m.Unlock()
}()
m.jobs = validJobs
log.Infof(log.DataHistory, "processing data history jobs")
for i := 0; (i < int(m.maxJobsPerCycle) || m.maxJobsPerCycle == -1) && i < len(m.jobs); i++ {
err := m.runJob(m.jobs[i])
if err != nil {
log.Error(log.DataHistory, err)
}
if m.verbose {
log.Debugf(log.DataHistory, "completed run of data history job %v", m.jobs[i].Nickname)
}
}
log.Infof(log.DataHistory, "completed run of data history jobs")
return nil
}
// runJob processes an active job, retrieves candle or trade data
// for a given date range and saves all results to the database
func (m *DataHistoryManager) runJob(job *DataHistoryJob) error {
if m == nil {
return ErrNilSubsystem
}
if atomic.LoadInt32(&m.started) == 0 {
return ErrSubSystemNotStarted
}
if job.Status != dataHistoryStatusActive {
return nil
}
var intervalsProcessed int64
if job.rangeHolder == nil || len(job.rangeHolder.Ranges) == 0 {
return fmt.Errorf("%s %w invalid start/end range %s-%s",
job.Nickname,
errJobInvalid,
job.StartDate.Format(common.SimpleTimeFormatWithTimezone),
job.EndDate.Format(common.SimpleTimeFormatWithTimezone),
)
}
exch := m.exchangeManager.GetExchangeByName(job.Exchange)
if exch == nil {
return fmt.Errorf("%s %w, cannot process job %s for %s %s",
job.Exchange,
errExchangeNotLoaded,
job.Nickname,
job.Asset,
job.Pair)
}
if m.verbose {
log.Debugf(log.DataHistory, "running data history job %v start: %s end: %s interval: %s datatype: %s",
job.Nickname,
job.StartDate,
job.EndDate,
job.Interval,
job.DataType)
}
ranges:
for i := range job.rangeHolder.Ranges {
isCompleted := true
for j := range job.rangeHolder.Ranges[i].Intervals {
if !job.rangeHolder.Ranges[i].Intervals[j].HasData {
isCompleted = false
break
}
}
if isCompleted ||
intervalsProcessed >= job.RunBatchLimit {
continue
}
var failures int64
hasDataInRange := false
resultLookup := job.Results[job.rangeHolder.Ranges[i].Start.Time]
for x := range resultLookup {
switch resultLookup[x].Status {
case dataHistoryIntervalMissingData:
continue ranges
case dataHistoryStatusFailed:
failures++
case dataHistoryStatusComplete:
// this can occur in the scenario where data is missing
// however no errors were encountered when data is missing
// eg an exchange only returns an empty slice
// or the exchange is simply missing the data and does not have an error
hasDataInRange = true
}
}
if failures >= job.MaxRetryAttempts {
// failure threshold reached, we should not attempt
// to check this interval again
for x := range resultLookup {
resultLookup[x].Status = dataHistoryIntervalMissingData
}
job.Results[job.rangeHolder.Ranges[i].Start.Time] = resultLookup
continue
}
if hasDataInRange {
continue
}
if m.verbose {
log.Debugf(log.DataHistory, "job %s processing range %v-%v", job.Nickname, job.rangeHolder.Ranges[i].Start, job.rangeHolder.Ranges[i].End)
}
intervalsProcessed++
id, err := uuid.NewV4()
if err != nil {
return err
}
result := DataHistoryJobResult{
ID: id,
JobID: job.ID,
IntervalStartDate: job.rangeHolder.Ranges[i].Start.Time,
IntervalEndDate: job.rangeHolder.Ranges[i].End.Time,
Status: dataHistoryStatusComplete,
Date: time.Now(),
}
// processing the job
switch job.DataType {
case dataHistoryCandleDataType:
candles, err := exch.GetHistoricCandlesExtended(job.Pair, job.Asset, job.rangeHolder.Ranges[i].Start.Time, job.rangeHolder.Ranges[i].End.Time, job.Interval)
if err != nil {
result.Result += "could not get candles: " + err.Error() + ". "
result.Status = dataHistoryStatusFailed
break
}
job.rangeHolder.SetHasDataFromCandles(candles.Candles)
for j := range job.rangeHolder.Ranges[i].Intervals {
if !job.rangeHolder.Ranges[i].Intervals[j].HasData {
result.Status = dataHistoryStatusFailed
result.Result += fmt.Sprintf("missing data from %v - %v. ",
job.rangeHolder.Ranges[i].Intervals[j].Start.Time.Format(common.SimpleTimeFormatWithTimezone),
job.rangeHolder.Ranges[i].Intervals[j].End.Time.Format(common.SimpleTimeFormatWithTimezone))
}
}
_, err = kline.StoreInDatabase(&candles, true)
if err != nil {
result.Result += "could not save results: " + err.Error() + ". "
result.Status = dataHistoryStatusFailed
}
case dataHistoryTradeDataType:
trades, err := exch.GetHistoricTrades(job.Pair, job.Asset, job.rangeHolder.Ranges[i].Start.Time, job.rangeHolder.Ranges[i].End.Time)
if err != nil {
result.Result += "could not get trades: " + err.Error() + ". "
result.Status = dataHistoryStatusFailed
break
}
candles, err := trade.ConvertTradesToCandles(job.Interval, trades...)
if err != nil {
result.Result += "could not convert candles to trades: " + err.Error() + ". "
result.Status = dataHistoryStatusFailed
break
}
job.rangeHolder.SetHasDataFromCandles(candles.Candles)
for j := range job.rangeHolder.Ranges[i].Intervals {
if !job.rangeHolder.Ranges[i].Intervals[j].HasData {
result.Status = dataHistoryStatusFailed
result.Result += fmt.Sprintf("missing data from %v - %v. ",
job.rangeHolder.Ranges[i].Intervals[j].Start.Time.Format(common.SimpleTimeFormatWithTimezone),
job.rangeHolder.Ranges[i].Intervals[j].End.Time.Format(common.SimpleTimeFormatWithTimezone))
}
}
err = trade.SaveTradesToDatabase(trades...)
if err != nil {
result.Result += "could not save results: " + err.Error() + ". "
result.Status = dataHistoryStatusFailed
}
default:
return errUnknownDataType
}
lookup := job.Results[result.IntervalStartDate]
lookup = append(lookup, result)
job.Results[result.IntervalStartDate] = lookup
}
completed := true
allResultsSuccessful := true
allResultsFailed := true
completionCheck:
for i := range job.rangeHolder.Ranges {
result, ok := job.Results[job.rangeHolder.Ranges[i].Start.Time]
if !ok {
completed = false
}
results:
for j := range result {
switch result[j].Status {
case dataHistoryIntervalMissingData:
allResultsSuccessful = false
break results
case dataHistoryStatusComplete:
allResultsFailed = false
break results
default:
completed = false
break completionCheck
}
}
}
if completed {
switch {
case allResultsSuccessful:
job.Status = dataHistoryStatusComplete
case allResultsFailed:
job.Status = dataHistoryStatusFailed
default:
job.Status = dataHistoryIntervalMissingData
}
log.Infof(log.DataHistory, "job %s finished! Status: %s", job.Nickname, job.Status)
}
dbJob := m.convertJobToDBModel(job)
err := m.jobDB.Upsert(dbJob)
if err != nil {
return fmt.Errorf("job %s failed to update database: %w", job.Nickname, err)
}
dbJobResults := m.convertJobResultToDBResult(job.Results)
err = m.jobResultDB.Upsert(dbJobResults...)
if err != nil {
return fmt.Errorf("job %s failed to insert job results to database: %w", job.Nickname, err)
}
return nil
}
// UpsertJob allows for GRPC interaction to upsert a job to be processed
func (m *DataHistoryManager) UpsertJob(job *DataHistoryJob, insertOnly bool) error {
if m == nil {
return ErrNilSubsystem
}
if !m.IsRunning() {
return ErrSubSystemNotStarted
}
if job == nil {
return errNilJob
}
if job.Nickname == "" {
return fmt.Errorf("upsert job %w", errNicknameUnset)
}
j, err := m.GetByNickname(job.Nickname, false)
if err != nil && !errors.Is(err, errJobNotFound) {
return err
}
if insertOnly && j != nil ||
(j != nil && j.Status != dataHistoryStatusActive) {
return fmt.Errorf("upsert job %w nickname: %s - status: %s ", errNicknameInUse, j.Nickname, j.Status)
}
m.m.Lock()
defer m.m.Unlock()
err = m.validateJob(job)
if err != nil {
return err
}
toUpdate := false
if !insertOnly {
for i := range m.jobs {
if !strings.EqualFold(m.jobs[i].Nickname, job.Nickname) {
continue
}
toUpdate = true
job.ID = m.jobs[i].ID
if job.Exchange != "" && m.jobs[i].Exchange != job.Exchange {
m.jobs[i].Exchange = job.Exchange
}
if job.Asset != "" && m.jobs[i].Asset != job.Asset {
m.jobs[i].Asset = job.Asset
}
if !job.Pair.IsEmpty() && !m.jobs[i].Pair.Equal(job.Pair) {
m.jobs[i].Pair = job.Pair
}
if !job.StartDate.IsZero() && !m.jobs[i].StartDate.Equal(job.StartDate) {
m.jobs[i].StartDate = job.StartDate
}
if !job.EndDate.IsZero() && !m.jobs[i].EndDate.Equal(job.EndDate) {
m.jobs[i].EndDate = job.EndDate
}
if job.Interval != 0 && m.jobs[i].Interval != job.Interval {
m.jobs[i].Interval = job.Interval
}
if job.RunBatchLimit != 0 && m.jobs[i].RunBatchLimit != job.RunBatchLimit {
m.jobs[i].RunBatchLimit = job.RunBatchLimit
}
if job.RequestSizeLimit != 0 && m.jobs[i].RequestSizeLimit != job.RequestSizeLimit {
m.jobs[i].RequestSizeLimit = job.RequestSizeLimit
}
if job.MaxRetryAttempts != 0 && m.jobs[i].MaxRetryAttempts != job.MaxRetryAttempts {
m.jobs[i].MaxRetryAttempts = job.MaxRetryAttempts
}
m.jobs[i].DataType = job.DataType
m.jobs[i].Status = job.Status
break
}
}
if job.ID == uuid.Nil {
job.ID, err = uuid.NewV4()
if err != nil {
return err
}
}
job.rangeHolder, err = kline.CalculateCandleDateRanges(job.StartDate, job.EndDate, job.Interval, uint32(job.RequestSizeLimit))
if err != nil {
return err
}
if !toUpdate {
m.jobs = append(m.jobs, job)
}
dbJob := m.convertJobToDBModel(job)
return m.jobDB.Upsert(dbJob)
}
func (m *DataHistoryManager) validateJob(job *DataHistoryJob) error {
if job == nil {
return errNilJob
}
if !job.Asset.IsValid() {
return fmt.Errorf("job %s %w %s", job.Nickname, asset.ErrNotSupported, job.Asset)
}
if job.Pair.IsEmpty() {
return fmt.Errorf("job %s %w", job.Nickname, errCurrencyPairUnset)
}
if !job.Status.Valid() {
return fmt.Errorf("job %s %w: %s", job.Nickname, errInvalidDataHistoryStatus, job.Status)
}
if !job.DataType.Valid() {
return fmt.Errorf("job %s %w: %s", job.Nickname, errInvalidDataHistoryDataType, job.DataType)
}
exch := m.exchangeManager.GetExchangeByName(job.Exchange)
if exch == nil {
return fmt.Errorf("job %s cannot process job: %s %w",
job.Nickname,
job.Exchange,
errExchangeNotLoaded)
}
pairs, err := exch.GetEnabledPairs(job.Asset)
if err != nil {
return fmt.Errorf("job %s exchange %s asset %s currency %s %w", job.Nickname, job.Exchange, job.Asset, job.Pair, err)
}
if !pairs.Contains(job.Pair, false) {
return fmt.Errorf("job %s exchange %s asset %s currency %s %w", job.Nickname, job.Exchange, job.Asset, job.Pair, errCurrencyNotEnabled)
}
if job.Results == nil {
job.Results = make(map[time.Time][]DataHistoryJobResult)
}
if job.RunBatchLimit <= 0 {
log.Warnf(log.DataHistory, "job %s has unset batch limit, defaulting to %v", job.Nickname, defaultDataHistoryBatchLimit)
job.RunBatchLimit = defaultDataHistoryBatchLimit
}
if job.MaxRetryAttempts <= 0 {
log.Warnf(log.DataHistory, "job %s has unset max retry limit, defaulting to %v", job.Nickname, defaultDataHistoryRetryAttempts)
job.MaxRetryAttempts = defaultDataHistoryRetryAttempts
}
if job.RequestSizeLimit <= 0 {
job.RequestSizeLimit = defaultDataHistoryRequestSizeLimit
}
if job.DataType == dataHistoryTradeDataType &&
(job.Interval >= kline.FourHour || job.Interval <= kline.TenMin) {
log.Warnf(log.DataHistory, "job %s interval %v outside limits, defaulting to %v", job.Nickname, job.Interval.Word(), defaultDataHistoryTradeInterval)
job.Interval = defaultDataHistoryTradeInterval
}
b := exch.GetBase()
if !b.Features.Enabled.Kline.Intervals[job.Interval.Word()] {
return fmt.Errorf("job %s %s %w %s", job.Nickname, job.Interval.Word(), kline.ErrUnsupportedInterval, job.Exchange)
}
job.StartDate = job.StartDate.Round(job.Interval.Duration())
job.EndDate = job.EndDate.Round(job.Interval.Duration())
if err := common.StartEndTimeCheck(job.StartDate, job.EndDate); err != nil {
return fmt.Errorf("job %s %w start: %v end %v", job.Nickname, err, job.StartDate, job.EndDate)
}
return nil
}
// GetByID returns a job's details from its ID
func (m *DataHistoryManager) GetByID(id uuid.UUID) (*DataHistoryJob, error) {
if m == nil {
return nil, ErrNilSubsystem
}
if atomic.LoadInt32(&m.started) == 0 {
return nil, ErrSubSystemNotStarted
}
if id == uuid.Nil {
return nil, errEmptyID
}
m.m.Lock()
for i := range m.jobs {
if m.jobs[i].ID == id {
cpy := *m.jobs[i]
m.m.Unlock()
return &cpy, nil
}
}
m.m.Unlock()
dbJ, err := m.jobDB.GetByID(id.String())
if err != nil {
return nil, fmt.Errorf("%w with id %s %s", errJobNotFound, id, err)
}
result, err := m.convertDBModelToJob(dbJ)
if err != nil {
return nil, fmt.Errorf("could not convert model with id %s %w", id, err)
}
return result, nil
}
// GetByNickname searches for jobs by name and returns it if found
// returns nil if not
// if fullDetails is enabled, it will retrieve all job history results from the database
func (m *DataHistoryManager) GetByNickname(nickname string, fullDetails bool) (*DataHistoryJob, error) {
if m == nil {
return nil, ErrNilSubsystem
}
if atomic.LoadInt32(&m.started) == 0 {
return nil, ErrSubSystemNotStarted
}
if fullDetails {
dbJ, err := m.jobDB.GetJobAndAllResults(nickname)
if err != nil {
return nil, fmt.Errorf("job %s could not load job from database: %w", nickname, err)
}
result, err := m.convertDBModelToJob(dbJ)
if err != nil {
return nil, fmt.Errorf("could not convert model with nickname %s %w", nickname, err)
}
return result, nil
}
m.m.Lock()
for i := range m.jobs {
if strings.EqualFold(m.jobs[i].Nickname, nickname) {
cpy := m.jobs[i]
m.m.Unlock()
return cpy, nil
}
}
m.m.Unlock()
// now try the database
j, err := m.jobDB.GetByNickName(nickname)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
// no need to display normal sql err to user
return nil, errJobNotFound
}
return nil, fmt.Errorf("job %s %w, %s", nickname, errJobNotFound, err)
}
job, err := m.convertDBModelToJob(j)
if err != nil {
return nil, err
}
return job, nil
}
// GetAllJobStatusBetween will return all jobs between two ferns
func (m *DataHistoryManager) GetAllJobStatusBetween(start, end time.Time) ([]*DataHistoryJob, error) {
if m == nil {
return nil, ErrNilSubsystem
}
if atomic.LoadInt32(&m.started) == 0 {
return nil, ErrSubSystemNotStarted
}
if err := common.StartEndTimeCheck(start, end); err != nil {
return nil, err
}
dbJobs, err := m.jobDB.GetJobsBetween(start, end)
if err != nil {
return nil, err
}
var results []*DataHistoryJob
for i := range dbJobs {
dbJob, err := m.convertDBModelToJob(&dbJobs[i])
if err != nil {
return nil, err
}
results = append(results, dbJob)
}
return results, nil
}
// DeleteJob helper function to assist in setting a job to deleted
func (m *DataHistoryManager) DeleteJob(nickname, id string) error {
if m == nil {
return ErrNilSubsystem
}
if atomic.LoadInt32(&m.started) == 0 {
return ErrSubSystemNotStarted
}
if nickname == "" && id == "" {
return errNicknameIDUnset
}
if nickname != "" && id != "" {
return errOnlyNicknameOrID
}
var dbJob *datahistoryjob.DataHistoryJob
var err error
m.m.Lock()
defer m.m.Unlock()
for i := range m.jobs {
if strings.EqualFold(m.jobs[i].Nickname, nickname) ||
m.jobs[i].ID.String() == id {
dbJob = m.convertJobToDBModel(m.jobs[i])
m.jobs = append(m.jobs[:i], m.jobs[i+1:]...)
break
}
}
if dbJob == nil {
if nickname != "" {
dbJob, err = m.jobDB.GetByNickName(nickname)
if err != nil {
return err
}
} else {
dbJob, err = m.jobDB.GetByID(id)
if err != nil {
return err
}
}
}
if dbJob.Status != int64(dataHistoryStatusActive) {
status := dataHistoryStatus(dbJob.Status)
return fmt.Errorf("job: %v status: %s error: %w", dbJob.Nickname, status, errCanOnlyDeleteActiveJobs)
}
dbJob.Status = int64(dataHistoryStatusRemoved)
err = m.jobDB.Upsert(dbJob)
if err != nil {
return err
}
log.Infof(log.DataHistory, "deleted job %v", dbJob.Nickname)
return nil
}
// GetActiveJobs returns all jobs with the status `dataHistoryStatusActive`
func (m *DataHistoryManager) GetActiveJobs() ([]DataHistoryJob, error) {
if m == nil {
return nil, ErrNilSubsystem
}
if !m.IsRunning() {
return nil, ErrSubSystemNotStarted
}
m.m.Lock()
defer m.m.Unlock()
var results []DataHistoryJob
for i := range m.jobs {
if m.jobs[i].Status == dataHistoryStatusActive {
results = append(results, *m.jobs[i])
}
}
return results, nil
}
// GenerateJobSummary returns a human readable summary of a job's status
func (m *DataHistoryManager) GenerateJobSummary(nickname string) (*DataHistoryJobSummary, error) {
if m == nil {
return nil, ErrNilSubsystem
}
job, err := m.GetByNickname(nickname, false)
if err != nil {
return nil, fmt.Errorf("job: %v %w", nickname, err)
}
err = m.compareJobsToData(job)
if err != nil {
return nil, err
}
return &DataHistoryJobSummary{
Nickname: job.Nickname,
Exchange: job.Exchange,
Asset: job.Asset,
Pair: job.Pair,
StartDate: job.StartDate,
EndDate: job.EndDate,
Interval: job.Interval,
Status: job.Status,
DataType: job.DataType,
ResultRanges: job.rangeHolder.DataSummary(true),
}, nil
}
// ----------------------------Lovely-converters----------------------------
func (m *DataHistoryManager) convertDBModelToJob(dbModel *datahistoryjob.DataHistoryJob) (*DataHistoryJob, error) {
id, err := uuid.FromString(dbModel.ID)
if err != nil {
return nil, err
}
cp, err := currency.NewPairFromString(fmt.Sprintf("%s-%s", dbModel.Base, dbModel.Quote))
if err != nil {
return nil, fmt.Errorf("job %s could not format pair %s-%s: %w", dbModel.Nickname, dbModel.Base, dbModel.Quote, err)
}
jobResults, err := m.convertDBResultToJobResult(dbModel.Results)
if err != nil {
return nil, fmt.Errorf("job %s could not convert database job: %w", dbModel.Nickname, err)
}
return &DataHistoryJob{
ID: id,
Nickname: dbModel.Nickname,
Exchange: dbModel.ExchangeName,
Asset: asset.Item(dbModel.Asset),
Pair: cp,
StartDate: dbModel.StartDate,
EndDate: dbModel.EndDate,
Interval: kline.Interval(dbModel.Interval),
RunBatchLimit: dbModel.BatchSize,
RequestSizeLimit: dbModel.RequestSizeLimit,
DataType: dataHistoryDataType(dbModel.DataType),
MaxRetryAttempts: dbModel.MaxRetryAttempts,
Status: dataHistoryStatus(dbModel.Status),
CreatedDate: dbModel.CreatedDate,
Results: jobResults,
}, nil
}
func (m *DataHistoryManager) convertDBResultToJobResult(dbModels []*datahistoryjobresult.DataHistoryJobResult) (map[time.Time][]DataHistoryJobResult, error) {
result := make(map[time.Time][]DataHistoryJobResult)
for i := range dbModels {
id, err := uuid.FromString(dbModels[i].ID)
if err != nil {
return nil, err
}
jobID, err := uuid.FromString(dbModels[i].JobID)
if err != nil {
return nil, err
}
lookup := result[dbModels[i].IntervalStartDate]
lookup = append(lookup, DataHistoryJobResult{
ID: id,
JobID: jobID,
IntervalStartDate: dbModels[i].IntervalStartDate,
IntervalEndDate: dbModels[i].IntervalEndDate,
Status: dataHistoryStatus(dbModels[i].Status),
Result: dbModels[i].Result,
Date: dbModels[i].Date,
})
result[dbModels[i].IntervalStartDate] = lookup
}
return result, nil
}
func (m *DataHistoryManager) convertJobResultToDBResult(results map[time.Time][]DataHistoryJobResult) []*datahistoryjobresult.DataHistoryJobResult {
var response []*datahistoryjobresult.DataHistoryJobResult
for _, v := range results {
for i := range v {
response = append(response, &datahistoryjobresult.DataHistoryJobResult{
ID: v[i].ID.String(),
JobID: v[i].JobID.String(),
IntervalStartDate: v[i].IntervalStartDate,
IntervalEndDate: v[i].IntervalEndDate,
Status: int64(v[i].Status),
Result: v[i].Result,
Date: v[i].Date,
})
}
}
return response
}
func (m *DataHistoryManager) convertJobToDBModel(job *DataHistoryJob) *datahistoryjob.DataHistoryJob {
model := &datahistoryjob.DataHistoryJob{
Nickname: job.Nickname,
ExchangeName: job.Exchange,
Asset: job.Asset.String(),
Base: job.Pair.Base.String(),
Quote: job.Pair.Quote.String(),
StartDate: job.StartDate,
EndDate: job.EndDate,
Interval: int64(job.Interval.Duration()),
RequestSizeLimit: job.RequestSizeLimit,
DataType: int64(job.DataType),
MaxRetryAttempts: job.MaxRetryAttempts,
Status: int64(job.Status),
CreatedDate: job.CreatedDate,
BatchSize: job.RunBatchLimit,
Results: m.convertJobResultToDBResult(job.Results),
}
if job.ID != uuid.Nil {
model.ID = job.ID.String()
}
return model
}

View File

@@ -0,0 +1,144 @@
# GoCryptoTrader package Datahistory manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
[![Build Status](https://github.com/thrasher-corp/gocryptotrader/actions/workflows/tests.yml/badge.svg?branch=master)](https://github.com/thrasher-corp/gocryptotrader/actions/workflows/tests.yml)
[![Software License](https://img.shields.io/badge/License-MIT-orange.svg?style=flat-square)](https://github.com/thrasher-corp/gocryptotrader/blob/master/LICENSE)
[![GoDoc](https://godoc.org/github.com/thrasher-corp/gocryptotrader?status.svg)](https://godoc.org/github.com/thrasher-corp/gocryptotrader/engine/datahistory_manager)
[![Coverage Status](http://codecov.io/github/thrasher-corp/gocryptotrader/coverage.svg?branch=master)](http://codecov.io/github/thrasher-corp/gocryptotrader?branch=master)
[![Go Report Card](https://goreportcard.com/badge/github.com/thrasher-corp/gocryptotrader)](https://goreportcard.com/report/github.com/thrasher-corp/gocryptotrader)
This datahistory_manager package is part of the GoCryptoTrader codebase.
## This is still in active development
You can track ideas, planned features and what's in progress on this Trello board: [https://trello.com/b/ZAhMhpOy/gocryptotrader](https://trello.com/b/ZAhMhpOy/gocryptotrader).
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Datahistory manager
+ The data history manager is an engine subsystem responsible for ensuring that the candle/trade history in the range you define is synchronised to your database
+ It is a long running synchronisation task designed to not overwhelm resources and ensure that all data requested is accounted for and saved to the database
+ The data history manager is disabled by default and requires a database connection to function
+ It can be enabled either via a runtime param, config modification or via RPC command `enablesubsystem`
+ The data history manager accepts jobs from RPC commands
+ A job is defined in the `Database tables` section below
+ Jobs will be addressed by the data history manager at an interval defined in your config, this is detailed below in the `Application run time parameters` table below
+ Jobs will fetch data at sizes you request (which can cater to hardware limitations such as low RAM)
+ Jobs are completed once all data has been fetched/attempted to be fetched in the time range
## What are the prerequisites?
+ Ensure you have a database setup, you can read about that [here](/database)
+ Ensure you have run dbmigrate under `/cmd/dbmigrate` via `dbmigrate -command=up`, you can read about that [here](/database#create-and-run-migrations)
+ Ensure you have seeded exchanges to the database via the application dbseed under `/cmd/dbseed`, you can read about it [here](/cmd/dbseed)
+ Ensure you have the database setup and enabled in your config, this can also be seen [here](/database)
+ Data retrieval can only be made on exchanges that support it, see the readmes for [candles](/docs/OHLCV.md) and [trades](/exchanges/trade#exchange-support-table)
+ Read below on how to enable the data history manager and add data history jobs
## What is a data history job?
A job is a set of parameters which will allow GoCryptoTrader to periodically retrieve historical data. Its purpose is to break up the process of retrieving large sets of data for multiple currencies and exchanges into more manageable chunks in a "set and forget" style.
For a breakdown of what a job consists of and what each parameter does, please review the database tables and the cycle details below.
## What happens during a data history cycle?
+ Once the checkInterval ticker timer has finished, the data history manager will process all jobs considered `active`.
+ A job's start and end time is broken down into intervals defined by the `interval` variable of a job. For a job beginning `2020-01-01` to `2020-01-02` with an interval of one hour will create 24 chunks to retrieve
+ The number of intervals it will then request from an API is defined by the `RequestSizeLimit`. A `RequestSizeLimit` of 2 will mean when processing a job, the data history manager will fetch 2 hours worth of data
+ When processing a job the `RunBatchLimit` defines how many `RequestSizeLimits` it will fetch. A `RunBatchLimit` of 3 means when processing a job, the history manager will fetch 3 lots of 2 hour chunks from the API in a run of a job
+ If the data is successfully retrieved, that chunk will be considered `complete` and saved to the database
+ The `MaxRetryAttempts` defines how many times the data history manager will attempt to fetch a chunk of data before flagging it as `failed`.
+ A chunk is only attempted once per processing time.
+ If it fails, the next attempt will be after the `checkInterval` has finished again.
+ The errors for retrieval failures are stored in the database, allowing you to understand why a certain chunk of time is unavailable (eg exchange downtime and missing data)
+ All results are saved to the database, the data history manager will analyse all results and ready jobs for the next round of processing
## How do I add one?
+ First ensure that the data history monitor is enabled, you can do this via the config (see table `dataHistoryManager` under Config parameters below), via run time parameter (see table Application run time parameters below) or via the RPC command `enablesubsystem --subsystemname="data_history_manager"`
+ The simplest way of adding a new data history job is via the GCTCLI under `/cmd/gctcli`.
+ Modify the following example command to your needs: `.\gctcli.exe datahistory upsertjob --nickname=binance-spot-bnb-btc-1h-candles --exchange=binance --asset=spot --pair=BNB-BTC --interval=3600 --start_date="2020-06-02 12:00:00" --end_date="2020-12-02 12:00:00" --request_size_limit=10 --data_type=0 --max_retry_attempts=3 --batch_size=3`
### Candle intervals and trade fetching
+ A candle interval is required for a job, even when fetching trade data. This is to appropriately break down requests into time interval chunks. However, it is restricted to only a small range of times. This is to prevent fetching issues as fetching trades over a period of days or weeks will take a significant amount of time. When setting a job to fetch trades, the allowable range is less than 4 hours and greater than 10 minutes.
### Application run time parameters
| Parameter | Description | Example |
| ------ | ----------- | ------- |
| datahistorymanager | A boolean value which determines if the data history manager is enabled. Defaults to `false` | `-datahistorymanager=true` |
### Config parameters
#### dataHistoryManager
| Config | Description | Example |
| ------ | ----------- | ------- |
| enabled | If enabled will run the data history manager on startup | `true` |
| checkInterval | A golang `time.Duration` interval of when to attempt to fetch all active jobs' data | `15000000000` |
| maxJobsPerCycle | Allows you to control how many jobs are processed after the `checkInterval` timer finishes. Useful if you have many jobs, but don't wish to constantly be retrieving data | `5` |
| verbose | Displays some extra logs to your logging output to help debug | `false` |
### RPC commands
The below table is a summary of commands. For more details, view the commands in `/cmd/gctcli` or `/gctrpc/rpc.swagger.json`
| Command | Description |
| ------ | ----------- |
| UpsertDataHistoryJob | Updates or Inserts a job to the manager and database |
| GetDataHistoryJobDetails | Returns a job's details via its nickname or ID. Can optionally return an array of all run results |
| GetActiveDataHistoryJobs | Will return all jobs that have an `active` status |
| DeleteJob | Will remove a job for processing. Data is preserved in the database for later reference |
| GetDataHistoryJobsBetween | Returns all jobs, of all status types between the dates provided |
| GetDataHistoryJobSummary | Will return an executive summary of the progress of your job by nickname |
### Database tables
#### datahistoryjob
| Field | Description | Example |
| ------ | ----------- | ------- |
| id | Unique ID of the job. Generated at creation | `deadbeef-dead-beef-dead-beef13371337` |
| nickname | A custom name for the job that is unique for lookups | `binance-xrp-doge-2017` |
| exchange_name_id | The exchange id to fetch data from. The ID should be generated via `/cmd/dbmigrate`. When creating a job, you only need to provide the exchange name | `binance` |
| asset | The asset type of the data to be fetching | `spot` |
| base | The currency pair base of the data to be fetching | `xrp` |
| quote | The currency pair quote of the data to be fetching | `doge` |
| start_time | When to begin fetching data | `01-01-2017T13:33:37Z` |
| end_time | When to finish fetching data | `01-01-2018T13:33:37Z` |
| interval | A golang `time.Duration` representation of the candle interval to use. | `30000000000` |
| data_type | The data type to fetch. `0` is candles and `1` is trades | `0` |
| request_size | The number of candles to fetch. eg if `500`, the data history manager will break up the request into the appropriate timeframe to ensure the data history run interval will fetch 500 candles to save to the database | `500` |
| max_retries | For an interval period, the amount of attempts the data history manager is allowed to attempt to fetch data before moving onto the next period. This can be useful for determining whether the exchange is missing the data in that time period or, if just one failure of three, just means that the data history manager couldn't finish one request | `3` |
| batch_count | The number of requests to make when processing a job | `3` |
| status | A numerical representation for the status. `0` is active, `1` is failed `2` is complete, `3` is removed and `4` is missing data | `0` |
| created | The date the job was created. | `2020-01-01T13:33:37Z` |
#### datahistoryjobresult
| Field | Description | Example |
| ------ | ----------- | ------- |
| id | Unique ID of the job status | `deadbeef-dead-beef-dead-beef13371337` |
| job_id | The job ID being referenced | `deadbeef-dead-beef-dead-beef13371337` |
| result | If there is an error, it will be detailed here | `exchange missing candle data for 2020-01-01 13:37Z` |
| status | A numerical representation of the job result status. `1` is failed, `2` is complete and `4` is missing data | `2` |
| interval_start_time | The start date of the period fetched | `2020-01-01T13:33:37Z` |
| interval_end_time | The end date of the period fetched | `2020-01-02T13:33:37Z` |
| run_time | The time the job was ran | `2020-01-03T13:33:37Z` |
### Please click GoDocs chevron above to view current GoDoc information for this package
## Contribution
Please feel free to submit any pull requests or suggest any desired features to be added.
When submitting a PR, please abide by our coding guidelines:
+ Code must adhere to the official Go [formatting](https://golang.org/doc/effective_go.html#formatting) guidelines (i.e. uses [gofmt](https://golang.org/cmd/gofmt/)).
+ Code must be documented adhering to the official Go [commentary](https://golang.org/doc/effective_go.html#commentary) guidelines.
+ Code must adhere to our [coding style](https://github.com/thrasher-corp/gocryptotrader/blob/master/doc/coding_style.md).
+ Pull requests need to be based on and opened against the `master` branch.
## Donations
<img src="https://github.com/thrasher-corp/gocryptotrader/blob/master/web/src/assets/donate.png?raw=true" hspace="70">
If this framework helped you in any way, or you would like to support the developers working on it, please donate Bitcoin to:
***bc1qk0jareu4jytc0cfrhr5wgshsq8282awpavfahc***

View File

@@ -0,0 +1,948 @@
package engine
import (
"database/sql"
"errors"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/gofrs/uuid"
"github.com/thrasher-corp/gocryptotrader/common"
"github.com/thrasher-corp/gocryptotrader/common/convert"
"github.com/thrasher-corp/gocryptotrader/config"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/database"
"github.com/thrasher-corp/gocryptotrader/database/repository/datahistoryjob"
"github.com/thrasher-corp/gocryptotrader/database/repository/datahistoryjobresult"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
)
func TestSetupDataHistoryManager(t *testing.T) {
t.Parallel()
_, err := SetupDataHistoryManager(nil, nil, nil)
if !errors.Is(err, errNilExchangeManager) {
t.Errorf("error '%v', expected '%v'", err, errNilConfig)
}
_, err = SetupDataHistoryManager(SetupExchangeManager(), nil, nil)
if !errors.Is(err, errNilDatabaseConnectionManager) {
t.Errorf("error '%v', expected '%v'", err, errNilDatabaseConnectionManager)
}
_, err = SetupDataHistoryManager(SetupExchangeManager(), &DatabaseConnectionManager{}, nil)
if !errors.Is(err, errNilConfig) {
t.Errorf("error '%v', expected '%v'", err, errNilConfig)
}
_, err = SetupDataHistoryManager(SetupExchangeManager(), &DatabaseConnectionManager{}, &config.DataHistoryManager{})
if !errors.Is(err, database.ErrNilInstance) {
t.Errorf("error '%v', expected '%v'", err, database.ErrNilInstance)
}
dbInst := &database.Instance{}
err = dbInst.SetConfig(&database.Config{Enabled: true})
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
dbInst.SetConnected(true)
dbCM := &DatabaseConnectionManager{
dbConn: dbInst,
started: 1,
}
err = dbInst.SetSQLiteConnection(&sql.DB{})
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
m, err := SetupDataHistoryManager(SetupExchangeManager(), dbCM, &config.DataHistoryManager{})
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if m == nil {
t.Fatal("expected manager")
}
}
func TestDataHistoryManagerIsRunning(t *testing.T) {
t.Parallel()
m := createDHM(t)
m.started = 0
if m.IsRunning() {
t.Error("expected false")
}
m.started = 1
if !m.IsRunning() {
t.Error("expected true")
}
m = nil
if m.IsRunning() {
t.Error("expected false")
}
}
func TestDataHistoryManagerStart(t *testing.T) {
t.Parallel()
m := createDHM(t)
m.started = 0
err := m.Start()
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
err = m.Start()
if !errors.Is(err, ErrSubSystemAlreadyStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemAlreadyStarted)
}
m = nil
err = m.Start()
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestDataHistoryManagerStop(t *testing.T) {
t.Parallel()
m := createDHM(t)
m.shutdown = make(chan struct{})
err := m.Stop()
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
err = m.Stop()
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
err = m.Stop()
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestUpsertJob(t *testing.T) {
t.Parallel()
m := createDHM(t)
err := m.UpsertJob(nil, false)
if !errors.Is(err, errNilJob) {
t.Errorf("error '%v', expected '%v'", err, errNilJob)
}
dhj := &DataHistoryJob{}
err = m.UpsertJob(dhj, false)
if !errors.Is(err, errNicknameUnset) {
t.Errorf("error '%v', expected '%v'", err, errNicknameUnset)
}
dhj.Nickname = "test1337"
err = m.UpsertJob(dhj, false)
if !errors.Is(err, asset.ErrNotSupported) {
t.Errorf("error '%v', expected '%v'", err, asset.ErrNotSupported)
}
dhj.Asset = asset.Spot
err = m.UpsertJob(dhj, false)
if !errors.Is(err, errCurrencyPairUnset) {
t.Errorf("error '%v', expected '%v'", err, errCurrencyPairUnset)
}
dhj.Exchange = strings.ToLower(testExchange)
dhj.Pair = currency.NewPair(currency.BTC, currency.USDT)
err = m.UpsertJob(dhj, false)
if !errors.Is(err, errCurrencyNotEnabled) {
t.Errorf("error '%v', expected '%v'", err, errCurrencyNotEnabled)
}
dhj.Pair = currency.NewPair(currency.BTC, currency.USD)
err = m.UpsertJob(dhj, false)
if !errors.Is(err, kline.ErrUnsupportedInterval) {
t.Errorf("error '%v', expected '%v'", err, kline.ErrUnsupportedInterval)
}
dhj.Interval = kline.OneHour
err = m.UpsertJob(dhj, false)
if !errors.Is(err, common.ErrDateUnset) {
t.Errorf("error '%v', expected '%v'", err, common.ErrDateUnset)
}
dhj.StartDate = time.Now().Add(-time.Hour)
dhj.EndDate = time.Now()
err = m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if len(m.jobs) != 1 {
t.Error("unexpected jerrb")
}
err = m.UpsertJob(dhj, true)
if !errors.Is(err, errNicknameInUse) {
t.Errorf("error '%v', expected '%v'", err, errNicknameInUse)
}
newJob := &DataHistoryJob{
Nickname: dhj.Nickname,
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: startDate,
EndDate: time.Now().Add(-time.Minute),
Interval: kline.FifteenMin,
RunBatchLimit: 1338,
RequestSizeLimit: 1337,
DataType: 2,
MaxRetryAttempts: 1337,
}
err = m.UpsertJob(newJob, false)
if !errors.Is(err, errInvalidDataHistoryDataType) {
t.Errorf("error '%v', expected '%v'", err, errInvalidDataHistoryDataType)
}
newJob.DataType = dataHistoryTradeDataType
err = m.UpsertJob(newJob, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if !m.jobs[0].StartDate.Equal(startDate) {
t.Error(err)
}
}
func TestDeleteJob(t *testing.T) {
t.Parallel()
m := createDHM(t)
dhj := &DataHistoryJob{
Nickname: "TestDeleteJob",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().Add(-time.Minute * 5),
EndDate: time.Now(),
Interval: kline.OneMin,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
err = m.DeleteJob("", "")
if !errors.Is(err, errNicknameIDUnset) {
t.Errorf("error '%v', expected '%v'", err, errNicknameIDUnset)
}
err = m.DeleteJob("1337", "1337")
if !errors.Is(err, errOnlyNicknameOrID) {
t.Errorf("error '%v', expected '%v'", err, errOnlyNicknameOrID)
}
err = m.DeleteJob(dhj.Nickname, "")
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if len(m.jobs) != 0 {
t.Error("expected 0")
}
err = m.DeleteJob("", dhj.ID.String())
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
atomic.StoreInt32(&m.started, 0)
err = m.DeleteJob("", dhj.ID.String())
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
err = m.DeleteJob("", dhj.ID.String())
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestGetByNickname(t *testing.T) {
t.Parallel()
m := createDHM(t)
dhj := &DataHistoryJob{
Nickname: "TestGetByNickname",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().Add(-time.Minute * 5),
EndDate: time.Now(),
Interval: kline.OneMin,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
_, err = m.GetByNickname(dhj.Nickname, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
_, err = m.GetByNickname(dhj.Nickname, true)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
m.jobs = []*DataHistoryJob{}
_, err = m.GetByNickname(dhj.Nickname, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
atomic.StoreInt32(&m.started, 0)
_, err = m.GetByNickname("test123", false)
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
_, err = m.GetByNickname("test123", false)
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestGetByID(t *testing.T) {
t.Parallel()
m := createDHM(t)
dhj := &DataHistoryJob{
Nickname: "TestGetByID",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().Add(-time.Minute * 5),
EndDate: time.Now(),
Interval: kline.OneMin,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
_, err = m.GetByID(dhj.ID)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
_, err = m.GetByID(uuid.UUID{})
if !errors.Is(err, errEmptyID) {
t.Errorf("error '%v', expected '%v'", err, errEmptyID)
}
m.jobs = []*DataHistoryJob{}
_, err = m.GetByID(dhj.ID)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
atomic.StoreInt32(&m.started, 0)
_, err = m.GetByID(dhj.ID)
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
_, err = m.GetByID(dhj.ID)
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestRetrieveJobs(t *testing.T) {
t.Parallel()
m := createDHM(t)
dhj := &DataHistoryJob{
Nickname: "TestRetrieveJobs",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().Add(-time.Minute * 5),
EndDate: time.Now(),
Interval: kline.OneMin,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
jobs, err := m.retrieveJobs()
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if len(jobs) != 1 {
t.Error("expected job")
}
atomic.StoreInt32(&m.started, 0)
_, err = m.retrieveJobs()
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
_, err = m.retrieveJobs()
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestGetActiveJobs(t *testing.T) {
t.Parallel()
m := createDHM(t)
jobs, err := m.GetActiveJobs()
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if len(jobs) != 0 {
t.Error("expected 0 jobs")
}
dhj := &DataHistoryJob{
Nickname: "TestGetActiveJobs",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().Add(-time.Minute * 5),
EndDate: time.Now(),
Interval: kline.OneMin,
}
err = m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
jobs, err = m.GetActiveJobs()
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if len(jobs) != 1 {
t.Error("expected 1 job")
}
dhj.Status = dataHistoryStatusFailed
jobs, err = m.GetActiveJobs()
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if len(jobs) != 0 {
t.Error("expected 0 jobs")
}
atomic.StoreInt32(&m.started, 0)
_, err = m.GetActiveJobs()
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
_, err = m.GetActiveJobs()
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestValidateJob(t *testing.T) {
t.Parallel()
m := createDHM(t)
err := m.validateJob(nil)
if !errors.Is(err, errNilJob) {
t.Errorf("error '%v', expected '%v'", err, errNilJob)
}
dhj := &DataHistoryJob{}
err = m.validateJob(dhj)
if !errors.Is(err, asset.ErrNotSupported) {
t.Errorf("error '%v', expected '%v'", err, asset.ErrNotSupported)
}
dhj.Asset = asset.Spot
err = m.validateJob(dhj)
if !errors.Is(err, errCurrencyPairUnset) {
t.Errorf("error '%v', expected '%v'", err, errCurrencyPairUnset)
}
dhj.Exchange = testExchange
dhj.Pair = currency.NewPair(currency.BTC, currency.USDT)
err = m.validateJob(dhj)
if !errors.Is(err, errCurrencyNotEnabled) {
t.Errorf("error '%v', expected '%v'", err, errCurrencyNotEnabled)
}
dhj.Pair = currency.NewPair(currency.BTC, currency.USD)
err = m.validateJob(dhj)
if !errors.Is(err, kline.ErrUnsupportedInterval) {
t.Errorf("error '%v', expected '%v'", err, kline.ErrUnsupportedInterval)
}
dhj.Interval = kline.OneMin
err = m.validateJob(dhj)
if !errors.Is(err, common.ErrDateUnset) {
t.Errorf("error '%v', expected '%v'", err, common.ErrDateUnset)
}
dhj.StartDate = time.Now().Add(time.Minute)
dhj.EndDate = time.Now().Add(time.Hour)
err = m.validateJob(dhj)
if !errors.Is(err, common.ErrStartAfterTimeNow) {
t.Errorf("error '%v', expected '%v'", err, errInvalidTimes)
}
dhj.StartDate = time.Now().Add(-time.Hour)
dhj.EndDate = time.Now().Add(-time.Minute)
err = m.validateJob(dhj)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
}
func TestGetAllJobStatusBetween(t *testing.T) {
t.Parallel()
m := createDHM(t)
dhj := &DataHistoryJob{
Nickname: "TestGetActiveJobs",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().Add(-time.Minute * 5),
EndDate: time.Now(),
Interval: kline.OneMin,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
jobs, err := m.GetAllJobStatusBetween(time.Now().Add(-time.Minute*5), time.Now().Add(time.Minute))
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if len(jobs) != 1 {
t.Error("expected 1 job")
}
_, err = m.GetAllJobStatusBetween(time.Now().Add(-time.Hour), time.Now().Add(-time.Minute*30))
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
m.started = 0
_, err = m.GetAllJobStatusBetween(time.Now().Add(-time.Hour), time.Now().Add(-time.Minute*30))
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
_, err = m.GetAllJobStatusBetween(time.Now().Add(-time.Hour), time.Now().Add(-time.Minute*30))
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestPrepareJobs(t *testing.T) {
t.Parallel()
m := createDHM(t)
jobs, err := m.PrepareJobs()
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if len(jobs) != 1 {
t.Errorf("expected 1 job, received %v", len(jobs))
}
m.started = 0
_, err = m.PrepareJobs()
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
_, err = m.PrepareJobs()
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestCompareJobsToData(t *testing.T) {
t.Parallel()
m := createDHM(t)
dhj := &DataHistoryJob{
Nickname: "TestGenerateJobSummary",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().Add(-time.Minute * 5),
EndDate: time.Now(),
Interval: kline.OneMin,
}
err := m.compareJobsToData(dhj)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
dhj.DataType = dataHistoryTradeDataType
err = m.compareJobsToData(dhj)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
dhj.DataType = 1337
err = m.compareJobsToData(dhj)
if !errors.Is(err, errUnknownDataType) {
t.Errorf("error '%v', expected '%v'", err, errUnknownDataType)
}
m.started = 0
err = m.compareJobsToData(dhj)
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
err = m.compareJobsToData(dhj)
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestRunJob(t *testing.T) {
t.Parallel()
m := createDHM(t)
dhj := &DataHistoryJob{
Nickname: "TestProcessJobs",
Exchange: "Binance",
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USDT),
StartDate: time.Now().Add(-time.Hour * 2),
EndDate: time.Now(),
Interval: kline.OneHour,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
err = m.runJob(dhj)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
dhj.Pair = currency.NewPair(currency.DOGE, currency.USDT)
err = m.runJob(dhj)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
dhjt := &DataHistoryJob{
Nickname: "TestProcessJobs2",
Exchange: "Binance",
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USDT),
StartDate: time.Now().Add(-time.Hour * 5),
EndDate: time.Now(),
Interval: kline.OneHour,
DataType: dataHistoryTradeDataType,
}
err = m.UpsertJob(dhjt, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
err = m.compareJobsToData(dhjt)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
err = m.runJob(dhjt)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
atomic.StoreInt32(&m.started, 0)
err = m.runJob(dhjt)
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
err = m.runJob(dhjt)
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestGenerateJobSummaryTest(t *testing.T) {
t.Parallel()
m := createDHM(t)
dhj := &DataHistoryJob{
Nickname: "TestGenerateJobSummary",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().Add(-time.Minute * 5),
EndDate: time.Now(),
Interval: kline.OneMin,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
summary, err := m.GenerateJobSummary("TestGenerateJobSummary")
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if len(summary.ResultRanges) == 0 {
t.Error("expected result ranges")
}
atomic.StoreInt32(&m.started, 0)
_, err = m.GenerateJobSummary("TestGenerateJobSummary")
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
_, err = m.GenerateJobSummary("TestGenerateJobSummary")
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestRunJobs(t *testing.T) {
t.Parallel()
m := createDHM(t)
err := m.runJobs()
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
atomic.StoreInt32(&m.started, 0)
err = m.runJobs()
if !errors.Is(err, ErrSubSystemNotStarted) {
t.Errorf("error '%v', expected '%v'", err, ErrSubSystemNotStarted)
}
m = nil
err = m.runJobs()
if !errors.Is(err, ErrNilSubsystem) {
t.Errorf("error '%v', expected '%v'", err, ErrNilSubsystem)
}
}
func TestConverters(t *testing.T) {
t.Parallel()
m := createDHM(t)
id, err := uuid.NewV4()
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
id2, err := uuid.NewV4()
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
dhj := &DataHistoryJob{
ID: id,
Nickname: "TestProcessJobs",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USDT),
StartDate: time.Now().Add(-time.Hour * 24),
EndDate: time.Now(),
Interval: kline.OneHour,
}
dbJob := m.convertJobToDBModel(dhj)
if dhj.ID.String() != dbJob.ID ||
dhj.Nickname != dbJob.Nickname ||
!dhj.StartDate.Equal(dbJob.StartDate) ||
int64(dhj.Interval.Duration()) != dbJob.Interval ||
dhj.Pair.Base.String() != dbJob.Base ||
dhj.Pair.Quote.String() != dbJob.Quote {
t.Error("expected matching job")
}
convertBack, err := m.convertDBModelToJob(dbJob)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if dhj.ID != convertBack.ID ||
dhj.Nickname != convertBack.Nickname ||
!dhj.StartDate.Equal(convertBack.StartDate) ||
dhj.Interval != convertBack.Interval ||
!dhj.Pair.Equal(convertBack.Pair) {
t.Error("expected matching job")
}
jr := DataHistoryJobResult{
ID: id,
JobID: id2,
IntervalStartDate: dhj.StartDate,
IntervalEndDate: dhj.EndDate,
Status: 0,
Result: "test123",
Date: time.Now(),
}
mapperino := make(map[time.Time][]DataHistoryJobResult)
mapperino[dhj.StartDate] = append(mapperino[dhj.StartDate], jr)
result := m.convertJobResultToDBResult(mapperino)
if jr.ID.String() != result[0].ID ||
jr.JobID.String() != result[0].JobID ||
jr.Result != result[0].Result ||
!jr.Date.Equal(result[0].Date) ||
!jr.IntervalStartDate.Equal(result[0].IntervalStartDate) ||
!jr.IntervalEndDate.Equal(result[0].IntervalEndDate) ||
jr.Status != dataHistoryStatus(result[0].Status) {
t.Error("expected matching job")
}
andBackAgain, err := m.convertDBResultToJobResult(result)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
if jr.ID != andBackAgain[dhj.StartDate][0].ID ||
jr.JobID != andBackAgain[dhj.StartDate][0].JobID ||
jr.Result != andBackAgain[dhj.StartDate][0].Result ||
!jr.Date.Equal(andBackAgain[dhj.StartDate][0].Date) ||
!jr.IntervalStartDate.Equal(andBackAgain[dhj.StartDate][0].IntervalStartDate) ||
!jr.IntervalEndDate.Equal(andBackAgain[dhj.StartDate][0].IntervalEndDate) ||
jr.Status != andBackAgain[dhj.StartDate][0].Status {
t.Error("expected matching job")
}
}
// test helper functions
func createDHM(t *testing.T) *DataHistoryManager {
em := SetupExchangeManager()
exch, err := em.NewExchangeByName(testExchange)
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
cp := currency.NewPair(currency.BTC, currency.USD)
exch.SetDefaults()
b := exch.GetBase()
b.CurrencyPairs.Pairs = make(map[asset.Item]*currency.PairStore)
b.CurrencyPairs.Pairs[asset.Spot] = &currency.PairStore{
Available: currency.Pairs{cp},
Enabled: currency.Pairs{cp},
AssetEnabled: convert.BoolPtr(true)}
em.Add(exch)
exch2, err := em.NewExchangeByName("Binance")
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
}
cp2 := currency.NewPair(currency.BTC, currency.USDT)
exch2.SetDefaults()
b = exch2.GetBase()
b.CurrencyPairs.Pairs = make(map[asset.Item]*currency.PairStore)
b.CurrencyPairs.Pairs[asset.Spot] = &currency.PairStore{
Available: currency.Pairs{cp2},
Enabled: currency.Pairs{cp2},
AssetEnabled: convert.BoolPtr(true),
ConfigFormat: &currency.PairFormat{Uppercase: true}}
em.Add(exch2)
m := &DataHistoryManager{
jobDB: dataHistoryJobService{},
jobResultDB: dataHistoryJobResultService{},
started: 1,
exchangeManager: em,
tradeLoader: dataHistoryTradeLoader,
candleLoader: dataHistoryCandleLoader,
interval: time.NewTicker(time.Minute),
}
return m
}
// these structs and function implementations are used
// to override database implementations as we are not testing those
// results here. see tests in the database folder
type dataHistoryJobService struct {
datahistoryjob.IDBService
}
type dataHistoryJobResultService struct {
datahistoryjobresult.IDBService
}
var (
jobID = "00a434e2-8502-4d6b-865f-e4243fd8b5a7"
startDate = time.Date(2020, 1, 1, 0, 0, 0, 0, time.Local)
endDate = time.Date(2021, 1, 1, 0, 0, 0, 0, time.Local)
)
func (d dataHistoryJobService) Upsert(_ ...*datahistoryjob.DataHistoryJob) error {
return nil
}
func (d dataHistoryJobService) GetByNickName(nickname string) (*datahistoryjob.DataHistoryJob, error) {
jc := j
jc.Nickname = nickname
return &jc, nil
}
func (d dataHistoryJobService) GetJobsBetween(_, _ time.Time) ([]datahistoryjob.DataHistoryJob, error) {
jc := j
return []datahistoryjob.DataHistoryJob{jc}, nil
}
func (d dataHistoryJobService) GetByID(id string) (*datahistoryjob.DataHistoryJob, error) {
jc := j
jc.ID = id
return &jc, nil
}
func (d dataHistoryJobService) GetAllIncompleteJobsAndResults() ([]datahistoryjob.DataHistoryJob, error) {
jc := j
return []datahistoryjob.DataHistoryJob{jc}, nil
}
func (d dataHistoryJobService) GetJobAndAllResults(nickname string) (*datahistoryjob.DataHistoryJob, error) {
jc := j
jc.Nickname = nickname
return &jc, nil
}
func (d dataHistoryJobResultService) Upsert(_ ...*datahistoryjobresult.DataHistoryJobResult) error {
return nil
}
func (d dataHistoryJobResultService) GetByJobID(_ string) ([]datahistoryjobresult.DataHistoryJobResult, error) {
return nil, nil
}
func (d dataHistoryJobResultService) GetJobResultsBetween(_ string, _, _ time.Time) ([]datahistoryjobresult.DataHistoryJobResult, error) {
return nil, nil
}
var j = datahistoryjob.DataHistoryJob{
ID: jobID,
Nickname: "datahistoryjob",
ExchangeName: testExchange,
Asset: "spot",
Base: "btc",
Quote: "usd",
StartDate: startDate,
EndDate: endDate,
Interval: int64(kline.OneHour.Duration()),
RequestSizeLimit: 3,
MaxRetryAttempts: 3,
BatchSize: 3,
CreatedDate: endDate,
Status: 0,
Results: []*datahistoryjobresult.DataHistoryJobResult{
{
ID: jobID,
JobID: jobID,
},
},
}
func dataHistoryTradeLoader(_, _, _, _ string, irh *kline.IntervalRangeHolder) error {
for i := range irh.Ranges {
for j := range irh.Ranges[i].Intervals {
irh.Ranges[i].Intervals[j].HasData = true
}
}
return nil
}
func dataHistoryCandleLoader(string, currency.Pair, asset.Item, kline.Interval, time.Time, time.Time) (kline.Item, error) {
return kline.Item{}, nil
}

View File

@@ -0,0 +1,163 @@
package engine
import (
"errors"
"sync"
"time"
"github.com/gofrs/uuid"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/database"
"github.com/thrasher-corp/gocryptotrader/database/repository/datahistoryjob"
"github.com/thrasher-corp/gocryptotrader/database/repository/datahistoryjobresult"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
)
const dataHistoryManagerName = "data_history_manager"
type dataHistoryStatus int64
type dataHistoryDataType int64
// Data type descriptors
const (
dataHistoryCandleDataType dataHistoryDataType = iota
dataHistoryTradeDataType
)
// DataHistoryJob status descriptors
const (
dataHistoryStatusActive dataHistoryStatus = iota
dataHistoryStatusFailed
dataHistoryStatusComplete
dataHistoryStatusRemoved
dataHistoryIntervalMissingData
)
// String stringifies iotas to readable
func (d dataHistoryStatus) String() string {
switch {
case int64(d) == 0:
return "active"
case int64(d) == 1:
return "failed"
case int64(d) == 2:
return "complete"
case int64(d) == 3:
return "removed"
case int64(d) == 4:
return "missing data"
}
return ""
}
// Valid ensures the value set is legitimate
func (d dataHistoryStatus) Valid() bool {
return int64(d) >= 0 && int64(d) <= 4
}
// String stringifies iotas to readable
func (d dataHistoryDataType) String() string {
switch {
case int64(d) == 0:
return "candles"
case int64(d) == 1:
return "trades"
}
return ""
}
// Valid ensures the value set is legitimate
func (d dataHistoryDataType) Valid() bool {
return int64(d) == 0 || int64(d) == 1
}
var (
errJobNotFound = errors.New("job not found")
errUnknownDataType = errors.New("job has invalid datatype set and cannot be processed")
errNilJob = errors.New("nil job received")
errNicknameIDUnset = errors.New("must set 'id' OR 'nickname'")
errEmptyID = errors.New("id not set")
errOnlyNicknameOrID = errors.New("can only set 'id' OR 'nickname'")
errNicknameInUse = errors.New("cannot continue as nickname already in use")
errNicknameUnset = errors.New("cannot continue as nickname unset")
errJobInvalid = errors.New("job has not been setup properly and cannot be processed")
errInvalidDataHistoryStatus = errors.New("unsupported data history status received")
errInvalidDataHistoryDataType = errors.New("unsupported data history data type received")
errCanOnlyDeleteActiveJobs = errors.New("can only delete active jobs")
// defaultDataHistoryTradeInterval is the default interval size used to verify whether there is any database data
// for a trade job
defaultDataHistoryTradeInterval = kline.FifteenMin
defaultDataHistoryMaxJobsPerCycle int64 = 5
defaultDataHistoryBatchLimit int64 = 3
defaultDataHistoryRetryAttempts int64 = 3
defaultDataHistoryRequestSizeLimit int64 = 10
defaultDataHistoryTicker = time.Minute
)
// DataHistoryManager is responsible for synchronising,
// retrieving and saving candle and trade data from loaded jobs
type DataHistoryManager struct {
exchangeManager iExchangeManager
databaseConnectionInstance database.IDatabase
started int32
processing int32
shutdown chan struct{}
interval *time.Ticker
jobs []*DataHistoryJob
m sync.Mutex
jobDB datahistoryjob.IDBService
jobResultDB datahistoryjobresult.IDBService
maxJobsPerCycle int64
verbose bool
tradeLoader func(string, string, string, string, *kline.IntervalRangeHolder) error
candleLoader func(string, currency.Pair, asset.Item, kline.Interval, time.Time, time.Time) (kline.Item, error)
}
// DataHistoryJob used to gather candle/trade history and save
// to the database
type DataHistoryJob struct {
ID uuid.UUID
Nickname string
Exchange string
Asset asset.Item
Pair currency.Pair
StartDate time.Time
EndDate time.Time
Interval kline.Interval
RunBatchLimit int64
RequestSizeLimit int64
DataType dataHistoryDataType
MaxRetryAttempts int64
Status dataHistoryStatus
CreatedDate time.Time
Results map[time.Time][]DataHistoryJobResult
rangeHolder *kline.IntervalRangeHolder
}
// DataHistoryJobResult contains details on
// the result of a history request
type DataHistoryJobResult struct {
ID uuid.UUID
JobID uuid.UUID
IntervalStartDate time.Time
IntervalEndDate time.Time
Status dataHistoryStatus
Result string
Date time.Time
}
// DataHistoryJobSummary is a human readable summary of the job
// for quickly understanding the status of a given job
type DataHistoryJobSummary struct {
Nickname string
Exchange string
Asset asset.Item
Pair currency.Pair
StartDate time.Time
EndDate time.Time
Interval kline.Interval
Status dataHistoryStatus
DataType dataHistoryDataType
ResultRanges []string
}

View File

@@ -44,6 +44,7 @@ type Engine struct {
gctScriptManager *gctscript.GctScriptManager
websocketRoutineManager *websocketRoutineManager
WithdrawManager *WithdrawManager
dataHistoryManager *DataHistoryManager
Settings Settings
uptime time.Time
ServicesWG sync.WaitGroup
@@ -139,6 +140,8 @@ func loadConfigWithSettings(settings *Settings, flagSet map[string]bool) (*confi
func validateSettings(b *Engine, s *Settings, flagSet map[string]bool) {
b.Settings = *s
b.Settings.EnableDataHistoryManager = (flagSet["datahistorymanager"] && b.Settings.EnableDatabaseManager) || b.Config.DataHistoryManager.Enabled
b.Settings.EnableGCTScriptManager = b.Settings.EnableGCTScriptManager &&
(flagSet["gctscriptmanager"] || b.Config.GCTScript.Enabled)
@@ -205,7 +208,7 @@ func validateSettings(b *Engine, s *Settings, flagSet map[string]bool) {
if b.Settings.GlobalHTTPTimeout <= 0 {
b.Settings.GlobalHTTPTimeout = b.Config.GlobalHTTPTimeout
}
common.HTTPClient = common.NewHTTPClientWithTimeout(b.Settings.GlobalHTTPTimeout)
common.SetHTTPClientWithTimeout(b.Settings.GlobalHTTPTimeout)
if b.Settings.GlobalHTTPUserAgent != "" {
common.HTTPUserAgent = b.Settings.GlobalHTTPUserAgent
@@ -223,6 +226,7 @@ func PrintSettings(s *Settings) {
gctlog.Debugf(gctlog.Global, "\t Enable all pairs: %v", s.EnableAllPairs)
gctlog.Debugf(gctlog.Global, "\t Enable coinmarketcap analaysis: %v", s.EnableCoinmarketcapAnalysis)
gctlog.Debugf(gctlog.Global, "\t Enable portfolio manager: %v", s.EnablePortfolioManager)
gctlog.Debugf(gctlog.Global, "\t Enable data history manager: %v", s.EnableDataHistoryManager)
gctlog.Debugf(gctlog.Global, "\t Portfolio manager sleep delay: %v\n", s.PortfolioManagerDelay)
gctlog.Debugf(gctlog.Global, "\t Enable gPRC: %v", s.EnableGRPC)
gctlog.Debugf(gctlog.Global, "\t Enable gRPC Proxy: %v", s.EnableGRPCProxy)
@@ -423,6 +427,20 @@ func (bot *Engine) Start() error {
}
}
if bot.Settings.EnableDataHistoryManager {
if bot.dataHistoryManager == nil {
bot.dataHistoryManager, err = SetupDataHistoryManager(bot.ExchangeManager, bot.DatabaseManager, &bot.Config.DataHistoryManager)
if err != nil {
gctlog.Errorf(gctlog.Global, "database history manager unable to setup: %s", err)
} else {
err = bot.dataHistoryManager.Start()
if err != nil {
gctlog.Errorf(gctlog.Global, "database history manager unable to start: %s", err)
}
}
}
}
bot.WithdrawManager, err = SetupWithdrawManager(bot.ExchangeManager, bot.portfolioManager, bot.Settings.EnableDryRun)
if err != nil {
return err
@@ -610,6 +628,12 @@ func (bot *Engine) Stop() {
}
}
if bot.dataHistoryManager.IsRunning() {
if err := bot.dataHistoryManager.Stop(); err != nil {
gctlog.Errorf(gctlog.DataHistory, "data history manager unable to stop. Error: %v", err)
}
}
if bot.DatabaseManager.IsRunning() {
if err := bot.DatabaseManager.Stop(); err != nil {
gctlog.Errorf(gctlog.Global, "Database manager unable to stop. Error: %v", err)

View File

@@ -76,14 +76,31 @@ func TestLoadConfigWithSettings(t *testing.T) {
func TestStartStopDoesNotCausePanic(t *testing.T) {
t.Parallel()
tempDir, err := ioutil.TempDir("", "")
if err != nil {
t.Fatalf("Problem creating temp dir at %s: %s\n", tempDir, err)
}
defer func() {
err = os.RemoveAll(tempDir)
if err != nil {
t.Error(err)
}
}()
botOne, err := NewFromSettings(&Settings{
ConfigFile: config.TestFile,
EnableDryRun: true,
DataDir: tempDir,
}, nil)
if err != nil {
t.Error(err)
}
botOne.Settings.EnableGRPCProxy = false
for i := range botOne.Config.Exchanges {
if botOne.Config.Exchanges[i].Name != testExchange {
// there is no need to load all exchanges for this test
botOne.Config.Exchanges[i].Enabled = false
}
}
if err = botOne.Start(); err != nil {
t.Error(err)
}

View File

@@ -20,6 +20,7 @@ type Settings struct {
EnableAllPairs bool
EnableCoinmarketcapAnalysis bool
EnablePortfolioManager bool
EnableDataHistoryManager bool
PortfolioManagerDelay time.Duration
EnableGRPC bool
EnableGRPCProxy bool

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Event_manager
# GoCryptoTrader package Event manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Event_manager
## Current Features for Event manager
+ The event manager subsystem is used to push events to communication systems such as Slack
+ The only configurable aspects of the event manager are the delays between receiving an event and pushing it and enabling verbose:

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Exchange_manager
# GoCryptoTrader package Exchange manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Exchange_manager
## Current Features for Exchange manager
+ The exchange manager subsystem is used load and store exchanges so that the engine Bot can use them to track orderbooks, submit orders etc etc
+ The exchange manager itself is not customisable, it is always enabled.
+ The exchange manager by default will load all exchanges that are enabled in your config, however, it will also load exchanges by request via GRPC commands

View File

@@ -55,6 +55,7 @@ func (bot *Engine) GetSubsystemsStatus() map[string]bool {
systems[DeprecatedName] = bot.Settings.EnableDeprecatedRPC
systems[WebsocketName] = bot.Settings.EnableWebsocketRPC
systems[dispatch.Name] = dispatch.IsRunning()
systems[dataHistoryManagerName] = bot.dataHistoryManager.IsRunning()
return systems
}
@@ -226,7 +227,17 @@ func (bot *Engine) SetSubsystem(subSystemName string, enable bool) error {
return bot.apiServer.StopWebsocketServer()
case grpcName, grpcProxyName:
return errors.New("cannot manage GRPC subsystem via GRPC. Please manually change your config")
case dataHistoryManagerName:
if enable {
if bot.dataHistoryManager == nil {
bot.dataHistoryManager, err = SetupDataHistoryManager(bot.ExchangeManager, bot.DatabaseManager, &bot.Config.DataHistoryManager)
if err != nil {
return err
}
}
return bot.dataHistoryManager.Start()
}
return bot.dataHistoryManager.Stop()
case vm.Name:
if enable {
if bot.gctScriptManager == nil {

View File

@@ -121,20 +121,10 @@ func TestGetAuthAPISupportedExchanges(t *testing.T) {
}
exch := e.ExchangeManager.GetExchangeByName(testExchange)
cfg, err := exch.GetDefaultConfig()
if err != nil {
t.Error(err)
}
cfg.Enabled = true
cfg.API.AuthenticatedSupport = true
cfg.API.AuthenticatedWebsocketSupport = true
cfg.API.Credentials.Key = "test"
cfg.API.Credentials.Secret = "test"
cfg.WebsocketTrafficTimeout = time.Minute
err = exch.Setup(cfg)
if err != nil {
t.Error(err)
}
b := exch.GetBase()
b.API.AuthenticatedWebsocketSupport = true
b.API.Credentials.Key = "test"
b.API.Credentials.Secret = "test"
if result := e.GetAuthAPISupportedExchanges(); len(result) != 1 {
t.Fatal("Unexpected result", result)
}

View File

@@ -123,7 +123,7 @@ func (m *ntpManager) FetchNTPTime() (time.Time, error) {
if atomic.LoadInt32(&m.started) == 0 {
return time.Time{}, fmt.Errorf("NTP manager %w", ErrSubSystemNotStarted)
}
return checkTimeInPools(m.pools), nil
return m.checkTimeInPools(), nil
}
// processTime determines the difference between system time and NTP time
@@ -154,11 +154,11 @@ func (m *ntpManager) processTime() error {
// checkTimeInPools returns local based on ntp servers provided timestamp
// if no server can be reached will return local time in UTC()
func checkTimeInPools(pool []string) time.Time {
for i := range pool {
con, err := net.DialTimeout("udp", pool[i], 5*time.Second)
func (m *ntpManager) checkTimeInPools() time.Time {
for i := range m.pools {
con, err := net.DialTimeout("udp", m.pools[i], 5*time.Second)
if err != nil {
log.Warnf(log.TimeMgr, "Unable to connect to hosts %v attempting next", pool[i])
log.Warnf(log.TimeMgr, "Unable to connect to hosts %v attempting next", m.pools[i])
continue
}

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Ntp_manager
# GoCryptoTrader package Ntp manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Ntp_manager
## Current Features for Ntp manager
+ The NTP manager subsystem is used highlight discrepancies between your system time and specified NTP server times
+ It is useful for debugging and understanding why a request to an exchange may be rejected
+ The NTP manager cannot update your system clock, so when it does alert you of issues, you must take it upon yourself to change your system time in the event your requests are being rejected for being too far out of sync

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Order_manager
# GoCryptoTrader package Order manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Order_manager
## Current Features for Order manager
+ The order manager subsystem stores and monitors all orders from enabled exchanges with API keys and `authenticatedSupport` enabled
+ It can be enabled or disabled via runtime command `-ordermanager=false` and defaults to true
+ All orders placed via GoCryptoTrader will be added to the order manager store

View File

@@ -30,6 +30,7 @@ type portfolioManager struct {
exchangeManager *ExchangeManager
shutdown chan struct{}
base *portfolio.Base
m sync.Mutex
}
// setupPortfolioManager creates a new portfolio manager
@@ -122,6 +123,8 @@ func (m *portfolioManager) processPortfolio() {
if !atomic.CompareAndSwapInt32(&m.processing, 0, 1) {
return
}
m.m.Lock()
defer m.m.Unlock()
data := m.base.GetPortfolioGroupedCoin()
for key, value := range data {
err := m.base.UpdatePortfolio(value, key)
@@ -270,6 +273,8 @@ func (m *portfolioManager) AddAddress(address, description string, coinType curr
if !m.IsRunning() {
return fmt.Errorf("portfolio manager %w", ErrSubSystemNotStarted)
}
m.m.Lock()
defer m.m.Unlock()
return m.base.AddAddress(address, description, coinType, balance)
}
@@ -281,6 +286,8 @@ func (m *portfolioManager) RemoveAddress(address, description string, coinType c
if !m.IsRunning() {
return fmt.Errorf("portfolio manager %w", ErrSubSystemNotStarted)
}
m.m.Lock()
defer m.m.Unlock()
return m.base.RemoveAddress(address, description, coinType)
}

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Portfolio_manager
# GoCryptoTrader package Portfolio manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Portfolio_manager
## Current Features for Portfolio manager
+ The portfolio manager subsystem is used to synchronise and monitor wallet addresses
+ It can read addresses specified in your config file
+ If you have set API keys for an enabled exchange and enabled `authenticatedSupport`, it will store your exchange addresses

View File

@@ -63,6 +63,7 @@ var (
errCurrencyNotEnabled = errors.New("currency not enabled")
errCurrencyPairInvalid = errors.New("currency provided is not found in the available pairs list")
errNoTrades = errors.New("no trades returned from supplied params")
errNilRequestData = errors.New("nil request data received, cannot continue")
)
// RPCServer struct
@@ -873,8 +874,9 @@ func (s *RPCServer) GetOrders(_ context.Context, r *gctrpc.GetOrdersRequest) (*g
return nil, err
}
}
if !start.IsZero() && !end.IsZero() && start.After(end) {
return nil, errInvalidTimes
err = common.StartEndTimeCheck(start, end)
if err != nil {
return nil, err
}
request := &order.GetOrdersRequest{
@@ -1513,17 +1515,20 @@ func (s *RPCServer) WithdrawalEventsByExchange(_ context.Context, r *gctrpc.With
// WithdrawalEventsByDate returns previous withdrawal request details by exchange
func (s *RPCServer) WithdrawalEventsByDate(_ context.Context, r *gctrpc.WithdrawalEventsByDateRequest) (*gctrpc.WithdrawalEventsByExchangeResponse, error) {
UTCStartTime, err := time.Parse(common.SimpleTimeFormat, r.Start)
start, err := time.Parse(common.SimpleTimeFormat, r.Start)
if err != nil {
return nil, err
return nil, fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
}
var UTCEndTime time.Time
UTCEndTime, err = time.Parse(common.SimpleTimeFormat, r.End)
end, err := time.Parse(common.SimpleTimeFormat, r.End)
if err != nil {
return nil, fmt.Errorf("%w cannot parse end time %v", errInvalidTimes, err)
}
err = common.StartEndTimeCheck(start, end)
if err != nil {
return nil, err
}
var ret []*withdraw.Response
ret, err = s.WithdrawManager.WithdrawEventByDate(r.Exchange, UTCStartTime, UTCEndTime, int(r.Limit))
ret, err = s.WithdrawManager.WithdrawEventByDate(r.Exchange, start, end, int(r.Limit))
if err != nil {
return nil, err
}
@@ -1894,16 +1899,19 @@ func (s *RPCServer) GetExchangeTickerStream(r *gctrpc.GetExchangeTickerStreamReq
// GetAuditEvent returns matching audit events from database
func (s *RPCServer) GetAuditEvent(_ context.Context, r *gctrpc.GetAuditEventRequest) (*gctrpc.GetAuditEventResponse, error) {
UTCStartTime, err := time.Parse(common.SimpleTimeFormat, r.StartDate)
start, err := time.Parse(common.SimpleTimeFormat, r.StartDate)
if err != nil {
return nil, fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
}
end, err := time.Parse(common.SimpleTimeFormat, r.EndDate)
if err != nil {
return nil, fmt.Errorf("%w cannot parse end time %v", errInvalidTimes, err)
}
err = common.StartEndTimeCheck(start, end)
if err != nil {
return nil, err
}
UTCEndTime, err := time.Parse(common.SimpleTimeFormat, r.EndDate)
if err != nil {
return nil, err
}
events, err := audit.GetEvent(UTCStartTime, UTCEndTime, r.OrderBy, int(r.Limit))
events, err := audit.GetEvent(start, end, r.OrderBy, int(r.Limit))
if err != nil {
return nil, err
}
@@ -1939,19 +1947,18 @@ func (s *RPCServer) GetAuditEvent(_ context.Context, r *gctrpc.GetAuditEventRequ
// GetHistoricCandles returns historical candles for a given exchange
func (s *RPCServer) GetHistoricCandles(_ context.Context, r *gctrpc.GetHistoricCandlesRequest) (*gctrpc.GetHistoricCandlesResponse, error) {
UTCStartTime, err := time.Parse(common.SimpleTimeFormat, r.Start)
start, err := time.Parse(common.SimpleTimeFormat, r.Start)
if err != nil {
return nil, fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
}
end, err := time.Parse(common.SimpleTimeFormat, r.End)
if err != nil {
return nil, fmt.Errorf("%w cannot parse end time %v", errInvalidTimes, err)
}
err = common.StartEndTimeCheck(start, end)
if err != nil {
return nil, err
}
var UTCEndTime time.Time
UTCEndTime, err = time.Parse(common.SimpleTimeFormat, r.End)
if err != nil {
return nil, err
}
if UTCStartTime.After(UTCEndTime) || UTCStartTime.Equal(UTCEndTime) {
return nil, errInvalidTimes
}
if r.Pair == nil {
return nil, errCurrencyPairUnset
}
@@ -1988,8 +1995,8 @@ func (s *RPCServer) GetHistoricCandles(_ context.Context, r *gctrpc.GetHistoricC
pair,
a,
interval,
UTCStartTime,
UTCEndTime)
start,
end)
if err != nil {
return nil, err
}
@@ -1997,14 +2004,14 @@ func (s *RPCServer) GetHistoricCandles(_ context.Context, r *gctrpc.GetHistoricC
if r.ExRequest {
klineItem, err = exch.GetHistoricCandlesExtended(pair,
a,
UTCStartTime,
UTCEndTime,
start,
end,
interval)
} else {
klineItem, err = exch.GetHistoricCandles(pair,
a,
UTCStartTime,
UTCEndTime,
start,
end,
interval)
}
}
@@ -2015,7 +2022,7 @@ func (s *RPCServer) GetHistoricCandles(_ context.Context, r *gctrpc.GetHistoricC
if r.FillMissingWithTrades {
var tradeDataKline *kline.Item
tradeDataKline, err = fillMissingCandlesWithStoredTrades(UTCStartTime, UTCEndTime, &klineItem)
tradeDataKline, err = fillMissingCandlesWithStoredTrades(start, end, &klineItem)
if err != nil {
return nil, err
}
@@ -2680,17 +2687,20 @@ func (s *RPCServer) GetSavedTrades(_ context.Context, r *gctrpc.GetSavedTradesRe
return nil, err
}
var UTCStartTime, UTCEndTime time.Time
UTCStartTime, err = time.Parse(common.SimpleTimeFormat, r.Start)
start, err := time.Parse(common.SimpleTimeFormat, r.Start)
if err != nil {
return nil, err
return nil, fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
}
UTCEndTime, err = time.Parse(common.SimpleTimeFormat, r.End)
end, err := time.Parse(common.SimpleTimeFormat, r.End)
if err != nil {
return nil, fmt.Errorf("%w cannot parse end time %v", errInvalidTimes, err)
}
err = common.StartEndTimeCheck(start, end)
if err != nil {
return nil, err
}
var trades []trade.Data
trades, err = trade.GetTradesInRange(r.Exchange, r.AssetType, r.Pair.Base, r.Pair.Quote, UTCStartTime, UTCEndTime)
trades, err = trade.GetTradesInRange(r.Exchange, r.AssetType, r.Pair.Base, r.Pair.Quote, start, end)
if err != nil {
return nil, err
}
@@ -2720,12 +2730,15 @@ func (s *RPCServer) ConvertTradesToCandles(_ context.Context, r *gctrpc.ConvertT
if r.End == "" || r.Start == "" || r.Exchange == "" || r.Pair == nil || r.AssetType == "" || r.Pair.String() == "" || r.TimeInterval == 0 {
return nil, errInvalidArguments
}
UTCStartTime, err := time.Parse(common.SimpleTimeFormat, r.Start)
start, err := time.Parse(common.SimpleTimeFormat, r.Start)
if err != nil {
return nil, err
return nil, fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
}
var UTCEndTime time.Time
UTCEndTime, err = time.Parse(common.SimpleTimeFormat, r.End)
end, err := time.Parse(common.SimpleTimeFormat, r.End)
if err != nil {
return nil, fmt.Errorf("%w cannot parse end time %v", errInvalidTimes, err)
}
err = common.StartEndTimeCheck(start, end)
if err != nil {
return nil, err
}
@@ -2747,7 +2760,7 @@ func (s *RPCServer) ConvertTradesToCandles(_ context.Context, r *gctrpc.ConvertT
}
var trades []trade.Data
trades, err = trade.GetTradesInRange(r.Exchange, r.AssetType, r.Pair.Base, r.Pair.Quote, UTCStartTime, UTCEndTime)
trades, err = trade.GetTradesInRange(r.Exchange, r.AssetType, r.Pair.Base, r.Pair.Quote, start, end)
if err != nil {
return nil, err
}
@@ -2814,12 +2827,15 @@ func (s *RPCServer) FindMissingSavedCandleIntervals(_ context.Context, r *gctrpc
return nil, err
}
var UTCStartTime, UTCEndTime time.Time
UTCStartTime, err = time.Parse(common.SimpleTimeFormat, r.Start)
start, err := time.Parse(common.SimpleTimeFormat, r.Start)
if err != nil {
return nil, err
return nil, fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
}
UTCEndTime, err = time.Parse(common.SimpleTimeFormat, r.End)
end, err := time.Parse(common.SimpleTimeFormat, r.End)
if err != nil {
return nil, fmt.Errorf("%w cannot parse end time %v", errInvalidTimes, err)
}
err = common.StartEndTimeCheck(start, end)
if err != nil {
return nil, err
}
@@ -2828,8 +2844,8 @@ func (s *RPCServer) FindMissingSavedCandleIntervals(_ context.Context, r *gctrpc
p,
a,
kline.Interval(r.Interval),
UTCStartTime,
UTCEndTime,
start,
end,
)
if err != nil {
return nil, err
@@ -2845,7 +2861,7 @@ func (s *RPCServer) FindMissingSavedCandleIntervals(_ context.Context, r *gctrpc
candleTimes = append(candleTimes, klineItem.Candles[i].Time)
}
var ranges []timeperiods.TimeRange
ranges, err = timeperiods.FindTimeRangesContainingData(UTCStartTime, UTCEndTime, klineItem.Interval.Duration(), candleTimes)
ranges, err = timeperiods.FindTimeRangesContainingData(start, end, klineItem.Interval.Duration(), candleTimes)
if err != nil {
return nil, err
}
@@ -2870,8 +2886,8 @@ func (s *RPCServer) FindMissingSavedCandleIntervals(_ context.Context, r *gctrpc
resp.Status = fmt.Sprintf("Found %v candles. Missing %v candles in requested timeframe starting %v ending %v",
foundCount,
len(resp.MissingPeriods),
UTCStartTime.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone),
UTCEndTime.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone))
start.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone),
end.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone))
}
return resp, nil
@@ -2898,22 +2914,24 @@ func (s *RPCServer) FindMissingSavedTradeIntervals(_ context.Context, r *gctrpc.
if err != nil {
return nil, err
}
var UTCStartTime, UTCEndTime time.Time
UTCStartTime, err = time.Parse(common.SimpleTimeFormat, r.Start)
start, err := time.Parse(common.SimpleTimeFormat, r.Start)
if err != nil {
return nil, fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
}
end, err := time.Parse(common.SimpleTimeFormat, r.End)
if err != nil {
return nil, fmt.Errorf("%w cannot parse end time %v", errInvalidTimes, err)
}
err = common.StartEndTimeCheck(start, end)
if err != nil {
return nil, err
}
UTCStartTime = UTCStartTime.Truncate(time.Hour)
UTCEndTime, err = time.Parse(common.SimpleTimeFormat, r.End)
if err != nil {
return nil, err
}
UTCEndTime = UTCEndTime.Truncate(time.Hour)
start = start.Truncate(time.Hour)
end = end.Truncate(time.Hour)
intervalMap := make(map[time.Time]bool)
iterationTime := UTCStartTime
for iterationTime.Before(UTCEndTime) {
iterationTime := start
for iterationTime.Before(end) {
intervalMap[iterationTime] = false
iterationTime = iterationTime.Add(time.Hour)
}
@@ -2924,8 +2942,8 @@ func (s *RPCServer) FindMissingSavedTradeIntervals(_ context.Context, r *gctrpc.
r.AssetType,
r.Pair.Base,
r.Pair.Quote,
UTCStartTime,
UTCEndTime,
start,
end,
)
if err != nil {
return nil, err
@@ -2941,7 +2959,7 @@ func (s *RPCServer) FindMissingSavedTradeIntervals(_ context.Context, r *gctrpc.
tradeTimes = append(tradeTimes, trades[i].Timestamp)
}
var ranges []timeperiods.TimeRange
ranges, err = timeperiods.FindTimeRangesContainingData(UTCStartTime, UTCEndTime, time.Hour, tradeTimes)
ranges, err = timeperiods.FindTimeRangesContainingData(start, end, time.Hour, tradeTimes)
if err != nil {
return nil, err
}
@@ -2966,8 +2984,8 @@ func (s *RPCServer) FindMissingSavedTradeIntervals(_ context.Context, r *gctrpc.
resp.Status = fmt.Sprintf("Found %v periods. Missing %v periods between %v and %v",
foundCount,
len(resp.MissingPeriods),
UTCStartTime.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone),
UTCEndTime.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone))
start.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone),
end.In(time.UTC).Format(common.SimpleTimeFormatWithTimezone))
}
return resp, nil
@@ -3009,13 +3027,15 @@ func (s *RPCServer) GetHistoricTrades(r *gctrpc.GetSavedTradesRequest, stream gc
return err
}
var trades []trade.Data
var UTCStartTime, UTCEndTime time.Time
UTCStartTime, err = time.Parse(common.SimpleTimeFormat, r.Start)
start, err := time.Parse(common.SimpleTimeFormat, r.Start)
if err != nil {
return err
return fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
}
UTCEndTime, err = time.Parse(common.SimpleTimeFormat, r.End)
end, err := time.Parse(common.SimpleTimeFormat, r.End)
if err != nil {
return fmt.Errorf("%w cannot parse end time %v", errInvalidTimes, err)
}
err = common.StartEndTimeCheck(start, end)
if err != nil {
return err
}
@@ -3025,7 +3045,7 @@ func (s *RPCServer) GetHistoricTrades(r *gctrpc.GetSavedTradesRequest, stream gc
Pair: r.Pair,
}
for iterateStartTime := UTCStartTime; iterateStartTime.Before(UTCEndTime); iterateStartTime = iterateStartTime.Add(time.Hour) {
for iterateStartTime := start; iterateStartTime.Before(end); iterateStartTime = iterateStartTime.Add(time.Hour) {
iterateEndTime := iterateStartTime.Add(time.Hour)
trades, err = exch.GetHistoricTrades(cp, a, iterateStartTime, iterateEndTime)
if err != nil {
@@ -3041,7 +3061,7 @@ func (s *RPCServer) GetHistoricTrades(r *gctrpc.GetSavedTradesRequest, stream gc
}
for i := range trades {
tradeTS := trades[i].Timestamp.In(time.UTC)
if tradeTS.After(UTCEndTime) {
if tradeTS.After(end) {
break
}
grpcTrades.Trades = append(grpcTrades.Trades, &gctrpc.SavedTrades{
@@ -3282,3 +3302,270 @@ func parseSingleEvents(ret *withdraw.Response) *gctrpc.WithdrawalEventsByExchang
Event: []*gctrpc.WithdrawalEventResponse{tempEvent},
}
}
// UpsertDataHistoryJob adds or updates a data history job for the data history manager
// It will upsert the entry in the database and allow for the processing of the job
func (s *RPCServer) UpsertDataHistoryJob(_ context.Context, r *gctrpc.UpsertDataHistoryJobRequest) (*gctrpc.UpsertDataHistoryJobResponse, error) {
if r == nil {
return nil, errNilRequestData
}
a, err := asset.New(r.Asset)
if err != nil {
return nil, err
}
p := currency.Pair{
Delimiter: r.Pair.Delimiter,
Base: currency.NewCode(r.Pair.Base),
Quote: currency.NewCode(r.Pair.Quote),
}
e := s.GetExchangeByName(r.Exchange)
err = checkParams(r.Exchange, e, a, p)
if err != nil {
return nil, err
}
start, err := time.Parse(common.SimpleTimeFormat, r.StartDate)
if err != nil {
return nil, fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
}
end, err := time.Parse(common.SimpleTimeFormat, r.EndDate)
if err != nil {
return nil, fmt.Errorf("%w cannot parse end time %v", errInvalidTimes, err)
}
err = common.StartEndTimeCheck(start, end)
if err != nil {
return nil, err
}
job := DataHistoryJob{
Nickname: r.Nickname,
Exchange: r.Exchange,
Asset: a,
Pair: p,
StartDate: start,
EndDate: end,
Interval: kline.Interval(r.Interval),
RunBatchLimit: r.BatchSize,
RequestSizeLimit: r.RequestSizeLimit,
DataType: dataHistoryDataType(r.DataType),
Status: dataHistoryStatusActive,
MaxRetryAttempts: r.MaxRetryAttempts,
}
err = s.dataHistoryManager.UpsertJob(&job, r.InsertOnly)
if err != nil {
return nil, err
}
result, err := s.dataHistoryManager.GetByNickname(r.Nickname, false)
if err != nil {
return nil, fmt.Errorf("%s %w", r.Nickname, err)
}
return &gctrpc.UpsertDataHistoryJobResponse{
JobId: result.ID.String(),
Message: "successfully upserted job: " + result.Nickname,
}, nil
}
// GetDataHistoryJobDetails returns a data history job's details
// can request all data history results with r.FullDetails
func (s *RPCServer) GetDataHistoryJobDetails(_ context.Context, r *gctrpc.GetDataHistoryJobDetailsRequest) (*gctrpc.DataHistoryJob, error) {
if r == nil {
return nil, errNilRequestData
}
if r.Id == "" && r.Nickname == "" {
return nil, errNicknameIDUnset
}
if r.Nickname != "" && r.Id != "" {
return nil, errOnlyNicknameOrID
}
var (
result *DataHistoryJob
err error
jobResults []*gctrpc.DataHistoryJobResult
)
if r.Id != "" {
var id uuid.UUID
id, err = uuid.FromString(r.Id)
if err != nil {
return nil, fmt.Errorf("%s %w", r.Id, err)
}
result, err = s.dataHistoryManager.GetByID(id)
if err != nil {
return nil, fmt.Errorf("%s %w", r.Id, err)
}
} else {
result, err = s.dataHistoryManager.GetByNickname(r.Nickname, r.FullDetails)
if err != nil {
return nil, fmt.Errorf("%s %w", r.Nickname, err)
}
if r.FullDetails {
for _, v := range result.Results {
for i := range v {
jobResults = append(jobResults, &gctrpc.DataHistoryJobResult{
StartDate: v[i].IntervalStartDate.Format(common.SimpleTimeFormat),
EndDate: v[i].IntervalEndDate.Format(common.SimpleTimeFormat),
HasData: v[i].Status == dataHistoryStatusComplete,
Message: v[i].Result,
RunDate: v[i].Date.Format(common.SimpleTimeFormat),
})
}
}
}
}
return &gctrpc.DataHistoryJob{
Id: result.ID.String(),
Nickname: result.Nickname,
Exchange: result.Exchange,
Asset: result.Asset.String(),
Pair: &gctrpc.CurrencyPair{
Delimiter: result.Pair.Delimiter,
Base: result.Pair.Base.String(),
Quote: result.Pair.Quote.String(),
},
StartDate: result.StartDate.Format(common.SimpleTimeFormat),
EndDate: result.EndDate.Format(common.SimpleTimeFormat),
Interval: int64(result.Interval.Duration()),
RequestSizeLimit: result.RequestSizeLimit,
DataType: result.DataType.String(),
MaxRetryAttempts: result.MaxRetryAttempts,
BatchSize: result.RunBatchLimit,
JobResults: jobResults,
Status: result.Status.String(),
}, nil
}
// DeleteDataHistoryJob deletes a data history job from the database
func (s *RPCServer) DeleteDataHistoryJob(_ context.Context, r *gctrpc.GetDataHistoryJobDetailsRequest) (*gctrpc.GenericResponse, error) {
if r == nil {
return nil, errNilRequestData
}
if r.Nickname == "" && r.Id == "" {
return nil, errNicknameIDUnset
}
if r.Nickname != "" && r.Id != "" {
return nil, errOnlyNicknameOrID
}
status := "success"
err := s.dataHistoryManager.DeleteJob(r.Nickname, r.Id)
if err != nil {
log.Error(log.GRPCSys, err)
status = "failed"
}
return &gctrpc.GenericResponse{Status: status}, err
}
// GetActiveDataHistoryJobs returns any active data history job details
func (s *RPCServer) GetActiveDataHistoryJobs(_ context.Context, _ *gctrpc.GetInfoRequest) (*gctrpc.DataHistoryJobs, error) {
jobs, err := s.dataHistoryManager.GetActiveJobs()
if err != nil {
return nil, err
}
var response []*gctrpc.DataHistoryJob
for i := range jobs {
response = append(response, &gctrpc.DataHistoryJob{
Id: jobs[i].ID.String(),
Nickname: jobs[i].Nickname,
Exchange: jobs[i].Exchange,
Asset: jobs[i].Asset.String(),
Pair: &gctrpc.CurrencyPair{
Delimiter: jobs[i].Pair.Delimiter,
Base: jobs[i].Pair.Base.String(),
Quote: jobs[i].Pair.Quote.String(),
},
StartDate: jobs[i].StartDate.Format(common.SimpleTimeFormat),
EndDate: jobs[i].EndDate.Format(common.SimpleTimeFormat),
Interval: int64(jobs[i].Interval.Duration()),
RequestSizeLimit: jobs[i].RequestSizeLimit,
DataType: jobs[i].DataType.String(),
MaxRetryAttempts: jobs[i].MaxRetryAttempts,
BatchSize: jobs[i].RunBatchLimit,
Status: jobs[i].Status.String(),
})
}
return &gctrpc.DataHistoryJobs{Results: response}, nil
}
// GetDataHistoryJobsBetween returns all jobs created between supplied dates
func (s *RPCServer) GetDataHistoryJobsBetween(_ context.Context, r *gctrpc.GetDataHistoryJobsBetweenRequest) (*gctrpc.DataHistoryJobs, error) {
if r == nil {
return nil, errNilRequestData
}
start, err := time.Parse(common.SimpleTimeFormat, r.StartDate)
if err != nil {
return nil, fmt.Errorf("%w cannot parse start time %v", errInvalidTimes, err)
}
end, err := time.Parse(common.SimpleTimeFormat, r.EndDate)
if err != nil {
return nil, fmt.Errorf("%w cannot parse end time %v", errInvalidTimes, err)
}
err = common.StartEndTimeCheck(start.Local(), end)
if err != nil {
return nil, err
}
jobs, err := s.dataHistoryManager.GetAllJobStatusBetween(start, end)
if err != nil {
return nil, err
}
var respJobs []*gctrpc.DataHistoryJob
for i := range jobs {
respJobs = append(respJobs, &gctrpc.DataHistoryJob{
Id: jobs[i].ID.String(),
Nickname: jobs[i].Nickname,
Exchange: jobs[i].Exchange,
Asset: jobs[i].Asset.String(),
Pair: &gctrpc.CurrencyPair{
Delimiter: jobs[i].Pair.Delimiter,
Base: jobs[i].Pair.Base.String(),
Quote: jobs[i].Pair.Quote.String(),
},
StartDate: jobs[i].StartDate.Format(common.SimpleTimeFormat),
EndDate: jobs[i].EndDate.Format(common.SimpleTimeFormat),
Interval: int64(jobs[i].Interval.Duration()),
RequestSizeLimit: jobs[i].RequestSizeLimit,
DataType: jobs[i].DataType.String(),
MaxRetryAttempts: jobs[i].MaxRetryAttempts,
BatchSize: jobs[i].RunBatchLimit,
Status: jobs[i].Status.String(),
})
}
return &gctrpc.DataHistoryJobs{
Results: respJobs,
}, nil
}
// GetDataHistoryJobSummary provides a general look at how a data history job is going with the "resultSummaries" property
func (s *RPCServer) GetDataHistoryJobSummary(_ context.Context, r *gctrpc.GetDataHistoryJobDetailsRequest) (*gctrpc.DataHistoryJob, error) {
if r == nil {
return nil, errNilRequestData
}
if r.Nickname == "" {
return nil, fmt.Errorf("get job summary %w", errNicknameUnset)
}
job, err := s.dataHistoryManager.GenerateJobSummary(r.Nickname)
if err != nil {
return nil, err
}
return &gctrpc.DataHistoryJob{
Nickname: job.Nickname,
Exchange: job.Exchange,
Asset: job.Asset.String(),
Pair: &gctrpc.CurrencyPair{
Delimiter: job.Pair.Delimiter,
Base: job.Pair.Base.String(),
Quote: job.Pair.Quote.String(),
},
StartDate: job.StartDate.Format(common.SimpleTimeFormat),
EndDate: job.EndDate.Format(common.SimpleTimeFormat),
Interval: int64(job.Interval.Duration()),
DataType: job.DataType.String(),
Status: job.Status.String(),
ResultSummaries: job.ResultRanges,
}, nil
}

View File

@@ -9,11 +9,14 @@ import (
"path/filepath"
"reflect"
"runtime"
"strings"
"sync"
"testing"
"time"
"github.com/gofrs/uuid"
"github.com/thrasher-corp/gocryptotrader/common"
"github.com/thrasher-corp/gocryptotrader/common/convert"
"github.com/thrasher-corp/gocryptotrader/config"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/database"
@@ -38,7 +41,7 @@ const (
unexpectedLackOfError = "unexpected lack of error"
migrationsFolder = "migrations"
databaseFolder = "database"
databaseName = "rpctestdb"
databaseName = "rpctestdb.db"
)
// fExchange is a fake exchange with function overrides
@@ -82,6 +85,7 @@ func (f fExchange) UpdateAccountInfo(a asset.Item) (account.Holdings, error) {
// Sets up everything required to run any function inside rpcserver
func RPCTestSetup(t *testing.T) *Engine {
t.Helper()
var err error
dbConf := database.Config{
Enabled: true,
@@ -102,6 +106,10 @@ func RPCTestSetup(t *testing.T) *Engine {
if err != nil {
log.Fatal(err)
}
err = engerino.LoadExchange("Binance", false, nil)
if err != nil {
log.Fatal(err)
}
engerino.Config.Database = dbConf
engerino.DatabaseManager, err = SetupDatabaseConnectionManager(&engerino.Config.Database)
if err != nil {
@@ -116,8 +124,15 @@ func RPCTestSetup(t *testing.T) *Engine {
if err != nil {
t.Fatalf("failed to run migrations %v", err)
}
uuider, _ := uuid.NewV4()
err = dbexchange.Insert(dbexchange.Details{Name: testExchange, UUID: uuider})
uuider, err := uuid.NewV4()
if err != nil {
t.Fatal(err)
}
uuider2, err := uuid.NewV4()
if err != nil {
t.Fatal(err)
}
err = dbexchange.InsertMany([]dbexchange.Details{{Name: testExchange, UUID: uuider}, {Name: "Binance", UUID: uuider2}})
if err != nil {
t.Fatalf("failed to insert exchange %v", err)
}
@@ -126,6 +141,7 @@ func RPCTestSetup(t *testing.T) *Engine {
}
func CleanRPCTest(t *testing.T, engerino *Engine) {
t.Helper()
err := engerino.DatabaseManager.Stop()
if err != nil {
t.Error(err)
@@ -388,8 +404,8 @@ func TestGetHistoricCandles(t *testing.T) {
Start: "2020-01-02 15:04:05",
End: "2020-01-02 15:04:05",
})
if !errors.Is(err, errInvalidTimes) {
t.Errorf("expected %v, received %v", errInvalidTimes, err)
if !errors.Is(err, common.ErrStartEqualsEnd) {
t.Errorf("received %v, expected %v", err, common.ErrStartEqualsEnd)
}
var results *gctrpc.GetHistoricCandlesResponse
// default run
@@ -875,11 +891,6 @@ func TestGetOrders(t *testing.T) {
t.Errorf("expected %v, received %v", errExchangeNotLoaded, err)
}
err = engerino.LoadExchange(exchName, false, nil)
if err != nil {
t.Error(err)
}
_, err = s.GetOrders(context.Background(), &gctrpc.GetOrdersRequest{
Exchange: exchName,
AssetType: asset.Spot.String(),
@@ -900,19 +911,19 @@ func TestGetOrders(t *testing.T) {
Exchange: exchName,
AssetType: asset.Spot.String(),
Pair: p,
StartDate: time.Now().Format(common.SimpleTimeFormat),
EndDate: time.Now().Add(-time.Hour).Format(common.SimpleTimeFormat),
StartDate: time.Now().UTC().Add(time.Second).Format(common.SimpleTimeFormat),
EndDate: time.Now().UTC().Add(-time.Hour).Format(common.SimpleTimeFormat),
})
if !errors.Is(err, errInvalidTimes) {
t.Errorf("expected %v, received %v", errInvalidTimes, err)
if !errors.Is(err, common.ErrStartAfterTimeNow) {
t.Errorf("received %v, expected %v", err, common.ErrStartAfterTimeNow)
}
_, err = s.GetOrders(context.Background(), &gctrpc.GetOrdersRequest{
Exchange: exchName,
AssetType: asset.Spot.String(),
Pair: p,
StartDate: time.Now().Format(common.SimpleTimeFormat),
EndDate: time.Now().Add(time.Hour).Format(common.SimpleTimeFormat),
StartDate: time.Now().UTC().Add(-time.Hour).Format(common.SimpleTimeFormat),
EndDate: time.Now().UTC().Add(time.Hour).Format(common.SimpleTimeFormat),
})
if !errors.Is(err, exchange.ErrAuthenticatedRequestWithoutCredentialsSet) {
t.Errorf("received '%v', expected '%v'", err, exchange.ErrAuthenticatedRequestWithoutCredentialsSet)
@@ -938,10 +949,16 @@ func TestGetOrders(t *testing.T) {
}
func TestGetOrder(t *testing.T) {
exchName := "binance"
exchName := "Binance"
engerino := RPCTestSetup(t)
defer CleanRPCTest(t, engerino)
s := RPCServer{Engine: engerino}
var wg sync.WaitGroup
var err error
engerino.OrderManager, err = SetupOrderManager(engerino.ExchangeManager, engerino.CommunicationsManager, &wg, false)
if !errors.Is(err, nil) {
t.Errorf("expected %v, received %v", errInvalidArguments, nil)
}
p := &gctrpc.CurrencyPair{
Delimiter: "-",
@@ -949,27 +966,21 @@ func TestGetOrder(t *testing.T) {
Quote: "USDT",
}
_, err := s.GetOrder(context.Background(), nil)
_, err = s.GetOrder(context.Background(), nil)
if !errors.Is(err, errInvalidArguments) {
t.Errorf("expected %v, received %v", errInvalidArguments, err)
}
_, err = s.GetOrder(context.Background(), &gctrpc.GetOrderRequest{
Exchange: exchName,
Exchange: "test123",
OrderId: "",
Pair: p,
Asset: "spot",
})
if !errors.Is(err, errExchangeNotLoaded) {
t.Errorf("expected %v, received %v", errExchangeNotLoaded, err)
}
err = engerino.LoadExchange(exchName, false, nil)
if err != nil {
t.Error(err)
}
_, err = s.GetOrder(context.Background(), &gctrpc.GetOrderRequest{
Exchange: exchName,
OrderId: "",
@@ -1178,3 +1189,279 @@ func TestParseEvents(t *testing.T) {
t.Fatal("Expected second entry in slice to return a Request.Type of Crypto")
}
}
func TestRPCServerUpsertDataHistoryJob(t *testing.T) {
t.Parallel()
m := createDHM(t)
em := SetupExchangeManager()
exch, err := em.NewExchangeByName(testExchange)
if err != nil {
t.Fatal(err)
}
exch.SetDefaults()
b := exch.GetBase()
cp := currency.NewPair(currency.BTC, currency.USD)
b.CurrencyPairs.Pairs = make(map[asset.Item]*currency.PairStore)
b.CurrencyPairs.Pairs[asset.Spot] = &currency.PairStore{
Available: currency.Pairs{cp},
Enabled: currency.Pairs{cp},
AssetEnabled: convert.BoolPtr(true)}
em.Add(exch)
s := RPCServer{Engine: &Engine{dataHistoryManager: m, ExchangeManager: em}}
_, err = s.UpsertDataHistoryJob(context.Background(), nil)
if !errors.Is(err, errNilRequestData) {
t.Errorf("received %v, expected %v", err, errNilRequestData)
}
_, err = s.UpsertDataHistoryJob(context.Background(), &gctrpc.UpsertDataHistoryJobRequest{})
if !errors.Is(err, asset.ErrNotSupported) {
t.Errorf("received %v, expected %v", err, asset.ErrNotSupported)
}
job := &gctrpc.UpsertDataHistoryJobRequest{
Nickname: "hellomoto",
Exchange: testExchange,
Asset: asset.Spot.String(),
Pair: &gctrpc.CurrencyPair{
Delimiter: "-",
Base: "BTC",
Quote: "USD",
},
StartDate: time.Now().Add(-time.Hour * 24).Format(common.SimpleTimeFormat),
EndDate: time.Now().Format(common.SimpleTimeFormat),
Interval: int64(kline.OneHour.Duration()),
RequestSizeLimit: 10,
DataType: int64(dataHistoryCandleDataType),
MaxRetryAttempts: 3,
BatchSize: 500,
}
_, err = s.UpsertDataHistoryJob(context.Background(), job)
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
}
func TestGetDataHistoryJobDetails(t *testing.T) {
t.Parallel()
m := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
dhj := &DataHistoryJob{
Nickname: "TestGetDataHistoryJobDetails",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().UTC().Add(-time.Minute * 2),
EndDate: time.Now().UTC(),
Interval: kline.OneMin,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
_, err = s.GetDataHistoryJobDetails(context.Background(), nil)
if !errors.Is(err, errNilRequestData) {
t.Errorf("received %v, expected %v", err, errNilRequestData)
}
_, err = s.GetDataHistoryJobDetails(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{})
if !errors.Is(err, errNicknameIDUnset) {
t.Errorf("received %v, expected %v", err, errNicknameIDUnset)
}
_, err = s.GetDataHistoryJobDetails(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Id: "123", Nickname: "123"})
if !errors.Is(err, errOnlyNicknameOrID) {
t.Errorf("received %v, expected %v", err, errOnlyNicknameOrID)
}
_, err = s.GetDataHistoryJobDetails(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Nickname: "TestGetDataHistoryJobDetails"})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
_, err = s.GetDataHistoryJobDetails(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Id: m.jobs[0].ID.String()})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
resp, err := s.GetDataHistoryJobDetails(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Nickname: "TestGetDataHistoryJobDetails", FullDetails: true})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
if resp == nil {
t.Fatal("expected job")
}
if !strings.EqualFold(resp.Nickname, "TestGetDataHistoryJobDetails") {
t.Errorf("received %v, expected %v", "TestGetDataHistoryJobDetails", resp.Nickname)
}
}
func TestDeleteDataHistoryJob(t *testing.T) {
t.Parallel()
m := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
dhj := &DataHistoryJob{
Nickname: "TestDeleteDataHistoryJob",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().UTC().Add(-time.Minute * 2),
EndDate: time.Now().UTC(),
Interval: kline.OneMin,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Fatalf("received %v, expected %v", err, nil)
}
_, err = s.DeleteDataHistoryJob(context.Background(), nil)
if !errors.Is(err, errNilRequestData) {
t.Errorf("received %v, expected %v", err, errNilRequestData)
}
_, err = s.DeleteDataHistoryJob(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{})
if !errors.Is(err, errNicknameIDUnset) {
t.Errorf("received %v, expected %v", err, errNicknameIDUnset)
}
_, err = s.DeleteDataHistoryJob(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Id: "123", Nickname: "123"})
if !errors.Is(err, errOnlyNicknameOrID) {
t.Errorf("received %v, expected %v", err, errOnlyNicknameOrID)
}
id := m.jobs[0].ID
_, err = s.DeleteDataHistoryJob(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Nickname: "TestDeleteDataHistoryJob"})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
dhj.ID = id
m.jobs = append(m.jobs, dhj)
_, err = s.DeleteDataHistoryJob(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Id: id.String()})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
if len(m.jobs) != 0 {
t.Errorf("received %v, expected %v", len(m.jobs), 0)
}
}
func TestGetActiveDataHistoryJobs(t *testing.T) {
t.Parallel()
m := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
dhj := &DataHistoryJob{
Nickname: "TestGetActiveDataHistoryJobs",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().UTC().Add(-time.Minute * 2),
EndDate: time.Now().UTC(),
Interval: kline.OneMin,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Fatalf("received %v, expected %v", err, nil)
}
r, err := s.GetActiveDataHistoryJobs(context.Background(), nil)
if !errors.Is(err, nil) {
t.Fatalf("received %v, expected %v", err, nil)
}
if len(r.Results) != 1 {
t.Fatalf("received %v, expected %v", len(r.Results), 1)
}
}
func TestGetDataHistoryJobsBetween(t *testing.T) {
t.Parallel()
m := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
dhj := &DataHistoryJob{
Nickname: "GetDataHistoryJobsBetween",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().UTC().Add(-time.Minute * 2),
EndDate: time.Now().UTC(),
Interval: kline.OneMin,
}
_, err := s.GetDataHistoryJobsBetween(context.Background(), nil)
if !errors.Is(err, errNilRequestData) {
t.Fatalf("received %v, expected %v", err, errNilRequestData)
}
_, err = s.GetDataHistoryJobsBetween(context.Background(), &gctrpc.GetDataHistoryJobsBetweenRequest{
StartDate: time.Now().UTC().Add(time.Minute).Format(common.SimpleTimeFormat),
EndDate: time.Now().UTC().Format(common.SimpleTimeFormat),
})
if !errors.Is(err, common.ErrStartAfterTimeNow) {
t.Fatalf("received %v, expected %v", err, common.ErrStartAfterTimeNow)
}
err = m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Fatalf("received %v, expected %v", err, nil)
}
r, err := s.GetDataHistoryJobsBetween(context.Background(), &gctrpc.GetDataHistoryJobsBetweenRequest{
StartDate: time.Now().Add(-time.Minute).UTC().Format(common.SimpleTimeFormat),
EndDate: time.Now().Add(time.Minute).UTC().Format(common.SimpleTimeFormat),
})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
if len(r.Results) != 1 {
t.Errorf("received %v, expected %v", len(r.Results), 1)
}
}
func TestGetDataHistoryJobSummary(t *testing.T) {
t.Parallel()
m := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
dhj := &DataHistoryJob{
Nickname: "TestGetDataHistoryJobSummary",
Exchange: testExchange,
Asset: asset.Spot,
Pair: currency.NewPair(currency.BTC, currency.USD),
StartDate: time.Now().UTC().Add(-time.Minute * 2),
EndDate: time.Now().UTC(),
Interval: kline.OneMin,
}
err := m.UpsertJob(dhj, false)
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
_, err = s.GetDataHistoryJobSummary(context.Background(), nil)
if !errors.Is(err, errNilRequestData) {
t.Errorf("received %v, expected %v", err, errNilRequestData)
}
_, err = s.GetDataHistoryJobSummary(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{})
if !errors.Is(err, errNicknameUnset) {
t.Errorf("received %v, expected %v", err, errNicknameUnset)
}
resp, err := s.GetDataHistoryJobSummary(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Nickname: "TestGetDataHistoryJobSummary"})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
if resp == nil {
t.Fatal("expected job")
}
if !strings.EqualFold(resp.Nickname, "TestGetDataHistoryJobSummary") {
t.Errorf("received %v, expected %v", "TestGetDataHistoryJobSummary", resp.Nickname)
}
if resp.ResultSummaries == nil {
t.Errorf("received %v, expected %v", nil, "result summaries slice")
}
}

View File

@@ -5,6 +5,7 @@ import (
"github.com/thrasher-corp/gocryptotrader/communications/base"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/database"
exchange "github.com/thrasher-corp/gocryptotrader/exchanges"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
"github.com/thrasher-corp/gocryptotrader/exchanges/order"
@@ -31,9 +32,11 @@ var (
// ErrSubSystemNotStarted message to return when subsystem not started
ErrSubSystemNotStarted = errors.New("subsystem not started")
// ErrNilSubsystem is returned when a subsystem hasn't had its Setup() func run
ErrNilSubsystem = errors.New("subsystem not setup")
errNilWaitGroup = errors.New("nil wait group received")
errNilExchangeManager = errors.New("cannot start with nil exchange manager")
ErrNilSubsystem = errors.New("subsystem not setup")
errNilWaitGroup = errors.New("nil wait group received")
errNilExchangeManager = errors.New("cannot start with nil exchange manager")
errNilDatabaseConnectionManager = errors.New("cannot start with nil database connection manager")
errNilConfig = errors.New("received nil config")
)
// iExchangeManager limits exposure of accessible functions to exchange manager
@@ -83,3 +86,8 @@ type iCurrencyPairSyncer interface {
PrintOrderbookSummary(*orderbook.Base, string, error)
Update(string, currency.Pair, asset.Item, int, error) error
}
// iDatabaseConnectionManager defines a limited scoped databaseConnectionManager
type iDatabaseConnectionManager interface {
GetInstance() database.IDatabase
}

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Subsystem_types
# GoCryptoTrader package Subsystem types
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Subsystem_types
## Current Features for Subsystem types
+ Subsystem contains subsystems that are used at run time by an `engine.Engine`, however they can be setup and run individually.
+ Subsystems are designed to be self contained
+ All subsystems have a public `Setup(...) (..., error)` function to return a valid subsystem ready for use

View File

@@ -32,7 +32,8 @@ var (
// DefaultSyncerWorkers limits the number of sync workers
DefaultSyncerWorkers = 15
// DefaultSyncerTimeoutREST the default time to switch from REST to websocket protocols without a response
DefaultSyncerTimeoutREST = time.Second * 15
DefaultSyncerTimeoutREST = time.Second * 15
// DefaultSyncerTimeoutWebsocket the default time to switch from websocket to REST protocols without a response
DefaultSyncerTimeoutWebsocket = time.Minute
errNoSyncItemsEnabled = errors.New("no sync items enabled")
errUnknownSyncItem = errors.New("unknown sync item")

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Sync_manager
# GoCryptoTrader package Sync manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Sync_manager
## Current Features for Sync manager
+ The currency pair syncer subsystem is used to keep all trades, tickers and orderbooks up to date for all enabled exchange asset currency pairs
+ It can sync data via a websocket connection or REST and will switch between them if there has been no updates
+ In order to modify the behaviour of the currency pair syncer subsystem, you can change runtime parameters as detailed below:

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Websocketroutine_manager
# GoCryptoTrader package Websocketroutine manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Websocketroutine_manager
## Current Features for Websocketroutine manager
+ The websocket routine manager subsystem is used process websocket data in a unified manner across enabled exchanges with websocket support
+ It can help process orders to the order manager subsystem when it receives new data
+ Logs output of ticker and orderbook updates

View File

@@ -1,4 +1,4 @@
# GoCryptoTrader package Withdraw_manager
# GoCryptoTrader package Withdraw manager
<img src="/common/gctlogo.png?raw=true" width="350px" height="350px" hspace="70">
@@ -18,7 +18,7 @@ You can track ideas, planned features and what's in progress on this Trello boar
Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader Slack](https://join.slack.com/t/gocryptotrader/shared_invite/enQtNTQ5NDAxMjA2Mjc5LTc5ZDE1ZTNiOGM3ZGMyMmY1NTAxYWZhODE0MWM5N2JlZDk1NDU0YTViYzk4NTk3OTRiMDQzNGQ1YTc4YmRlMTk)
## Current Features for Withdraw_manager
## Current Features for Withdraw manager
+ The withdraw manager subsystem is responsible for the processing of withdrawal requests and submitting them to exchanges
+ The withdraw manager can be interacted with via GRPC commands such as `WithdrawFiatRequest` and `WithdrawCryptoRequest`
+ Supports caching of responses to allow for quick viewing of withdrawal events via GRPC