From 2a2c0d0e1c6eb7da33aed6182192a47d6dcd1926 Mon Sep 17 00:00:00 2001 From: Luis Fernando Molina Date: Tue, 25 Feb 2020 08:17:02 +0100 Subject: [PATCH] OHLCV --- Historic-OHLCV/User.Bot.js | 452 +++++++++++++++++++++++++++++++++++++ Project.njsproj | 4 + 2 files changed, 456 insertions(+) create mode 100644 Historic-OHLCV/User.Bot.js diff --git a/Historic-OHLCV/User.Bot.js b/Historic-OHLCV/User.Bot.js new file mode 100644 index 000000000..ed11c653b --- /dev/null +++ b/Historic-OHLCV/User.Bot.js @@ -0,0 +1,452 @@ + +exports.newUserBot = function newUserBot(bot, logger, COMMONS, UTILITIES, fileStorage, STATUS_REPORT, EXCHANGE_API) { + + const FULL_LOG = true; + const LOG_FILE_CONTENT = false; + const GMT_SECONDS = ':00.000 GMT+0000'; + const GMT_MILI_SECONDS = '.000 GMT+0000'; + const MODULE_NAME = "User Bot"; + const TRADES_FOLDER_NAME = "OHLCVs"; + + thisObject = { + initialize: initialize, + start: start + }; + + let utilities = UTILITIES.newCloudUtilities(bot, logger) + let statusDependencies + + const ONE_MINUTE = 60000 + const MAX_TRADES_PER_EXECUTION = 100000 + const symbol = bot.market.baseAsset + '/' + bot.market.quotedAsset + const ccxt = require('ccxt') + + let allOHLCVs = [] + let thisReport; + let since + let initialProcessTimestamp + let beginingOfMarket + let lastFileSaved + let exchangeId + let options = {} + let firstId + let rateLimit + let exchange + let uiStartDate = new Date(bot.uiStartDate) + + const limit = 1000 + + return thisObject; + + function initialize(pStatusDependencies, callBackFunction) { + try { + + logger.fileName = MODULE_NAME; + logger.initialize(); + + statusDependencies = pStatusDependencies; + + exchangeId = bot.exchange.toLowerCase() + + /* Applying the parameters defined by the user at the Exchange node */ + if (bot.exchangeNode.code.API !== undefined) { + for (let i = 0; i < bot.exchangeNode.code.API.length; i++) { + if (bot.exchangeNode.code.API[i].method === 'fetchOHLCVs') { + if (bot.exchangeNode.code.API[i].class !== undefined) { + exchangeId = bot.exchangeNode.code.API[i].class + } + if (bot.exchangeNode.code.API[i].fetchOHLCVsMethod !== undefined) { + options = { + 'fetchOHLCVsMethod': bot.exchangeNode.code.API[i].fetchOHLCVsMethod + } + } + if (bot.exchangeNode.code.API[i].firstId !== undefined) { + firstId = bot.exchangeNode.code.API[i].firstId + } + if (bot.exchangeNode.code.API[i].rateLimit !== undefined) { + rateLimit = bot.exchangeNode.code.API[i].rateLimit + } + } + } + } + + let key = process.env.KEY + let secret = process.env.SECRET + + if (key === "undefined") { key = undefined } + if (secret === "undefined") { secret = undefined } + + + const exchangeClass = ccxt[exchangeId] + const exchangeConstructorParams = { + 'apiKey': key, + 'secret': secret, + 'timeout': 30000, + 'enableRateLimit': true, + verbose: false, + options: options + } + if (rateLimit !== undefined) { + exchangeConstructorParams.rateLimit = rateLimit + } + + exchange = new exchangeClass(exchangeConstructorParams) + + callBackFunction(global.DEFAULT_OK_RESPONSE); + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] initialize -> err = " + err.stack); + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + } + } + + function start(callBackFunction) { + try { + + if (global.STOP_TASK_GRACEFULLY === true) { + callBackFunction(global.DEFAULT_OK_RESPONSE); + return + } + + let abort = false + begin() + + async function begin() { + + getContextVariables() + if (abort === true) { return } + await getOHLCVs() + if (abort === true) { return } + if (global.STOP_TASK_GRACEFULLY === true) { + callBackFunction(global.DEFAULT_OK_RESPONSE); + return + } + await saveOHLCVs() + } + + function getContextVariables() { + + try { + let reportKey; + + reportKey = "AAMasters" + "-" + "AACharly" + "-" + "Historic-OHLCVs" + "-" + "dataSet.V1"; + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] start -> getContextVariables -> reportKey = " + reportKey); } + + if (statusDependencies.statusReports.get(reportKey).status === "Status Report is corrupt.") { + logger.write(MODULE_NAME, "[ERROR] start -> getContextVariables -> Can not continue because dependecy Status Report is corrupt. "); + callBackFunction(global.DEFAULT_RETRY_RESPONSE); + return; + } + + thisReport = statusDependencies.statusReports.get(reportKey) + + if (thisReport.file.beginingOfMarket !== undefined) { // This means this is not the first time this process run. + beginingOfMarket = new Date(thisReport.file.beginingOfMarket.year + "-" + thisReport.file.beginingOfMarket.month + "-" + thisReport.file.beginingOfMarket.days + " " + thisReport.file.beginingOfMarket.hours + ":" + thisReport.file.beginingOfMarket.minutes + GMT_SECONDS); + lastFileSaved = new Date(thisReport.file.lastFileSaved.year + "-" + thisReport.file.lastFileSaved.month + "-" + thisReport.file.lastFileSaved.days + " " + thisReport.file.lastFileSaved.hours + ":" + thisReport.file.lastFileSaved.minutes + GMT_SECONDS); + } else { // This means this is the first time this process run. + beginingOfMarket = new Date(uiStartDate.valueOf()) + } + + defineSince() + function defineSince() { + if (thisReport.file.uiStartDate === undefined) { + thisReport.file.uiStartDate = uiStartDate + } else { + thisReport.file.uiStartDate = new Date(thisReport.file.uiStartDate) + } + + if (uiStartDate.valueOf() !== thisReport.file.uiStartDate.valueOf()) { + since = uiStartDate.valueOf() + initialProcessTimestamp = since + beginingOfMarket = new Date(uiStartDate.valueOf()) + } else { + if (lastFileSaved !== undefined) { + since = lastFileSaved.valueOf() + initialProcessTimestamp = lastFileSaved.valueOf() + } else { + since = uiStartDate.valueOf() + initialProcessTimestamp = uiStartDate.valueOf() + } + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] start -> getContextVariables -> err = " + err.stack); + if (err.message === "Cannot read property 'file' of undefined") { + logger.write(MODULE_NAME, "[HINT] start -> getContextVariables -> Check the bot Status Dependencies. "); + logger.write(MODULE_NAME, "[HINT] start -> getContextVariables -> Dependencies loaded -> keys = " + JSON.stringify(statusDependencies.keys)); + } + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + abort = true + } + } + + async function getOHLCVs() { + + try { + let lastTradeKey = '' + let params = undefined + let previousSince + + while (true) { + + /* Reporting we are doing well */ + let processingDate = new Date(since) + processingDate = processingDate.getUTCFullYear() + '-' + utilities.pad(processingDate.getUTCMonth() + 1, 2) + '-' + utilities.pad(processingDate.getUTCDate(), 2); + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] start -> getOHLCVs -> Fetching OHLCVs @ " + processingDate + "-> exchange = " + bot.exchange + " -> symbol = " + symbol + " -> since = " + since + " -> limit = " + limit) } + bot.processHeartBeat("Fetching " + allOHLCVs.length.toFixed(0) + " OHLCVs from " + bot.exchange + " " + symbol + " @ " + processingDate) // tell the world we are alive and doing well + + /* Fetching the OHLCVs from the exchange.*/ + const OHLCVs = await exchange.fetchOHLCVs(symbol, '1m', since, limit, params) + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] start -> getOHLCVs -> OHLCVs Fetched = " + OHLCVs.length) } + if (OHLCVs.length > 0) { + let beginDate = new Date(OHLCVs[0].timestamp) + let endDate = new Date(OHLCVs[OHLCVs.length - 1].timestamp) + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] start -> getOHLCVs -> OHLCVs Fetched From " + beginDate) } + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] start -> getOHLCVs -> OHLCVs Fetched to " + endDate) } + } + + if (OHLCVs.length > 1) { + previousSince = since + since = OHLCVs[OHLCVs.length - 1]['timestamp'] + if (since === previousSince) { + since++ // this prevents requesting in a loop OHLCVs with the same timestamp, that can happen when all the records fetched comes with exactly the same timestamp. + } + + for (let i = 0; i < OHLCVs.length; i++) { + let OHLCV = OHLCVs[i] + let OHLCVKey = OHLCV.id + '-' + OHLCV.timestamp + '-' + OHLCV.side + '-' + OHLCV.price.toFixed(16) + '-' + OHLCV.amount.toFixed(16) + if (OHLCVKey !== lastTradeKey) { + allOHLCVs.push([OHLCV.timestamp, OHLCV.side, OHLCV.price, OHLCV.amount, OHLCV.id]) + } + lastTradeKey = OHLCVKey + } + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] start -> getOHLCVs -> Fetching OHLCVs @ " + processingDate + "-> exchange = " + bot.exchange + " -> symbol = " + symbol + " -> since = " + since + " -> limit = " + limit) } + bot.processHeartBeat("Fetching " + allOHLCVs.length.toFixed(0) + " OHLCVs from " + bot.exchange + " " + symbol + " @ " + processingDate) // tell the world we are alive and doing well + + } + + if ( + OHLCVs.length < limit - 1 || + global.STOP_TASK_GRACEFULLY === true || + allOHLCVs.length >= MAX_TRADES_PER_EXECUTION + ) { + break + } + } + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] start -> getOHLCVs -> Retrying Later -> err = " + err.stack); + callBackFunction(global.DEFAULT_RETRY_RESPONSE); + abort = true + } + } + + async function saveOHLCVs() { + + try { + + let fileContent = '[' + let previousRecordMinute = Math.trunc((initialProcessTimestamp - ONE_MINUTE) / ONE_MINUTE) + let currentRecordMinute + let needSeparator = false + let error + let separator + let heartBeatCounter = 0 + + let i = -1 + + controlLoop() + + function loop() { + + let filesToCreate = 0 + let filesCreated = 0 + + let record = allOHLCVs[i] + let OHLCV = { + timestamp: record[0], + side: record[1], + price: record[2], + amount: record[3] + } + + let processingDate = new Date(OHLCV.timestamp) + processingDate = processingDate.getUTCFullYear() + '-' + utilities.pad(processingDate.getUTCMonth() + 1, 2) + '-' + utilities.pad(processingDate.getUTCDate(), 2); + + /* Reporting we are doing well */ + heartBeatCounter-- + if (heartBeatCounter <= 0) { + heartBeatCounter = 1000 + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] start -> saveOHLCVs -> Saving OHLCVs @ " + processingDate + " -> i = " + i + " -> total = " + allOHLCVs.length) } + bot.processHeartBeat("Saving " + i.toFixed(0) + " OHLCVs from " + bot.exchange + " " + symbol + " @ " + processingDate) // tell the world we are alive and doing well + } + + /* Saving the OHLCVs in Files*/ + currentRecordMinute = Math.trunc(OHLCV.timestamp / ONE_MINUTE) + + if ( + currentRecordMinute !== previousRecordMinute + ) { + /* There are no more OHLCVs at this minute or it is the last OHLCV, so we save the file.*/ + saveFile(previousRecordMinute) + } + + if (needSeparator === false) { + needSeparator = true; + separator = ''; + } else { + separator = ','; + } + + /* Add the OHLCV to the file content.*/ + fileContent = fileContent + separator + '[' + OHLCV.timestamp + ',"' + OHLCV.side + '",' + OHLCV.price + ',' + OHLCV.amount + ']'; + + if (i === allOHLCVs.length - 1) { + /* This is the last OHLCV, so we save the file.*/ + saveFile(currentRecordMinute) + /* It might happen that there are several minutes after the last OHLCV without OHLCVs. We need to record empty files for them.*/ + if (allOHLCVs.length < MAX_TRADES_PER_EXECUTION) { + let currentTimeMinute = Math.trunc((new Date()).valueOf() / ONE_MINUTE) + if (currentTimeMinute - currentRecordMinute > 1) { + createMissingEmptyFiles(currentRecordMinute, currentTimeMinute) + } + } + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] start -> saveOHLCVs -> Saving OHLCVs @ " + processingDate + " -> i = " + i + " -> total = " + allOHLCVs.length) } + bot.processHeartBeat("Saving " + i.toFixed(0) + " OHLCVs from " + bot.exchange + " " + symbol + " @ " + processingDate) // tell the world we are alive and doing well + + return + } + previousRecordMinute = currentRecordMinute + if (error) { + callBackFunction(error); + return; + } + if (filesToCreate === 0) { + controlLoop() + } + function saveFile(minute) { + fileContent = fileContent + ']' + if (currentRecordMinute - previousRecordMinute > 1) { + createMissingEmptyFiles(previousRecordMinute, currentRecordMinute) + } + let fileName = bot.market.baseAsset + '_' + bot.market.quotedAsset + '.json' + if (previousRecordMinute >= initialProcessTimestamp / ONE_MINUTE) { + filesToCreate++ + fileStorage.createTextFile(getFilePath(minute * ONE_MINUTE) + '/' + fileName, fileContent + '\n', onFileCreated); + } + fileContent = '[' + needSeparator = false + + } + function createMissingEmptyFiles(begin, end) { + + /* + If this range is too wide, we will consider this means that the begin is before the begining of this market at this exchange. + In that case we will change the begin and the beginingOfMarket + */ + + if ((end - begin) / 60 / 24 > 7) { + begin = end + beginingOfMarket = new Date(end * ONE_MINUTE) + } + + for (let j = begin + 1; j < end; j++) { + let fileName = bot.market.baseAsset + '_' + bot.market.quotedAsset + '.json' + filesToCreate++ + fileStorage.createTextFile(getFilePath(j * ONE_MINUTE) + '/' + fileName, "[]" + '\n', onFileCreated); + + } + } + function onFileCreated(err) { + if (err.result !== global.DEFAULT_OK_RESPONSE.result) { + logger.write(MODULE_NAME, "[ERROR] start -> OHLCVsReadyToBeSaved -> onFileBCreated -> err = " + JSON.stringify(err)); + error = err // This allows the loop to be breaked. + return; + } + filesCreated++ + lastFileSaved = new Date((currentRecordMinute * ONE_MINUTE)) + if (filesCreated === filesToCreate) { + controlLoop() + } + } + function getFilePath(timestamp) { + let datetime = new Date(timestamp) + let dateForPath = datetime.getUTCFullYear() + '/' + + utilities.pad(datetime.getUTCMonth() + 1, 2) + '/' + + utilities.pad(datetime.getUTCDate(), 2) + '/' + + utilities.pad(datetime.getUTCHours(), 2) + '/' + + utilities.pad(datetime.getUTCMinutes(), 2) + let filePath = bot.filePathRoot + "/Output/" + TRADES_FOLDER_NAME + '/' + dateForPath; + return filePath + } + + + } + function controlLoop() { + if (global.STOP_TASK_GRACEFULLY === true) { + callBackFunction(global.DEFAULT_OK_RESPONSE); + return + } + + i++ + if (i < allOHLCVs.length) { + setImmediate(loop) + } else { + writeStatusReport() + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] start -> saveOHLCVs -> err = " + err.stack); + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + abort = true + } + } + + function writeStatusReport() { + try { + if (lastFileSaved === undefined) { return } + thisReport.file = { + lastFileSaved: { + year: lastFileSaved.getUTCFullYear(), + month: (lastFileSaved.getUTCMonth() + 1), + days: lastFileSaved.getUTCDate(), + hours: lastFileSaved.getUTCHours(), + minutes: lastFileSaved.getUTCMinutes() + }, + beginingOfMarket: { + year: beginingOfMarket.getUTCFullYear(), + month: (beginingOfMarket.getUTCMonth() + 1), + days: beginingOfMarket.getUTCDate(), + hours: beginingOfMarket.getUTCHours(), + minutes: beginingOfMarket.getUTCMinutes() + }, + uiStartDate: uiStartDate.toUTCString() + }; + + thisReport.save(onSaved); + + function onSaved(err) { + if (err.result !== global.DEFAULT_OK_RESPONSE.result) { + logger.write(MODULE_NAME, "[ERROR] start -> writeStatusReport -> onSaved -> err = " + err.stack); + callBackFunction(err); + return; + } + callBackFunction(global.DEFAULT_OK_RESPONSE); + } + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] start -> writeStatusReport -> err = " + err.stack); + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + } + } + + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] start -> err = " + err.stack); + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + } + } +}; diff --git a/Project.njsproj b/Project.njsproj index b6e16399a..e9672baf3 100644 --- a/Project.njsproj +++ b/Project.njsproj @@ -18,11 +18,15 @@ + + Code + +