fez5587/Superalgos
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remote Data Implemented

Browse Source
This commit is contained in:
Luis Fernando Molina committed 2020-04-24 12:03:46 +02:00
1 parent 480ab1b78e
commit 2c5bcefc39
11 files changed
+223 -48
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

DataDependencies.js
+12 -1
View File
@@ -10,7 +10,8 @@
nodeArray: undefined,
dataSetsModulesArray: [],
isItADepenency: isItADepenency,
initialize: initialize
initialize: initialize,
finalize: finalize
};
let filter = new Map()
@@ -103,6 +104,16 @@
}
}
function finalize() {
for (let i = 0; i < dataSetsModulesArray.length; i++) {
dataSetsModule = dataSetsModulesArray[i]
dataSetModule.finalize()
}
filter = undefined
bot = undefined
thisObject = undefined
}
function isItADepenency(timeFrame, product) {
let key = timeFrame + '-' + product
DataSet.js
+29 -14
View File
@@ -8,6 +8,7 @@ exports.newDataSet = function newDataSet(BOT, logger) {
node: undefined,
networkNode: undefined,
initialize: initialize,
finalize: finalize,
getTextFile: getTextFile,
createTextFile: createTextFile
};
@@ -78,29 +79,36 @@ exports.newDataSet = function newDataSet(BOT, logger) {
}
/* Now we will see where do we need to fetch this data from. */
let network = global.TASK_NETWORK.parentNode.parentNode.parentNode.parentNode.parentNode
let network = global.TASK_NETWORK
let datasetProductDefinition = thisObject.node.parentNode
let datasetNetworkNode
for (let i = 0; i < network.networkNodes.length; i++) {
let networkNode = network.networkNodes[i]
if (networkNode.dataStorage !== undefined) {
if (networkNode.dataStorage.sessionIndependentData !== undefined) {
for (let j = 0; i < networkNode.dataStorage.sessionIndependentData.exchangeDataProducts.length; i++) {
let exchangeDataProduct = exchangeDataProducts[j]
for (let j = 0; j < networkNode.dataStorage.sessionIndependentData.exchangeDataProducts.length; j++) {
let exchangeDataProduct = networkNode.dataStorage.sessionIndependentData.exchangeDataProducts[j]
for (let k = 0; k < exchangeDataProduct.singleMarketData.length; k++) {
let singleMarketData = exchangeDataProduct.singleMarketData[k]
for (let m = 0; m < singleMarketData.dataProducts.length; m++) {
let dataProduct = singleMarketData.dataProducts[m]
if (dataProduct.referenceParent !== undefined) {
let productDefinition = dataProduct.referenceParent
if (datasetProductDefinition.id === productDefinition.id) {
if (singleMarketData.referenceParent !== undefined) {
let market = singleMarketData.referenceParent
let currentProcessMarket = bot.processNode.marketReference.referenceParent
/* We found where the data is located on the network. */
thisObject.networkNode = networkNode
fileStorage = FILE_STORAGE.newFileStorage(logger, networkNode.code.host, networkNode.code.webPort);
callBackFunction(global.DEFAULT_OK_RESPONSE);
return
if (currentProcessMarket.id === market.id) {
for (let m = 0; m < singleMarketData.dataProducts.length; m++) {
let dataProduct = singleMarketData.dataProducts[m]
if (dataProduct.referenceParent !== undefined) {
let productDefinition = dataProduct.referenceParent
if (datasetProductDefinition.id === productDefinition.id) {
/* We found where the data is located on the network. */
if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Retrieving data from " + networkNode.name + " -> host = " + networkNode.code.host + ' -> port = ' + networkNode.code.webPort + '.'); }
fileStorage = FILE_STORAGE.newFileStorage(logger, networkNode.code.host, networkNode.code.webPort);
callBackFunction(global.DEFAULT_OK_RESPONSE);
return
}
}
}
}
}
@@ -119,6 +127,13 @@ exports.newDataSet = function newDataSet(BOT, logger) {
}
}
function finalize() {
fileStorage = undefined
thisObject.networkNode = undefined
bot = undefined
thisObject = undefined
}
function getTextFile(pFolderPath, pFileName, callBackFunction) {
try {
EventServerClient.js
+10 -3
View File
@@ -1,4 +1,4 @@
exports.newEventsServerClient = function newEventsServerClient() {
exports.newEventsServerClient = function newEventsServerClient(host, port) {
const MODULE = "Task Server "
const ERROR_LOG = true
@@ -19,7 +19,14 @@
let WEB_SOCKETS_CLIENT
const WEB_SOCKET = require('ws')
let port = process.env.WEB_SOCKETS_SERVER_PORT
if (host === undefined) {
host = 'localhost'
}
if (port === undefined) {
port = process.env.WEB_SOCKETS_SERVER_PORT
}
return thisObject
@@ -32,7 +39,7 @@
function setuptWebSockets(callBackFunction) {
try {
WEB_SOCKETS_CLIENT = new WEB_SOCKET('ws://localhost:' + port )
WEB_SOCKETS_CLIENT = new WEB_SOCKET('ws://' + host + ':' + port )
WEB_SOCKETS_CLIENT.onerror = error => {
console.log('[ERROR] Task Server -> Event Server Client -> setuptWebSockets -> On connection error -> error = ' + JSON.stringify(error))
FileStorage.js
+10 -3
View File
@@ -52,7 +52,7 @@ exports.newFileStorage = function newFileStorage(logger, host, port) {
logger.write(MODULE_NAME, '[INFO] FileStorage -> getTextFile -> fileLocation: ' + fileLocation)
/* Here we actually reaad the file. */
if (host === undefined) {
if (host === undefined || host === 'localhost') {
/* We read the file from the local file system. */
const fs = require('fs')
fs.readFile(fileLocation, onFileRead)
@@ -124,7 +124,14 @@ exports.newFileStorage = function newFileStorage(logger, host, port) {
if (canUsePrevious === true) {
logger.write(MODULE_NAME, '[WARN] FileStorage -> getTextFile -> onFileRead -> Could read the file, but could not parse it as it is not a valid JSON. Will try to read the PREVIOUS version instead. -> file = ' + fileLocation)
fs.readFile(fileLocation + '.Previous.json', onPreviousFileRead)
if (host === undefined || host === 'localhost') {
/* We read the file from the local file system. */
const fs = require('fs')
fs.readFile(fileLocation + '.Previous.json', onPreviousFileRead)
} else {
/* We read the file via a web server over http */
getFileViaHTTP(fileLocation + '.Previous.json', onPreviousFileRead)
}
function onPreviousFileRead(err, text) {
@@ -452,7 +459,7 @@ exports.newFileStorage = function newFileStorage(logger, host, port) {
function onEnd() {
let fileContent = Buffer.concat(chunks).toString('utf8')
callback(global.DEFAULT_OK_RESPONSE, fileContent)
callback(null, fileContent)
}
}
IndicatorBotProcessMainLoop.js
+7 -3
View File
@@ -776,10 +776,13 @@
function onFinished(err) {
try {
if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessFramework -> onFinished -> Entering function."); }
processFramework.finalize()
processFramework = undefined
dataDependencies.finalize()
dataDependencies = undefined
statusDependencies.finalize()
statusDependencies = undefined
switch (err.result) {
case global.DEFAULT_OK_RESPONSE.result: {
@@ -876,11 +879,12 @@
processExecutionEvents.finish(onFinished);
function onFinished(err) {
try {
if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Entering function."); }
processExecutionEvents.finalize()
processExecutionEvents = undefined
switch (err.result) {
case global.DEFAULT_OK_RESPONSE.result: {
logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Execution finished well.");
ProcessExecutionEvents.js
+76 -6
View File
@@ -8,27 +8,33 @@
*/
const MODULE_NAME = "Process Execution Events";
logger.fileName = MODULE_NAME
let bot = BOT;
let thisObject = {
networkNode: undefined,
initialize, initialize,
finalize: finalize,
start: start,
finish: finish
};
let processThisDependsOn; // This is the process this process needs to wait for it to finishes in order to start.
let currentProcessKey
let EVENT_SERVER_CLIENT = require('./EventServerClient.js');
return thisObject;
function initialize(callBackFunction) {
try {
if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Entering function."); }
let name = 'Not Depends on any Process'
logger.fileName = MODULE_NAME + "." + name;
if (bot.processNode.referenceParent !== undefined) {
@@ -48,9 +54,63 @@
}
}
}
/* We need to find at which network node is running the process that we need to hear from when it finished. */
let network = global.TASK_NETWORK
for (let i = 0; i < network.networkNodes.length; i++) {
let networkNode = network.networkNodes[i]
if (networkNode.dataMining !== undefined) {
for (let j = 0; j < networkNode.dataMining.exchangeTasks.length; j++) {
let exchangeTasks = networkNode.dataMining.exchangeTasks[j]
for (let k = 0; k < exchangeTasks.taskManagers.length; k++) {
let taskManager = exchangeTasks.taskManagers[k]
for (let m = 0; m < taskManager.tasks.length; m++) {
let task = taskManager.tasks[m]
if (task.bot !== undefined) {
for (let n = 0; n < task.bot.processes.length; n++) {
let process = task.bot.processes[n]
if (process.marketReference !== undefined) {
if (process.marketReference.referenceParent !== undefined) {
let market = process.marketReference.referenceParent
let currentProcessMarket = bot.processNode.marketReference.referenceParent
if (currentProcessMarket.id === market.id) {
if (process.referenceParent !== undefined) {
let processDefinition = process.referenceParent
if (processThisDependsOn.id === processDefinition.id) {
/* We found where the task that runs the process definition we are waiting for is located on the network. */
thisObject.networkNode = networkNode
if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Connecting to Websockets Server " + networkNode.name + " -> host = " + networkNode.code.host + ' -> port = ' + networkNode.code.webSocketsPort + '.'); }
EVENT_SERVER_CLIENT = EVENT_SERVER_CLIENT.newEventsServerClient(networkNode.code.host, networkNode.code.webSocketsPort)
EVENT_SERVER_CLIENT.initialize(onConnected)
function onConnected() {
if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Connected to Websockets Server " + networkNode.name + " -> host = " + networkNode.code.host + ' -> port = ' + networkNode.code.webSocketsPort + '.'); }
callBackFunction(global.DEFAULT_OK_RESPONSE);
return
}
return
}
}
}
}
}
}
}
}
}
}
}
}
callBackFunction(global.DEFAULT_FAIL_RESPONSE);
logger.write(MODULE_NAME, "[ERROR] initialize -> Initialization Failed because we could not find where the task that runs the Process Definition we are waiting for is located within the network. Check the logs for more info.");
return
}
logger.fileName = MODULE_NAME + "." + name;
callBackFunction(global.DEFAULT_OK_RESPONSE);
} catch (err) {
@@ -59,6 +119,16 @@
}
}
function finalize() {
EVENT_SERVER_CLIENT.finalize()
EVENT_SERVER_CLIENT = undefined
processThisDependsOn = undefined
currentProcessKey = undefined
thisObject.networkNode = undefined
bot = undefined
thisObject = undefined
}
function start(callBackFunction) {
try {
@@ -81,9 +151,9 @@
if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> waitForDependantProcess -> key = " + key); }
if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> waitForDependantProcess -> callerId = " + callerId); }
global.EVENT_SERVER_CLIENT.listenToEvent(key, 'Process Execution Finished', undefined, callerId, responseCallBack, eventsCallBack)
EVENT_SERVER_CLIENT.listenToEvent(key, 'Process Execution Finished', undefined, callerId, responseCallBack, eventsCallBack)
bot.processHeartBeat(undefined, undefined, "Waiting for " + processThisDependsOn.parentNode.parentNode.name + "->" + processThisDependsOn.parentNode.code.codeName + "->" + processThisDependsOn.code.codeName)
bot.processHeartBeat(undefined, undefined, "Waiting for " + thisObject.networkNode.name + "->" + processThisDependsOn.parentNode.parentNode.name + "->" + processThisDependsOn.parentNode.code.codeName + "->" + processThisDependsOn.code.codeName)
function responseCallBack(message) {
if (message.result !== global.DEFAULT_OK_RESPONSE.result) {
@@ -102,7 +172,7 @@
/* We continue the normal flow after we learn the dependent process has finished its execution. */
if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> eventsCallBack -> stopListening to Process Execution Finished. "); }
if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> eventsCallBack -> subscriptionId = " + subscriptionId); }
global.EVENT_SERVER_CLIENT.stopListening(key, 'Process Execution Finished', subscriptionId)
EVENT_SERVER_CLIENT.stopListening(key, 'Process Execution Finished', subscriptionId)
if (message.event.err.result === global.DEFAULT_OK_RESPONSE.result) {
/* We emit our own event that the process started. */
SensorBotProcessMainLoop.js
+3 -2
View File
@@ -584,11 +584,12 @@
processExecutionEvents.finish(onFinished);
function onFinished(err) {
try {
if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Entering function."); }
processExecutionEvents.finalize()
processExecutionEvents = undefined
switch (err.result) {
case global.DEFAULT_OK_RESPONSE.result: {
logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Execution finished well.");
StatusDependencies.js
+9
View File
@@ -11,6 +11,7 @@
statusReports: new Map(),
reportsByMainUtility: new Map(),
initialize: initialize,
finalize: finalize,
keys: []
};
@@ -165,4 +166,12 @@
}
}
function finalize() {
statusReports.forEach(forEachStatusDependency)
function forEachStatusDependency(statusDependency) {
statusDependency.finalize()
}
bot = undefined
thisObject = undefined
}
};
StatusReport.js
+59 -13
View File
@@ -13,9 +13,11 @@
let bot = BOT;
let thisObject = {
networkNode: undefined,
mainUtility: undefined,
file: undefined, // Here we have the JSON object representing the file content.
initialize: initialize,
finalize: finalize,
load: load,
save: save,
status: undefined
@@ -30,7 +32,7 @@
/* Storage account to be used here. */
const FILE_STORAGE = require('./FileStorage.js');
let fileStorage = FILE_STORAGE.newFileStorage(logger);
let fileStorage
let sessionPath = ''
@@ -93,17 +95,6 @@
statusDependencyNode.bottype = statusDependencyNode.referenceParent.parentNode.parentNode.type
statusDependencyNode.dataMine = statusDependencyNode.referenceParent.parentNode.parentNode.parentNode.code.codeName
/* Let's see if the process is run monthly or not. */
if (statusDependencyNode.referenceParent.parentNode.code.startMode !== undefined) {
if (statusDependencyNode.referenceParent.parentNode.code.startMode.allMonths !== undefined) {
if (statusDependencyNode.referenceParent.parentNode.code.startMode.oneMonth !== undefined) {
if (statusDependencyNode.referenceParent.parentNode.code.startMode.allMonths.run === "true" || statusDependencyNode.referenceParent.parentNode.code.startMode.oneMonth.run === "true") {
statusDependencyNode.processRunMonthly = true
}
}
}
}
/* We retrieve the report main utility */
if (statusDependencyNode.code !== undefined) {
if (statusDependencyNode.code.mainUtility !== undefined) {
@@ -122,7 +113,55 @@
sessionPath = bot.SESSION.folderName + "/"
}
callBackFunction(global.DEFAULT_OK_RESPONSE);
/* Now we will see where do we need to fetch this status report from. */
let network = global.TASK_NETWORK
let processThisDependsOn = statusDependencyNode.referenceParent.parentNode
for (let i = 0; i < network.networkNodes.length; i++) {
let networkNode = network.networkNodes[i]
if (networkNode.dataMining !== undefined) {
for (let j = 0; j < networkNode.dataMining.exchangeTasks.length; j++) {
let exchangeTasks = networkNode.dataMining.exchangeTasks[j]
for (let k = 0; k < exchangeTasks.taskManagers.length; k++) {
let taskManager = exchangeTasks.taskManagers[k]
for (let m = 0; m < taskManager.tasks.length; m++) {
let task = taskManager.tasks[m]
if (task.bot !== undefined) {
for (let n = 0; n < task.bot.processes.length; n++) {
let process = task.bot.processes[n]
if (process.marketReference !== undefined) {
if (process.marketReference.referenceParent !== undefined) {
let market = process.marketReference.referenceParent
let currentProcessMarket = bot.processNode.marketReference.referenceParent
if (currentProcessMarket.id === market.id) {
if (process.referenceParent !== undefined) {
let processDefinition = process.referenceParent
if (processThisDependsOn.id === processDefinition.id) {
/* We found where the task that runs the process definition this status report depends on and where it is located on the network. */
thisObject.networkNode = networkNode
if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Retrieving status report from " + networkNode.name + " -> host = " + networkNode.code.host + ' -> port = ' + networkNode.code.webPort + '.'); }
fileStorage = FILE_STORAGE.newFileStorage(logger, networkNode.code.host, networkNode.code.webPort);
callBackFunction(global.DEFAULT_OK_RESPONSE);
return
}
}
}
}
}
}
}
}
}
}
}
}
callBackFunction(global.DEFAULT_FAIL_RESPONSE);
logger.write(MODULE_NAME, "[ERROR] initialize -> Initialization Failed because we could not find where the data of this status report is located within the network. Check the logs for more info.");
} catch (err) {
logger.write(MODULE_NAME, "[ERROR] initialize -> err = "+ err.stack);
@@ -130,6 +169,13 @@
}
}
function finalize() {
fileStorage = undefined
thisObject.networkNode = undefined
bot = undefined
thisObject = undefined
}
function load(callBackFunction) {
try {
TradingBotProcessMainLoop.js
+6 -1
View File
@@ -694,6 +694,10 @@
if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessFramework -> onFinished -> Entering function."); }
processFramework.finalize()
processFramework = undefined
dataDependencies.finalize()
dataDependencies = undefined
statusDependencies.finalize()
statusDependencies = undefined
switch (err.result) {
case global.DEFAULT_OK_RESPONSE.result: {
@@ -792,8 +796,9 @@
processExecutionEvents.finish(onFinished);
function onFinished(err) {
try {
processExecutionEvents.finalize()
processExecutionEvents = undefined
if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Entering function."); }
server.js
+2 -2
View File
@@ -131,7 +131,7 @@ function preLoader() {
}
} catch (err) {
console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE -> ' + err.stack)
console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE = ' + JSON.stringify(global.TASK_NODE))
console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE = ' + JSON.stringify(global.TASK_NODE).substring(0, 1000))
}
}
else { /* This process was started not by the Task Manager, but independently (most likely for debugging purposes). In this case we listen to an event with the Task Info that should be emitted at the UI */
@@ -146,7 +146,7 @@ function preLoader() {
}
} catch (err) {
console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE -> ' + err.stack)
console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE = ' + JSON.stringify(global.TASK_NODE))
console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE = ' + JSON.stringify(global.TASK_NODE).substring(0, 1000))
}
}
}