From 4f4b80789b862548e4cfe31f95be82da75b61a6b Mon Sep 17 00:00:00 2001 From: Luis Fernando Molina Date: Thu, 8 Feb 2018 17:38:44 +0100 Subject: [PATCH] Hole Fixing modules taken as example --- .vs/AABruce/v15/.suo | Bin 3584 -> 12288 bytes AABruce.njsproj | 172 +- AABruce.sln | 25 + One-Min-Daily-Candles/Interval.js | 1748 +++++++++++++++++ .../One Min Daily Candles.js | 418 +--- One-Min-Daily-Candles/run.js | 60 +- 6 files changed, 1994 insertions(+), 429 deletions(-) create mode 100644 AABruce.sln create mode 100644 One-Min-Daily-Candles/Interval.js diff --git a/.vs/AABruce/v15/.suo b/.vs/AABruce/v15/.suo index 90574b27a81ff197191b7ee1b7e7a6acdbf19340..006109b6142a0abf4011042dff86788fec7e66e9 100644 GIT binary patch literal 12288 zcmeHN&2Jk;6d$J_w1ont^n(JeDWxsYHFmx6NBR{zPLxWCQre`gz!x{(q)uHsvg5Q> z6Nv*791(|#W088O1P8cq=mBxyf`ll7e*gsc9=K3dwETXvJMq}{I_t!3NXfI_?(EFl zd2imldGp?^f4;rxx6i-c@w=#lqhgh~|6sjXtJV9ovUpy|HrCyL@ZbTPSVnOLmXQWd ziXwhf!Y_J70r!%)!m0EOTg95WIoEBaQ3qfBbkn?d{a4?!Xzbo(Q=9iJ!F>#+ zUU9k6tFCohMTbvUw=1F6kU|;eD`7}fFzUR>fj@CvoWgSkXh~TPI&^NgJ>dTg8dm`+ zhZ|je{j5`=#?Nt2LXINzVF9aB)W2E#O#ZEopJR`9*!?_(+dp~-PMK?Z9op6dHvpdo za-181Y0k|3XB5(_EE08oR@ZZ-!JJ)~B{s0@OhwC%<_WzCbzXlE8IgQHx z7rXvx&eXQkZPOcR+qnMc{GV6;6i%)GrhF_ z%A7y>bKPi!XM04KNQ)6smSIO3f3AdIQ0Ucs{iiBACI5->xX^2f7$ZAxqk1Q zV_GRZ77gg@U-^FC85io$EHnPAuXBt4wt;rLM*lL(uK-^K?gZ`v`hl+jcLQGsvhO!= zrxEM}?gzdJd<*zC@Bq*N27vDXgTN4wF?twi0y}{bU=$bw#)0)RA4T0UU^g%UOagm= zimu*FpXBFe`A-UNpai`%1Lhyh36zeS(W0bHo0PQr57BB>>o4~d`4qo+hd|G-?o6hYD z`;++`_Y1H5UjnVe9z64M1(vZAalYGgW(45+C)%8CDYvZX)1~z8_ zYpuSK#np1)_-)SiwYIhKQ!D)EI}h6n{@iPM=JF`UpMqAVvCDFAZ3M8tMd`mEv!0c@ zFTnHVF&lkyMQ}DsnxARJn|0Jb&z#g8xY6nLy8e@x`BC_P+CPI``kn#D9-6It)gSe* zcze}9*DK=|LgV$%b6EO?A)E;|^L~tlTGv1K3!c>xYVCa(d@IoU)ZBS&N$$)_2KycT zO8?7oGq&?OeqH~xuXErzjvr_IjP!V;*oC<$nd1&xkDV%9{<; zKlv(XO6)W?v;RS8{UW&XJB%I+xvhnT_+NgfakIb1_ODA)J3?~-<7epQHvcv!dLa8t zNCGQi67dP6Pey{Q)q(b>bkDoq=2m8`OeE+d8^pS26l{B<0DL2Du^`t$ANXculv+Cz z;O}Ej%kg;S&u>M{r3v=<`2A@V{VP9v99$Xsj>8Y?(YF1Leue*OdHjsM7%vl+`uxY1 zLdAnJDdfe;4ppoBoqF_4?B30STiU@}#lpG}i>cH5dDLa!BDJs9;X>u4{`-OLImT>V zip9=9V?iQeA$`G4Hi$d7-e28$cw6^}XaAhI)AP|Msv|o3dEgJU+x+p8PwcEc*bTrB zO~4aP!u#n%}e?_EVmVSZa0WX zFE95#`(F<{eXGFcRBj@~5&8cLfvGk@e-%QQIO$E|mq3Kt2crnXEW)TW&=N#;6-6l~ zsbVPgq7-pxP0+>A9tYhlbd<%JLoHuB>(aS$h1Ak0+3eFX(yo)|*6oyyhTZPbp2VWz z!r`f>C1${(fJNnp_D;Z30{9OOGh&{8o<}JU?dWTS`A4g=FpEh*PR2~r7)uXi zAD7aulmAk2@0sVX3V05<84n%ehA-6>#P4d*>`W#Ty?sWwC){b6!BEtQnVm7C*NmXn z49BAJ;OtBykt|h4tw3R-JXtDE%=C1nN6J>IoF1s;%IR{kQW~|=C2O)+PCKOOp@L-$ z<_bnClgnQ*dNPG<-YTa{l|o>mJTr4OWTryVcqC->^>&5~Gh(KUcqkMyddx&$D3pq( z;_<}nH%lZXWcNdMzm(nY$lik7@QFV9MMAE7dFZ-Bq?U)EYi%1kwn!UX?E42Y*1>2d zaxQ3CVG}FoU@T_DBiS*d)5?s6byY8QT=9*X=@nN2yl84cU-dt~S5j zZPC2D;E8&CJ+@&o_3)G52l(BV`5N=zcEQ=(+y@p8d$7yZet|4wH$08lfKfz1UYqsK z{N+9DJ&)#36n0+xL7Vz-C8*{(Z~^0I1ZjuPHrM$qRT~FqwV|t diff --git a/AABruce.njsproj b/AABruce.njsproj index 623ced6ca..aaa85903c 100644 --- a/AABruce.njsproj +++ b/AABruce.njsproj @@ -1,89 +1,89 @@  - - Debug - 2.0 - {045bbaef-3c35-4017-8458-e46aef438790} - - ShowAllFiles - Azure File Storage.js - . - . - {3AF33F2E-1136-4D97-BBB7-1795711AC8B8};{349c5851-65df-11da-9384-00065b846f21};{9092AA53-FB77-4645-B42D-1CCCA6BD08BD} - 14.0 - $(MSBuildExtensionsPath32)\Microsoft\VisualStudio\v$(VisualStudioVersion) - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - False - True - 0 - / - http://localhost:48022/ - False - True - http://localhost:1337 - False - - - - - - - CurrentPage - True - False - False - False - - - - - - - - - False - False - - - - + + Debug + 2.0 + {045bbaef-3c35-4017-8458-e46aef438790} + + ShowAllFiles + One-Min-Daily-Candles\run.js + C:\Users\Luis\source\repos\AABruce\One-Min-Daily-Candles\ + . + {3AF33F2E-1136-4D97-BBB7-1795711AC8B8};{349c5851-65df-11da-9384-00065b846f21};{9092AA53-FB77-4645-B42D-1CCCA6BD08BD} + 14.0 + $(MSBuildExtensionsPath32)\Microsoft\VisualStudio\v$(VisualStudioVersion) + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + False + True + 0 + / + http://localhost:48022/ + False + True + http://localhost:1337 + False + + + + + + + CurrentPage + True + False + False + False + + + + + + + + + False + False + + + + \ No newline at end of file diff --git a/AABruce.sln b/AABruce.sln new file mode 100644 index 000000000..e5011cdca --- /dev/null +++ b/AABruce.sln @@ -0,0 +1,25 @@ + +Microsoft Visual Studio Solution File, Format Version 12.00 +# Visual Studio 15 +VisualStudioVersion = 15.0.26730.10 +MinimumVisualStudioVersion = 10.0.40219.1 +Project("{9092AA53-FB77-4645-B42D-1CCCA6BD08BD}") = "AABruce", "AABruce.njsproj", "{045BBAEF-3C35-4017-8458-E46AEF438790}" +EndProject +Global + GlobalSection(SolutionConfigurationPlatforms) = preSolution + Debug|Any CPU = Debug|Any CPU + Release|Any CPU = Release|Any CPU + EndGlobalSection + GlobalSection(ProjectConfigurationPlatforms) = postSolution + {045BBAEF-3C35-4017-8458-E46AEF438790}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {045BBAEF-3C35-4017-8458-E46AEF438790}.Debug|Any CPU.Build.0 = Debug|Any CPU + {045BBAEF-3C35-4017-8458-E46AEF438790}.Release|Any CPU.ActiveCfg = Release|Any CPU + {045BBAEF-3C35-4017-8458-E46AEF438790}.Release|Any CPU.Build.0 = Release|Any CPU + EndGlobalSection + GlobalSection(SolutionProperties) = preSolution + HideSolutionNode = FALSE + EndGlobalSection + GlobalSection(ExtensibilityGlobals) = postSolution + SolutionGuid = {2A5094EC-199A-4833-894F-166CB38B4715} + EndGlobalSection +EndGlobal diff --git a/One-Min-Daily-Candles/Interval.js b/One-Min-Daily-Candles/Interval.js new file mode 100644 index 000000000..744a74635 --- /dev/null +++ b/One-Min-Daily-Candles/Interval.js @@ -0,0 +1,1748 @@ +exports.newInterval = function newInterval(BOT) { + + let bot = BOT; + const ROOT_DIR = '../'; + const GMT_SECONDS = ':00.000 GMT+0000'; + const GMT_MILI_SECONDS = '.000 GMT+0000'; + + const MODULE_NAME = "Interval"; + const LOG_INFO = true; + + const EXCHANGE_NAME = "Poloniex"; + const EXCHANGE_ID = 1; + + const TRADES_FOLDER_NAME = "Trades"; + + const GO_RANDOM = false; + const FORCE_MARKET = 2; // This allows to debug the execution of an specific market. Not intended for production. + + const MARKETS_MODULE = require(ROOT_DIR + 'Markets'); + + const POLONIEX_CLIENT_MODULE = require(ROOT_DIR + 'Poloniex API Client'); + + const DEBUG_MODULE = require(ROOT_DIR + 'Debug Log'); + const logger = DEBUG_MODULE.newDebugLog(); + logger.fileName = MODULE_NAME; + logger.bot = bot; + + interval = { + initialize: initialize, + start: start + }; + + let markets; + + const AZURE_FILE_STORAGE = require(ROOT_DIR + 'Azure File Storage'); + let azureFileStorage = AZURE_FILE_STORAGE.newAzureFileStorage(bot); + + const UTILITIES = require(ROOT_DIR + 'Utilities'); + let utilities = UTILITIES.newUtilities(bot); + + let year; + let month; + + return interval; + + function initialize(yearAssigend, monthAssigned, callBackFunction) { + + try { + + year = yearAssigend; + month = monthAssigned; + + month = utilities.pad(month, 2); // Adding a left zero when needed. + + logger.fileName = MODULE_NAME + "-" + year + "-" + month; + + const logText = "[INFO] initialize - Entering function 'initialize' " + " @ " + year + "-" + month; + console.log(logText); + logger.write(logText); + + azureFileStorage.initialize(); + + markets = MARKETS_MODULE.newMarkets(bot); + markets.initialize(callBackFunction); + + + } catch (err) { + + const logText = "[ERROR] initialize - ' ERROR : " + err.message; + console.log(logText); + logger.write(logText); + + } + } + +/* + +This process is going to do the following: + +It will iterate through all the markets and will try to find "holes" on the transaction history and fix them by retrieving the missing transactions from the exchange. + +What are holes? + + Holes are missing transactions in the history of the market and the reason there can be holes are the following: + + 1) The "Lives Trades" process can be stopped and later started again. The process will only retrieve live trades, leaving a "hole" between it was stopped and restarted. + 2) The "Historic Trades" process might have received a "hole" of transactions directly from the exchange. + 3) The "Historic Trades" process failed to write some files to the storage and now they are missing, together with the transaction in them. + 4) The "Historic Trades" process wrote some files to the storage with corrupted content. + 5) The "Historic Trades" process wrote files with transaction not properly ordered by Trade Id (sometimes the exchange sent this wrong). + +How does the process work? + + Transactions have a unique id generated by the exchange. It is a secuencial number. This process scans the transaction history looking for missing ids. When it finds + some group of missing ids, it request the exchange the missing records and it writes the necesary files to fix the hole. But some things might happen: + + 1) The exchange responds, but does not provides the missing records. In this case the process will flag this hole for retry later, up to 3 times. After that it will flag + the hole as "permanent". + 2) The exchange call timeouts: the process will continue retrying later. + 3) The exchange returns an unexpected error: The process will retry up to 3 times later under this condition. + 4) The exchange returns neighboring transactions but not the missing ones. The process will retry 3 more times and mark the hole as permanent if the exchange refuses to send the missing trades. + +Since the process is run in an infinite loop but can be restarted any time, the status of is running is recorded in the Process.Status file. + +What is the lastFile pointer? + + It is the Datetime since the begining of the hitory of a market that is considered without holes (or only with permanent ones, unfixables). In other words, the process + starts from the begining of the history of a market, and only when its history is signaled as "Complete" by the "Historic Trades" process. From there, the process is advancing + in time, validating the history and fixing holes, and moving the lastFile pointer forward as it gets sure that all the previous trades files are with no holes or unfixable ones. + This pointer will later be used by Indicators bots that depends of these trades, to calculate their final version of indicators up to the lastFile pointer datetime. + + +*/ + + function start(callBackFunction) { + + try { + + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'start', with year = " + year + " and month = " + month); + } + + let processDate = new Date(year + "-" + month + "-1 00:00:00.000 GMT+0000"); + + let lastMinuteOfMonth = new Date(year + "-" + month + "-1 00:00:00.000 GMT+0000"); + + lastMinuteOfMonth.setUTCMonth(lastMinuteOfMonth.getUTCMonth() + 1); // First we go 1 month into the future. + lastMinuteOfMonth.setUTCSeconds(lastMinuteOfMonth.getUTCSeconds() - 30); // Then we go back 30 seconds, or to the last minute of the original month. + + let thisDatetime = new Date(); + + if ((year === thisDatetime.getUTCFullYear() && month > thisDatetime.getUTCMonth() + 1) || year > thisDatetime.getUTCFullYear()) { + + logger.write("[INFO] We are too far in the future. Interval will not execute. Sorry."); + return; + + } + + let nextIntervalExecution = false; // This tell weather the Interval module will be executed again or not. By default it will not unless some hole have been found in the current execution. + + let currentDate; // This will hold the current datetime of each execution. + let cursorDatetime; // This holds the datetime we are using to request records from, backwards. + + let marketQueue; // This is the queue of all markets to be procesesd at each interval. + let market = { // This is the current market being processed after removing it from the queue. + id: 0, + assetA: "", + assetB: "" + }; + + let dateForPath; + + let filePath; + + let exchangeCallTime; + + const MAX_EXCHANGE_CALL_RETRIES = 3; + let exchangeCallRetries = 0; + + const MAX_HOLE_FIXING_RETRIES = 3; + + const FIRST_TRADE_RECORD_ID = -1; + const UNKNOWN_TRADE_RECORD_ID = -2; + + let tradesWithHole = []; // File content of the file where a hole was discovered. + + let currentTradeId; // This points to the last Trade Id that is ok. + let currentDatetime; // This points to the last Trade datetime that is ok. + + /* The next 3 variables hold the information read from varios Status Reports. */ + + let lastLiveTradeFile; // Datetime of the last complete trades file written by the Live Trades process. + let lastHistoricTradeFile; // Datatime of the last trades file written by the Historic Trades process. + let lastHoleFixingFile; // Datetime of the last file certified by the Hole Fixing process as without permanent holes. + + /* The next 4 variables hold the results of the search of the next hole. */ + + let holeInitialId; // This is the Id just before the hole. + let holeInitialDatetime; // This is the Datetime just before the hole. + + let holeFinalId; // This is the Id just after the hole. + let holeFinalDatetime; // This is the Datetime just after the hole. + + let holeFixingStatusReport; // Current hole Fixing Status Report. + + marketsLoop(); + + /* + + At every run, the process needs to loop through all the markets at this exchange. + The problem is that we can not fire all the requests at once to the exchange or we would get banned. + For that reason we create a queue with the markets to process and we do one at the time. + The following functions marketsLoop(), openMarket(), closeMarket() and closeAndOpenMarket() controls the serialization of this processing. + + */ + + function marketsLoop() { + + try { + + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'marketsLoop'"); + } + + currentDate = new Date(); + markets.getMarketsByExchange(EXCHANGE_ID, onMarketsReady); + + function onMarketsReady(marketsArray) { + + marketQueue = JSON.parse(marketsArray); + + openMarket(); // First execution and entering into the real loop. + + } + } + catch (err) { + const logText = "[ERROR] 'marketsLoop' - ERROR : " + err.message; + logger.write(logText); + } + } + + function openMarket() { + + // To open a Market means picking a new market from the queue. + + try { + + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'openMarket'"); + } + + + if (marketQueue.length === 0) { + + if (LOG_INFO === true) { + logger.write("[INFO] 'openMarket' - marketQueue.length === 0"); + } + + const logText = "[WARN] We processed all the markets."; + logger.write(logText); + + callBackFunction(nextIntervalExecution); + + return; + } + + if (GO_RANDOM === true) { + const index = parseInt(Math.random() * (marketQueue.length - 1)); + + market.id = marketQueue[index][0]; + market.assetA = marketQueue[index][1]; + market.assetB = marketQueue[index][2]; + market.status = marketQueue[index][3]; + + marketQueue.splice(index, 1); + } else { + let marketRecord = marketQueue.shift(); + + market.id = marketRecord[0]; + market.assetA = marketRecord[1]; + market.assetB = marketRecord[2]; + market.status = marketRecord[3]; + + if (FORCE_MARKET > 0) { + if (FORCE_MARKET !== market.id) { + closeAndOpenMarket(); + return; + } + } + } + + if (LOG_INFO === true) { + logger.write("[INFO] 'openMarket' - marketQueue.length = " + marketQueue.length); + logger.write("[INFO] 'openMarket' - market sucessfully opened : " + market.assetA + "_" + market.assetB); + } + + if (market.status === markets.ENABLED) { + + exchangeCallRetries = 0; // For each market we will count the retries, so we need to reset this here. + + getStatusReport(); + + } else { + + logger.write("[INFO] 'openMarket' - market " + market.assetA + "_" + market.assetB + " skipped because its status is not valid. Status = " + market.status); + closeAndOpenMarket(); + return; + + } + + + } + catch (err) { + const logText = "[ERROR] 'openMarket' - ERROR : " + err.message; + logger.write(logText); + closeMarket(); + } + } + + function closeMarket() { + + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'closeMarket'"); + } + + } + + function closeAndOpenMarket() { + + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'closeAndOpenMarket'"); + } + + openMarket(); + } + + + /* + + The following code executes for each market, trying to fix the hole in trades history. + + */ + + + function getStatusReport() { + + /* + + In order to know where to start the process (which datetime) we need to read the StatusReport of the process. This file will tell + us where we left after the last execution. If this file does not exist, this could mean that this process was never sucessfuly ran + for this market. This is normal when the market does not have the "complete" flag at the Trades History status report. + + We will also get the status report from the Live Trade process, to know which is highest possible datetime the process can work on. + + */ + + try { + + let reportFilePath; + let fileName = "Status.Report." + market.assetA + '_' + market.assetB + ".json" + + getLivesTrades(); + + function getLivesTrades() { + + reportFilePath = EXCHANGE_NAME + "/Processes/" + "Live-Trades"; + + azureFileStorage.getTextFile(reportFilePath, fileName, onStatusReportReceived, true); + + function onStatusReportReceived(text) { + + if (text === undefined) { + + /* The file does not exist, so this means it is too early to run this process. */ + + const logText = "[INFO] 'getStatusReport' - Market " + market.assetA + '_' + market.assetB + " is too early too prcess it, since there are no Live Trades reports about it yet. Skipping it. "; + logger.write(logText); + + closeAndOpenMarket(); + + } else { + + let statusReport; + + try { + + statusReport = JSON.parse(text); + + lastLiveTradeFile = new Date(statusReport.lastFile.year + "-" + statusReport.lastFile.month + "-" + statusReport.lastFile.days + " " + statusReport.lastFile.hours + ":" + statusReport.lastFile.minutes + GMT_SECONDS); + + } catch (err) { + + /* + + It might happen the the file contect is corrupt if it is read exactly at the moment the other process is writting it. + In this situation we just assume the current datetime is a good engough value. + + */ + + lastLiveTradeFile = new Date(); + + } + + /* We get from the file the datetime of the last file created. */ + + getHistoricTrades(); + + } + } + } + + function getHistoricTrades() { + + reportFilePath = EXCHANGE_NAME + "/Processes/" + "Historic-Trades"; + + azureFileStorage.getTextFile(reportFilePath, fileName, onStatusReportReceived, true); + + function onStatusReportReceived(text) { + + if (text === undefined) { + + /* The file does not exist, so this means it is too early to run this process. */ + + const logText = "[INFO] 'getStatusReport' - Market " + market.assetA + '_' + market.assetB + " is too early too prcess it, since there are no Historic Trades reports about it yet. Skipping it. "; + logger.write(logText); + + closeAndOpenMarket(); + + } else { + + let statusReport = JSON.parse(text); + + /* We get from the file to know if this markets history is complete or not. */ + + if (statusReport.completeHistory === true) { + + lastHistoricTradeFile = new Date(statusReport.lastFile.year + "-" + statusReport.lastFile.month + "-" + statusReport.lastFile.days + " " + statusReport.lastFile.hours + ":" + statusReport.lastFile.minutes + GMT_SECONDS); + + /* + + Before processing this month we need to check if it is not too far in the past. + + */ + + if ( + processDate.getUTCFullYear() < lastHistoricTradeFile.getUTCFullYear() + || + (processDate.getUTCFullYear() === lastHistoricTradeFile.getUTCFullYear() && processDate.getUTCMonth() < lastHistoricTradeFile.getUTCMonth()) + ) { + + const logText = "[INFO] 'getStatusReport' - The current year / month is before the start of the market history for market " + market.assetA + '_' + market.assetB + " . Skipping it. "; + logger.write(logText); + + closeAndOpenMarket(); + + } else { + + getHoleFixing(); + + } + + } else { + + const logText = "[INFO] 'getStatusReport' - Market " + market.assetA + '_' + market.assetB + " is history is not complete. Skipping it. "; + logger.write(logText); + + closeAndOpenMarket(); + + } + + } + } + } + + function getHoleFixing() { + + reportFilePath = EXCHANGE_NAME + "/Processes/" + "Hole-Fixing" + "/" + year + "/" + month; + + azureFileStorage.getTextFile(reportFilePath, fileName, onStatusReportReceived, true); + + function onStatusReportReceived(text) { + + try { + holeFixingStatusReport = JSON.parse(text); + } + catch (err) { + text = undefined; // If the content of the file is corrupt, this equals as if the file did not exist. + } + + if (text === undefined) { + + /* + + The file does not exist, so this means this is the first time we are running this process. + Then, we might me under two different possible situations: + + 1. We are at the first month of the market. In this case, the starting date is not the begining of the month but the first date of + the history of the market. + + 2. We are at some other month different than the first one. In this case, the starting date is the first minute of the market. + + */ + + if (processDate.valueOf() < lastHistoricTradeFile.valueOf()) { + + lastHoleFixingFile = new Date(lastHistoricTradeFile.valueOf() - 60 * 1000); // One minute less that the begining of market history. + + currentTradeId = FIRST_TRADE_RECORD_ID; + currentDatetime = new Date(lastHistoricTradeFile.valueOf()); + + } else { + + lastHoleFixingFile = new Date(processDate.valueOf() - 60 * 1000); // One minute less that the begining of the month. + + currentTradeId = UNKNOWN_TRADE_RECORD_ID; + currentDatetime = new Date(processDate.valueOf()); + + } + + findNextHole(); + + } else { + + if (holeFixingStatusReport.monthChecked === true) { + + const logText = "[INFO] 'getStatusReport' - The current year / month was already fully checked for market " + market.assetA + '_' + market.assetB + " . Skipping it. "; + logger.write(logText); + + closeAndOpenMarket(); + + } else { + + /* We get from the file the datetime of the last file without holes. */ + + lastHoleFixingFile = new Date(holeFixingStatusReport.lastFile.year + "-" + holeFixingStatusReport.lastFile.month + "-" + holeFixingStatusReport.lastFile.days + " " + holeFixingStatusReport.lastFile.hours + ":" + holeFixingStatusReport.lastFile.minutes + GMT_SECONDS); + + currentTradeId = holeFixingStatusReport.lastTrade.id; + currentDatetime = new Date(holeFixingStatusReport.lastTrade.year + "-" + holeFixingStatusReport.lastTrade.month + "-" + holeFixingStatusReport.lastTrade.days + " " + holeFixingStatusReport.lastTrade.hours + ":" + holeFixingStatusReport.lastTrade.minutes + ":" + holeFixingStatusReport.lastTrade.seconds + GMT_MILI_SECONDS); + + findNextHole(); + + } + } + } + } + + } + catch (err) { + const logText = "[ERROR] 'getStatusReport' - ERROR : " + err.message; + logger.write(logText); + closeMarket(); + } + } + + function findNextHole() { + try { + + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'findNextHole'"); + } + + /* + + To find the next hole, we will read first the current file (which we know if fully ok) and get from it the current Trade Id. + After that, we will continue reading the next files, checking all the ids, until we find a hole. + + */ + + let filePath; + let fileName = '' + market.assetA + '_' + market.assetB + '.json'; + let date; // This is pointing to each Trades File + + let fileCheckedCounter = 0; + + date = new Date(lastHoleFixingFile.valueOf()); + + readNextFile(); + + function readNextFile() { + + date = new Date(date.valueOf() + 60 * 1000); + + if (LOG_INFO === true) { + const logText = "[INFO] Entering function 'readNextFile' with date = " + date.toUTCString(); + logger.write(logText); + } + + if (date.valueOf() > lastLiveTradeFile.valueOf()) { + + /* This mean we reached the forward end of the market */ + + if (LOG_INFO === true) { + logger.write("[INFO] readNextFile - End of the market reached at date = " + date.toUTCString()); + } + + nextIntervalExecution = true; // Even if we didn-t find a hole, we need to continue the execution of this month interval. + + writeStatusReport(currentDatetime, currentTradeId, false, true, onStatusReportWritten); + + function onStatusReportWritten() { + + closeAndOpenMarket(); + + } + + return; + + } + + /* Lets check if we have reached the end of the month. */ + + if (date.valueOf() > lastMinuteOfMonth.valueOf()) { + + /* This mean we reached the forward end of the market */ + + if (LOG_INFO === true) { + logger.write("[INFO] readNextFile - End of the month reached at date = " + date.toUTCString()); + } + + writeStatusReport(currentDatetime, currentTradeId, true, false, onStatusReportWritten); + + function onStatusReportWritten() { + + closeAndOpenMarket(); + + } + + return; + + } + + dateForPath = date.getUTCFullYear() + '/' + utilities.pad(date.getUTCMonth() + 1, 2) + '/' + utilities.pad(date.getUTCDate(), 2) + '/' + utilities.pad(date.getUTCHours(), 2) + '/' + utilities.pad(date.getUTCMinutes(), 2); + + filePath = EXCHANGE_NAME + "/Output/" + TRADES_FOLDER_NAME + '/' + dateForPath; + + azureFileStorage.getTextFile(filePath, fileName, onNextFileReceived, true); + + function onNextFileReceived(text) { + + if (text === undefined) { + + /* The file does not exist, so this means there is a hole!!! */ + + if (LOG_INFO === true) { + logger.write("[INFO] Hole by missing file detected. Date = " + date.toUTCString()); + } + + holeInitialId = currentTradeId; + holeInitialDatetime = new Date(currentDatetime.valueOf()); // Field #5 contains the seconds. + + findEndOfHole(); + + } else { + + try { + + let tradesTest = JSON.parse(text); + + } catch (err) { + + /* + + If the file is corrupt, then we are in a similar situation as if it does not exist. + + */ + + if (LOG_INFO === true) { + logger.write("[INFO] Hole by corrupt file detected. Date = " + date.toUTCString()); + } + + holeInitialId = currentTradeId; + holeInitialDatetime = new Date(currentDatetime.valueOf()); // Field #5 contains the seconds. + + findEndOfHole(); + + return; + } + + checkHolesInFile(text); + + } + } + } + + function checkHolesInFile(text) { + + let trades = JSON.parse(text); + + /* + + tradesWithHole variable: + + Until verified, this trades in this file becomes potentially the last set of trades with hole. If it is not, then this variable will be overwritten later + by the one. + + + We will need these trades at the end of the process. + + */ + + tradesWithHole = trades; + + if (LOG_INFO === true) { + const logText = "[INFO] Entering function 'checkHolesInFile' to check File '" + fileName + "' @ " + filePath + " - " + trades.length + " records in it."; + logger.write(logText); + console.log(logText); + } + + if (currentTradeId === FIRST_TRADE_RECORD_ID || currentTradeId === UNKNOWN_TRADE_RECORD_ID) { + + /* + + Here we dont know the currentTradeId, so we might take it directly from the file. There will be a problem though if the market starts + with some empty files. In those cases, we need to jump to the next file until we find one with some records. + + */ + + if (trades.length > 0) { + + currentTradeId = trades[0][0] - 1; + + } else { + + readNextFile(); + return; + + } + + } + + for (let i = 0; i < trades.length; i++) { + + let fileTradeId = trades[i][0]; // First position in each record. + + if (currentTradeId + 1 > fileTradeId) { + + /* + + This happens when the process resumes execution, reads the first file and the first trades have lowers ids that the ones the process already + checked during the last execution. + + It also happens when we find a non valid id, as the one used in an empty record signaling that the file is incomplete. (zero). + + */ + + continue; // we simply jump to the next trade. + + } + + if (currentTradeId + 1 < fileTradeId) { + + /* + + We should usually try to fix the hole, but there is an exception. If the we tried this 3 times already, we must declare the problem + unsolvable and move forward. + + */ + + let lastRecordedTradeId = 0; + let lastRecordedCounter = 0; + + if (holeFixingStatusReport !== undefined) { // The whole could have benn found before the monthly report was created. + + lastRecordedTradeId = holeFixingStatusReport.lastTrade.id; + lastRecordedCounter = holeFixingStatusReport.lastTrade.counter; + + } + + if (currentTradeId === lastRecordedTradeId && lastRecordedCounter >= MAX_HOLE_FIXING_RETRIES) { + + if (LOG_INFO === true) { + logger.write("[INFO] Hole by non consecutive ID detected. MAX_HOLE_FIXING_RETRIES reched, giving up with this validation. Date = " + date.toUTCString()); + } + + /* We advance anyway to the next Id since there is no other solution. */ + + currentTradeId = fileTradeId; + currentDatetime = new Date(date.valueOf() + trades[i][5] * 1000); + + + } else { + + /* Here we have a hole that needs to be fixed !!! */ + + if (LOG_INFO === true) { + logger.write("[INFO] Hole by non consecutive ID detected. Date = " + date.toUTCString()); + } + + holeInitialId = currentTradeId; + holeInitialDatetime = new Date(currentDatetime.valueOf()); + + holeFinalId = fileTradeId; + holeFinalDatetime = new Date(date.valueOf() + trades[i][5] * 1000); // Field #5 contains the seconds. + + getTheTrades(); + + break; + + } + + + } else { + + /* We keep here the last Trade Id and Datetime that are allright. */ + + currentTradeId = fileTradeId; + currentDatetime = new Date(date.valueOf() + trades[i][5] * 1000); + + } + } + + if (holeInitialId === undefined) { + + fileCheckedCounter++; + + if (fileCheckedCounter === 60) { // Every hour checked we write a Status Report so that if the process is terminated, it can resume later from there. + + writeStatusReport(currentDatetime, currentTradeId, false, false, onStatusReportWritten); + + function onStatusReportWritten() { + + fileCheckedCounter = 0; + readNextFile(); + } + + } else { + + readNextFile(); + + } + } + } + + + function findEndOfHole() { + + /* Here we will enter a loop where will try to find the next available file recorded and extract from it the Id and Datetime from the first record. */ + + date = new Date(date.valueOf() + 60 * 1000); + + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'findEndOfHole' with date = " + date.toUTCString()); + } + + if (date.valueOf() > lastLiveTradeFile.valueOf()) { + + /* + + In this case we have an open hole produced by a missing file, and because live trades files contains zero records or are missing, we reached the + forward side of the market. The situation in unsolvable for now, we will leave it of future execution. + + */ + + if (LOG_INFO === true) { + logger.write("[INFO] findEndOfHole - End of the market reached at date = " + date.toUTCString()); + } + + nextIntervalExecution = true; // Even if we didn-t find the end of the hole, we need to continue the execution of this month interval. + + writeStatusReport(currentDatetime, currentTradeId, false, false, onStatusReportWritten); + + function onStatusReportWritten() { + + closeAndOpenMarket(); + + } + + return; + + } + + dateForPath = date.getUTCFullYear() + '/' + utilities.pad(date.getUTCMonth() + 1, 2) + '/' + utilities.pad(date.getUTCDate(), 2) + '/' + utilities.pad(date.getUTCHours(), 2) + '/' + utilities.pad(date.getUTCMinutes(), 2); + + filePath = EXCHANGE_NAME + "/Output/" + TRADES_FOLDER_NAME + '/' + dateForPath; + + azureFileStorage.getTextFile(filePath, fileName, onNextFileReceived, true); + + function onNextFileReceived(text) { + + if (text === undefined) { + + /* The file does not exist, so this means we need to move forward */ + + findEndOfHole(); + + return; + + } else { + + let trades; + + try { + + trades = JSON.parse(text); + + } catch(err) { + /* + + If the file is corrupt, then we are in a similar situation as if it does not exist. + + */ + + if (LOG_INFO === true) { + logger.write("[INFO] findEndOfHole - Corrupt file with no records found at date = " + date.toUTCString()); + } + + findEndOfHole(); + + return; + + } + + if (trades.length === 0) { + + /* This is the same situation that if there is no file, move forward */ + + if (LOG_INFO === true) { + logger.write("[INFO] findEndOfHole - File with no records found at date = " + date.toUTCString()); + } + + findEndOfHole(); + + return; + + } + + if (LOG_INFO === true) { + logger.write("[INFO] findEndOfHole - Next available record found at date = " + date.toUTCString()); + } + + let fileTradeId = trades[0][0]; // First position in each record. + + holeFinalId = fileTradeId; + holeFinalDatetime = new Date(date.valueOf() + trades[0][5] * 1000); // Field #5 contains the seconds. + + getTheTrades(); + + return; + + } + } + } + + } + catch (err) { + const logText = "[ERROR] 'findNextHole' - ERROR : " + err.message; + logger.write(logText); + closeMarket(); + } + + } + + function getTheTrades() { + + try { + + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'getTheTrades' with input variables: holeInitialDatetime = " + holeInitialDatetime.toUTCString() + ", holeInitialId = " + holeInitialId + ", holeFinalDatetime = " + holeFinalDatetime.toUTCString() + ", holeFinalId = " + holeFinalId ); + } + + /* We request to the Exchange API some more records than needed, anyway we will discard records out of the range we need. To do this we substract some seconds and add some others + seconds to the already calculated UNIX timestamps. */ + + const startTime = parseInt(holeInitialDatetime.valueOf() / 1000 - 65); + const endTime = parseInt(holeFinalDatetime.valueOf() / 1000 + 65); + + exchangeCallTime = new Date(); + + let poloniexApiClient = new POLONIEX_CLIENT_MODULE(); + + poloniexApiClient.returnPublicTradeHistory(market.assetA, market.assetB, startTime, endTime, onExchangeCallReturned); + + } + catch (err) { + const logText = "[ERROR] 'getTheTrades' - ERROR : " + err.message; + logger.write(logText); + closeMarket(); + } + } + + function onExchangeCallReturned(err, tradesRequested) { + + try { + + if (LOG_INFO === true) { + + let exchangeResponseTime = new Date(); + let timeDifference = (exchangeResponseTime.valueOf() - exchangeCallTime.valueOf()) / 1000; + logger.write("[INFO] Entering function 'onExchangeCallReturned' - Call time recorded = " + timeDifference + " seconds."); + } + + if (err || tradesRequested.error !== undefined) { + try { + + if (err.message.indexOf("ETIMEDOUT") > 0) { + + exchangeCallRetries++; + + if (exchangeCallRetries > MAX_EXCHANGE_CALL_RETRIES) { + + const logText = "[WARN] onExchangeCallReturned - Timeout reached while trying to access the Exchange API. MAX_EXCHANGE_CALL_RETRIES reached. Giving up. : ERROR = " + err.message; + logger.write(logText); + + closeAndOpenMarket(); + return; + + } else { + + const logText = "[WARN] onExchangeCallReturned - Timeout reached while trying to access the Exchange API. Trying again. : ERROR = " + err.message; + logger.write(logText); + + /* We try to reconnect to the exchange and fetch the data again. */ + + getTheTrades(); + return; + + } + + } else { + + if (err.message.indexOf("ECONNRESET") > 0) { + + exchangeCallRetries++; + + if (exchangeCallRetries > MAX_EXCHANGE_CALL_RETRIES) { + + const logText = "[WARN] onExchangeCallReturned - The exchange reseted the connection. MAX_EXCHANGE_CALL_RETRIES reached. Giving up. : ERROR = " + err.message; + logger.write(logText); + + closeAndOpenMarket(); + return; + + } else { + + const logText = "[WARN] onExchangeCallReturned - The exchange reseted the connection. Trying again. : ERROR = " + err.message; + logger.write(logText); + + /* We try to reconnect to the exchange and fetch the data again. */ + + getTheTrades(); + return; + + } + + } else { + + const logText = "[ERROR] onExchangeCallReturned - Unexpected error trying to contact the Exchange. Giving up. : ERROR = " + err.message; + logger.write(logText); + closeMarket(); + return; + } + } + + } catch (err) { + const logText = "[ERROR] onExchangeCallReturned : ERROR : tradesRequested.error = " + tradesRequested.error; + logger.write(logText); + + if (tradesRequested.error === "Invalid currency pair.") { + + markets.disableMarket(EXCHANGE_ID, market.id, onMarketDeactivated) + + function onMarketDeactivated() { + + logger.write("[INFO] Market " + market.assetA + "_" + market.assetB + " deactivated. Id = " + market.id); + + closeAndOpenMarket(); + return; + + } + + } + + closeMarket(); + return; + } + + closeMarket(); + return; + + } else { + + tradesReadyToBeSaved(tradesRequested); + } + } + catch (err) { + const logText = "[ERROR] 'onExchangeCallReturned' - ERROR : " + err.message; + logger.write(logText); + closeMarket(); + } + + } + + function tradesReadyToBeSaved(tradesRequested) { + + try { + + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'tradesReadyToBeSaved'"); + } + + nextIntervalExecution = true; // When there is at least one hole, we activate this flag to get an extra Interval execution when this one ends. + + /* + + We have learnt that the records from the exchange dont always come in the right order, sorted by TradeId. That means the we need to sort them + by ourselves if we want that our verification of holes work. + + */ + + let iterations = tradesRequested.length; + + for (let i = 0; i < iterations; i++) { + + for (let j = 0; j < iterations - 1; j++) { + + if (tradesRequested[j].tradeID < tradesRequested[j + 1].tradeID) { + + let trade = tradesRequested[j + 1]; + + tradesRequested.splice(j + 1, 1); // Remove that trade from the array. + + tradesRequested.splice(j, 0, trade); // Insert the trade removed. + + + } + + } + + } + + /* + + The trades received from the exchange might or might not be enough to fix the hole. We wont worry about that at this point. We will simple record the trades received + in the range where records where missing. + + We only have to take into account that the lowest id we have is already on a file that exist and it is partially verified, so we have to be carefull to overwrite this file + without introducing new holes. + + */ + + let fileRecordCounter = 0; + + let needSeparator; + let separator; + + let lastProcessMinute; // Stores the previous record minute during each iteration + let filesToSave = []; // Array where we will store all the content to be written to files + + needSeparator = false; + + let fileContent = ""; + + let currentProcessMinute = Math.trunc(holeFinalDatetime.valueOf() / 1000 / 60); // Number of minutes since the begining of time, where the process is pointing to. + let holeStartsMinute = Math.trunc(holeInitialDatetime.valueOf() / 1000 / 60); // Number of minutes since the begining of time, where the hole started. + + /* We will iterate through all the records received from the exchange. We know Poloniex sends the older records first, so this is going to be going back in time as we advance. */ + + for (let i = 0; i < tradesRequested.length; i++) { + + let record = tradesRequested[i]; + + const trade = { + tradeIdAtExchange: record.tradeID, + marketIdAtExchange: record.globalTradeID, + type: record.type, + rate: record.rate, + amountA: record.total, + amountB: record.amount, + datetime: new Date(record.date + GMT_MILI_SECONDS) + }; + + trade.seconds = trade.datetime.getUTCSeconds(); + + let currentRecordMinute = Math.trunc(trade.datetime.valueOf() / 1000 / 60); // This are the number of minutes since the begining of time of this trade. + + if (currentRecordMinute > currentProcessMinute) { + + /* We discard this trade, since it happened after the minute we want to record in the current file. */ + + continue; + } + + if (currentRecordMinute < currentProcessMinute) { + + /* + + The information is older that the current time. + We must store the current info and reset the pointer to the current time to match the one on the information currently being processd. + We know this can lead to a 'hole' or some empty files being skipped, but we solve that problem with the next loop. + + */ + + let blackMinutes = currentProcessMinute - currentRecordMinute; + + for (let j = 1; j <= blackMinutes; j++) { + + storeFileContent(); + currentProcessMinute--; + + } + } + + if (currentRecordMinute === currentProcessMinute) { + + if (needSeparator === false) { + + needSeparator = true; + separator = ''; + + } else { + separator = ','; + } + + if (trade.tradeIdAtExchange > holeInitialId) { + + /* We only add trades with ids bigger that the last id verified without holes. */ + + fileContent = '[' + trade.tradeIdAtExchange + ',"' + trade.type + '",' + trade.rate + ',' + trade.amountA + ',' + trade.amountB + ',' + trade.seconds + ']' + separator + fileContent; + + fileRecordCounter++; + + } + } + } + + if (fileContent !== "") { + + /* + + Usually the last file Content must be discarded since it could belong to an incomplete file. But there is one exception: it a hole is found at a file and the previous minute is empty + then this will produce the exception in which the fileContent needs to saved. To figure out if we are in this situation we do the following: + + */ + + if (currentProcessMinute === holeStartsMinute) { + + storeFileContent(); + + } + + + } + + function storeFileContent() { + + let existingFileContent = ""; + let separator = ""; + + if (currentProcessMinute === holeStartsMinute) { + + /* + + Here we are at the situation that the content already generated has to be added to the content already existing on the file where the hole was found. + + */ + + for (let i = 0; i < tradesWithHole.length; i++) { + + if (tradesWithHole[i][0] <= holeInitialId && tradesWithHole[i][0] !== 0) { // 0 because of the empty trade record signaling an incomplete file. + + /* We only add trades with ids smallers that the last id verified without holes. */ + + existingFileContent = existingFileContent + separator + '[' + tradesWithHole[i][0] + ',"' + tradesWithHole[i][1] + '",' + tradesWithHole[i][2] + ',' + tradesWithHole[i][3] + ',' + tradesWithHole[i][4] + ',' + tradesWithHole[i][5] + ']'; + + fileRecordCounter++; + + if (separator === "") { + + separator = ","; + + } + } + } + } + + if (existingFileContent === "") { + + fileContent = '[' + fileContent + ']'; + + } else { + + if (fileContent === "") { + + fileContent = '[' + existingFileContent + ']'; + + } else { + + fileContent = '[' + existingFileContent + "," + fileContent + ']'; + } + } + + let fileRecord = { + datetime: currentProcessMinute, + content: fileContent, + records: fileRecordCounter + }; + + filesToSave.push(fileRecord); + + fileRecordCounter = 0; + needSeparator = false; + fileContent = ""; + } + + /* + + Now it is time to process all the information we stored at filesToSave. + + */ + + let i = 0; + let date; + + nextRecord(); + + function nextRecord() { + + let fileName = '' + market.assetA + '_' + market.assetB + '.json'; + + date = new Date(filesToSave[i].datetime * 60 * 1000); + fileRecordCounter = filesToSave[i].records; + fileContent = filesToSave[i].content; + + dateForPath = date.getUTCFullYear() + '/' + utilities.pad(date.getUTCMonth() + 1, 2) + '/' + utilities.pad(date.getUTCDate(), 2) + '/' + utilities.pad(date.getUTCHours(), 2) + '/' + utilities.pad(date.getUTCMinutes(), 2); + + filePath = EXCHANGE_NAME + "/Output/" + TRADES_FOLDER_NAME + '/' + dateForPath; + + utilities.createFolderIfNeeded(filePath, azureFileStorage, onFolderCreated); + + function onFolderCreated() { + + azureFileStorage.createTextFile(filePath, fileName, fileContent + '\n', onFileCreated); + + function onFileCreated() { + + const logText = "[WARN] Finished with File @ " + market.assetA + "_" + market.assetB + ", " + fileRecordCounter + " records inserted into " + filePath + "/" + fileName + ""; + console.log(logText); + logger.write(logText); + + controlLoop(); + } + } + } + + + function controlLoop() { + + i++; + + if (i < filesToSave.length) { + + nextRecord(); + + } else { + + if (LOG_INFO === true) { + logger.write("[INFO] Leaving function 'tradesReadyToBeSaved'"); + } + + writeStatusReport(undefined, undefined, false, false, onStatusReportWritten); + + function onStatusReportWritten() { + + closeAndOpenMarket(); + + } + + } + } + } + catch (err) { + const logText = "[ERROR] 'tradesReadyToBeSaved' - ERROR : " + err.message; + logger.write(logText); + closeMarket(); + } + } + + function writeStatusReport(lastTradeDatetime, lastTradeId, monthChecked, atHeadOfMarket, callBack) { + + /* + + If no parameters are provided, that means that last good information is the begining of the hole. If they are provided is because no hole was detected until the + forward end of the market. + + */ + + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'writeStatusReport'"); + } + + if (lastTradeId === undefined) { + + lastTradeId = holeInitialId; + lastTradeDatetime = holeInitialDatetime; + + } + + let lastFileWithoutHoles = new Date(lastTradeDatetime.valueOf() - 60 * 1000); // It is the previous file where the last verified trade is. + + try { + + let reportFilePath = EXCHANGE_NAME + "/Processes/" + bot.process + "/" + year + "/" + month; + + utilities.createFolderIfNeeded(reportFilePath, azureFileStorage, onFolderCreated); + + function onFolderCreated() { + + try { + + /* + + Here we will calculate the "counter". The counter keeps track of how many times the process tried to fix the same hole. This allows + the process to know when a hole is not fixable. To do that we need to compre the current status report with the information we ve got + about the hole. If it is the same, we add to the counter. + + */ + + let counter = 0; + + try { + + if (holeFixingStatusReport.lastTrade.id === lastTradeId) { + + counter = holeFixingStatusReport.lastTrade.counter; + + if (counter === undefined) { counter = 0; } + counter++; + + } + + } catch (err) { // we are here when the status report did not exist. + + counter = 0; + + } + + let fileName = "Status.Report." + market.assetA + '_' + market.assetB + ".json"; + + let report = { + lastFile: { + year: lastFileWithoutHoles.getUTCFullYear(), + month: (lastFileWithoutHoles.getUTCMonth() + 1), + days: lastFileWithoutHoles.getUTCDate(), + hours: lastFileWithoutHoles.getUTCHours(), + minutes: lastFileWithoutHoles.getUTCMinutes() + }, + lastTrade: { + year: lastTradeDatetime.getUTCFullYear(), + month: (lastTradeDatetime.getUTCMonth() + 1), + days: lastTradeDatetime.getUTCDate(), + hours: lastTradeDatetime.getUTCHours(), + minutes: lastTradeDatetime.getUTCMinutes(), + seconds: lastTradeDatetime.getUTCSeconds(), + id: lastTradeId, + counter: counter + }, + monthChecked: monthChecked, + atHeadOfMarket: atHeadOfMarket + }; + + let fileContent = JSON.stringify(report); + + azureFileStorage.createTextFile(reportFilePath, fileName, fileContent + '\n', onFileBCreated); + + function onFileBCreated() { + + if (LOG_INFO === true) { + logger.write("[INFO] 'writeStatusReport' - Content written: " + fileContent); + } + + /* + + If we are at the same month of the begining of the market, we need to create the main status report file. + We will re-create it even every time the month status report is created. When this month check finished, other months later + will update it. + + */ + + if (processDate.getUTCMonth() === lastHistoricTradeFile.getUTCMonth() && processDate.getUTCFullYear() === lastHistoricTradeFile.getUTCFullYear()) { + + createMainStatusReport(lastTradeDatetime, lastTradeId, onMainReportCreated); + + function onMainReportCreated () { + + verifyMarketComplete(monthChecked, callBack); + + } + + } else { + + verifyMarketComplete(monthChecked, callBack); + + } + + } + + } + catch (err) { + const logText = "[ERROR] 'writeStatusReport - onFolderCreated' - ERROR : " + err.message; + logger.write(logText); + closeMarket(); + } + } + + } + catch (err) { + const logText = "[ERROR] 'writeStatusReport' - ERROR : " + err.message; + logger.write(logText); + closeMarket(); + } + + } + + function createMainStatusReport(lastTradeDatetime, lastTradeId,callBack) { + try { + + let reportFilePath = EXCHANGE_NAME + "/Processes/" + bot.process; + + utilities.createFolderIfNeeded(reportFilePath, azureFileStorage, onFolderCreated); + + function onFolderCreated() { + + try { + + let lastFileWithoutHoles = new Date(lastTradeDatetime.valueOf() - 60 * 1000); // It is the previous file where the last verified trade is. + + let fileName = "Status.Report." + market.assetA + '_' + market.assetB + ".json"; + + let report = { + lastFile: { + year: lastFileWithoutHoles.getUTCFullYear(), + month: (lastFileWithoutHoles.getUTCMonth() + 1), + days: lastFileWithoutHoles.getUTCDate(), + hours: lastFileWithoutHoles.getUTCHours(), + minutes: lastFileWithoutHoles.getUTCMinutes() + }, + lastTrade: { + year: lastTradeDatetime.getUTCFullYear(), + month: (lastTradeDatetime.getUTCMonth() + 1), + days: lastTradeDatetime.getUTCDate(), + hours: lastTradeDatetime.getUTCHours(), + minutes: lastTradeDatetime.getUTCMinutes(), + seconds: lastTradeDatetime.getUTCSeconds(), + id: lastTradeId + } + }; + + let fileContent = JSON.stringify(report); + + azureFileStorage.createTextFile(reportFilePath, fileName, fileContent + '\n', onFileBCreated); + + function onFileBCreated() { + + if (LOG_INFO === true) { + logger.write("[INFO] 'createMainStatusReport' - Content written: " + fileContent); + } + + callBack(); + + } + + } + catch (err) { + const logText = "[ERROR] 'createMainStatusReport - onFolderCreated' - ERROR : " + err.message; + logger.write(logText); + closeMarket(); + } + } + + } + catch (err) { + const logText = "[ERROR] 'createMainStatusReport' - ERROR : " + err.message; + logger.write(logText); + closeMarket(); + } + } + + + function verifyMarketComplete(isMonthComplete, callBack) { + + if (isMonthComplete === true) { + + logger.write("[INFO] 'writeStatusReport' - verifyMarketComplete - Month Completed !!! "); + + /* + + The mission of this function is to update the main status report. This report contains the date of the last file sucessfully checked + but in a consecutive way. + + For example: if the market starts in March, and March, April and June are checked, then the file will be the last of June even if September is also checked. + + */ + + let initialYear; + let initialMonth; + + let finalYear = (new Date()).getUTCFullYear(); + let finalMonth = (new Date()).getUTCMonth() + 1; + + let reportFilePath = EXCHANGE_NAME + "/Processes/" + bot.process; + let fileName = "Status.Report." + market.assetA + '_' + market.assetB + ".json"; + + /* Lets read the main status report */ + + azureFileStorage.getTextFile(reportFilePath, fileName, onFileReceived, true); + + function onFileReceived(text) { + + if (text === undefined) { + + /* The first month of the market didnt create this file yet. Aborting verification. */ + + logger.write("[INFO] 'writeStatusReport' - verifyMarketComplete - Verification ABORTED since the main status report does not exist. "); + + callBack(); + + } else { + + let statusReport = JSON.parse(text); + + initialYear = statusReport.lastFile.year; + initialMonth = statusReport.lastFile.month; + + loopCycle(); + } + + } + + function loopCycle() { + + /* + + Here we read the status report file of each month / year to verify if it is complete or not. + + */ + let paddedInitialMonth = utilities.pad(initialMonth, 2); + + let reportFilePath = EXCHANGE_NAME + "/Processes/" + bot.process + "/" + initialYear + "/" + paddedInitialMonth; + let fileName = "Status.Report." + market.assetA + '_' + market.assetB + ".json"; + + azureFileStorage.getTextFile(reportFilePath, fileName, onStatusReportFileReceived, true); + + function onStatusReportFileReceived(text) { + + if (text === undefined) { + + /* If any of the files do not exist, it means that the continuity has ben broken and this checking procedure must be aborted */ + + logger.write("[INFO] 'writeStatusReport' - verifyMarketComplete - Verification ABORTED since the status report for year " + initialYear + " and month " + initialMonth + " did not exist. "); + + callBack(); + + } else { + + let statusReport = JSON.parse(text); + + if (statusReport.monthChecked === true) { + + readAndWriteNewReport(statusReport); + + } else { + + /* If any of the files says that month is not checked then it is enough to know the market continuity is broken. */ + + logger.write("[INFO] 'writeStatusReport' - verifyMarketComplete - Verification ABORTED since the status report for year " + initialYear + " and month " + initialMonth + " is not hole checked. "); + + callBack(); + + } + } + } + } + + + + function readAndWriteNewReport(monthlyStatusReport) { + + /* We will read the current file to preserve its data, and save it again with the new lastFile */ + + let reportFilePath = EXCHANGE_NAME + "/Processes/" + bot.process; + let fileName = "Status.Report." + market.assetA + '_' + market.assetB + ".json"; + + azureFileStorage.getTextFile(reportFilePath, fileName, onFileReceived, true); + + function onFileReceived(text) { + + let statusReport = JSON.parse(text); + + /* + + Only if the last trade ID of the month being evaluated is bigger the trade Id at the status report we do replace it, otherwise not. + + */ + + if (monthlyStatusReport.lastTrade.id > statusReport.lastTrade.id) { + + statusReport.lastFile = monthlyStatusReport.lastFile; + statusReport.lastTrade = monthlyStatusReport.lastTrade; + statusReport.lastTrade.counter = undefined; + + let fileContent = JSON.stringify(statusReport); + + azureFileStorage.createTextFile(reportFilePath, fileName, fileContent + '\n', onMasterFileCreated); + + function onMasterFileCreated() { + + logger.write("[INFO] 'writeStatusReport' - verifyMarketComplete - Main Status Report Updated - Content written: " + fileContent); + + loop(); // Lets see the next month. + + } + + } else { + + logger.write("[INFO] 'writeStatusReport' - verifyMarketComplete - Main Status Report Not Updated since current Trade Id (" + monthlyStatusReport.lastTrade.id + ") is <= than Id at main status report file. (" + statusReport.lastTrade.id + ")" ); + + loop(); // Lets see the next month. + + } + } + } + + + function loop() { + + initialMonth++; + + if (initialMonth > 12) { + + initialMonth = 1; + initialYear++; + + } + + if ((initialYear === finalYear && initialMonth > finalMonth) || (initialYear > finalYear)) { + + /* We arrived to the point where we have checked all the status reports of every month and they are all complete. */ + + logger.write("[INFO] 'writeStatusReport' - verifyMarketComplete - Verification Finished. "); + + callBack(); + + return; + } + + loopCycle(); + + } + + } else { + + logger.write("[INFO] 'writeStatusReport' - verifyMarketComplete - Verification ABORTED: Month is not checked. "); + callBack(); + + } + + } + + + + + } + catch (err) { + const logText = "[ERROR] 'Start' - ERROR : " + err.message; + logger.write(logText); + } + } +}; diff --git a/One-Min-Daily-Candles/One Min Daily Candles.js b/One-Min-Daily-Candles/One Min Daily Candles.js index 25c35796f..7c67dc32a 100644 --- a/One-Min-Daily-Candles/One Min Daily Candles.js +++ b/One-Min-Daily-Candles/One Min Daily Candles.js @@ -1,66 +1,54 @@ - - - -exports.newPloniexOrderBooks = function newPloniexOrderBooks(BOT) { +exports.newPloniexHoleFixing = function newPloniexHoleFixing(BOT) { let bot = BOT; const ROOT_DIR = '../'; const INTERVAL_LAPSE = 60 * 1000; - const MODULE_NAME = "Poloniex Order Books"; - const LOG_INFO = false; + const MODULE_NAME = "One Min Daily Candles"; + const LOG_INFO = true; - const EXCHANGE_NAME = "Poloniex"; - const EXCHANGE_ID = 1; - - const RAW_FOLDER_NAME = "order-books"; - const AGG_FOLDER_NAME = "aggregatted-order-books"; - - const ORDER_BOOK_DEPTH = 100000; // This is the max amount of BIDs or ASKs we want the Exchange API to return us from each market. - - const MARKETS_MODULE = require(ROOT_DIR + 'Markets'); - - const PloniexClient = require(ROOT_DIR + 'Poloniex API Client'); + const INTERVAL_MODULE = require('./Interval'); const DEBUG_MODULE = require(ROOT_DIR + 'Debug Log'); const logger = DEBUG_MODULE.newDebugLog(); logger.fileName = MODULE_NAME; logger.bot = bot; - poloniexOrderBooks = { + const INTERVAL_MAX_LOOPS = -1; // -1 for infinite loop. + + poloniexHoleFixing = { initialize: initialize, start: start }; - let markets; - - const AZURE_FILE_STORAGE = require(ROOT_DIR + 'Azure File Storage'); - let azureFileStorage = AZURE_FILE_STORAGE.newAzureFileStorage(bot); + const SHUTDOWN_EVENT = require(ROOT_DIR + 'Azure Web Job Shutdown'); + let shutdownEvent = SHUTDOWN_EVENT.newShutdownEvent(bot); const UTILITIES = require(ROOT_DIR + 'Utilities'); let utilities = UTILITIES.newUtilities(bot); - const SHUTDOWN_EVENT = require(ROOT_DIR + 'Azure Web Job Shutdown'); - let shutdownEvent = SHUTDOWN_EVENT.newShutdownEvent(bot); + let year; + let month; - return poloniexOrderBooks; + return poloniexHoleFixing; - function initialize(callBackFunction) { + function initialize(yearAssigend, monthAssigned, callBackFunction) { try { - const logText = "[INFO] initialize - Entering function 'initialize' "; + year = yearAssigend; + month = monthAssigned; + + month = utilities.pad(month, 2); // Adding a left zero when needed. + + logger.fileName = MODULE_NAME + "-" + year + "-" + month; + + const logText = "[INFO] initialize - Entering function 'initialize' " + " @ " + year + "-" + month; console.log(logText); logger.write(logText); - poloniexApiClient = new PloniexClient(); - - azureFileStorage.initialize(); - - markets = MARKETS_MODULE.newMarkets(bot); - markets.initialize(callBackFunction); - + callBackFunction(); } catch (err) { @@ -72,82 +60,57 @@ exports.newPloniexOrderBooks = function newPloniexOrderBooks(BOT) { } - function start() { - let currentDate; - try { - let intervalId = setInterval(requestNewData, INTERVAL_LAPSE); + if (LOG_INFO === true) { + logger.write("[INFO] Entering function 'start', with year = " + year + " and month = " + month); + } - requestNewData(); // The first execution we do it without waiting for the interval call. + let intervalId; - function requestNewData() { + if (INTERVAL_MAX_LOOPS === 0) { + + intervalId = setInterval(startNewInterval, INTERVAL_LAPSE); + + } + + let intervalLoopCounter = 0; + + startNewInterval(); // First, inmadiate run. + + function startNewInterval() { try { - const logText = "[INFO] start - Entering function 'requestNewData' , next execution in " + INTERVAL_LAPSE / 1000 / 60 + " minutes."; - console.log(logText); - logger.write(logText); + if (LOG_INFO === true) { + + if (INTERVAL_MAX_LOOPS !== 0) { + + logger.write("[INFO] Entering function 'startNewInterval', next execution inmediately after the current Interval finishes."); + + } else { + + logger.write("[INFO] Entering function 'startNewInterval', next execution in " + INTERVAL_LAPSE / 1000 / 60 + " minutes."); + + } + + } + if (shutdownEvent.isShuttingDown() === false) { - currentDate = new Date(); + let fileProcessingInterval = INTERVAL_MODULE.newInterval(bot); - getOrderBookFromExchangeApi(); + fileProcessingInterval.initialize(year, month, onReadyToStart); - function getOrderBookFromExchangeApi() { + function onReadyToStart() { - poloniexApiClient.returnOrderBooks(ORDER_BOOK_DEPTH, onReturnOrderBooks); + fileProcessingInterval.start(loopControl); } - function onReturnOrderBooks(err, apiResponse) { - - try { - - if (err || apiResponse.error !== undefined) { - try { - - if (err.message.indexOf("ETIMEDOUT") > 0) { - - const logText = "[WARN] requestNewData - onReturnOrderBooks - Timeout reached while trying to access the Exchange API."; - console.log(logText); - logger.write(logText); - - /* We try to reconnect to the exchange and fetch the data again. */ - - getOrderBookFromExchangeApi(); - - } else { - const logText = "[ERROR] requestNewData - onReturnOrderBooks - err.message.indexOf(ETIMEDOUT) <= 0 ' ERROR : " + err.message; - console.log(logText); - logger.write(logText); - return; - } - - } catch (err) { - const logText = "[ERROR] requestNewData - onReturnOrderBooks ' RECEIVED ERROR : " + apiResponse.error; - console.log(logText); - logger.write(logText); - } - return; - - } else { - processNextMessage(apiResponse); - } - - } catch (err) { - - const logText = "[ERROR] requestNewData - onReturnOrderBooks ' ERROR : " + err.message; - console.log(logText); - logger.write(logText); - - } - } - - } else { @@ -159,254 +122,47 @@ exports.newPloniexOrderBooks = function newPloniexOrderBooks(BOT) { } - - } catch (err) { - - const logText = "[ERROR] requestNewData ' ERROR : " + err.message; - console.log(logText); + } + catch (err) { + const logText = "[ERROR] 'startNewInterval' - ERROR : " + err.message; logger.write(logText); + } + } + + function loopControl(moreIntervalExecutionsNeeded) { + + if (moreIntervalExecutionsNeeded === true) { + + intervalLoopCounter++; + + if (intervalLoopCounter === INTERVAL_MAX_LOOPS) { + + const logText = "[WARN] 'INTERVAL_MAX_LOOPS' reached. Stopping execution. "; + logger.write(logText); + + } else { + + const logText = "[WARN] 'loopControl'. Staring new Interval. "; + logger.write(logText); + + setTimeout(startNewInterval, INTERVAL_LAPSE);// Looped Execution. + + } + + } else { + + logger.write("[WARN] Received the signal from last Interval execution to not call it anymore. Stopping execution. "); } } - - function processNextMessage(messageReceived) { - - try { - - let orderBooksMessage = messageReceived; - - let dateForPath = currentDate.getUTCFullYear() + '/' + utilities.pad(currentDate.getUTCMonth() + 1, 2) + '/' + utilities.pad(currentDate.getUTCDate(), 2) + '/' + utilities.pad(currentDate.getUTCHours(), 2) + '/' + utilities.pad(currentDate.getUTCMinutes(), 2); - - let rawFilePath = RAW_FOLDER_NAME + '/' + dateForPath; - let aggregatedFilePath = AGG_FOLDER_NAME + '/' + dateForPath; - - utilities.createFolderIfNeeded(rawFilePath, azureFileStorage, onFirstFolderCreated); - - function onFirstFolderCreated() { - - utilities.createFolderIfNeeded(aggregatedFilePath, azureFileStorage, processEachMarket); - - } - - function processEachMarket() { - - for (let marketName in orderBooksMessage) { - - let market = orderBooksMessage[marketName]; - - markets.getExistingOrNewMarketId(marketName, market, EXCHANGE_ID, onMarketIdReady); - - function onMarketIdReady(marketName, market, pMarketId) { - - try { - - let bidsCounter = 0; - let asksCounter = 0; - - let newDate = new Date(); - - let asks = market.asks; - let bids = market.bids; - - let marketId = pMarketId; - - let fileName = '' + EXCHANGE_ID + '_' + marketId + '.json'; - - let needSeparator; - let separator; - - needSeparator = false; - - let maxRate = bids[0][0] * 1.1; - let presicion = utilities.calculatePresicion(maxRate); - - let truncatedRate; - let asksMap = new Map(); - let bidsMap = new Map(); - - let fileContent = ""; - - generateFirstFile(); - - function generateFirstFile() { - - /* First we create the raw file and aggregate the date for the aggregated file */ - - fileContent = fileContent + '['; - fileContent = fileContent + '['; - - for (let i = 0; i < asks.length; i++) { - - if (needSeparator === false) { - - needSeparator = true; - separator = ''; - - } else { - separator = ','; - } - - let rate = asks[i][0]; - let amount = asks[i][1]; - - fileContent = fileContent + separator + '[' + rate + ',' + amount + ']'; - - roundedRate = Math.round(rate / presicion) * presicion; - - let currentAmount = asksMap.get(roundedRate) || 0; - currentAmount = currentAmount + amount; - - asksMap.set(roundedRate, currentAmount); - - asksCounter = i; - } - - fileContent = fileContent + '],['; - - needSeparator = false; - - for (let i = 0; i < bids.length; i++) { - - if (needSeparator === false) { - - needSeparator = true; - separator = ''; - - } else { - separator = ','; - } - - let rate = bids[i][0]; - let amount = bids[i][1]; - - fileContent = fileContent + separator + '[' + rate + ',' + amount + ']'; - - truncatedRate = Math.trunc(rate / presicion) * presicion; - - let currentAmount = bidsMap.get(truncatedRate) || 0; - currentAmount = currentAmount + amount; - - bidsMap.set(truncatedRate, currentAmount); - - bidsCounter = i; - } - - fileContent = fileContent + ']'; - fileContent = fileContent + ']'; - - azureFileStorage.createTextFile(rawFilePath, fileName, fileContent + '\n', onFirstFileCreated); - - function onFirstFileCreated() { - - generateSecondFile(); - - } - - } - - function generateSecondFile() { - - - fileContent = ""; - - fileContent = fileContent + '['; - fileContent = fileContent + '['; - - needSeparator = false; - - asksMap.forEach(addAsks); - - function addAsks(value, key, map) { - - if (needSeparator === false) { - - needSeparator = true; - separator = ''; - - } else { - separator = ','; - } - - value = Math.trunc(value * 1000000) / 1000000; - - fileContent = fileContent + separator + '[' + key + ',' + value + ']'; - } - - fileContent = fileContent + '],['; - - needSeparator = false; - - bidsMap.forEach(addBids); - - function addBids(value, key, map) { - - if (needSeparator === false) { - - needSeparator = true; - separator = ''; - - } else { - separator = ','; - } - - value = Math.trunc(value * 1000000) / 1000000; - - fileContent = fileContent + separator + '[' + key + ',' + value + ']'; - } - - fileContent = fileContent + ']'; - fileContent = fileContent + ']'; - - azureFileStorage.createTextFile(aggregatedFilePath, fileName, fileContent + '\n', onSecondFileCreated); - - function onSecondFileCreated() { - - const logText = "[WARN] Finished with " + marketName + ", " + asksCounter + " aks and " + bidsCounter + " bids written to file " + fileName; - console.log(logText); - logger.write(logText); - - } - } - - } catch (err) { - - const logText = "[ERROR] processNextMessage - onMarketIdReady ' ERROR : " + err.message; - console.log(logText); - logger.write(logText); - - } - } - - } - - - } - - - - messageReceived = null; - - } catch (err) { - - const logText = "[ERROR] processNextMessage ' ERROR : " + err.message; - console.log(logText); - logger.write(logText); - - } - } - - } catch (err) { - - const logText = "[ERROR] Start - ' ERROR : " + err.message; - console.log(logText); + } + catch (err) { + const logText = "[ERROR] 'Start' - ERROR : " + err.message; logger.write(logText); - } } - - }; diff --git a/One-Min-Daily-Candles/run.js b/One-Min-Daily-Candles/run.js index f292cb690..652bbb52e 100644 --- a/One-Min-Daily-Candles/run.js +++ b/One-Min-Daily-Candles/run.js @@ -11,9 +11,10 @@ const ROOT_DIR = '../'; const MODULE_NAME = "run"; -const START_ONE_MIN_DAILY_CANDLES = true; +const START_POLONIEX_HOLE_FIXING = true; +const START_POLONIEX_HOLE_FIXING_AT_ONE_MONTH = false; -const POLONIEX_ORDE_BOOKS_MODULE = require('./One Min Daily Candles'); +const POLONIEX_HOLE_FIXING_MODULE = require('./Poloniex Hole Fixing'); const DEBUG_MODULE = require(ROOT_DIR + 'Debug Log'); const logger = DEBUG_MODULE.newDebugLog(); @@ -21,7 +22,7 @@ logger.fileName = MODULE_NAME; logger.bot = bot; process.on('uncaughtException', function (err) { - logger.write('uncaughtException' + err.message); + logger.write('uncaughtException - ' + err.message); }); @@ -35,27 +36,62 @@ process.on('exit', function (code) { }); - try { - if (START_ONE_MIN_DAILY_CANDLES === true) { + if (START_POLONIEX_HOLE_FIXING === true) { - const poloniexOrderBooks = POLONIEX_ORDE_BOOKS_MODULE.newPloniexOrderBooks(bot); + for (let year = 2018; year > 2014; year--) { - poloniexOrderBooks.initialize(onInitializeReady); + for (let month = 12; month > 0; month--) { - function onInitializeReady() { + let timeDelay = Math.random() * 100 * 1000; // We need a delay so as not to get banned from the exchange for requesting to many things at the same time. - poloniexOrderBooks.start(); + setTimeout(startProcess, timeDelay); + + function startProcess() { + + const newPloniexHoleFixing = POLONIEX_HOLE_FIXING_MODULE.newPloniexHoleFixing(bot); + + newPloniexHoleFixing.initialize(year, month, onInitializeReady); + + function onInitializeReady() { + + newPloniexHoleFixing.start(); + + } + + } + + } } } + + if (START_POLONIEX_HOLE_FIXING_AT_ONE_MONTH === true) { + + startProcess(); + + function startProcess() { + + const newPloniexHoleFixing = POLONIEX_HOLE_FIXING_MODULE.newPloniexHoleFixing(bot); + + let month = 2; + let year = 2015; + + newPloniexHoleFixing.initialize(year, month, onInitializeReady); + + function onInitializeReady() { + + newPloniexHoleFixing.start(); + + } + } + } + + } catch (err) { console.log(err.message); logger.write(err.message); } - - -