d0zingcat/gocryptotrader
1
0
Fork 0
mirror of https://github.com/d0zingcat/gocryptotrader.git synced 2026-05-24 07:26:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feature: Candle conversion & Candle validation (#716)

Browse Source 
* Remove old concept. Introduce new job types and candle scaling

* Adds extra processing, commands

* new concept for queued jobs. Jobs can pause. New commands to manage status

* =End of day commit designing tables and implementing prerequisites further.

* Adds postgres data history relations

* Fixes table design for sqlite. Fixes all issues from merge

* Fixes craziness of database design. Adds some functions to get related jobs

* Fixes errors

* Updates some documentation, manages prerequisite jobs a little better, adds rpc funcs

* Fixes database design and adjust repo functions

* Tests database relationship

* Test coverage of new job functions

* Finishes coverage of new functions

* Commands and RPC coverage

* New database modifications for new job types

* Adds db support of new columns. Adds conversion validation. lint

* command blurb changes

* Allows websocket test to pass consistently

* Fixes merge issue preventing datahistorymanager from starting via config

* Minor fixes for different job type processing

* Fixes rangeholder issue, fixes validation, does not address jobs not starting or wrong status

* Fixes database tests, but at what cost. Fixes dhm tests

* Fixes dhj completion issue. Adds prerequisite by nickname

* Fixes validation processing. Adds db tests and validation

* Fixes validation job processing range

* Fixes trade sql. Reduces defaults. Validation processing and errors

* Updates cli job commands. adds validation decimal. fix job validation

* Expands run job handling and tests

* Validation work

* Fixes validation processing

* candle relations. new job type. updating database design

* Adds secondary exchange support. Sets stage for candle override

* Re adds accidentally deleted relationship

* Updates loading and saving candles to have relationship data when relevant

* Now validates and replaces candle data appropriately

* Fixes getting and setting datahistory data. Neatens DHM

* Test coverage

* Updates proto for new db types. New test coverage. Secondary exchange work

* Investigation into never-ending validation jobs. Now that intervals are ruled out, now need to complete the job....

* Fixes issues with validation job completion. Fixes validation volume issue for secondary exchange

* Adds candle warning support to the backtester

* Fixes warnings

* lint and begin docs

* Documentation updates. Final testing changes

* Minor fixes

* docs, prerequisite checks, more testing

* Fixes binance trade test. Rename err

* Documentation fixes. Figure fixes

* documentation update

* Fixes remote PSQL tests

* Fix binance mock test

* Remove unnecessary JSON

* regen proto

* Some minor nit fixes

* Var usage, query sorting, log improving, sql mirroring

* Extra coverage

* Experimental removal of m.jobs and mutex. Fix messaging

* Fixes error

* Lint fixes, command description improvements. More isRunning gates

* description improvements

* Lint

* BUFF regenerate

* Rough concept to fix insertions taking up long periods of time

* New calculation for trade data. Adds batch saving

This also adds an experimental request feature to shut down lingering requests. However, its uncertain whether or not this is having any impact. Initially thought it was the trades that was taking time and not SQL. Will investigate further

* Removes experimental requester. Adds documentation. Fixes typo

* rm unused error

* re-adds more forgotten contributors

* Now with proper commit count
This commit is contained in:
Scott
2021-08-05 10:27:27 +10:00
committed by GitHub
parent 3b1fe81d8b
commit 48434dfd46
67 changed files with 24651 additions and 12894 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1310
engine/datahistory_manager.go
View File

File diff suppressed because it is too large Load Diff

121
engine/datahistory_manager.md
View File

@@ -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 Datahistory manager
## What is the data history 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
@@ -29,7 +29,19 @@ Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader
+ 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?
## Current Features for the data history manager
+ Retrieval and storage of exchange API candle data
+ Retrieval and storage of exchange API trade data
+ Conversion of stored trade data into custom candle data
+ Conversion of stored candle data into custom candle data
+ Validation of stored candle data against exchange API data
+ Optionally can replace data when an issue is found on a customisable threshold
+ Validation of stored candle data against a secondary exchange's API data
+ Pausing and unpause jobs
+ Queue jobs via prerequisite jobs
+ GRPC command support for creating/modifying/checking jobs
## What are the requirements for the data history manager?
+ 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)
@@ -38,9 +50,28 @@ Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader
+ 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.
A job is a set of parameters which will allow GoCryptoTrader to periodically retrieve, convert or validate 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 kind of data jobs are there?
A breakdown of each type is under the Add Jobs command list below
### What are the different job status types?
| Job Status | Description | Representative value |
| ---------- | ----------- | -------------------- |
| active | A job that is ready to processed | 0 |
| failed | The job has failed to retrieve/covert/validate the data you have specified. See the associated data history job results to understand why | 1 |
| complete | The job has successfully retrieved/converted/validated all data you have specified | 2 |
| removed | The job has been deleted. No data is removed, but the job can no longer be processed | 3 |
| missing data | The job is complete, however there is some missing data. See the associated data history job results to understand why | 4 |
| paused | The job has been paused and will not be processed. Either it has a prerequisite job that needs to be completed, or a user must unpause the job | 5 |
### How do I add a job?
+ 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 addjob savecandles --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 --max_retry_attempts=3 --batch_size=3`
## 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
@@ -53,45 +84,75 @@ For a breakdown of what a job consists of and what each parameter does, please r
+ 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
## Job queuing and prerequisite jobs
You can add jobs which will be paused by default by using the `prerequisite` subcommand containing the associated job nickname. The prerequisite job will be checked to ensure it exists and has not yet completed and add the relationship.
+ Once you have set a prerequisite job, when the prerequisite job status is set to `complete`, the data history manager will search for any jobs which are pending its completion and update their status to `active`.
+ If the prerequisite job is deleted or fails, the upcoming job will _not_ be run.
+ Multiple jobs can use the same prerequisite job, but a job cannot have multiple prerequisites
+ Attempting to add a new prerequisite will overwrite the existing prerequisite
+ You can chain many queued jobs together allowing for automated and large scale data retrieval projects. For example:
| Job Type | Job Name | Description | Prerequisite Job |
| -------- | -------- | ----------- | ---------------- |
| savetrades | save-trades | Save jobs between 01-01-2021 and 01-02-2021 | |
| converttrades | convert-trades | Convert trades to 5m candles | save-trades |
| validatecandles | validate-candles | Ensure the converted trades match the exchange's API data | convert-trades |
| convertcandles | convert-candles-10m | Now that we have confidence in conversion, convert candle to 10m | validate-candles |
| convertcandles | convert-candles-25m | Now that we have confidence in conversion, convert candle to 25m | validate-candles |
| convertcandles | convert-candles-1d | Now that we have confidence in conversion, convert candle to 1d | validate-candles |
## 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 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` |
| maxResultInsertions | When saving candle/trade results, loop it in batches of this number. | `10000` |
| verbose | Displays some extra logs to your logging output to help debug | `false` |
### RPC commands
## 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 |
| AddJob | Shows a list of subcommands to add a new job type, detailed in the next table |
| 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 |
| PauseDataHistoryJob | Will set a job's status to paused |
| UnpauseDataHistoryJob | Will se a job's status to `active` |
### Database tables
#### datahistoryjob
### AddJob commands
| Command | Description | DataHistoryJobDataType |
| ------- | ----------- | ---------------------- |
| savecandles | Will fetch candle data from an exchange and save it to the database | 0 |
| savetrades | Will fetch trade data from an exchange and save it to the database | 1 |
| converttrades | Convert trades saved to the database to any candle resolution eg 30min | 2 |
| convertcandles | Convert candles saved to the database to a new resolution eg 1min -> 5min | 3 |
| validatecandles | Will compare database candle data with API candle data - useful for validating converted trades and candles | 4 |
| secondaryvalidatecandles | Will compare database candle data with a different exchange's API candle data | 5 |
## Database tables
The following is a screenshot of the relationship between relevant data history job tables
![image](https://user-images.githubusercontent.com/9261323/125889821-954b730a-01fd-4eb0-839d-3cc623178f20.png)
### datahistoryjob
| Field | Description | Example |
| ------ | ----------- | ------- |
@@ -104,14 +165,20 @@ The below table is a summary of commands. For more details, view the commands in
| 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` |
| data_type | The data type to fetch. See job types in table `AddJob commands` above | `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` |
| status | A numerical representation for the status. See data history job status subsection | `0` |
| created | The date the job was created. | `2020-01-01T13:33:37Z` |
| conversion_interval | When converting data as a job, this determines the resulting interval | `86400000000000` |
| overwrite_data | If data already exists, the setting allows you to overwrite it | `true` |
| secondary_exchange_id | For a `secondaryvalidatecandles` job, the exchange id of the exchange to compare data to. | `ftx` |
| decimal_place_comparison | When validating API candles, this will round the data to the supplied decimal point to check for equality | `3` |
| replace_on_issue | When there is an issue validating candles for a `validatecandles` job, the API data will overwrite the existing candle data | `false` |
### datahistoryjobresult
#### datahistoryjobresult
| Field | Description | Example |
| ------ | ----------- | ------- |
| id | Unique ID of the job status | `deadbeef-dead-beef-dead-beef13371337` |
@@ -122,6 +189,24 @@ The below table is a summary of commands. For more details, view the commands in
| 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` |
### datahistoryjobrelations
| Field | Description | Example |
| ------ | ----------- | ------- |
| prerequisite_job_id | The job that must be completed before `job_id` can be run | `deadbeef-dead-beef-dead-beef13371337` |
| job_id | The job that will be run after `prerequisite_job_id` completes | `deadbeef-dead-beef-dead-beef13371337` |
### candle
The candle table also has relationships to data history jobs. Only the relevant columns are listed below:
| Field | Description | Example |
| ------ | ----------- | ------- |
| source_job_id | The source job id for where the candle data came from. | `deadbeef-dead-beef-dead-beef13371337` |
| validation_job_id | When job id for what job validated the candle data | `deadbeef-dead-beef-dead-beef13371337` |
| validation_issues | If any discrepancies are found, the data will be written to the column | `issues found at 2020-07-08 00:00:00, Open api: 9262.62 db: 9262.69 diff: 3%, replacing database candle data with API data` |
### Please click GoDocs chevron above to view current GoDoc information for this package
## Contribution

1011
engine/datahistory_manager_test.go
View File

File diff suppressed because it is too large Load Diff

114
engine/datahistory_manager_types.go
View File

@@ -2,7 +2,6 @@ package engine
import (
"errors"
"sync"
"time"
"github.com/gofrs/uuid"
@@ -12,6 +11,7 @@ import (
"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"
)
const dataHistoryManagerName = "data_history_manager"
@@ -23,6 +23,10 @@ type dataHistoryDataType int64
const (
dataHistoryCandleDataType dataHistoryDataType = iota
dataHistoryTradeDataType
dataHistoryConvertTradesDataType
dataHistoryConvertCandlesDataType
dataHistoryCandleValidationDataType
dataHistoryCandleValidationSecondarySourceType
)
// DataHistoryJob status descriptors
@@ -31,7 +35,8 @@ const (
dataHistoryStatusFailed
dataHistoryStatusComplete
dataHistoryStatusRemoved
dataHistoryIntervalMissingData
dataHistoryIntervalIssuesFound
dataHistoryStatusPaused
)
// String stringifies iotas to readable
@@ -46,30 +51,41 @@ func (d dataHistoryStatus) String() string {
case int64(d) == 3:
return "removed"
case int64(d) == 4:
return "missing data"
return "issues found"
case int64(d) == 5:
return "paused"
}
return ""
}
// Valid ensures the value set is legitimate
func (d dataHistoryStatus) Valid() bool {
return int64(d) >= 0 && int64(d) <= 4
return int64(d) >= 0 && int64(d) <= 5
}
// String stringifies iotas to readable
func (d dataHistoryDataType) String() string {
switch {
case int64(d) == 0:
n := int64(d)
switch n {
case 0:
return "candles"
case int64(d) == 1:
case 1:
return "trades"
case 2:
return "trade conversion"
case 3:
return "candle conversion"
case 4:
return "conversion validation"
case 5:
return "conversion validation secondary source"
}
return ""
}
// Valid ensures the value set is legitimate
func (d dataHistoryDataType) Valid() bool {
return int64(d) == 0 || int64(d) == 1
return int64(d) >= 0 && int64(d) <= 5
}
var (
@@ -79,20 +95,27 @@ var (
errNicknameIDUnset = errors.New("must set 'id' OR 'nickname'")
errEmptyID = errors.New("id not set")
errOnlyNicknameOrID = errors.New("can only set 'id' OR 'nickname'")
errBadStatus = errors.New("cannot set job status")
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")
errNilResult = errors.New("received nil job result")
errJobMustBeActiveOrPaused = errors.New("job must be active or paused to be set as a prerequisite")
errNilCandles = errors.New("received nil candles")
// 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
defaultMaxResultInsertions int64 = 10000
defaultDataHistoryBatchLimit int64 = 3
defaultDataHistoryRetryAttempts int64 = 3
defaultDataHistoryRequestSizeLimit int64 = 10
defaultDataHistoryRequestSizeLimit int64 = 500
defaultDataHistoryTicker = time.Minute
defaultDataHistoryTradeRequestSize int64 = 10
defaultDecimalPlaceComparison int64 = 3
)
// DataHistoryManager is responsible for synchronising,
@@ -104,35 +127,45 @@ type DataHistoryManager struct {
processing int32
shutdown chan struct{}
interval *time.Ticker
jobs []*DataHistoryJob
m sync.Mutex
jobDB datahistoryjob.IDBService
jobResultDB datahistoryjobresult.IDBService
maxJobsPerCycle int64
maxResultInsertions 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)
tradeLoader func(string, string, string, string, time.Time, time.Time) ([]trade.Data, error)
tradeSaver func(...trade.Data) error
candleSaver func(*kline.Item, bool) (uint64, 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
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
OverwriteExistingData bool
ConversionInterval kline.Interval
DecimalPlaceComparison int64
SecondaryExchangeSource string
IssueTolerancePercentage float64
ReplaceOnIssue bool
// Prerequisites mean this job is paused until the prerequisite job is completed
PrerequisiteJobID uuid.UUID
PrerequisiteJobNickname string
}
// DataHistoryJobResult contains details on
@@ -150,14 +183,17 @@ type DataHistoryJobResult struct {
// 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
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
OverwriteExistingData bool
ConversionInterval kline.Interval
PrerequisiteJobNickname string
}

179
engine/rpcserver.go
View File

@@ -3426,18 +3426,25 @@ func (s *RPCServer) UpsertDataHistoryJob(_ context.Context, r *gctrpc.UpsertData
}
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,
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),
MaxRetryAttempts: r.MaxRetryAttempts,
Status: dataHistoryStatusActive,
OverwriteExistingData: r.OverwriteExistingData,
ConversionInterval: kline.Interval(r.ConversionInterval),
DecimalPlaceComparison: r.DecimalPlaceComparison,
SecondaryExchangeSource: r.SecondaryExchangeName,
IssueTolerancePercentage: r.IssueTolerancePercentage,
ReplaceOnIssue: r.ReplaceOnIssue,
PrerequisiteJobNickname: r.PrerequisiteJobNickname,
}
err = s.dataHistoryManager.UpsertJob(&job, r.InsertOnly)
@@ -3513,39 +3520,25 @@ func (s *RPCServer) GetDataHistoryJobDetails(_ context.Context, r *gctrpc.GetDat
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(),
StartDate: result.StartDate.Format(common.SimpleTimeFormat),
EndDate: result.EndDate.Format(common.SimpleTimeFormat),
Interval: int64(result.Interval.Duration()),
RequestSizeLimit: result.RequestSizeLimit,
MaxRetryAttempts: result.MaxRetryAttempts,
BatchSize: result.RunBatchLimit,
Status: result.Status.String(),
DataType: result.DataType.String(),
ConversionInterval: int64(result.ConversionInterval.Duration()),
OverwriteExistingData: result.OverwriteExistingData,
PrerequisiteJobNickname: result.PrerequisiteJobNickname,
DecimalPlaceComparison: result.DecimalPlaceComparison,
SecondaryExchangeName: result.SecondaryExchangeSource,
IssueTolerancePercentage: result.IssueTolerancePercentage,
ReplaceOnIssue: result.ReplaceOnIssue,
JobResults: jobResults,
}, 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()
@@ -3565,14 +3558,21 @@ func (s *RPCServer) GetActiveDataHistoryJobs(_ context.Context, _ *gctrpc.GetInf
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(),
StartDate: jobs[i].StartDate.Format(common.SimpleTimeFormat),
EndDate: jobs[i].EndDate.Format(common.SimpleTimeFormat),
Interval: int64(jobs[i].Interval.Duration()),
RequestSizeLimit: jobs[i].RequestSizeLimit,
MaxRetryAttempts: jobs[i].MaxRetryAttempts,
BatchSize: jobs[i].RunBatchLimit,
Status: jobs[i].Status.String(),
DataType: jobs[i].DataType.String(),
ConversionInterval: int64(jobs[i].ConversionInterval.Duration()),
OverwriteExistingData: jobs[i].OverwriteExistingData,
PrerequisiteJobNickname: jobs[i].PrerequisiteJobNickname,
DecimalPlaceComparison: jobs[i].DecimalPlaceComparison,
SecondaryExchangeName: jobs[i].SecondaryExchangeSource,
IssueTolerancePercentage: jobs[i].IssueTolerancePercentage,
ReplaceOnIssue: jobs[i].ReplaceOnIssue,
})
}
return &gctrpc.DataHistoryJobs{Results: response}, nil
@@ -3612,14 +3612,21 @@ func (s *RPCServer) GetDataHistoryJobsBetween(_ context.Context, r *gctrpc.GetDa
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(),
StartDate: jobs[i].StartDate.Format(common.SimpleTimeFormat),
EndDate: jobs[i].EndDate.Format(common.SimpleTimeFormat),
Interval: int64(jobs[i].Interval.Duration()),
RequestSizeLimit: jobs[i].RequestSizeLimit,
MaxRetryAttempts: jobs[i].MaxRetryAttempts,
BatchSize: jobs[i].RunBatchLimit,
Status: jobs[i].Status.String(),
DataType: jobs[i].DataType.String(),
ConversionInterval: int64(jobs[i].ConversionInterval.Duration()),
OverwriteExistingData: jobs[i].OverwriteExistingData,
PrerequisiteJobNickname: jobs[i].PrerequisiteJobNickname,
DecimalPlaceComparison: jobs[i].DecimalPlaceComparison,
SecondaryExchangeName: jobs[i].SecondaryExchangeSource,
IssueTolerancePercentage: jobs[i].IssueTolerancePercentage,
ReplaceOnIssue: jobs[i].ReplaceOnIssue,
})
}
return &gctrpc.DataHistoryJobs{
@@ -3648,12 +3655,15 @@ func (s *RPCServer) GetDataHistoryJobSummary(_ context.Context, r *gctrpc.GetDat
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,
StartDate: job.StartDate.Format(common.SimpleTimeFormat),
EndDate: job.EndDate.Format(common.SimpleTimeFormat),
Interval: int64(job.Interval.Duration()),
Status: job.Status.String(),
DataType: job.DataType.String(),
ConversionInterval: int64(job.ConversionInterval.Duration()),
OverwriteExistingData: job.OverwriteExistingData,
PrerequisiteJobNickname: job.PrerequisiteJobNickname,
ResultSummaries: job.ResultRanges,
}, nil
}
@@ -3665,3 +3675,44 @@ func (s *RPCServer) unixTimestamp(x time.Time) int64 {
}
return x.Unix()
}
// SetDataHistoryJobStatus sets a data history job's status
func (s *RPCServer) SetDataHistoryJobStatus(_ context.Context, r *gctrpc.SetDataHistoryJobStatusRequest) (*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.SetJobStatus(r.Nickname, r.Id, dataHistoryStatus(r.Status))
if err != nil {
log.Error(log.GRPCSys, err)
status = "failed"
}
return &gctrpc.GenericResponse{Status: status}, err
}
// UpdateDataHistoryJobPrerequisite sets or removes a prerequisite job for an existing job
// if the prerequisite job is "", then the relationship is removed
func (s *RPCServer) UpdateDataHistoryJobPrerequisite(_ context.Context, r *gctrpc.UpdateDataHistoryJobPrerequisiteRequest) (*gctrpc.GenericResponse, error) {
if r == nil {
return nil, errNilRequestData
}
if r.Nickname == "" {
return nil, errNicknameUnset
}
status := "success"
err := s.dataHistoryManager.SetJobRelationship(r.PrerequisiteJobNickname, r.Nickname)
if err != nil {
return nil, err
}
if r.PrerequisiteJobNickname == "" {
return &gctrpc.GenericResponse{Status: status, Data: fmt.Sprintf("Removed prerequisite from job '%v'", r.Nickname)}, nil
}
return &gctrpc.GenericResponse{Status: status, Data: fmt.Sprintf("Set job '%v' prerequisite job to '%v' and set status to paused", r.Nickname, r.PrerequisiteJobNickname)}, nil
}

73
engine/rpcserver_test.go
View File

@@ -1318,7 +1318,7 @@ func TestParseEvents(t *testing.T) {
func TestRPCServerUpsertDataHistoryJob(t *testing.T) {
t.Parallel()
m := createDHM(t)
m, _ := createDHM(t)
em := SetupExchangeManager()
exch, err := em.NewExchangeByName(testExchange)
if err != nil {
@@ -1370,7 +1370,7 @@ func TestRPCServerUpsertDataHistoryJob(t *testing.T) {
func TestGetDataHistoryJobDetails(t *testing.T) {
t.Parallel()
m := createDHM(t)
m, _ := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
dhj := &DataHistoryJob{
@@ -1407,7 +1407,7 @@ func TestGetDataHistoryJobDetails(t *testing.T) {
t.Errorf("received %v, expected %v", err, nil)
}
_, err = s.GetDataHistoryJobDetails(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Id: m.jobs[0].ID.String()})
_, err = s.GetDataHistoryJobDetails(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Id: dhj.ID.String()})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
@@ -1424,9 +1424,9 @@ func TestGetDataHistoryJobDetails(t *testing.T) {
}
}
func TestDeleteDataHistoryJob(t *testing.T) {
func TestSetDataHistoryJobStatus(t *testing.T) {
t.Parallel()
m := createDHM(t)
m, j := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
dhj := &DataHistoryJob{
@@ -1442,40 +1442,49 @@ func TestDeleteDataHistoryJob(t *testing.T) {
if !errors.Is(err, nil) {
t.Fatalf("received %v, expected %v", err, nil)
}
_, err = s.DeleteDataHistoryJob(context.Background(), nil)
_, err = s.SetDataHistoryJobStatus(context.Background(), nil)
if !errors.Is(err, errNilRequestData) {
t.Errorf("received %v, expected %v", err, errNilRequestData)
}
_, err = s.DeleteDataHistoryJob(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{})
_, err = s.SetDataHistoryJobStatus(context.Background(), &gctrpc.SetDataHistoryJobStatusRequest{})
if !errors.Is(err, errNicknameIDUnset) {
t.Errorf("received %v, expected %v", err, errNicknameIDUnset)
}
_, err = s.DeleteDataHistoryJob(context.Background(), &gctrpc.GetDataHistoryJobDetailsRequest{Id: "123", Nickname: "123"})
_, err = s.SetDataHistoryJobStatus(context.Background(), &gctrpc.SetDataHistoryJobStatusRequest{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"})
id := dhj.ID
_, err = s.SetDataHistoryJobStatus(context.Background(), &gctrpc.SetDataHistoryJobStatusRequest{Nickname: "TestDeleteDataHistoryJob", Status: int64(dataHistoryStatusRemoved)})
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()})
j.Status = int64(dataHistoryStatusActive)
_, err = s.SetDataHistoryJobStatus(context.Background(), &gctrpc.SetDataHistoryJobStatusRequest{Id: id.String(), Status: int64(dataHistoryStatusRemoved)})
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)
_, err = s.SetDataHistoryJobStatus(context.Background(), &gctrpc.SetDataHistoryJobStatusRequest{Id: id.String(), Status: int64(dataHistoryStatusActive)})
if !errors.Is(err, errBadStatus) {
t.Errorf("received %v, expected %v", err, errBadStatus)
}
j.Status = int64(dataHistoryStatusActive)
_, err = s.SetDataHistoryJobStatus(context.Background(), &gctrpc.SetDataHistoryJobStatusRequest{Id: id.String(), Status: int64(dataHistoryStatusPaused)})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
if j.Status != int64(dataHistoryStatusPaused) {
t.Errorf("received %v, expected %v", dataHistoryStatus(j.Status), dataHistoryStatusPaused)
}
}
func TestGetActiveDataHistoryJobs(t *testing.T) {
t.Parallel()
m := createDHM(t)
m, _ := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
dhj := &DataHistoryJob{
@@ -1504,7 +1513,7 @@ func TestGetActiveDataHistoryJobs(t *testing.T) {
func TestGetDataHistoryJobsBetween(t *testing.T) {
t.Parallel()
m := createDHM(t)
m, _ := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
dhj := &DataHistoryJob{
@@ -1549,7 +1558,7 @@ func TestGetDataHistoryJobsBetween(t *testing.T) {
func TestGetDataHistoryJobSummary(t *testing.T) {
t.Parallel()
m := createDHM(t)
m, _ := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
dhj := &DataHistoryJob{
@@ -1782,3 +1791,33 @@ func TestRPCServer_GetTicker_LastUpdatedNanos(t *testing.T) {
t.Errorf("have %d, want %d", two.LastUpdated, want)
}
}
func TestUpdateDataHistoryJobPrerequisite(t *testing.T) {
t.Parallel()
m, _ := createDHM(t)
s := RPCServer{Engine: &Engine{dataHistoryManager: m}}
_, err := s.UpdateDataHistoryJobPrerequisite(context.Background(), nil)
if !errors.Is(err, errNilRequestData) {
t.Errorf("received %v, expected %v", err, errNilRequestData)
}
_, err = s.UpdateDataHistoryJobPrerequisite(context.Background(), &gctrpc.UpdateDataHistoryJobPrerequisiteRequest{})
if !errors.Is(err, errNicknameUnset) {
t.Errorf("received %v, expected %v", err, errNicknameUnset)
}
_, err = s.UpdateDataHistoryJobPrerequisite(context.Background(), &gctrpc.UpdateDataHistoryJobPrerequisiteRequest{
Nickname: "test456",
})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
_, err = s.UpdateDataHistoryJobPrerequisite(context.Background(), &gctrpc.UpdateDataHistoryJobPrerequisiteRequest{
Nickname: "test456",
PrerequisiteJobNickname: "test123",
})
if !errors.Is(err, nil) {
t.Errorf("received %v, expected %v", err, nil)
}
}